logo

开发者热搜

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

作者：狼烟四起2025.09.18 16:00浏览量：0

简介：本文详细介绍了如何通过Python调用天眼查API接口实现企业信息的自动化查询，涵盖API认证、请求构造、数据处理及异常处理等关键环节，帮助开发者快速构建高效的企业信息检索系统。

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

一、天眼查API核心价值解析

天眼查作为国内领先的企业信息查询平台，其API接口为开发者提供了结构化、标准化的企业数据获取通道。相较于传统网页爬取方式，API调用具有三大优势：

  1. 数据合规性：严格遵循《网络安全法》要求，确保数据来源合法
  2. 访问稳定性：提供QPS（每秒查询数）控制机制，避免IP被封禁
  3. 数据完整性：覆盖工商信息、司法风险、经营状况等20+维度数据字段

典型应用场景包括：

  • 金融机构贷前尽调自动化
  • 供应链企业资质核验
  • 商业情报分析系统建设
  • 法律风险预警系统开发

二、Python环境准备与依赖安装

2.1 基础环境配置

  1. # 推荐Python 3.7+版本
  2. import sys
  3. print(sys.version)  # 应显示3.7.x或更高版本

2.2 核心依赖库安装

  1. pip install requests pandas jsonpath-ng
库名称 用途说明 版本要求
requests HTTP请求处理 ≥2.24.0
pandas 数据结构化处理 ≥1.1.0
jsonpath-ng JSON数据解析 ≥1.5.2

三、API认证机制详解

3.1 认证方式对比

认证方式 安全性 适用场景 配置复杂度
API Key 中 简单查询场景 ★
OAuth2.0 高 长期集成项目 ★★★
签名验证 极高 金融级应用 ★★★★

3.2 签名验证实现示例

  1. import hashlib
  2. import time
  3. import hmac
  5. def generate_sign(secret_key, params):
  6.     """
  7.     生成HMAC-SHA256签名
  8.     :param secret_key: 分配的密钥
  9.     :param params: 请求参数字典
  10.     :return: 十六进制签名
  11.     """
  12.     sorted_params = sorted(params.items(), key=lambda x: x[0])
  13.     param_str = '&'.join([f"{k}={v}" for k, v in sorted_params])
  14.     timestamp = str(int(time.time()))
  15.     message = f"{timestamp}&{param_str}"
  16.     return hmac.new(secret_key.encode(), message.encode(), hashlib.sha256).hexdigest()

四、核心查询接口实现

4.1 企业基础信息查询

  1. import requests
  2. import json
  4. def get_company_info(api_key, company_name):
  5.     """
  6.     获取企业基础信息
  7.     :param api_key: 天眼查分配的API Key
  8.     :param company_name: 企业名称（支持模糊查询）
  9.     :return: 解析后的企业信息字典
  10.     """
  11.     url = "https://api.tianyancha.com/services/open/ic/v2/company/search"
  12.     headers = {
  13.         "X-Auth-Token": api_key,
  14.         "Content-Type": "application/json"
  15.     }
  16.     params = {
  17.         "keyword": company_name,
  18.         "pageSize": 1
  19.     }
  21.     try:
  22.         response = requests.get(url, headers=headers, params=params)
  23.         response.raise_for_status()
  24.         data = response.json()
  26.         # 使用jsonpath提取核心字段
  27.         from jsonpath_ng import parse
  28.         expr = parse('$.data.items[*].legalPersonName')
  29.         legal_person = [match.value for match in expr.find(data)]
  31.         return {
  32.             "company_name": data.get("data", {}).get("items", [{}])[0].get("name"),
  33.             "legal_person": legal_person[0] if legal_person else None,
  34.             "registered_capital": data.get("data", {}).get("items", [{}])[0].get("regCapital")
  35.         }
  36.     except requests.exceptions.RequestException as e:
  37.         print(f"请求失败: {str(e)}")
  38.         return None

4.2 司法风险信息查询

  1. def get_judicial_risks(api_key, company_id):
  2.     """
  3.     获取企业司法风险信息
  4.     :param api_key: API密钥
  5.     :param company_id: 企业唯一标识
  6.     :return: 风险信息列表
  7.     """
  8.     url = f"https://api.tianyancha.com/services/open/ic/v2/company/{company_id}/judicial"
  9.     headers = {"X-Auth-Token": api_key}
  11.     response = requests.get(url, headers=headers)
  12.     if response.status_code == 200:
  13.         data = response.json()
  14.         risks = []
  15.         for item in data.get("data", {}).get("items", []):
  16.             risks.append({
  17.                 "case_type": item.get("caseType"),
  18.                 "court": item.get("court"),
  19.                 "case_reason": item.get("caseReason"),
  20.                 "publish_date": item.get("publishDate")
  21.             })
  22.         return risks
  23.     else:
  24.         print(f"错误响应: {response.status_code}")
  25.         return []

五、高级功能实现技巧

5.1 批量查询优化

  1. import pandas as pd
  2. from concurrent.futures import ThreadPoolExecutor
  4. def batch_query(api_key, company_names, max_workers=5):
  5.     """
  6.     并发批量查询企业信息
  7.     :param api_key: API密钥
  8.     :param company_names: 企业名称列表
  9.     :param max_workers: 最大并发数
  10.     :return: 包含查询结果的DataFrame
  11.     """
  12.     results = []
  14.     def query_single(name):
  15.         info = get_company_info(api_key, name)
  16.         if info:
  17.             results.append(info)
  19.     with ThreadPoolExecutor(max_workers=max_workers) as executor:
  20.         executor.map(query_single, company_names)
  22.     return pd.DataFrame(results)

5.2 数据缓存机制

  1. import pickle
  2. import os
  3. from functools import wraps
  5. def cache_results(cache_dir="./.tianyan_cache"):
  6.     """
  7.     查询结果缓存装饰器
  8.     :param cache_dir: 缓存目录
  9.     """
  10.     if not os.path.exists(cache_dir):
  11.         os.makedirs(cache_dir)
  13.     def decorator(func):
  14.         @wraps(func)
  15.         def wrapper(api_key, *args, **kwargs):
  16.             cache_key = f"{func.__name__}_{'_'.join(map(str, args))}.pkl"
  17.             cache_path = os.path.join(cache_dir, cache_key)
  19.             if os.path.exists(cache_path):
  20.                 with open(cache_path, "rb") as f:
  21.                     return pickle.load(f)
  23.             result = func(api_key, *args, **kwargs)
  24.             with open(cache_path, "wb") as f:
  25.                 pickle.dump(result, f)
  27.             return result
  28.         return wrapper
  29.     return decorator

六、最佳实践与注意事项

6.1 频率控制策略

  • 基础版API：≤5次/秒
  • 专业版API：≤20次/秒
  • 实现建议：
    ```python
    import time

class RateLimiter:
def init(self, max_calls, period):
self.max_calls = max_calls
self.period = period
self.call_times = []

  1. def wait(self):
  2.     now = time.time()
  3.     self.call_times = [t for t in self.call_times if now - t < self.period]
  4.     if len(self.call_times) >= self.max_calls:
  5.         oldest = self.call_times[0]
  6.         sleep_time = self.period - (now - oldest)
  7.         if sleep_time > 0:
  8.             time.sleep(sleep_time)
  9.     self.call_times.append(time.time())
  2. ### 6.2 错误处理机制
  3. ```python
  4. def handle_api_error(response):
  5.     """
  6.     统一API错误处理
  7.     :param response: requests响应对象
  8.     :return: 错误信息字典或None
  9.     """
  10.     if response.status_code == 401:
  11.         return {"error": "认证失败", "code": 401}
  12.     elif response.status_code == 429:
  13.         retry_after = int(response.headers.get("Retry-After", 60))
  14.         return {"error": "频率限制", "code": 429, "retry_after": retry_after}
  15.     elif response.status_code == 500:
  16.         return {"error": "服务端错误", "code": 500}
  17.     else:
  18.         try:
  19.             error_data = response.json()
  20.             return {"error": error_data.get("message"), "code": error_data.get("code")}
  21.         except ValueError:
  22.             return {"error": "未知错误", "code": response.status_code}

七、完整项目示例

  1. # main.py 完整示例
  2. import pandas as pd
  3. from functools import partial
  5. class TianYanChaClient:
  6.     def __init__(self, api_key):
  7.         self.api_key = api_key
  8.         self.rate_limiter = RateLimiter(max_calls=10, period=1)
  10.     def query_company(self, company_name):
  11.         self.rate_limiter.wait()
  12.         return get_company_info(self.api_key, company_name)
  14.     def batch_query(self, company_names):
  15.         return batch_query(self.api_key, company_names)
  17. # 使用示例
  18. if __name__ == "__main__":
  19.     API_KEY = "your_api_key_here"
  20.     client = TianYanChaClient(API_KEY)
  22.     # 单个查询
  23.     company_info = client.query_company("阿里巴巴")
  24.     print("单个查询结果:", company_info)
  26.     # 批量查询
  27.     companies = ["腾讯", "百度", "字节跳动"]
  28.     batch_result = client.batch_query(companies)
  29.     print("\n批量查询结果:")
  30.     print(batch_result.to_markdown(index=False))

八、进阶建议

  1. 数据清洗：使用pandas的fillna()和apply()方法处理缺失值
  2. 可视化分析：结合pyecharts生成企业关系图谱
  3. 定时任务：使用APScheduler实现每日数据更新
  4. 异常监控：集成Sentry进行错误报警

通过系统化的API调用方案，开发者可以构建高效、稳定的企业信息查询系统。建议从基础版API开始，随着业务需求增长逐步升级到专业版，同时注意遵守天眼查的API使用条款，确保数据获取的合规性。

相关文章推荐

发表评论

最热文章

关于作者

  • 被阅读数
  • 被赞数
  • 被收藏数