logo

开发者热搜

如何用Python高效查询天眼查企业信息：技术实现与实战指南

作者：梅琳marlin2025.09.18 16:01浏览量：1

简介：本文详解如何通过Python自动化查询天眼查企业信息，涵盖API调用、网页解析、反爬策略及数据存储方案，提供可复用的代码示例与最佳实践。

一、技术选型与前期准备

1.1 查询方式对比

天眼查企业信息查询主要有三种技术路径：

  • 官方API接口：需申请企业级账号，支持结构化数据返回（推荐指数★★★★☆）
  • 网页爬取：通过解析HTML获取数据，需处理动态加载与反爬机制（推荐指数★★★☆☆）
  • 浏览器自动化：使用Selenium模拟人工操作，适合复杂交互场景（推荐指数★★☆☆☆）

关键决策点：

  • 官方API提供最高稳定性，但需承担商业授权费用
  • 爬虫方案零成本但需持续维护，适合个人开发者或小规模应用
  • 浏览器自动化仅作为备选方案，性能效率较低

1.2 环境配置清单

  1. # 基础环境要求
  2. Python 3.8+
  3. requests 2.28.1+  # HTTP请求库
  4. beautifulsoup4 4.11.1+  # HTML解析
  5. selenium 4.3.0+  # 浏览器自动化
  6. fake_useragent 1.1.1+  # 请求头伪装
  7. pymongo 4.2.0+  # MongoDB存储（可选）

二、官方API实现方案

2.1 认证流程详解

  1. 账号注册：访问天眼查开发者平台完成企业认证
  2. API密钥获取：在控制台创建应用获取appKey和appSecret
  3. 权限配置：申请企业信息查询权限（需提供使用场景说明）

2.2 核心代码实现

  1. import requests
  2. import hashlib
  3. import time
  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.tianyancha.com/services/v3/open/"
  11.     def generate_sign(self, params):
  12.         """生成请求签名"""
  13.         sorted_params = sorted(params.items(), key=lambda x: x[0])
  14.         param_str = "&".join([f"{k}={v}" for k, v in sorted_params])
  15.         sign_str = f"{param_str}{self.app_secret}"
  16.         return hashlib.md5(sign_str.encode()).hexdigest()
  18.     def query_company(self, company_name):
  19.         """企业基本信息查询"""
  20.         timestamp = str(int(time.time()))
  21.         params = {
  22.             "appKey": self.app_key,
  23.             "timestamp": timestamp,
  24.             "name": company_name,
  25.             "pageSize": 10
  26.         }
  27.         params["sign"] = self.generate_sign(params)
  29.         try:
  30.             response = requests.get(
  31.                 f"{self.base_url}company/searchBaseInfoList",
  32.                 params=params
  33.             )
  34.             response.raise_for_status()
  35.             return response.json()
  36.         except requests.exceptions.RequestException as e:
  37.             print(f"API请求失败: {e}")
  38.             return None
  40. # 使用示例
  41. api_client = TianYanChaAPI("your_app_key", "your_app_secret")
  42. result = api_client.query_company("阿里巴巴")
  43. print(result)

2.3 返回数据解析

典型响应结构包含：

  1. {
  2.     "code": 200,
  3.     "message": "success",
  4.     "result": {
  5.         "total": 1,
  6.         "items": [
  7.             {
  8.                 "id": 123456,
  9.                 "name": "阿里巴巴集团",
  10.                 "legalPersonName": "张勇",
  11.                 "registeredCapital": "100000万人民币",
  12.                 "establishTime": "1999-09-09"
  13.             }
  14.         ]
  15.     }
  16. }

三、网页爬取实现方案

3.1 反爬机制分析

天眼查主要反爬策略：

  • IP频率限制：单IP每分钟请求不超过5次
  • 行为检测：监测鼠标轨迹、点击间隔等特征
  • 参数加密：关键查询参数采用JS动态生成

3.2 绕过策略实现

  1. from fake_useragent import UserAgent
  2. import random
  4. class TianYanChaSpider:
  5.     def __init__(self):
  6.         self.ua = UserAgent()
  7.         self.session = requests.Session()
  8.         self.proxies = [
  9.             "http://proxy1.example.com:8080",
  10.             "http://proxy2.example.com:8080"
  11.         ]
  13.     def get_random_proxy(self):
  14.         """随机轮换代理IP"""
  15.         return {"http": random.choice(self.proxies)}
  17.     def get_company_info(self, company_name):
  18.         """获取企业详情页信息"""
  19.         headers = {
  20.             "User-Agent": self.ua.random,
  21.             "Referer": "https://www.tianyancha.com/"
  22.         }
  24.         # 第一步：获取搜索结果页
  25.         search_url = f"https://www.tianyancha.com/search?key={company_name}"
  26.         try:
  27.             response = self.session.get(
  28.                 search_url,
  29.                 headers=headers,
  30.                 proxies=self.get_random_proxy(),
  31.                 timeout=10
  32.             )
  33.             # 解析企业ID（示例简化）
  34.             company_id = self._parse_company_id(response.text)
  36.             # 第二步：访问企业详情页
  37.             detail_url = f"https://www.tianyancha.com/company/{company_id}"
  38.             detail_resp = self.session.get(
  39.                 detail_url,
  40.                 headers=headers,
  41.                 proxies=self.get_random_proxy()
  42.             )
  43.             return self._parse_detail_page(detail_resp.text)
  45.         except Exception as e:
  46.             print(f"请求失败: {e}")
  47.             return None
  49.     def _parse_company_id(self, html):
  50.         """从搜索结果解析企业ID（需根据实际HTML结构调整）"""
  51.         # 实际实现需使用BeautifulSoup或正则表达式
  52.         pass

3.3 动态参数处理

对于加密参数_token的获取方案：

  1. 使用Selenium加载页面获取完整DOM
  2. 执行JS代码提取加密参数：
    ```python
    from selenium import webdriver

def get_dynamic_token(company_name):
driver = webdriver.Chrome()
driver.get(“https://www.tianyancha.com/search“)

  1. # 输入搜索词并触发查询
  2. search_box = driver.find_element("name", "key")
  3. search_box.send_keys(company_name)
  4. search_box.submit()
  6. # 等待结果加载后提取token
  7. token = driver.execute_script("""
  8.     return window._token || document.querySelector('input[name="_token"]').value;
  9. """)
  10. driver.quit()
  11. return token
  2. # 四、数据存储与处理
  3. ## 4.1 结构化存储方案
  4. 推荐MongoDB存储方案：
  5. ```python
  6. from pymongo import MongoClient
  8. class DataStorage:
  9.     def __init__(self):
  10.         self.client = MongoClient("mongodb://localhost:27017/")
  11.         self.db = self.client["tianyancha"]
  12.         self.companies = self.db["companies"]
  14.     def save_company(self, company_data):
  15.         """存储企业信息"""
  16.         filter_cond = {"id": company_data["id"]}
  17.         update_op = {
  18.             "$set": {
  19.                 "name": company_data["name"],
  20.                 "legal_person": company_data["legalPersonName"],
  21.                 "update_time": datetime.now()
  22.             }
  23.         }
  24.         self.companies.update_one(filter_cond, update_op, upsert=True)

4.2 数据清洗建议

  1. 字段标准化：
    • 注册资本统一转换为数值（处理”万人民币”、”美元”等单位）
    • 日期字段统一为YYYY-MM-DD格式
  2. 异常值处理：
    • 过滤无效的法人名称（如”—“、”未知”等）
    • 校验联系电话格式

五、最佳实践与注意事项

5.1 法律合规要点

  1. 严格遵守《网络安全法》和《数据安全法》
  2. 查询频率控制在合理范围（建议QPS≤1）
  3. 不得将获取的数据用于商业竞争分析等敏感用途

5.2 性能优化建议

  1. 异步处理：使用aiohttp实现并发请求
  2. 缓存机制：对重复查询结果进行本地缓存
  3. 分布式架构：使用Scrapy-Redis实现分布式爬取

5.3 错误处理策略

  1. def robust_query(company_name, max_retries=3):
  2.     """带重试机制的查询函数"""
  3.     for attempt in range(max_retries):
  4.         try:
  5.             # 实际查询逻辑
  6.             result = api_client.query_company(company_name)
  7.             if result and result.get("code") == 200:
  8.                 return result
  9.         except Exception as e:
  10.             wait_time = 2 ** attempt  # 指数退避
  11.             time.sleep(wait_time)
  12.     raise RuntimeError("查询失败")

六、进阶应用场景

6.1 批量查询实现

  1. def batch_query(company_list):
  2.     """批量查询企业信息"""
  3.     results = []
  4.     with ThreadPoolExecutor(max_workers=5) as executor:
  5.         futures = [
  6.             executor.submit(api_client.query_company, name)
  7.             for name in company_list
  8.         ]
  9.         for future in as_completed(futures):
  10.             results.append(future.result())
  11.     return results

6.2 变更监控系统

  1. 定时任务：使用Celery设置每日凌晨执行
  2. 差异检测：对比当日与历史数据的关键字段
  3. 告警机制：通过邮件/短信通知变更

6.3 数据可视化应用

推荐使用Pyecharts生成企业关系图谱：

  1. from pyecharts import options as opts
  2. from pyecharts.charts import Graph
  4. def generate_relation_graph(company_data):
  5.     nodes = [
  6.         {"name": "阿里巴巴", "symbolSize": 50},
  7.         {"name": "蚂蚁集团", "symbolSize": 40}
  8.     ]
  9.     links = [{"source": "阿里巴巴", "target": "蚂蚁集团"}]
  11.     graph = (
  12.         Graph()
  13.         .add("", nodes, links, repulsion=1000)
  14.         .set_global_opts(title_opts=opts.TitleOpts(title="企业关系图谱"))
  15.     )
  16.     return graph.render("relation_graph.html")

七、总结与展望

本方案提供了从基础查询到高级应用的完整实现路径，开发者可根据实际需求选择API或爬虫方案。未来可拓展方向包括：

  1. 结合NLP技术实现企业风险预警
  2. 构建企业知识图谱支持智能查询
  3. 开发微信小程序实现移动端查询

建议开发者持续关注天眼查的反爬策略更新，保持代码库的维护频率。对于生产环境应用，推荐优先采用官方API方案以确保服务稳定性。

相关文章推荐

发表评论

最热文章

关于作者

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