logo

开发者热搜

如何使用Python高效查询天眼查企业信息

作者：公子世无双2025.09.25 23:53浏览量：0

简介：本文详细介绍如何通过Python调用天眼查API实现企业信息自动化查询，涵盖API申请、请求封装、数据解析及异常处理全流程，提供可复用的代码示例和优化建议。

如何使用Python高效查询天眼查企业信息

一、天眼查API核心价值与适用场景

天眼查作为国内领先的企业信息查询平台，其开放API为开发者提供了结构化的企业数据获取能力。通过Python调用天眼查API，可实现企业基础信息、工商信息、司法风险、经营状况等数据的自动化采集，适用于以下场景：

  1. 金融风控：批量核查企业征信数据，构建风险评估模型
  2. 市场调研：获取行业企业分布数据，分析市场竞争格局
  3. 供应链管理：验证供应商资质，监控合作伙伴经营状态
  4. 法律服务：快速调取企业涉诉记录，辅助案件分析

相较于网页爬取，API方式具有数据准确性高、请求稳定性强、合规风险低等优势。天眼查API返回JSON格式数据，包含企业名称、统一社会信用代码、法定代表人、注册资本、成立日期、经营范围、股东信息、司法风险等200+字段。

二、Python调用天眼查API全流程

1. API权限申请与配置

访问天眼查开放平台（https://open.tianyancha.com/），完成企业认证后申请API密钥。获取以下关键参数：

  • API_KEY：接口调用凭证
  • SECRET_KEY：用于生成签名
  • APP_ID：应用标识

建议将密钥存储在环境变量中，避免硬编码在代码里：

  1. import os
  2. from dotenv import load_dotenv
  4. load_dotenv()  # 从.env文件加载环境变量
  6. API_KEY = os.getenv('TIANYANCHA_API_KEY')
  7. SECRET_KEY = os.getenv('TIANYANCHA_SECRET_KEY')
  8. APP_ID = os.getenv('TIANYANCHA_APP_ID')

2. 请求签名生成算法

天眼查API要求每个请求必须包含签名参数sign，生成步骤如下：

  1. 按字典序拼接参数名和参数值
  2. 拼接SECRET_KEY作为盐值
  3. 对拼接字符串进行MD5加密

实现代码：

  1. import hashlib
  2. import time
  3. import random
  4. from urllib.parse import urlencode
  6. def generate_sign(params, secret_key):
  7.     # 参数按key排序
  8.     sorted_params = sorted(params.items(), key=lambda x: x[0])
  9.     param_str = ''.join([f"{k}{v}" for k, v in sorted_params])
  10.     param_str += secret_key
  11.     return hashlib.md5(param_str.encode('utf-8')).hexdigest()
  13. def build_request_url(api_path, params):
  14.     base_params = {
  15.         'appId': APP_ID,
  16.         'timestamp': str(int(time.time())),
  17.         'nonce': str(random.randint(10000, 99999))
  18.     }
  19.     base_params.update(params)
  20.     base_params['sign'] = generate_sign(base_params, SECRET_KEY)
  21.     return f"https://api.tianyancha.com{api_path}?{urlencode(base_params)}"

3. 企业信息查询实现

以查询企业基础信息为例，API路径为/services/v3/open/search，关键参数：

  • keyword：企业名称或关键词
  • pageSize：每页返回数量（最大50）

完整实现：

  1. import requests
  2. import json
  4. def query_enterprise_info(keyword):
  5.     api_path = '/services/v3/open/search'
  6.     params = {
  7.         'keyword': keyword,
  8.         'pageSize': 10
  9.     }
  10.     url = build_request_url(api_path, params)
  12.     try:
  13.         response = requests.get(url, timeout=10)
  14.         response.raise_for_status()
  15.         data = response.json()
  17.         if data.get('code') != 0:
  18.             raise Exception(f"API错误: {data.get('message')}")
  20.         return data['data']['items']
  21.     except requests.exceptions.RequestException as e:
  22.         print(f"请求失败: {str(e)}")
  23.         return None
  25. # 示例调用
  26. if __name__ == "__main__":
  27.     results = query_enterprise_info("阿里巴巴")
  28.     if results:
  29.         for item in results:
  30.             print(f"企业名称: {item['name']}")
  31.             print(f"统一信用代码: {item['creditCode']}")
  32.             print(f"法定代表人: {item['legalPersonName']}")

三、高级功能实现技巧

1. 批量查询优化

对于需要查询大量企业的情况，建议：

  1. 使用多线程/异步请求提高吞吐量
  2. 实现请求频率控制（天眼查API限制QPS≤10）
  3. 添加缓存机制减少重复查询

异步实现示例：

  1. import asyncio
  2. import aiohttp
  4. async def async_query(keyword):
  5.     api_path = '/services/v3/open/search'
  6.     params = {'keyword': keyword, 'pageSize': 1}
  7.     url = build_request_url(api_path, params)
  9.     async with aiohttp.ClientSession() as session:
  10.         async with session.get(url) as response:
  11.             return await response.json()
  13. async def batch_query(keywords):
  14.     tasks = [async_query(kw) for kw in keywords]
  15.     return await asyncio.gather(*tasks)

2. 数据解析与清洗

天眼查返回的数据可能包含嵌套结构，建议使用pydantic进行模型验证：

  1. from pydantic import BaseModel
  2. from typing import Optional, List
  4. class Shareholder(BaseModel):
  5.     name: str
  6.     shareRatio: float
  8. class Enterprise(BaseModel):
  9.     name: str
  10.     creditCode: str
  11.     legalPersonName: str
  12.     registeredCapital: str
  13.     shareholders: Optional[List[Shareholder]]
  15. def parse_enterprise(data):
  16.     return Enterprise(
  17.         name=data['name'],
  18.         creditCode=data['creditCode'],
  19.         legalPersonName=data['legalPersonName'],
  20.         registeredCapital=data['registeredCapital'],
  21.         shareholders=[
  22.             Shareholder(name=s['name'], shareRatio=s['shareRatio'])
  23.             for s in data.get('shareholders', [])
  24.         ]
  25.     )

3. 异常处理机制

完善的异常处理应包括：

  • 网络请求异常
  • API返回错误
  • 数据格式异常
  • 频率限制处理

增强版查询函数：

  1. from requests.exceptions import HTTPError, Timeout, ConnectionError
  3. def robust_query(keyword, max_retries=3):
  4.     api_path = '/services/v3/open/search'
  5.     params = {'keyword': keyword, 'pageSize': 1}
  7.     for attempt in range(max_retries):
  8.         try:
  9.             url = build_request_url(api_path, params)
  10.             response = requests.get(url, timeout=10)
  11.             response.raise_for_status()
  12.             data = response.json()
  14.             if data.get('code') != 0:
  15.                 if data.get('code') == 403 and attempt < max_retries - 1:
  16.                     time.sleep(2 ** attempt)  # 指数退避
  17.                     continue
  18.                 raise Exception(data.get('message', '未知错误'))
  20.             return data['data']['items'][0] if data['data']['items'] else None
  22.         except (HTTPError, Timeout, ConnectionError) as e:
  23.             if attempt == max_retries - 1:
  24.                 raise
  25.             time.sleep(1)

四、最佳实践与注意事项

  1. 合规使用：严格遵守天眼查API使用条款，不得用于非法用途
  2. 请求频率：控制请求速率，避免触发IP限制（建议≤5QPS）
  3. 数据存储：对查询结果进行脱敏处理，符合数据安全法规
  4. 错误监控：建立日志系统，记录API调用失败情况
  5. 版本管理：关注API版本更新，及时调整调用参数

五、扩展应用场景

  1. 企业关系图谱：通过/services/v3/open/graph/invest接口获取企业投资关系
  2. 变更监控：设置定时任务，跟踪目标企业工商变更
  3. 行业分析：批量采集行业企业数据，进行统计分析
  4. 风险预警：结合司法风险数据，构建企业风险预警系统

通过Python与天眼查API的深度集成，开发者可以构建高效、稳定的企业信息查询系统，为各类商业决策提供数据支持。建议在实际应用中结合具体业务场景，进一步优化查询策略和数据解析逻辑。

相关文章推荐

发表评论

最热文章

关于作者

  • 被阅读数
  • 被赞数
  • 被收藏数
活动
咨询