logo

开发者热搜

Python实战：高效爬取天眼查企业数据的全流程指南

作者：问答酱2025.09.18 16:01浏览量：3

简介：本文详解如何使用Python爬取天眼查企业数据，涵盖反爬机制破解、数据解析与存储等核心环节，提供可复用的完整代码方案。

一、天眼查数据价值与爬取前提

天眼查作为国内领先的企业信息查询平台，其公开的企业工商信息、司法风险、经营状况等数据对市场分析、竞品调研、投资决策具有重要价值。但直接爬取存在三大挑战：动态加密参数、高频请求拦截、数据结构复杂化。开发者需明确：仅限爬取公开非敏感数据，遵守《网络安全法》与平台服务条款，建议控制单日请求量在200次以内。

二、技术栈选型与工具准备

  1. 核心库组合

    • requests-html：支持动态渲染的HTTP客户端（优于纯requests）
    • parsel：基于XPath/CSS选择器的高效解析器
    • selenium-wire：应对加密参数的中间件捕获工具
    • pymongo：MongoDB存储方案（适合结构化数据）

  2. 代理与IP池
    使用scrapy-rotating-proxies中间件，配置付费动态住宅IP（如BrightData），避免被封禁。示例代理配置：

    1. ROTATING_PROXY_LIST = [
    2.     'http://user:pass@ip1:port',
    3.     'http://user:pass@ip2:port'
    4. ]

三、反爬机制深度破解

1. 参数加密逆向

天眼查的请求参数包含动态生成的token与_sign，通过以下步骤破解：

  • 步骤1：使用selenium-wire捕获真实请求
    1. from seleniumwire import webdriver
    2. driver = webdriver.Chrome()
    3. driver.get('https://www.tianyancha.com/search?key=阿里巴巴')
    4. for request in driver.requests:
    5.     if request.url.startswith('https://api.tianyancha.com'):
    6.         print(request.body)  # 获取加密参数样本
  • 步骤2：分析参数生成逻辑（通常为JS加密）
    通过Chrome开发者工具的Sources面板定位加密JS文件，使用pyexecjs调用：
    1. import execjs
    2. with open('encrypt.js', 'r') as f:
    3.     js_code = f.read()
    4. ctx = execjs.compile(js_code)
    5. encrypted_param = ctx.call('generateSign', '原始参数')

2. 行为模拟策略

  • 请求头伪装：完善User-Agent、Referer、X-Requested-With等字段
    1. headers = {
    2.     'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    3.     'Referer': 'https://www.tianyancha.com/',
    4.     'X-Requested-With': 'XMLHttpRequest'
    5. }
  • 请求间隔控制：采用指数退避算法
    1. import time
    2. import random
    3. def request_with_delay(url):
    4.     delay = 1 + random.uniform(0, 2) * (2 ** retry_count)
    5.     time.sleep(delay)
    6.     # 执行请求

四、完整爬取流程实现

1. 企业列表页抓取

  1. from requests_html import HTMLSession
  2. session = HTMLSession()
  4. def get_company_list(keyword, page=1):
  5.     url = f'https://www.tianyancha.com/search?key={keyword}&pn={page}'
  6.     r = session.get(url, headers=headers)
  7.     companies = []
  8.     for item in r.html.find('.search-result-single'):
  9.         name = item.find('.name', first=True).text
  10.         link = item.find('.name a', first=True).attrs['href']
  11.         companies.append({'name': name, 'link': link})
  12.     return companies

2. 企业详情页解析

  1. import parsel
  3. def parse_company_detail(url):
  4.     r = session.get(url, headers=headers)
  5.     selector = parsel.Selector(r.text)
  7.     # 基础信息
  8.     base_info = {
  9.         'phone': selector.css('.phone::text').get(),
  10.         'email': selector.css('.email::text').get(),
  11.         'address': selector.css('.address::text').get()
  12.     }
  14.     # 股东信息（处理嵌套表格）
  15.     shareholders = []
  16.     for tr in selector.css('.shareholder-table tr')[1:]:
  17.         name = tr.css('td:nth-child(1)::text').get()
  18.         ratio = tr.css('td:nth-child(2)::text').get()
  19.         shareholders.append({'name': name, 'ratio': ratio})
  21.     return {'base_info': base_info, 'shareholders': shareholders}

3. 数据存储方案

  1. from pymongo import MongoClient
  3. client = MongoClient('mongodb://localhost:27017/')
  4. db = client['tianyancha_db']
  6. def save_to_mongo(data):
  7.     collection = db['companies']
  8.     # 处理重复数据
  9.     if not collection.find_one({'name': data['name']}):
  10.         collection.insert_one(data)

五、进阶优化技巧

  1. 分布式爬取
    使用Scrapy-Redis实现多节点协作，配置如下：

    1. # scrapy_redis.py
    2. REDIS_URL = 'redis://localhost:6379/0'
    3. SCHEDULER = "scrapy_redis.scheduler.Scheduler"
    4. DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

  2. 增量更新机制
    通过企业最后更新时间字段实现增量抓取：

    1. def get_updated_companies(last_check_time):
    2.     query = {'update_time': {'$gt': last_check_time}}
    3.     return list(db.companies.find(query))

  3. 异常处理增强 

    1. from requests.exceptions import RequestException
    2. import logging
    4. logging.basicConfig(filename='crawler.log', level=logging.ERROR)
    6. def safe_request(url):
    7.     try:
    8.         return session.get(url, headers=headers, timeout=10)
    9.     except RequestException as e:
    10.         logging.error(f'Request failed: {e}')
    11.         return None

六、法律与伦理规范

  1. 合规要点

    • 严格遵守robots.txt协议（检查https://www.tianyancha.com/robots.txt）
    • 禁止爬取用户隐私数据（如身份证号、银行账户）
    • 商业用途需获得平台授权

  2. 数据使用建议

    • 对爬取数据进行脱敏处理
    • 建立内部数据使用审批流程
    • 定期清理过期数据（建议保留不超过1年）

七、完整案例代码

  1. # tianyancha_crawler.py
  2. import time
  3. from requests_html import HTMLSession
  4. from pymongo import MongoClient
  5. import logging
  7. logging.basicConfig(filename='tianyancha.log', level=logging.INFO)
  9. class TianYanChaCrawler:
  10.     def __init__(self):
  11.         self.session = HTMLSession()
  12.         self.headers = {
  13.             'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
  14.             'Referer': 'https://www.tianyancha.com/'
  15.         }
  16.         self.client = MongoClient('mongodb://localhost:27017/')
  17.         self.db = self.client['tianyancha_db']
  19.     def search_companies(self, keyword, max_pages=5):
  20.         all_companies = []
  21.         for page in range(1, max_pages + 1):
  22.             url = f'https://www.tianyancha.com/search?key={keyword}&pn={page}'
  23.             r = self.safe_request(url)
  24.             if not r:
  25.                 continue
  27.             companies = []
  28.             for item in r.html.find('.search-result-single'):
  29.                 name = item.find('.name', first=True).text
  30.                 link = item.find('.name a', first=True).attrs['href']
  31.                 companies.append({'name': name, 'link': link})
  33.             all_companies.extend(companies)
  34.             time.sleep(2)  # 礼貌性延迟
  36.         return all_companies
  38.     def parse_detail(self, url):
  39.         r = self.safe_request(url)
  40.         if not r:
  41.             return None
  43.         # 解析逻辑（同前文示例）
  44.         # ...
  46.         return company_data
  48.     def safe_request(self, url):
  49.         try:
  50.             return self.session.get(url, headers=self.headers, timeout=10)
  51.         except Exception as e:
  52.             logging.error(f'Error fetching {url}: {str(e)}')
  53.             return None
  55. if __name__ == '__main__':
  56.     crawler = TianYanChaCrawler()
  57.     companies = crawler.search_companies('人工智能', max_pages=3)
  59.     for company in companies:
  60.         detail = crawler.parse_detail(company['link'])
  61.         if detail:
  62.             crawler.db.companies.insert_one(detail)

八、常见问题解决方案

  1. 验证码触发

    • 现象：返回403且HTML包含验证码图片
    • 解决方案：接入第三方打码平台（如超级鹰），或降低请求频率

  2. 数据缺失

    • 检查是否登录后才能查看的数据（需模拟登录）
    • 确认CSS选择器是否随页面更新而改变

  3. 内存溢出

    • 对大规模数据采用流式处理
    • 使用生成器代替列表存储：
      1. def generate_companies(keyword):
      2.     for page in range(1, 100):
      3.         # 抓取逻辑
      4.         yield company_data

通过本文提供的完整方案，开发者可在遵守法律法规的前提下，高效构建天眼查数据爬取系统。实际部署时建议结合具体业务需求调整爬取频率与数据范围，并定期监控爬虫运行状态。

相关文章推荐

发表评论

最热文章

关于作者

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