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

写一个能落地的小脚本按产品关键词检索工厂翻页拉取对候选拉档案最后写成 CSV。数据源用天下工厂开放平台——一个覆盖全国 480 万家工厂的数据库做了工厂身份识别只收真实从事生产的工厂排除贸易商与空壳类主体对外提供 REST 和 MCP 两种等价接入方式。本文走 REST纯requests实现没有 SDK 依赖。准备密钥在开发者控制台https://www.tianxiagongchang.com/open/console自助签发注册即送体验额度。调试阶段可以先用公开沙箱密钥sk-tx-test-1685549fb3710c1b36e4d75dc2d0f42a返回示例数据、不计费把代码跑通再换正式密钥。importcsvimportosimporttimeimportrequests BASEhttps://open.tianxiagongchang.com/open/v1/capabilitiesKEYos.environ[TIANXIA_API_KEY]HEADERS{Authorization:fBearer{KEY},Content-Type:application/json,}defcall(capability:str,body:dict,timeout:int30)-dict:resprequests.post(f{BASE}/{capability},jsonbody,headersHEADERS,timeouttimeout)enveloperesp.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:strNone,max_pages:int5):items[]forpageinrange(1,max_pages1):body{keyword:keyword,page:page,per_page:20}ifprovince:body[province]province envcall(factory_search,body)dataenv[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 factoriessearch_all(注塑模具,province浙江省)注意关键词和地区都用简体中文全称浙江省带后缀底层索引是中文的英文关键词召回会差很多。每个响应都带credits_charged和credits_balance把它打出来脚本跑到哪、花了多少一目了然——按量计费单次以角计价。拉档案并落地 CSVdefenrich_and_dump(factories:list,out_path:str,top_n:int20):rows[]forfinfactories[:top_n]:envcall(factory_detail,{company_id:f[id]})detailenv[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,encodingutf-8-sig)asfp:writercsv.DictWriter(fp,fieldnamesrows[0].keys())writer.writeheader()writer.writerows(rows)enrich_and_dump(factories,factories.csv)两个细节encodingutf-8-sig是为了让 Excel 直接双击打开不乱码time.sleep(0.3)是对常规限流每分钟 300 次的保守处理批量任务里比事后处理42900更省心。真收到42900时按 1 秒、2 秒、4 秒指数退避重试即可超限的请求不扣费。别踩的坑未知字段会被拒绝。入参是严格校验多传一个拼错的字段直接返回40000改参数再发别原样重试。长任务另设超时。如果要用factory_deepdiveAI 深挖单厂或factory_agent_search自然语言检索单次执行 30 到 90 秒timeout要给到 120 秒以上并且这两个能力共享每分钟 6 次的慢速限流只能串行。联系电话放漏斗末端。factory_contact单价最高同一家重复解锁免费先用检索和档案把名单筛小再对入围的调用。完整字段说明在官方文档 https://www.tianxiagongchang.com/open/docsOpenAPI 3.1 规范可以在GET https://open.tianxiagongchang.com/open/v1/meta/openapi.json匿名获取交给代码生成器直接出客户端也行。上面这一百来行已经是一个能跑的最小管道改改关键词就能用在你自己的选品或行业调研脚本里。