ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀,每一条都经过真实交付验证。

Python 调用工厂数据 API 实战:从关键词检索到落地 CSV

2026/8/3 3:38:49 拓冰建站 浏览量
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_chargedcredits_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匿名获取,交给代码生成器直接出客户端也行。上面这一百来行已经是一个能跑的最小管道,改改关键词就能用在你自己的选品或行业调研脚本里。