构建电商数据中台：基于淘宝 API 关键词搜索接口的设计与实现

数据中台架构分层：
├── 数据源层：淘宝开放平台API
├── 数据采集层：API接入模块（本文核心）
│   ├── 鉴权模块：AppKey/AppSecret签名
│   ├── 请求模块：关键词搜索接口调用
│   ├── 数据清洗模块：格式标准化、脏数据过滤
│   └── 数据落地模块：写入数据中台存储（MySQL/ClickHouse）
├── 数据加工层：ETL、数据建模
├── 数据服务层：API接口、可视化报表
└── 应用层：运营分析、智能决策

pip install requests python-dotenv pymysql

# 淘宝API配置
TAOBAO_APP_KEY=你的AppKey
TAOBAO_APP_SECRET=你的AppSecret
# MySQL配置
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=root
MYSQL_PASSWORD=你的密码
MYSQL_DB=ecommerce_platform

import pymysql
from pymysql.err import OperationalError

class MySQLStorage:
    def __init__(self):
        self.host = os.getenv("MYSQL_HOST")
        self.port = int(os.getenv("MYSQL_PORT"))
        self.user = os.getenv("MYSQL_USER")
        self.password = os.getenv("MYSQL_PASSWORD")
        self.db = os.getenv("MYSQL_DB")
        self.conn = None
        self._connect()

    def _connect(self):
        """建立数据库连接"""
        try:
            self.conn = pymysql.connect(
                host=self.host,
                port=self.port,
                user=self.user,
                password=self.password,
                db=self.db,
                charset="utf8mb4"
            )
        except OperationalError as e:
            print(f"数据库连接失败：{str(e)}")
            raise

    def create_table(self):
        """创建商品数据表"""
        create_sql = """
        CREATE TABLE IF NOT EXISTS taobao_items (
            id INT AUTO_INCREMENT PRIMARY KEY,
            item_id VARCHAR(64) NOT NULL COMMENT '商品ID',
            title VARCHAR(512) NOT NULL COMMENT '商品标题',
            price DECIMAL(10,2) NOT NULL COMMENT '商品价格',
            sales INT NOT NULL COMMENT '销量',
            shop_name VARCHAR(128) COMMENT '店铺名称',
            category VARCHAR(64) COMMENT '商品分类',
            commission_rate DECIMAL(5,4) COMMENT '佣金比例',
            create_time DATETIME NOT NULL COMMENT '采集时间',
            UNIQUE KEY uk_item_id (item_id)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='淘宝商品数据表';
        """
        try:
            with self.conn.cursor() as cursor:
                cursor.execute(create_sql)
            self.conn.commit()
            print("数据表创建/验证成功")
        except Exception as e:
            print(f"创建表失败：{str(e)}")
            self.conn.rollback()

    def batch_insert(self, data):
        """批量插入商品数据"""
        if not data:
            print("无数据可插入")
            return

        insert_sql = """
        INSERT INTO taobao_items (item_id, title, price, sales, shop_name, category, commission_rate, create_time)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE
        price=VALUES(price),
        sales=VALUES(sales),
        commission_rate=VALUES(commission_rate),
        create_time=VALUES(create_time);
        """
        try:
            with self.conn.cursor() as cursor:
                cursor.executemany(insert_sql, [
                    (
                        item["item_id"],
                        item["title"],
                        item["price"],
                        item["sales"],
                        item["shop_name"],
                        item["category"],
                        item["commission_rate"],
                        item["create_time"]
                    ) for item in data
                ])
            self.conn.commit()
            print(f"成功插入/更新 {cursor.rowcount} 条数据")
        except Exception as e:
            print(f"批量插入失败：{str(e)}")
            self.conn.rollback()

    def close(self):
        """关闭数据库连接"""
        if self.conn:
            self.conn.close()

def main():
    # 1. 初始化组件
    taobao_api = TaobaoAPI()
    mysql_storage = MySQLStorage()

    # 2. 创建数据表
    mysql_storage.create_table()

    # 3. 定义搜索关键词与采集参数
    keywords = ["夏季连衣裙", "男士运动鞋", "儿童玩具"]  # 待搜索关键词
    max_page = 3  # 采集页数

    # 4. 批量采集并存储数据
    for keyword in keywords:
        print(f"\n开始采集关键词：{keyword}")
        for page in range(1, max_page + 1):
            print(f"采集第 {page} 页数据...")
            # 调用API获取数据
            item_data = taobao_api.search_items(
                keyword=keyword,
                page_no=page,
                page_size=40,  # 每页最大40条
                sort="total_sales"
            )
            # 写入数据库
            mysql_storage.batch_insert(item_data)
            # 避免QPS超限，休眠1秒
            time.sleep(1)

    # 5. 关闭数据库连接
    mysql_storage.close()
    print("\n所有关键词采集完成")

if __name__ == "__main__":
    main()