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

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

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

典型应用场景包括：

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

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

2.1 准备工作

1. API密钥申请：

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

2. 开发环境配置：

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

2.2 核心接口调用流程

2.2.1 身份认证

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

import hmac
import hashlib
import base64
import time
from urllib.parse import quote
def generate_sign(secret, method, path, params, timestamp):
    raw_str = f"{method}\n{path}\n{params}\n{timestamp}"
    secret_bytes = secret.encode('utf-8')
    raw_bytes = raw_str.encode('utf-8')
    sign = hmac.new(secret_bytes, raw_bytes, hashlib.sha256).digest()
    return base64.b64encode(sign).decode('utf-8')
# 示例调用
api_key = "your_api_key"
api_secret = "your_api_secret"
timestamp = str(int(time.time()))
path = "/api/company/search"
params = "keyword=阿里巴巴"
method = "GET"
sign = generate_sign(api_secret, method, path, params, timestamp)

2.2.2 企业搜索接口

import requests
def search_company(keyword, page=1, size=10):
    url = "https://api.tianyancha.com/api/company/search"
    headers = {
        "X-TYC-APIKEY": api_key,
        "X-TYC-TIMESTAMP": timestamp,
        "X-TYC-SIGN": sign,
        "Content-Type": "application/json"
    }
    params = {
        "keyword": keyword,
        "page": page,
        "size": size
    }
    try:
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None
# 示例：搜索包含"腾讯"的企业
result = search_company("腾讯")

2.2.3 企业详情接口

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

三、进阶应用技巧

3.1 批量查询优化

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

3.2 数据持久化方案

import sqlite3
def save_to_db(company_data):
    conn = sqlite3.connect("company_data.db")
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS companies (
            id TEXT PRIMARY KEY,
            name TEXT,
            legal_person TEXT,
            registered_capital TEXT,
            establish_date TEXT,
            business_status TEXT
        )
    """)
    for item in company_data.get("items", []):
        cursor.execute("""
            INSERT OR REPLACE INTO companies VALUES (?,?,?,?,?,?)
        """, (
            item["id"],
            item["name"],
            item["legalPersonName"],
            item["registeredCapital"],
            item["establishTime"],
            item["businessStatus"]
        ))
    conn.commit()
    conn.close()
# 示例：保存搜索结果
search_result = search_company("华为")
if search_result:
    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):

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

  ```

4.2 数据解析异常处理

def parse_company_data(raw_data):
    try:
        if not raw_data or "code" not in raw_data or raw_data["code"] != 0:
            raise ValueError("无效的响应数据")
        company = {
            "name": raw_data["data"]["name"],
            "legal_person": raw_data["data"].get("legalPersonName", "未知"),
            "status": raw_data["data"]["businessStatus"],
            "shareholders": [
                {"name": s["name"], "ratio": s["stockPercent"]}
                for s in raw_data["data"].get("stockHolders", [])
            ]
        }
        return company
    except (KeyError, TypeError) as e:
        print(f"数据解析错误: {e}")
        return None

五、最佳实践建议

1. 缓存策略：

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

2. 错误监控：

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

3. 数据更新机制：

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

4. 安全规范：

   • API密钥存储在环境变量或密钥管理服务中
   • 实现请求日志脱敏处理

六、总结与展望

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

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

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

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

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