多店铺跨境电商服务器管理：香港VPS统一管理Amazon/eBay/TikTok Shop数据抓取与分析

多平台多店铺运营是跨境电商规模化的必经之路，但随之而来的是数据分散问题：Amazon 后台一个数据、eBay 另一套报表、TikTok Shop 又一套——每天花 2～3 小时人工汇总数据，既低效又容易出错。在香港 VPS 上搭建多店铺数据中台，通过官方 API 自动抓取各平台数据，统一可视化展示，让选品和运营决策有数据支撑。

一、系统架构

<code">
Amazon SP-API ──┐
eBay API ───────┤──→ 香港VPS数据中台 ──→ PostgreSQL 数据仓库 ──→ Grafana 大盘
TikTok Shop ────┘    （Python采集服务）      (统一数据结构)        (多维分析)
                             │
                        定时任务调度
                      (每小时/每天采集)

二、香港 VPS 环境准备

<code"># 安装 Python 3.11 + 必要依赖
apt update && apt install -y python3.11 python3.11-venv python3-pip \
  postgresql postgresql-contrib

# 创建项目目录和虚拟环境
mkdir -p /opt/ecom-dashboard
cd /opt/ecom-dashboard
python3.11 -m venv venv
source venv/bin/activate

pip install requests pandas psycopg2-binary \
  python-dotenv schedule sqlalchemy \
  sp-api  # Amazon SP-API SDK

<code"># 创建 PostgreSQL 数据库
sudo -u postgres psql << 'EOF'
CREATE USER ecom WITH PASSWORD 'ecom_db_pass';
CREATE DATABASE ecom_dashboard OWNER ecom;
GRANT ALL PRIVILEGES ON DATABASE ecom_dashboard TO ecom;
EOF

三、数据库统一表结构设计

<code"># /opt/ecom-dashboard/models.py
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Text
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class Order(Base):
    """统一订单表——所有平台订单汇入此表"""
    __tablename__ = 'orders'

    id = Column(Integer, primary_key=True)
    platform = Column(String(20), nullable=False)    # amazon/ebay/tiktok
    shop_name = Column(String(100))                  # 店铺名称
    order_id = Column(String(100), unique=True)
    order_date = Column(DateTime)
    sku = Column(String(100))
    product_title = Column(Text)
    quantity = Column(Integer)
    unit_price = Column(Float)
    total_amount = Column(Float)
    currency = Column(String(10))
    status = Column(String(50))
    customer_country = Column(String(50))
    created_at = Column(DateTime, default=datetime.utcnow)

class DailySummary(Base):
    """每日汇总表——用于趋势分析"""
    __tablename__ = 'daily_summary'

    id = Column(Integer, primary_key=True)
    date = Column(String(10))
    platform = Column(String(20))
    shop_name = Column(String(100))
    order_count = Column(Integer, default=0)
    revenue_usd = Column(Float, default=0)
    units_sold = Column(Integer, default=0)
    avg_order_value = Column(Float, default=0)

# 创建表
engine = create_engine('postgresql://ecom:ecom_db_pass@localhost/ecom_dashboard')
Base.metadata.create_all(engine)

四、Amazon SP-API 数据采集

<code"># /opt/ecom-dashboard/collectors/amazon.py
from sp_api.api import Orders
from sp_api.base import Marketplaces
from datetime import datetime, timedelta
import os

class AmazonCollector:
    def __init__(self, shop_config):
        self.shop_name = shop_config['shop_name']
        self.client = Orders(
            credentials=dict(
                refresh_token=shop_config['refresh_token'],
                lwa_app_id=shop_config['lwa_app_id'],
                lwa_client_secret=shop_config['lwa_client_secret'],
                aws_access_key=os.getenv('AWS_ACCESS_KEY'),
                aws_secret_key=os.getenv('AWS_SECRET_KEY'),
                role_arn=shop_config['role_arn']
            ),
            marketplace=Marketplaces.US
        )

    def fetch_recent_orders(self, days=1):
        """获取最近N天的订单"""
        created_after = datetime.utcnow() - timedelta(days=days)
        response = self.client.get_orders(
            CreatedAfter=created_after.isoformat(),
            OrderStatuses=['Shipped', 'Unshipped', 'PartiallyShipped']
        )

        orders = []
        for order in response.payload.get('Orders', []):
            orders.append({
                'platform': 'amazon',
                'shop_name': self.shop_name,
                'order_id': order['AmazonOrderId'],
                'order_date': order['PurchaseDate'],
                'total_amount': float(order.get('OrderTotal', {}).get('Amount', 0)),
                'currency': order.get('OrderTotal', {}).get('CurrencyCode', 'USD'),
                'status': order['OrderStatus'],
                'customer_country': order.get('ShipServiceLevel', ''),
            })
        return orders

五、TikTok Shop API 数据采集

<code"># /opt/ecom-dashboard/collectors/tiktok.py
import requests, hashlib, hmac, time, json

class TikTokShopCollector:
    def __init__(self, shop_config):
        self.shop_name = shop_config['shop_name']
        self.app_key = shop_config['app_key']
        self.app_secret = shop_config['app_secret']
        self.access_token = shop_config['access_token']
        self.base_url = "https://open-api.tiktokglobalshop.com"

    def _sign(self, path, params):
        """TikTok Shop API 签名算法"""
        timestamp = int(time.time())
        params['app_key'] = self.app_key
        params['timestamp'] = timestamp
        sorted_params = ''.join(f"{k}{v}" for k, v in sorted(params.items()))
        sign_str = f"{self.app_secret}{path}{sorted_params}{self.app_secret}"
        sign = hmac.new(
            self.app_secret.encode(),
            sign_str.encode(),
            hashlib.sha256
        ).hexdigest().upper()
        params['sign'] = sign
        params['timestamp'] = timestamp
        return params

    def fetch_orders(self, days=1):
        """获取最近订单"""
        path = "/api/orders/search"
        end_time = int(time.time())
        start_time = end_time - days * 86400
        params = self._sign(path, {
            'create_time_from': start_time,
            'create_time_to': end_time,
            'page_size': 100,
        })
        headers = {'x-tts-access-token': self.access_token}
        response = requests.get(
            self.base_url + path,
            params=params, headers=headers, timeout=10
        )
        data = response.json()
        orders = []
        for order in data.get('data', {}).get('order_list', []):
            orders.append({
                'platform': 'tiktok',
                'shop_name': self.shop_name,
                'order_id': order['order_id'],
                'order_date': datetime.fromtimestamp(order['create_time']),
                'total_amount': float(order.get('payment', {}).get('total_amount', 0)) / 100,
                'currency': 'USD',
                'status': order['order_status'],
            })
        return orders

六、定时采集调度

<code"># /opt/ecom-dashboard/scheduler.py
import schedule, time, logging
from collectors.amazon import AmazonCollector
from collectors.tiktok import TikTokShopCollector
from database import save_orders

logging.basicConfig(level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s')

# 店铺配置（实际使用中从环境变量或加密配置文件读取）
SHOPS = {
    'amazon': [
        {'shop_name': 'US_Store_A', 'refresh_token': '...', ...},
        {'shop_name': 'US_Store_B', 'refresh_token': '...', ...},
    ],
    'tiktok': [
        {'shop_name': 'TT_Shop_A', 'app_key': '...', ...},
    ]
}

def collect_all():
    logging.info("开始采集所有店铺数据...")
    total = 0
    for config in SHOPS['amazon']:
        orders = AmazonCollector(config).fetch_recent_orders(days=1)
        save_orders(orders)
        total += len(orders)
        logging.info(f"Amazon {config['shop_name']}: {len(orders)} 条订单")

    for config in SHOPS['tiktok']:
        orders = TikTokShopCollector(config).fetch_orders(days=1)
        save_orders(orders)
        total += len(orders)
        logging.info(f"TikTok {config['shop_name']}: {len(orders)} 条订单")

    logging.info(f"采集完成，共 {total} 条订单")

# 每小时采集一次
schedule.every().hour.do(collect_all)

if __name__ == '__main__':
    collect_all()    # 启动时立即执行一次
    while True:
        schedule.run_pending()
        time.sleep(60)

<code"># 创建 systemd 服务
cat > /etc/systemd/system/ecom-collector.service << 'EOF'
[Unit]
Description=Ecom Multi-Store Data Collector
After=network.target postgresql.service

[Service]
User=www-data
WorkingDirectory=/opt/ecom-dashboard
ExecStart=/opt/ecom-dashboard/venv/bin/python scheduler.py
Restart=on-failure
RestartSec=30

[Install]
WantedBy=multi-user.target
EOF
systemctl enable --now ecom-collector

七、Grafana 多店铺运营大盘

<code"># 启动 Grafana（可与第二阶段文章5的监控栈复用同一实例）
docker run -d --name grafana \
  -p 127.0.0.1:3000:3000 \
  -e GF_SECURITY_ADMIN_PASSWORD=StrongPass \
  -v grafana_data:/var/lib/grafana \
  grafana/grafana:latest

# 在 Grafana 中添加 PostgreSQL 数据源，然后创建以下面板：

<code">-- 面板1：今日各平台收入对比（柱状图）
SELECT platform, shop_name, SUM(total_amount) as revenue
FROM orders
WHERE order_date >= CURRENT_DATE
GROUP BY platform, shop_name
ORDER BY revenue DESC;

-- 面板2：过去30天日收入趋势（折线图）
SELECT DATE(order_date) as day, platform,
       SUM(total_amount) as daily_revenue
FROM orders
WHERE order_date >= NOW() - INTERVAL '30 days'
GROUP BY day, platform
ORDER BY day;

-- 面板3：最畅销SKU Top10
SELECT sku, product_title, SUM(quantity) as units,
       SUM(total_amount) as revenue
FROM orders
WHERE order_date >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY sku, product_title
ORDER BY units DESC LIMIT 10;

八、总结

香港 VPS 的国际 API 访问能力是搭建多店铺数据中台的关键——Amazon、TikTok Shop、eBay 的 API 在大陆访问不稳定，在香港则完全无障碍。一台 4核8G 香港 VPS（¥200/月）即可支撑 10～20 个店铺的数据采集和可视化，ROI 极高。