增強版 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
- 商品詳情同步：確保本地商品信息完整
- 訂單增量同步：只同步指定時間段的訂單
- 推廣鏈接搜索：方便生成傭金鏈接、獨立商城分銷
- 統一簽名：自動處理拼多多簽名，無需手動拼接
