天眼查API助力Python：高效查询企业信息的实践指南

作者：暴富20212025.09.18 16:00浏览量：59

简介：本文详细介绍如何使用Python调用天眼查API实现企业信息查询，涵盖API接入流程、参数配置、错误处理及代码示例，助力开发者快速构建企业信息检索系统。

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

天眼查作为国内领先的企业信息查询平台，其API接口为开发者提供了结构化、高时效的企业数据服务。相较于网页爬取，API调用具有三大核心优势：

  1. 数据合法性：通过官方授权接口获取数据，规避法律风险
  2. 效率提升：单次请求可获取企业基础信息、股东结构、司法风险等20+维度数据
  3. 稳定性保障：官方维护的接口服务，避免网页结构变更导致的解析失败

典型应用场景包括：

  • 金融风控系统中的企业征信核查
  • 供应链管理系统中的供应商资质审查
  • 商业智能工具中的竞品分析模块
  • 法律服务平台的案件关联企业检索

二、Python调用天眼查API的技术实现

2.1 准备工作

  1. API密钥申请：

    • 登录天眼查开发者平台（https://open.tianyancha.com）
    • 完成企业认证后获取API Key和Secret
    • 注意：免费版每日调用限额为100次，商业应用需购买套餐

  2. 开发环境配置：

    1. # 推荐使用requests库处理HTTP请求
    2. pip install requests
    3. # 数据解析建议使用pandas
    4. pip install pandas

2.2 核心接口调用流程

2.2.1 身份认证

天眼查API采用HMAC-SHA256签名认证机制：

  1. import hmac
  2. import hashlib
  3. import base64
  4. import time
  5. from urllib.parse import quote
  7. def generate_sign(secret, method, path, params, timestamp):
  8.     raw_str = f"{method}\n{path}\n{params}\n{timestamp}"
  9.     secret_bytes = secret.encode('utf-8')
  10.     raw_bytes = raw_str.encode('utf-8')
  11.     sign = hmac.new(secret_bytes, raw_bytes, hashlib.sha256).digest()
  12.     return base64.b64encode(sign).decode('utf-8')
  14. # 示例调用
  15. api_key = "your_api_key"
  16. api_secret = "your_api_secret"
  17. timestamp = str(int(time.time()))
  18. path = "/api/company/search"
  19. params = "keyword=阿里巴巴"
  20. method = "GET"
  22. sign = generate_sign(api_secret, method, path, params, timestamp)

2.2.2 企业搜索接口

  1. import requests
  3. def search_company(keyword, page=1, size=10):
  4.     url = "https://api.tianyancha.com/api/company/search"
  5.     headers = {
  6.         "X-TYC-APIKEY": api_key,
  7.         "X-TYC-TIMESTAMP": timestamp,
  8.         "X-TYC-SIGN": sign,
  9.         "Content-Type": "application/json"
  10.     }
  11.     params = {
  12.         "keyword": keyword,
  13.         "page": page,
  14.         "size": size
  15.     }
  17.     try:
  18.         response = requests.get(url, headers=headers, params=params)
  19.         response.raise_for_status()
  20.         return response.json()
  21.     except requests.exceptions.RequestException as e:
  22.         print(f"请求失败: {e}")
  23.         return None
  25. # 示例：搜索包含"腾讯"的企业
  26. result = search_company("腾讯")

2.2.3 企业详情接口

  1. def get_company_detail(company_id):
  2.     url = f"https://api.tianyancha.com/api/company/{company_id}"
  3.     headers = {
  4.         "X-TYC-APIKEY": api_key,
  5.         "X-TYC-TIMESTAMP": str(int(time.time())),
  6.         # 需重新生成签名
  7.         "X-TYC-SIGN": generate_sign(api_secret, "GET", f"/api/company/{company_id}", "", str(int(time.time()))),
  8.     }
  10.     try:
  11.         response = requests.get(url, headers=headers)
  12.         response.raise_for_status()
  13.         return response.json()
  14.     except requests.exceptions.RequestException as e:
  15.         print(f"获取详情失败: {e}")
  16.         return None
  18. # 示例：获取企业ID为123456的详情
  19. detail = get_company_detail("123456")

三、进阶应用技巧

3.1 批量查询优化

  1. import pandas as pd
  2. from concurrent.futures import ThreadPoolExecutor
  4. def batch_search(keywords, max_workers=5):
  5.     results = []
  6.     with ThreadPoolExecutor(max_workers=max_workers) as executor:
  7.         futures = [executor.submit(search_company, kw) for kw in keywords]
  8.         for future in futures:
  9.             data = future.result()
  10.             if data:
  11.                 results.extend(data.get("items", []))
  12.     return pd.DataFrame(results)
  14. # 示例：批量查询多个关键词
  15. keywords = ["字节跳动", "美团", "京东"]
  16. df = batch_search(keywords)

3.2 数据持久化方案

  1. import sqlite3
  3. def save_to_db(company_data):
  4.     conn = sqlite3.connect("company_data.db")
  5.     cursor = conn.cursor()
  6.     cursor.execute("""
  7.         CREATE TABLE IF NOT EXISTS companies (
  8.             id TEXT PRIMARY KEY,
  9.             name TEXT,
  10.             legal_person TEXT,
  11.             registered_capital TEXT,
  12.             establish_date TEXT,
  13.             business_status TEXT
  14.         )
  15.     """)
  17.     for item in company_data.get("items", []):
  18.         cursor.execute("""
  19.             INSERT OR REPLACE INTO companies VALUES (?,?,?,?,?,?)
  20.         """, (
  21.             item["id"],
  22.             item["name"],
  23.             item["legalPersonName"],
  24.             item["registeredCapital"],
  25.             item["establishTime"],
  26.             item["businessStatus"]
  27.         ))
  28.     conn.commit()
  29.     conn.close()
  31. # 示例：保存搜索结果
  32. search_result = search_company("华为")
  33. if search_result:
  34.     save_to_db(search_result)

四、常见问题处理

4.1 接口调用频率限制

  • 免费版：100次/日，QPS≤5

  • 解决方案：

    • 实现请求队列，控制并发数
    • 错误码429时，自动重试（间隔递增）
      ```python
      import time
      from requests.exceptions import HTTPError

    def safe_request(func, max_retries=3):

    1. for i in range(max_retries):
    2.     try:
    3.         return func()
    4.     except HTTPError as e:
    5.         if e.response.status_code == 429 and i < max_retries - 1:
    6.             sleep_time = (i + 1) * 2
    7.             time.sleep(sleep_time)
    8.         else:
    9.             raise

    ```

4.2 数据解析异常处理

  1. def parse_company_data(raw_data):
  2.     try:
  3.         if not raw_data or "code" not in raw_data or raw_data["code"] != 0:
  4.             raise ValueError("无效的响应数据")
  6.         company = {
  7.             "name": raw_data["data"]["name"],
  8.             "legal_person": raw_data["data"].get("legalPersonName", "未知"),
  9.             "status": raw_data["data"]["businessStatus"],
  10.             "shareholders": [
  11.                 {"name": s["name"], "ratio": s["stockPercent"]}
  12.                 for s in raw_data["data"].get("stockHolders", [])
  13.             ]
  14.         }
  15.         return company
  16.     except (KeyError, TypeError) as e:
  17.         print(f"数据解析错误: {e}")
  18.         return None

五、最佳实践建议

  1. 缓存策略：

    • 对高频查询的企业ID建立本地缓存（Redis/SQLite）
    • 设置合理的TTL（如企业基本信息缓存24小时）

  2. 错误监控：

    • 记录所有API调用日志（请求参数、响应时间、错误码）
    • 设置异常报警阈值（如连续5次429错误）

  3. 数据更新机制：

    • 对关键企业信息（如司法风险）实现增量更新
    • 使用ETag或Last-Modified头实现条件请求

  4. 安全规范：

六、总结与展望

通过Python调用天眼查API，开发者可以高效构建企业信息查询系统。关键实施要点包括：

  1. 严格遵循API调用规范，做好身份认证
  2. 实现健壮的错误处理和重试机制
  3. 结合业务场景设计合理的数据缓存策略
  4. 持续监控接口调用指标，优化调用频率

未来随着企业数据服务的发展，建议关注：

  • 天眼查新推出的深度分析接口
  • 结合OCR技术实现非结构化数据解析
  • 探索企业关联图谱的可视化应用

通过系统化的API调用方案，开发者能够为企业风控、供应链管理等场景提供可靠的数据支持，创造显著的业务价值。

最热文章