)
1688开放平台的基础API商品搜索、详情、订单、物流查询对企业开发者完全免费只要完成企业实名认证并创建「自用型应用」即可直接调用。限制主要体现在QPS每秒查询率和单日调用总量理解这两个阈值才能做好系统调度设计。一、 免费额度的准确数字2026版参考项目免费标准自用型应用说明QPS每秒请求数通常10~20次/秒视接口不同商品搜索多数为10订单查询可到20超出返回ISP_FLOW_CONTROL_LIMIT日调用量无明确公开硬性上限但异常高频会触发风控审查建议单应用≤50万次/天资源包提额购买后QPS可升至50/100/200基础功能仍免费只是提频需付费部分实时库存高级接口、跨境增值、数据推送普通B2B同步不涉及✅结论中小企业做商品同步每日几千~几万次 订单回写免费额度完全够用。二、 受控调用的Python封装含QPS限速 限流重试下面代码在之前API客户端基础上加入令牌桶限速和限流自动退避确保不触发1688流控# ali1688_free_api.py import hashlib import time import requests import urllib.parse from typing import Dict, List, Optional from datetime import datetime # 封装好API供应商demo urlhttps://console.open.onebound.cn/console/?iLex # ───────────────────────────────────────────── # 简易令牌桶控制QPS不超过免费限额 # ───────────────────────────────────────────── class TokenBucket: def __init__(self, rate: float, capacity: int None): self.rate rate # 令牌生成速率 允许QPS self.capacity capacity or int(rate) self.tokens float(self.capacity) self.last_time time.monotonic() def consume(self, n: int 1): now time.monotonic() elapsed now - self.last_time self.tokens min(self.capacity, self.tokens elapsed * self.rate) self.last_time now if self.tokens n: self.tokens - n return True else: # 计算需等待时间 wait (n - self.tokens) / self.rate time.sleep(wait 0.01) self.tokens 0 return True class Ali1688FreeClient: 1688 基础免费API客户端 支持商品搜索(alibaba.offer.search)、商品详情(alibaba.item.get) 订单列表(alibaba.trade.buyer.list)、订单详情(alibaba.trade.get) GATEWAY_SEARCH https://gw.open.1688.com/openapi/param2/2/alibaba.offer.search/2.0 GATEWAY_COMMON https://gw.open.1688.com/openapi/http/2/1 def __init__(self, app_key: str, app_secret: str, access_token: str None, qps: int 8): self.app_key app_key self.app_secret app_secret self.access_token access_token self.bucket TokenBucket(rateqps) # 默认按8 QPS限速留余量 # ───────────── 签名 ───────────── def _sign(self, params: Dict) - str: filtered sorted((k, v) for k, v in params.items() if v is not None) qs .join(f{k}{v} for k, v in filtered) raw f{self.app_secret}{qs}{self.app_secret} return hashlib.md5(raw.encode(utf-8)).hexdigest().upper() def _call_get(self, url: str, method: str, biz: Dict) - Dict: self.bucket.consume() # ← QPS控制点 api_params { method: method, app_key: self.app_key, timestamp: str(int(time.time() * 1000)), format: json, v: 2.0, sign_method: md5, } if self.access_token: api_params[session] self.access_token api_params[param2 if param2 in url else param] \ urllib.parse.quote_plus(str(biz).replace(, )) api_params[sign] self._sign(api_params) for attempt in range(3): resp requests.get(url, paramsapi_params, timeout15) resp.raise_for_status() data resp.json() # 限流检测 → 指数退避重试 if error_response in data: err_code str(data[error_response].get(code, )) if FLOW_CONTROL in err_code or err_code 429: wait 2 ** attempt print(f⚠️ 触发限流{wait}s后重试(第{attempt1}次)...) time.sleep(wait) continue else: err data[error_response] raise Exception(f1688 Err[{err.get(code)}]: {err.get(msg)}) # 提取结果 result_key [k for k in data if k ! error_response] if result_key: return data[result_key[0]] return data raise Exception(API调用失败持续被限流请降低QPS或购买资源包) # ───────────── 商品搜索 ───────────── def search_products(self, keyword: str, page_no: int 1, page_size: int 40, price_minNone, price_maxNone) - Dict: biz { keywords: keyword, pageNo: page_no, pageSize: min(page_size, 50), sortType: booked } if price_min is not None: biz[beginPrice] str(int(price_min * 100)) if price_max is not None: biz[endPrice] str(int(price_max * 100)) return self._call_get(self.GATEWAY_SEARCH, alibaba.offer.search, biz) # ───────────── 商品详情 ───────────── def get_product_detail(self, offer_id: str, fields: str None) - Dict: biz {item_id: offer_id} if fields: biz[fields] fields return self._call_get( self.GATEWAY_COMMON, alibaba.item.get, biz ).get(alibaba_item_get_response, {}).get(item, {}) # ───────────── 订单列表简易封装──────────── def list_orders(self, status: str waitsellersend, hours_back: int 48, page_no: int 1) - List[Dict]: biz { orderStatus: status, gmtCreateStart: (datetime.now() .replace(microsecond0).__class__.__sub__(datetime.now(), hourshours_back) .strftime(%Y-%m-%d %H:%M:%S).replace(__sub__, -)), gmtCreateEnd: datetime.now().strftime(%Y-%m-%d %H:%M:%S), pageNo: page_no, pageSize: 50 } # 简化实际建议用之前专用订单client from datetime import timedelta biz[gmtCreateStart] (datetime.now() - timedelta(hourshours_back)).strftime(%Y-%m-%d %H:%M:%S) res self._call_get(self.GATEWAY_COMMON, alibaba.trade.buyer.list, biz) return res.get(alibaba_trade_buyer_list_response, {}).get(tradeModelList, []) or [] # 封装好API供应商demo urlhttps://console.open.onebound.cn/console/?iLex # # 使用示例 # if __name__ __main__: client Ali1688FreeClient( app_keyYOUR_APP_KEY, app_secretYOUR_APP_SECRET, # access_tokenNone # 商品搜索可不传订单查询需传session key qps8 # 留余量低于免费上限10 ) try: # 搜索 result client.search_products(不锈钢保温杯, page_no1, price_min15, price_max60) offers result.get(offers, []) total result.get(totalResult, 0) print(f✅ 找到 {total} 条本页 {len(offers)} 条) if offers: offer_id offers[0].get(offerId) # 详情 detail client.get_product_detail(str(offer_id)) print(f商品: {detail.get(title)} | 起批价: {detail.get(price)}) except Exception as e: print(f❌ {e})三、 免费调用四大注意事项避坑QPS必须主动限速免费应用无弹性超频立刻返回ISP_FLOW_CONTROL_LIMIT。上述TokenBucket按8 QPS限速是安全值。Access Token区分场景商品搜索/详情 →可不传公开数据订单/物流/购物车 →必须传买家登录态 session key企业认证是前提个人开发者应用无法调用订单类接口且部分商品字段被裁剪。全量同步错峰每日全量商品同步放凌晨2~4点分页步进sleep≥0.2s避免短时突发超QPS。四、 什么时候需要买资源包现象建议商品搜索QPS长期接近10且影响同步时效买基础包提至50 QPS需要实时可售库存非页面展示价买含库存查询的资源包日调用量 100万且频繁触发人工审核警告联系1688商务谈企业套餐绝大多数中小ERP做商品主数据同步 订单自动回写不用花钱。如需我补充OAuth2获取Access Token完整代码或每日增量同步定时任务(Crontab/APScheduler)直接说