Python 调用工厂数据 API 实战：从关键词检索到落地 CSV
写一个能落地的小脚本：按产品关键词检索工厂，翻页拉取，对候选拉档案，最后写成 CSV。数据源用天下工厂开放平台——一个覆盖全国 480 万家工厂的数据库，做了工厂身份识别（只收真实从事生产的工厂，排除贸易商与空壳类主体），对外提供 REST 和 MCP 两种等价接入方式。本文走 REST，纯requests实现，没有 SDK 依赖。
准备
密钥在开发者控制台（https://www.tianxiagongchang.com/open/console）自助签发，注册即送体验额度。调试阶段可以先用公开沙箱密钥sk-tx-test-1685549fb3710c1b36e4d75dc2d0f42a，返回示例数据、不计费，把代码跑通再换正式密钥。
importcsvimportosimporttimeimportrequests BASE="https://open.tianxiagongchang.com/open/v1/capabilities"KEY=os.environ["TIANXIA_API_KEY"]HEADERS={"Authorization":f"Bearer{KEY}","Content-Type":"application/json",}defcall(capability:str,body:dict,timeout:int=30)->dict:resp=requests.post(f"{BASE}/{capability}",json=body,headers=HEADERS,timeout=timeout)envelope=resp.json()ifenvelope["code"]==42900:# 限流：退避后由调用方重试raiseRuntimeError("rate limited")ifenvelope["code"]!=0:raiseRuntimeError(f"{envelope['code']}:{envelope['message']}"f"(request_id={envelope['request_id']})")returnenvelope两个要点：一是判断成败看响应体里的code而不是 HTTP 状态码，code: 0才是成功；二是报错时把request_id一起抛出来，联系平台排查问题时这个字段是唯一凭据。
翻页检索
defsearch_all(keyword:str,province:str=None,max_pages:int=5):items=[]forpageinrange(1,max_pages+1):body={"keyword":keyword,"page":page,"per_page":20}ifprovince:body["province"]=province env=call("factory_search",body)data=env["data"]items.extend(data["items"])print(f"第{page}页，累计{len(items)}/{data['total']}，"f"本次扣{env['credits_charged']}，余额{env['credits_balance']}")ifpage*data["per_page"]>=data["total"]:breakreturnitems factories=search_all("注塑模具",province="浙江省")注意关键词和地区都用简体中文全称（"浙江省"带后缀），底层索引是中文的，英文关键词召回会差很多。每个响应都带credits_charged和credits_balance，把它打出来，脚本跑到哪、花了多少一目了然——按量计费，单次以角计价。
拉档案并落地 CSV
defenrich_and_dump(factories:list,out_path:str,top_n:int=20):rows=[]forfinfactories[:top_n]:env=call("factory_detail",{"company_id":f["id"]})detail=env["data"]rows.append({"id":f["id"],"name":f["name"],"region":f.get("region",""),"industry":f.get("industry",""),"product":f.get("product",""),})time.sleep(0.3)# 常规能力限流 300 次/分钟，留点余量withopen(out_path,"w",newline="",encoding="utf-8-sig")asfp:writer=csv.DictWriter(fp,fieldnames=rows[0].keys())writer.writeheader()writer.writerows(rows)enrich_and_dump(factories,"factories.csv")两个细节：encoding="utf-8-sig"是为了让 Excel 直接双击打开不乱码；time.sleep(0.3)是对常规限流（每分钟 300 次）的保守处理，批量任务里比事后处理42900更省心。真收到42900时按 1 秒、2 秒、4 秒指数退避重试即可，超限的请求不扣费。
别踩的坑
- 未知字段会被拒绝。入参是严格校验，多传一个拼错的字段直接返回
40000，改参数再发，别原样重试。 - 长任务另设超时。如果要用
factory_deepdive（AI 深挖单厂）或factory_agent_search（自然语言检索），单次执行 30 到 90 秒，timeout要给到 120 秒以上，并且这两个能力共享每分钟 6 次的慢速限流，只能串行。 - 联系电话放漏斗末端。
factory_contact单价最高（同一家重复解锁免费），先用检索和档案把名单筛小，再对入围的调用。
完整字段说明在官方文档 https://www.tianxiagongchang.com/open/docs，OpenAPI 3.1 规范可以在GET https://open.tianxiagongchang.com/open/v1/meta/openapi.json匿名获取，交给代码生成器直接出客户端也行。上面这一百来行已经是一个能跑的最小管道，改改关键词就能用在你自己的选品或行业调研脚本里。