Serverless 赋能：无需管理服务器，快速部署与监控 1688 API 数据拉取服务

CREATE DATABASE IF NOT EXISTS 1688_data_db DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

USE 1688_data_db;

CREATE TABLE IF NOT EXISTS product_info (
    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
    product_id VARCHAR(64) NOT NULL COMMENT '1688商品ID',
    product_name VARCHAR(256) NOT NULL COMMENT '商品名称',
    supplier_name VARCHAR(128) NOT NULL COMMENT '供应商名称',
    price DECIMAL(10,2) NOT NULL COMMENT '商品价格（元）',
    sales_volume INT DEFAULT 0 COMMENT '销量',
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '数据拉取时间',
    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '数据更新时间',
    UNIQUE KEY uk_product_id (product_id) COMMENT '商品ID唯一索引'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT '1688商品信息表';

requests==2.31.0
pymysql==1.1.0
cryptography==41.0.7

import os
import json
import time
import requests
import pymysql
from datetime import datetime

# -------------------------- 配置信息（建议通过FC环境变量配置，避免硬编码） --------------------------
# 1688 API配置
APP_KEY = os.environ.get("APP_KEY")  # 从FC环境变量获取App Key
APP_SECRET = os.environ.get("APP_SECRET")  # 从FC环境变量获取App Secret
ACCESS_TOKEN = os.environ.get("ACCESS_TOKEN")  # 从FC环境变量获取Access Token
1688_API_URL = "https://gw.open.1688.com/openapi/param2/1/cn.alibaba.open/offer.get/{}".format(APP_KEY)  # 商品详情API

# RDS MySQL配置
RDS_HOST = os.environ.get("RDS_HOST")  # RDS实例地址
RDS_PORT = int(os.environ.get("RDS_PORT", 3306))  # RDS端口
RDS_USER = os.environ.get("RDS_USER")  # RDS用户名
RDS_PASSWORD = os.environ.get("RDS_PASSWORD")  # RDS密码
RDS_DB = os.environ.get("RDS_DB", "1688_data_db")  # 数据库名

# -------------------------- 工具函数：RDS数据库连接 --------------------------
def get_rds_connection():
    """
    获取RDS MySQL数据库连接
    """
    try:
        conn = pymysql.connect(
            host=RDS_HOST,
            port=RDS_PORT,
            user=RDS_USER,
            password=RDS_PASSWORD,
            database=RDS_DB,
            charset="utf8mb4",
            connect_timeout=10
        )
        return conn
    except Exception as e:
        print(f"[ERROR] 获取RDS连接失败：{str(e)}")
        raise e

# -------------------------- 工具函数：1688 API签名（简化版，完整签名参考1688开放平台文档） --------------------------
def generate_1688_sign(params, app_secret):
    """
    生成1688 API调用签名（简化版，实际生产环境需严格按照1688签名规则实现）
    """
    # 按参数名升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 拼接字符串
    sign_str = app_secret
    for key, value in sorted_params:
        sign_str += f"{key}{value}"
    sign_str += app_secret
    # MD5加密（实际需按1688要求编码）
    import hashlib
    sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
    return sign

# -------------------------- 核心函数：拉取1688商品数据 --------------------------
def pull_1688_product_data(product_id):
    """
    调用1688 API拉取单个商品数据
    :param product_id: 1688商品ID
    :return: 结构化商品数据
    """
    try:
        # 1. 构造API请求参数
        params = {
            "access_token": ACCESS_TOKEN,
            "offerId": product_id,
            "format": "json",
            "timestamp": str(int(time.time() * 1000)),
            "v": "2.0"
        }
        # 2. 生成签名
        params["sign"] = generate_1688_sign(params, APP_SECRET)
        
        # 3. 发送API请求
        response = requests.get(1688_API_URL, params=params, timeout=30)
        response.raise_for_status()  # 抛出HTTP请求异常
        api_result = response.json()
        
        # 4. 校验API返回结果
        if "error" in api_result:
            print(f"[ERROR] 1688 API调用失败：{api_result['error']['msg']}（错误码：{api_result['error']['code']}）")
            raise Exception(f"1688 API错误：{api_result['error']['msg']}")
        
        # 5. 数据结构化处理
        product_info = api_result.get("result", {})
        structured_data = {
            "product_id": product_id,
            "product_name": product_info.get("subject", "").strip(),
            "supplier_name": product_info.get("seller", {}).get("companyName", "").strip(),
            "price": float(product_info.get("price", 0) or 0),
            "sales_volume": int(product_info.get("sales", {}).get("volume", 0) or 0)
        }
        print(f"[INFO] 成功拉取商品{product_id}数据：{structured_data}")
        return structured_data
    except Exception as e:
        print(f"[ERROR] 拉取商品{product_id}数据失败：{str(e)}")
        raise e

# -------------------------- 核心函数：存储数据到RDS --------------------------
def save_product_data_to_rds(product_data):
    """
    将结构化商品数据存储到RDS MySQL
    :param product_data: 结构化商品数据
    """
    conn = None
    cursor = None
    try:
        # 1. 获取数据库连接
        conn = get_rds_connection()
        cursor = conn.cursor()
        
        # 2. 构造SQL（存在则更新，不存在则插入）
        sql = """
            INSERT INTO product_info (product_id, product_name, supplier_name, price, sales_volume)
            VALUES (%s, %s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE
            product_name = VALUES(product_name),
            supplier_name = VALUES(supplier_name),
            price = VALUES(price),
            sales_volume = VALUES(sales_volume),
            update_time = CURRENT_TIMESTAMP
        """
        # 3. 执行SQL
        cursor.execute(sql, (
            product_data["product_id"],
            product_data["product_name"],
            product_data["supplier_name"],
            product_data["price"],
            product_data["sales_volume"]
        ))
        conn.commit()
        print(f"[INFO] 商品{product_data['product_id']}数据已成功存储/更新到RDS")
    except Exception as e:
        if conn:
            conn.rollback()
        print(f"[ERROR] 存储商品{product_data['product_id']}数据到RDS失败：{str(e)}")
        raise e
    finally:
        # 4. 关闭连接
        if cursor:
            cursor.close()
        if conn:
            conn.close()

# -------------------------- FC函数入口 --------------------------
def handler(event, context):
    """
    阿里云FC函数入口函数
    :param event: 触发事件（定时触发/HTTP触发参数）
    :param context: 函数运行上下文
    :return: 执行结果
    """
    try:
        # 1. 解析触发参数（示例：从event中获取需要拉取的商品ID列表）
        # 定时触发时，可直接配置固定商品ID列表；HTTP触发时，可从请求参数中获取
        event_data = json.loads(event) if isinstance(event, str) else event
        product_id_list = event_data.get("product_id_list", ["1234567890", "0987654321"])  # 替换为实际商品ID
        
        # 2. 批量拉取并存储数据
        success_count = 0
        fail_count = 0
        for product_id in product_id_list:
            try:
                product_data = pull_1688_product_data(product_id)
                save_product_data_to_rds(product_data)
                success_count += 1
                # 避免API调用频率超限，添加短暂延时
                time.sleep(0.5)
            except Exception as e:
                fail_count += 1
                continue
        
        # 3. 构造返回结果
        result = {
            "status": "success",
            "message": f"数据拉取完成：成功{success_count}条，失败{fail_count}条",
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "success_count": success_count,
            "fail_count": fail_count
        }
        print(f"[INFO] 本次数据拉取任务执行完成：{result}")
        return json.dumps(result, ensure_ascii=False)
    except Exception as e:
        error_result = {
            "status": "fail",
            "message": f"任务执行失败：{str(e)}",
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        }
        print(f"[ERROR] 本次数据拉取任务执行异常：{error_result}")
        return json.dumps(error_result, ensure_ascii=False)

{
    "product_id_list": ["1234567890", "0987654321"]
}