logo

开发者热搜

使用Python调用天眼查API查询企业信息全攻略

作者：快去debug2025.09.25 23:53浏览量：94

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

一、天眼查API基础认知

天眼查作为国内领先的企业信息查询平台，其API接口为开发者提供了结构化的企业数据获取能力。通过API可获取包括但不限于企业基础信息（名称、法人、注册资本）、工商变更记录、司法风险信息、经营状况等核心数据。

1.1 API服务类型

天眼查提供两类主要API接口：

  • 公开数据API：免费但调用频次受限，适合个人开发者和小规模应用
  • 商业数据API：需付费订阅，提供高并发支持和深度数据，适合企业级应用

1.2 认证机制

所有API调用需通过AppKey+AppSecret的双重认证：

  1. import hashlib
  2. import time
  4. def generate_sign(app_secret, params):
  5.     """生成API请求签名"""
  6.     sorted_params = sorted(params.items(), key=lambda x: x[0])
  7.     param_str = "&".join([f"{k}={v}" for k, v in sorted_params])
  8.     sign_str = f"{param_str}&{app_secret}"
  9.     return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()

二、Python环境准备

2.1 基础依赖安装

  1. pip install requests pandas jsonpath-rw
  • requests：处理HTTP请求
  • pandas：数据结构化处理
  • jsonpath-rw：复杂JSON解析

2.2 封装请求类

  1. import requests
  2. import json
  3. from datetime import datetime
  5. class TianYanChaAPI:
  6.     def __init__(self, app_key, app_secret):
  7.         self.app_key = app_key
  8.         self.app_secret = app_secret
  9.         self.base_url = "https://open.api.tianyancha.com"
  11.     def _get_headers(self):
  12.         return {
  13.             "X-TYC-APP-KEY": self.app_key,
  14.             "X-TYC-TIMESTAMP": str(int(datetime.now().timestamp() * 1000)),
  15.             "Content-Type": "application/json"
  16.         }
  18.     def call_api(self, method, path, params=None):
  19.         url = f"{self.base_url}{path}"
  20.         params = params or {}
  21.         params.update({
  22.             "appKey": self.app_key,
  23.             "timestamp": self._get_headers()["X-TYC-TIMESTAMP"]
  24.         })
  25.         params["sign"] = generate_sign(self.app_secret, params)
  27.         response = requests.get(url, params=params, headers=self._get_headers())
  28.         return self._handle_response(response)
  30.     def _handle_response(self, response):
  31.         try:
  32.             data = response.json()
  33.             if data.get("code") != 200:
  34.                 raise Exception(f"API Error: {data.get('message')}")
  35.             return data["data"]
  36.         except json.JSONDecodeError:
  37.             raise Exception("Invalid JSON response")

三、核心查询功能实现

3.1 企业基础信息查询

  1. def get_company_basic(api_client, company_name):
  2.     """查询企业基础信息"""
  3.     path = "/services/open/ic/search/v5"
  4.     params = {
  5.         "keyword": company_name,
  6.         "pageSize": 1
  7.     }
  8.     result = api_client.call_api("GET", path, params)
  9.     if result and result.get("businessEntities"):
  10.         return result["businessEntities"][0]
  11.     return None

3.2 工商变更记录查询

  1. def get_change_records(api_client, company_id):
  2.     """获取企业工商变更记录"""
  3.     path = f"/services/open/ic/changeinfo/v2/{company_id}"
  4.     return api_client.call_api("GET", path)

3.3 司法风险信息查询

  1. def get_legal_risks(api_client, company_id):
  2.     """查询企业司法风险信息"""
  3.     path = f"/services/open/ic/lawSuit/v4/{company_id}"
  4.     return api_client.call_api("GET", path)

四、高级应用场景

4.1 批量查询优化

  1. import concurrent.futures
  3. def batch_query(api_client, company_names):
  4.     """并发批量查询企业信息"""
  5.     results = []
  6.     with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
  7.         future_to_name = {
  8.             executor.submit(get_company_basic, api_client, name): name 
  9.             for name in company_names
  10.         }
  11.         for future in concurrent.futures.as_completed(future_to_name):
  12.             name = future_to_name[future]
  13.             try:
  14.                 results.append((name, future.result()))
  15.             except Exception as exc:
  16.                 results.append((name, {"error": str(exc)}))
  17.     return results

4.2 数据持久化存储

  1. import sqlite3
  3. def save_to_db(company_data):
  4.     """将查询结果存入SQLite数据库"""
  5.     conn = sqlite3.connect('company_data.db')
  6.     cursor = conn.cursor()
  7.     cursor.execute('''
  8.         CREATE TABLE IF NOT EXISTS companies (
  9.             id TEXT PRIMARY KEY,
  10.             name TEXT,
  11.             legal_person TEXT,
  12.             reg_capital TEXT,
  13.             est_date TEXT,
  14.             status TEXT
  15.         )
  16.     ''')
  18.     for data in company_data:
  19.         if data and "id" in data:
  20.             cursor.execute('''
  21.                 INSERT OR REPLACE INTO companies VALUES (?,?,?,?,?,?)
  22.             ''', (
  23.                 data["id"],
  24.                 data["name"],
  25.                 data.get("legalPersonName"),
  26.                 data.get("regCapital"),
  27.                 data.get("estiblishTime"),
  28.                 data.get("companyStatus")
  29.             ))
  30.     conn.commit()
  31.     conn.close()

五、最佳实践建议

5.1 调用频率控制

  • 免费版API建议QPS≤5
  • 商业版可通过X-TYC-RATE-LIMIT头信息获取剩余配额
  • 实现指数退避重试机制：
    ```python
    import time
    import random

def retry_api_call(api_func, max_retries=3):
for attempt in range(max_retries):
try:
return api_func()
except Exception as e:
if attempt == max_retries - 1:
raise
wait_time = min(2 ** attempt + random.uniform(0, 1), 10)
time.sleep(wait_time)

  2. ## 5.2 数据质量校验
  3. ```python
  4. def validate_company_data(data):
  5.     """验证企业数据完整性"""
  6.     required_fields = ["id", "name", "companyStatus"]
  7.     missing = [field for field in required_fields if field not in data]
  8.     if missing:
  9.         raise ValueError(f"Missing required fields: {missing}")
  10.     return True

5.3 异常处理体系

  1. class TYCAPIError(Exception):
  2.     """天眼查API基础异常类"""
  3.     pass
  5. class RateLimitError(TYCAPIError):
  6.     """调用频率超限异常"""
  7.     pass
  9. class AuthError(TYCAPIError):
  10.     """认证失败异常"""
  11.     pass
  13. def handle_api_errors(response):
  14.     if response.status_code == 403:
  15.         raise AuthError("Invalid API credentials")
  16.     elif response.status_code == 429:
  17.         raise RateLimitError("API rate limit exceeded")
  18.     elif response.status_code != 200:
  19.         raise TYCAPIError(f"API request failed with status {response.status_code}")

六、完整示例流程

  1. if __name__ == "__main__":
  2.     # 初始化API客户端
  3.     api_client = TianYanChaAPI(
  4.         app_key="your_app_key",
  5.         app_secret="your_app_secret"
  6.     )
  8.     # 查询单个企业
  9.     company = get_company_basic(api_client, "阿里巴巴")
  10.     if company:
  11.         print(f"企业名称: {company['name']}")
  12.         print(f"法人代表: {company.get('legalPersonName')}")
  13.         print(f"注册资本: {company.get('regCapital')}")
  15.         # 查询工商变更
  16.         changes = get_change_records(api_client, company["id"])
  17.         print(f"变更记录数: {len(changes.get('changeInfos', []))}")
  19.     # 批量查询示例
  20.     companies = ["腾讯", "百度", "字节跳动"]
  21.     batch_results = batch_query(api_client, companies)
  22.     for name, data in batch_results:
  23.         print(f"{name}: {'成功' if data else '失败'}")

七、注意事项

  1. 合规性要求：严格遵守《网络安全法》和《数据安全法》，不得非法获取或使用企业数据
  2. 服务条款：商业用途需签订正式API服务协议
  3. 数据更新：天眼查数据存在T+1的更新延迟，关键决策需结合最新工商信息
  4. 缓存策略：建议对查询结果实施24小时缓存，减少重复调用

通过系统化的API调用和数据处理，Python开发者可以高效构建企业信息查询系统，为风险控制、商业分析等场景提供可靠的数据支持。实际开发中应持续关注天眼查API的版本更新和接口调整，确保系统的长期稳定性。

相关文章推荐

发表评论

最热文章

关于作者

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