Python高效查询天眼查企业信息：方法、实践与优化策略

一、天眼查数据价值与查询场景

天眼查作为国内领先的企业信息查询平台，覆盖超过3亿家市场主体数据，包含工商信息、司法风险、经营状况等200+维度数据。开发者通过Python自动化查询这些数据，可应用于企业尽调、风险监控、供应链管理等场景。例如金融机构可通过实时获取企业司法涉诉信息评估信贷风险，投资机构可分析目标企业股权结构变化辅助决策。

二、Python查询技术实现路径

1. 官方API接口调用（推荐方案）

天眼查提供标准化的RESTful API接口，开发者需申请API Key并遵守《天眼查数据服务协议》。典型查询流程如下：

import requests
def query_company_info(api_key, company_name):
    url = "https://api.tianyancha.com/services/open/ic/searchV2"
    params = {
        "key": api_key,
        "name": company_name,
        "pageSize": 10
    }
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    response = requests.get(url, params=params, headers=headers)
    if response.status_code == 200:
        return response.json()
    else:
        return {"error": "API调用失败"}
# 使用示例
api_key = "your_api_key_here"
result = query_company_info(api_key, "阿里巴巴")
print(result["data"][0]["legalPersonName"])  # 输出法定代表人

优势：数据实时性强，结构化程度高，支持批量查询。注意：需控制调用频率（建议QPS≤5），避免触发IP限流。

2. 网页爬取方案（备选方案）

当API调用受限时，可采用Selenium+BeautifulSoup组合实现网页数据采集。以查询企业基本信息为例：

from selenium import webdriver
from bs4 import BeautifulSoup
import time
def scrape_company_page(company_name):
    driver = webdriver.Chrome()
    driver.get("https://www.tianyancha.com/search?key={}".format(company_name))
    time.sleep(3)  # 等待页面加载
    # 解析搜索结果页
    soup = BeautifulSoup(driver.page_source, 'html.parser')
    company_link = soup.find("a", class_="name")["href"]
    # 进入企业详情页
    driver.get("https://www.tianyancha.com" + company_link)
    time.sleep(5)
    # 提取关键信息
    detail_soup = BeautifulSoup(driver.page_source, 'html.parser')
    legal_person = detail_soup.find("div", class_="legalPersonName").text.strip()
    registered_capital = detail_soup.find("div", class_="registeredCapital").text.strip()
    driver.quit()
    return {
        "legal_person": legal_person,
        "registered_capital": registered_capital
    }
# 使用示例
info = scrape_company_page("腾讯")
print(info)

优化建议：

• 使用webdriver.ChromeOptions()配置无头模式提升效率
• 结合random.uniform(1,3)实现动态等待
• 通过代理IP池应对反爬机制

三、数据解析与结构化处理

1. JSON数据解析

API返回的JSON数据需进行深度解析：

import json
def parse_api_response(json_str):
    data = json.loads(json_str)
    company_list = data.get("data", [])
    for company in company_list:
        yield {
            "name": company.get("name"),
            "status": company.get("status"),
            "legal_person": company.get("legalPersonName"),
            "registered_address": company.get("registeredAddress")
        }
# 使用示例
api_response = '{"data":[{"name":"字节跳动","status":"存续","legalPersonName":"张一鸣"}]}'
for item in parse_api_response(api_response):
    print(item)

2. 网页数据清洗

爬取的HTML数据需进行标准化处理：

import re
def clean_text(raw_text):
    # 去除特殊字符
    text = re.sub(r'[\t\n\r]', '', raw_text)
    # 去除多余空格
    text = ' '.join(text.split())
    return text.strip()
# 示例应用
dirty_text = "  法定代表人：\n张三  \t"
clean_text = clean_text(dirty_text)  # 输出"法定代表人：张三"

四、性能优化与异常处理

1. 并发查询策略

使用concurrent.futures实现多线程查询：

from concurrent.futures import ThreadPoolExecutor
def batch_query(api_key, company_names):
    results = []
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(query_company_info, api_key, name) for name in company_names]
        for future in futures:
            try:
                results.append(future.result())
            except Exception as e:
                results.append({"error": str(e)})
    return results
# 使用示例
companies = ["华为", "小米", "美团"]
batch_results = batch_query(api_key, companies)

2. 异常处理机制

import requests
from requests.exceptions import RequestException
def safe_query(api_key, company_name):
    try:
        response = requests.get(
            "https://api.tianyancha.com/services/open/ic/searchV2",
            params={"key": api_key, "name": company_name},
            timeout=10
        )
        response.raise_for_status()
        return response.json()
    except RequestException as e:
        return {"error": f"请求失败: {str(e)}"}
    except json.JSONDecodeError:
        return {"error": "数据解析异常"}

五、合规使用与最佳实践

1. 数据使用规范：

   • 严格遵守《网络安全法》《数据安全法》
   • 仅用于合法商业目的，不得用于非法集资监控等场景
   • 对获取的数据进行脱敏处理

2. 反爬应对策略：

   • 设置合理的请求间隔（建议≥2秒/次）
   • 使用User-Agent轮换
   • 结合Cookie管理机制

3. 数据存储建议：

   • 对频繁查询的数据建立本地缓存
   • 使用MongoDB等文档数据库存储非结构化数据
   • 定期更新数据保证时效性

六、进阶应用场景

1. 企业关系图谱构建：
   通过递归查询股东信息，使用NetworkX构建股权结构图：
   ```python
   import networkx as nx
   import matplotlib.pyplot as plt

def build_relation_graph(company_data):
G = nx.DiGraph()
for shareholder in company_data[“shareholders”]:
G.add_edge(shareholder[“name”], company_data[“name”], ratio=shareholder[“ratio”])
nx.draw(G, with_labels=True)
plt.show()

2. **风险预警系统**：
实时监控企业司法涉诉变化：
```python
def monitor_legal_risk(company_id, api_key):
    while True:
        response = query_company_detail(api_key, company_id)
        new_cases = response["legalCases"]
        # 与本地存储的历史数据对比
        # 触发预警逻辑...
        time.sleep(86400)  # 每日检查一次

七、技术选型建议

场景	推荐方案	替代方案
实时数据查询	官方API	网页爬取
大规模历史数据采集	网页爬取+分布式队列	API批量导出
移动端应用集成	官方API+轻量级HTTP客户端	本地数据库查询
高并发场景	API+消息队列	多线程爬取

通过合理选择技术方案，开发者可在合规前提下高效获取天眼查企业数据。建议优先使用官方API，当遇到调用限制时，再结合网页爬取技术作为补充。在实际项目中，应建立完善的数据更新机制和异常处理流程，确保系统的稳定性和数据的准确性。