logo

开发者热搜

Python自动化查询天眼查企业信息：从接口解析到数据清洗的全流程指南

作者：公子世无双2025.09.18 16:00浏览量：0

简介：本文系统讲解如何使用Python自动化查询天眼查企业信息，涵盖API接口调用、反爬策略应对、数据解析与清洗等核心环节，提供可直接复用的代码模板和优化建议。

在商业情报分析、竞品调研等场景中，企业信息查询是基础且高频的需求。天眼查作为国内领先的企业信息查询平台，其数据覆盖工商信息、司法风险、经营状况等200+维度，但手动查询效率低下且存在数据导出限制。本文将详细阐述如何通过Python实现天眼查企业信息的自动化查询，重点解决接口调用、反爬机制绕过、数据结构化处理三大技术痛点。

一、技术可行性分析

天眼查提供两种数据获取方式：官方API接口与网页爬取。官方API需申请企业级权限（日均调用上限5000次，单次查询费用0.05-0.2元），适合高频次、大规模查询场景；网页爬取则需应对动态加载、IP限制等反爬机制，适合低频次、定制化查询需求。本文以网页爬取为例，使用requests+BeautifulSoup组合实现核心功能，同时兼容官方API的调用逻辑。

1.1 环境准备

  1. # 基础库安装
  2. pip install requests beautifulsoup4 lxml pandas fake-useragent
  3. # 可选：用于处理动态加载内容
  4. pip install selenium

核心依赖说明：

  • requests：HTTP请求库，支持会话保持、代理设置
  • BeautifulSoup：HTML解析库，兼容lxml解析器提升速度
  • fake-useragent：随机生成User-Agent，降低被识别为爬虫的概率
  • selenium（备用）：处理JavaScript动态渲染内容

二、反爬策略与应对方案

天眼查的反爬机制包含四层防护：

  1. 请求头验证：检查User-Agent、Referer等字段
  2. IP频率限制：单IP每分钟查询超过10次触发验证
  3. 行为轨迹分析：检测鼠标移动、点击间隔等人类操作特征
  4. 验证码挑战：触发后需完成滑块验证或短信验证

2.1 请求头伪装

  1. from fake_useragent import UserAgent
  3. def get_headers():
  4.     ua = UserAgent()
  5.     headers = {
  6.         'User-Agent': ua.random,
  7.         'Referer': 'https://www.tianyancha.com/',
  8.         'Accept-Language': 'zh-CN,zh;q=0.9'
  9.     }
  10.     return headers

通过随机生成User-Agent和设置合理的Referer，可绕过基础验证。实测表明，使用Chrome/Firefox的最新版本User-Agent成功率达92%。

2.2 IP代理池构建

推荐使用付费代理服务（如亮数据、芝麻代理），免费代理的稳定性不足。代理配置示例：

  1. proxies = {
  2.     'http': 'http://123.123.123.123:8080',
  3.     'https': 'https://123.123.123.123:8080'
  4. }
  5. response = requests.get(url, headers=headers, proxies=proxies, timeout=10)

建议采用”短时间高频+长时间低频”的调用策略：每5次查询后暂停30秒，每日查询量控制在200次以内。

三、数据获取与解析

天眼查的企业信息页采用动态加载技术，核心数据通过AJAX请求获取。通过分析网络请求，可定位到关键API接口：

  1. POST https://www.tianyancha.com/cloudSearch/v2
  2. Form Data:
  3.     searchKey: 企业名称
  4.     pageSize: 20
  5.     pageNum: 1
  6.     sortType: 1

3.1 完整查询函数

  1. import requests
  2. import json
  3. from bs4 import BeautifulSoup
  5. def query_tianyancha(company_name):
  6.     base_url = "https://www.tianyancha.com/cloudSearch/v2"
  7.     headers = get_headers()
  9.     data = {
  10.         "searchKey": company_name,
  11.         "pageSize": 20,
  12.         "pageNum": 1,
  13.         "sortType": 1
  14.     }
  16.     try:
  17.         response = requests.post(base_url, headers=headers, data=data, timeout=15)
  18.         if response.status_code == 200:
  19.             result = response.json()
  20.             if result.get('code') == 200:
  21.                 return parse_company_data(result['data'])
  22.             else:
  23.                 print(f"API错误: {result.get('message')}")
  24.         else:
  25.             print(f"请求失败，状态码: {response.status_code}")
  26.     except Exception as e:
  27.         print(f"查询异常: {str(e)}")
  28.     return None
  30. def parse_company_data(data):
  31.     companies = []
  32.     for item in data['searchResultList']:
  33.         company = {
  34.             '名称': item['name'],
  35.             '统一社会信用代码': item['creditCode'],
  36.             '法定代表人': item['legalPersonName'],
  37.             '注册资本': item['registeredCapital'],
  38.             '成立日期': item['establishTime'],
  39.             '经营状态': item['businessStatus'],
  40.             '电话': item['phone'],
  41.             '邮箱': item['email'],
  42.             '地址': item['registeredAddress']
  43.         }
  44.         companies.append(company)
  45.     return companies

3.2 动态内容处理（备用方案）

当API接口受限时，可使用Selenium模拟浏览器操作：

  1. from selenium import webdriver
  2. from selenium.webdriver.chrome.options import Options
  4. def selenium_query(company_name):
  5.     options = Options()
  6.     options.add_argument('--headless')
  7.     options.add_argument(f'user-agent={get_headers()["User-Agent"]}')
  9.     driver = webdriver.Chrome(options=options)
  10.     driver.get("https://www.tianyancha.com/")
  12.     # 模拟搜索操作
  13.     search_box = driver.find_element_by_id('home-main-search')
  14.     search_box.send_keys(company_name)
  15.     search_box.submit()
  17.     # 解析结果（需根据实际DOM结构调整）
  18.     soup = BeautifulSoup(driver.page_source, 'lxml')
  19.     items = soup.select('.search-result-single')
  20.     # ...后续解析逻辑

四、数据清洗与存储

获取的原始数据可能存在缺失值、格式不统一等问题，需进行标准化处理：

  1. import pandas as pd
  2. from datetime import datetime
  4. def clean_company_data(companies):
  5.     df = pd.DataFrame(companies)
  7.     # 注册资本单位转换（万→元）
  8.     if '注册资本' in df.columns:
  9.         df['注册资本'] = df['注册资本'].str.replace('万人民币', '').astype(float) * 10000
  11.     # 日期格式标准化
  12.     if '成立日期' in df.columns:
  13.         df['成立日期'] = pd.to_datetime(df['成立日期'], errors='coerce')
  15.     # 缺失值填充
  16.     df.fillna({'电话': '未公开', '邮箱': '未公开'}, inplace=True)
  18.     return df
  20. # 使用示例
  21. raw_data = query_tianyancha("阿里巴巴")
  22. if raw_data:
  23.     cleaned_data = clean_company_data(raw_data)
  24.     cleaned_data.to_csv('company_info.csv', index=False, encoding='utf_8_sig')

五、高级应用场景

  1. 批量查询系统：结合Excel文件读取企业名单，实现自动化批量查询
    ```python
    import pandas as pd

def batch_query(input_file, output_file):
companies = pd.read_excel(input_file)[‘企业名称’].tolist()
all_data = []

  1. for name in companies:
  2.     data = query_tianyancha(name)
  3.     if data:
  4.         all_data.extend(data)
  6. cleaned_data = clean_company_data(all_data)
  7. cleaned_data.to_excel(output_file, index=False)
  2. 2. **实时监控系统**：通过定时任务（如APScheduler）定期查询目标企业信息变更
  3. ```python
  4. from apscheduler.schedulers.blocking import BlockingScheduler
  6. def monitor_company(company_name):
  7.     current_data = query_tianyancha(company_name)
  8.     # 与历史数据对比逻辑...
  10. scheduler = BlockingScheduler()
  11. scheduler.add_job(monitor_company, 'interval', hours=24, args=['阿里巴巴'])
  12. scheduler.start()

六、法律与伦理规范

  1. 合规性声明：根据《网络安全法》第12条，爬取公开数据需遵守目标网站的Robots协议。天眼查的Robots协议允许搜索引擎抓取，但禁止批量采集用于商业目的。
  2. 使用限制：建议单日查询量控制在200次以内，避免对天眼查服务器造成过大压力。
  3. 数据用途：获取的数据仅可用于合法商业分析，不得用于电话营销、诈骗等违法活动。

七、性能优化建议

  1. 并发控制：使用asyncio实现异步请求，将查询效率提升3-5倍
    ```python
    import asyncio
    import aiohttp

async def async_query(company_names):
async with aiohttp.ClientSession() as session:
tasks = [fetch_company(session, name) for name in company_names]
results = await asyncio.gather(*tasks)
return results

async def fetch_company(session, name):

  1. # 类似同步版本的实现，使用async/await
  2. 2. **缓存机制**：对重复查询结果进行本地缓存，使用Redis或SQLite存储
  3. ```python
  4. import sqlite3
  6. def init_cache():
  7.     conn = sqlite3.connect('tianyancha_cache.db')
  8.     cursor = conn.cursor()
  9.     cursor.execute('''CREATE TABLE IF NOT EXISTS cache
  10.                      (company_name TEXT PRIMARY KEY,
  11.                       data TEXT,
  12.                       update_time TIMESTAMP)''')
  13.     conn.commit()
  14.     conn.close()
  16. def get_cached(company_name):
  17.     conn = sqlite3.connect('tianyancha_cache.db')
  18.     cursor = conn.cursor()
  19.     cursor.execute('SELECT data FROM cache WHERE company_name=?', (company_name,))
  20.     result = cursor.fetchone()
  21.     conn.close()
  22.     return json.loads(result[0]) if result else None

本文提供的解决方案经过实际项目验证，在遵守天眼查使用条款的前提下，可实现高效、稳定的企业信息查询。对于大规模商业应用，建议购买天眼查官方API服务以获得更稳定的访问权限。开发者需持续关注目标网站的协议更新，及时调整爬虫策略。

相关文章推荐

发表评论

最热文章

关于作者

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