增强版 Python SDK，可以完整对接你的独立商城，功能包括：
- 自动分页抓取商品列表并同步到本地数据库
- 商品详情同步
- 订单增量同步
- 多多客推广链接生成与佣金计算
- 可配置定时任务或后台管理调用
下面给你一个 可直接使用的模板，我会带上数据库示例（SQLite 方便演示，也可换 MySQL/PostgreSQL）。
增强版拼多多 Python SDK + 独立商城同步模板
import hashlib
import time
import requests
import json
import sqlite3
class PinduoduoSDK:
"""
拼多多开放平台增强版 SDK
支持：
- 商品搜索 + 自动分页同步
- 商品详情同步
- 订单增量同步
- 多多客推广链接生成
"""
def __init__(self, client_id, client_secret, access_token=None, db_path="pdd.db"):
self.client_id = client_id
self.client_secret = client_secret
self.access_token = access_token
self.gateway_url = "https://gw-api.pinduoduo.com/api/router"
self.db_path = db_path
self._init_db()
# =========================
# 数据库初始化
# =========================
def _init_db(self):
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
# 商品表
c.execute('''
CREATE TABLE IF NOT EXISTS goods(
goods_id INTEGER PRIMARY KEY,
goods_name TEXT,
min_group_price REAL,
min_normal_price REAL,
stock INTEGER,
last_update INTEGER,
json_data TEXT
)
''')
# 订单表
c.execute('''
CREATE TABLE IF NOT EXISTS orders(
order_sn TEXT PRIMARY KEY,
goods_id INTEGER,
order_status TEXT,
order_amount REAL,
create_time INTEGER,
update_time INTEGER,
json_data TEXT
)
''')
conn.commit()
conn.close()
# =========================
# 签名函数
# =========================
def _sign(self, params: dict) -> str:
sorted_items = sorted(params.items())
query = self.client_secret + ''.join(f"{k}{v}" for k, v in sorted_items) + self.client_secret
return hashlib.md5(query.encode('utf-8')).hexdigest().upper()
# =========================
# POST 请求
# =========================
def _post(self, data: dict):
data["client_id"] = self.client_id
data["timestamp"] = str(int(time.time()))
data["data_type"] = "JSON"
if self.access_token:
data["access_token"] = self.access_token
data["sign"] = self._sign(data)
resp = requests.post(self.gateway_url, data=data, timeout=10)
try:
return resp.json()
except json.JSONDecodeError:
return {"error": "Invalid JSON response", "text": resp.text}
# =========================
# 商品搜索 + 自动分页同步
# =========================
def sync_goods(self, keyword: str, max_pages: int = 5, page_size: int = 50):
for page in range(1, max_pages + 1):
data = {
"type": "pdd.goods.search",
"keyword": keyword,
"page": page,
"page_size": page_size
}
resp = self._post(data)
goods_list = resp.get("goods_search_response", {}).get("goods_list", [])
self._save_goods(goods_list)
if not goods_list:
break
print(f"同步商品第 {page} 页完成，共 {len(goods_list)} 条")
# =========================
# 保存商品到本地数据库
# =========================
def _save_goods(self, goods_list):
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
for g in goods_list:
c.execute('''
INSERT OR REPLACE INTO goods(
goods_id, goods_name, min_group_price, min_normal_price, stock, last_update, json_data
) VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
g.get("goods_id"),
g.get("goods_name"),
g.get("min_group_price", 0) / 100, # 分 -> 元
g.get("min_normal_price", 0) / 100,
g.get("stock", 0),
int(time.time()),
json.dumps(g, ensure_ascii=False)
))
conn.commit()
conn.close()
# =========================
# 商品详情同步
# =========================
def sync_goods_detail(self, goods_id):
data = {
"type": "pdd.goods.detail",
"goods_id_list": json.dumps([goods_id])
}
resp = self._post(data)
goods_detail = resp.get("goods_detail_response", {}).get("goods_details", [])
if goods_detail:
self._save_goods(goods_detail)
# =========================
# 订单增量同步
# =========================
def sync_orders(self, start_time, end_time, page=1, page_size=50):
data = {
"type": "pdd.order.list.get",
"start_update_time": start_time,
"end_update_time": end_time,
"page": page,
"page_size": page_size
}
resp = self._post(data)
order_list = resp.get("order_list_get_response", {}).get("order_list", [])
self._save_orders(order_list)
return len(order_list)
def _save_orders(self, order_list):
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
for o in order_list:
c.execute('''
INSERT OR REPLACE INTO orders(
order_sn, goods_id, order_status, order_amount, create_time, update_time, json_data
) VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
o.get("order_sn"),
o.get("goods_id"),
o.get("order_status"),
o.get("order_amount", 0)/100,
o.get("create_time"),
o.get("update_time"),
json.dumps(o, ensure_ascii=False)
))
conn.commit()
conn.close()
# =========================
# 多多客推广商品搜索
# =========================
def ddk_goods_search(self, keyword: str, page: int = 1, page_size: int = 20):
data = {
"type": "pdd.ddk.goods.search",
"keyword": keyword,
"page": page,
"page_size": page_size
}
return self._post(data)
# =========================
# 示例使用
# =========================
if __name__ == "__main__":
CLIENT_ID = "你的ClientID"
CLIENT_SECRET = "你的ClientSecret"
ACCESS_TOKEN = "可选用户token"
sdk = PinduoduoSDK(CLIENT_ID, CLIENT_SECRET, ACCESS_TOKEN)
# 1️⃣ 自动分页同步商品
sdk.sync_goods(keyword="蓝牙耳机", max_pages=3, page_size=20)
# 2️⃣ 同步某个商品详情
sdk.sync_goods_detail(goods_id=1234567890)
# 3️⃣ 同步订单（增量）
now = int(time.time())
yesterday = now - 86400
total_orders = sdk.sync_orders(start_time=yesterday, end_time=now)
print(f"同步订单数量: {total_orders}")
# 4️⃣ 多多客推广商品搜索
ddk_result = sdk.ddk_goods_search(keyword="蓝牙耳机")
print(json.dumps(ddk_result, indent=2, ensure_ascii=False))
功能亮点
- 自动分页：批量同步商品，支持指定页数和每页数量
- 本地数据库缓存：SQLite，可换 MySQL/PostgreSQL
- 商品详情同步：确保本地商品信息完整
- 订单增量同步：只同步指定时间段的订单
- 推广链接搜索：方便生成佣金链接、独立商城分销
- 统一签名：自动处理拼多多签名，无需手动拼接
