logo

开发者热搜

Python高效爬取企查查工商信息指南

作者：渣渣辉2025.09.18 16:01浏览量：2

简介：本文详细解析如何使用Python爬取企查查工商信息，涵盖反爬机制应对、数据解析与存储，提供完整代码示例及合规建议。

一、企查查数据价值与爬取背景

企查查作为国内领先的商业信息查询平台，聚合了超过2亿家企业的工商注册、股东信息、司法风险等核心数据。这些结构化数据对于市场调研、竞品分析、风险控制等场景具有重要价值。例如金融机构可通过企业股权结构判断关联交易风险，咨询公司可基于行业数据分布制定市场策略。

传统获取方式存在三大痛点：手动查询效率低下（单企业查询需15-30秒）、批量数据获取成本高昂（API接口年费数万元）、数据维度受限（免费版仅展示基础信息）。Python爬虫技术可实现分钟级获取千家企业数据，成本降低90%以上，且支持自定义数据维度。

二、爬取技术架构设计

1. 请求头伪装技术

企查查反爬系统通过User-Agent检测、Referer校验、Cookie追踪三重机制识别爬虫。需构建动态请求头：

  1. import random
  2. from fake_useragent import UserAgent
  4. def generate_headers():
  5.     ua = UserAgent()
  6.     headers = {
  7.         'User-Agent': ua.random,
  8.         'Referer': 'https://www.qcc.com/',
  9.         'Accept-Language': 'zh-CN,zh;q=0.9',
  10.         'Cookie': '你的Cookie值'  # 需定期更新
  11.     }
  12.     return headers

2. 动态参数解析

搜索接口采用多重加密参数，包括：

  • searchKey：Base64编码的搜索词
  • token：动态生成的JS变量
  • timestamp：毫秒级时间戳

通过逆向分析发现，核心参数可通过以下方式获取：

  1. import base64
  2. import time
  3. import execjs
  5. def get_search_params(keyword):
  6.     # 模拟JS加密过程
  7.     js_code = """
  8.     function encrypt(key) {
  9.         var arr = key.split('');
  10.         for(var i=0; i<arr.length; i++) {
  11.             arr[i] = String.fromCharCode(arr[i].charCodeAt(0)^0x66);
  12.         }
  13.         return btoa(arr.join(''));
  14.     }
  15.     """
  16.     ctx = execjs.compile(js_code)
  17.     encrypted_key = ctx.call('encrypt', keyword)
  19.     params = {
  20.         'searchKey': base64.b64encode(encrypted_key.encode()).decode(),
  21.         'token': '动态获取的token值',
  22.         't': int(time.time() * 1000)
  23.     }
  24.     return params

3. 分页处理策略

企查查采用滚动加载机制，每页显示20条数据。通过分析XHR请求，发现分页参数pageNum与pageSize组合控制数据加载。实现自动翻页的代码示例：

  1. import requests
  2. from urllib.parse import urlencode
  4. def fetch_company_list(keyword, max_pages=5):
  5.     base_url = 'https://www.qcc.com/webSearch'
  6.     companies = []
  8.     for page in range(1, max_pages+1):
  9.         params = {
  10.             'key': keyword,
  11.             'pageNum': page,
  12.             'pageSize': 20,
  13.             'sortType': 0
  14.         }
  16.         response = requests.get(
  17.             f'{base_url}?{urlencode(params)}',
  18.             headers=generate_headers()
  19.         )
  20.         data = response.json()
  22.         if not data.get('dataList'):
  23.             break
  25.         companies.extend(data['dataList'])
  26.         time.sleep(random.uniform(1, 3))  # 防封禁
  28.     return companies

三、数据解析与存储

1. JSON结构解析

返回数据采用嵌套JSON结构，核心字段包括：

  • companyName：企业名称
  • legalPersonName：法人代表
  • regCapital：注册资本（万元）
  • estabTime：成立日期
  • phone：联系电话（需二次验证）

解析示例：

  1. def parse_company_info(raw_data):
  2.     parsed = {
  3.         '企业名称': raw_data.get('companyName'),
  4.         '统一社会信用代码': raw_data.get('creditCode'),
  5.         '注册资本': raw_data.get('regCapital'),
  6.         '成立日期': raw_data.get('estabTime'),
  7.         '经营范围': raw_data.get('businessScope'),
  8.         '股东信息': []
  9.     }
  11.     # 解析股东信息
  12.     for shareholder in raw_data.get('shareholderList', []):
  13.         parsed['股东信息'].append({
  14.             '股东名称': shareholder.get('name'),
  15.             '认缴出资': shareholder.get('subConAm'),
  16.             '出资比例': shareholder.get('fundedRatio')
  17.         })
  19.     return parsed

2. 数据存储方案

根据使用场景选择存储方式：

  • CSV存储：适合快速分析
    ```python
    import csv

def save_to_csv(data_list, filename=’qcc_data.csv’):
with open(filename, ‘w’, newline=’’, encoding=’utf-8-sig’) as f:
writer = csv.DictWriter(f, fieldnames=data_list[0].keys())
writer.writeheader()
writer.writerows(data_list)

  2. - **MySQL存储**：适合长期维护
  3. ```python
  4. import pymysql
  6. def save_to_mysql(data):
  7.     conn = pymysql.connect(
  8.         host='localhost',
  9.         user='root',
  10.         password='your_password',
  11.         database='company_db',
  12.         charset='utf8mb4'
  13.     )
  15.     try:
  16.         with conn.cursor() as cursor:
  17.             sql = """
  18.             INSERT INTO company_info 
  19.             (name, credit_code, reg_capital, estab_date, scope)
  20.             VALUES (%s, %s, %s, %s, %s)
  21.             """
  22.             cursor.execute(sql, (
  23.                 data['企业名称'],
  24.                 data['统一社会信用代码'],
  25.                 data['注册资本'],
  26.                 data['成立日期'],
  27.                 data['经营范围']
  28.             ))
  29.         conn.commit()
  30.     finally:
  31.         conn.close()

四、反爬机制应对策略

1. IP代理池建设

推荐使用付费代理服务（如亮数据、芝麻代理），配合自动切换：

  1. import requests
  2. from proxy_pool import ProxyPool  # 假设的代理池类
  4. class AntiCrawler:
  5.     def __init__(self):
  6.         self.proxy_pool = ProxyPool()
  8.     def get_proxied_session(self):
  9.         proxy = self.proxy_pool.get_proxy()
  10.         proxies = {
  11.             'http': f'http://{proxy}',
  12.             'https': f'https://{proxy}'
  13.         }
  14.         session = requests.Session()
  15.         session.proxies.update(proxies)
  16.         return session

2. 行为模拟技术

通过Selenium模拟真实用户操作：

  1. from selenium import webdriver
  2. from selenium.webdriver.common.keys import Keys
  4. def selenium_search(keyword):
  5.     options = webdriver.ChromeOptions()
  6.     options.add_argument('--disable-blink-features=AutomationControlled')
  7.     driver = webdriver.Chrome(options=options)
  9.     try:
  10.         driver.get('https://www.qcc.com/')
  11.         search_box = driver.find_element_by_id('searchKey')
  12.         search_box.send_keys(keyword)
  13.         search_box.send_keys(Keys.RETURN)
  15.         # 等待结果加载
  16.         time.sleep(3)
  18.         # 提取数据（需根据实际DOM结构调整）
  19.         companies = []
  20.         items = driver.find_elements_by_class_name('search-result-item')
  21.         for item in items:
  22.             companies.append({
  23.                 'name': item.find_element_by_class_name('company-name').text,
  24.                 'legal_person': item.find_element_by_class_name('legal-person').text
  25.             })
  27.         return companies
  28.     finally:
  29.         driver.quit()

五、合规使用建议

  1. 数据使用边界：仅用于个人研究或内部分析，不得用于商业竞争或非法用途
  2. 频率控制：单IP每小时请求不超过100次，每日不超过500次
  3. 数据脱敏：处理包含个人信息的字段（如法人身份证号）
  4. 备份机制：定期备份数据，防止账号被封禁导致数据丢失

六、完整项目示例

  1. import requests
  2. import json
  3. import time
  4. import random
  5. from datetime import datetime
  7. class QCCCrawler:
  8.     def __init__(self):
  9.         self.base_url = 'https://www.qcc.com/webSearch'
  10.         self.session = requests.Session()
  11.         self.headers = {
  12.             'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
  13.             'Referer': 'https://www.qcc.com/'
  14.         }
  16.     def search_company(self, keyword, max_pages=3):
  17.         all_companies = []
  19.         for page in range(1, max_pages+1):
  20.             params = {
  21.                 'key': keyword,
  22.                 'pageNum': page,
  23.                 'pageSize': 20
  24.             }
  26.             try:
  27.                 response = self.session.get(
  28.                     f'{self.base_url}?{urlencode(params)}',
  29.                     headers=self.headers,
  30.                     timeout=10
  31.                 )
  32.                 data = response.json()
  34.                 if data.get('code') != 0:
  35.                     print(f"请求失败: {data.get('message')}")
  36.                     break
  38.                 companies = data.get('dataList', [])
  39.                 if not companies:
  40.                     break
  42.                 all_companies.extend(companies)
  43.                 print(f"已获取第{page}页数据，共{len(companies)}条")
  45.                 # 随机延迟
  46.                 time.sleep(random.uniform(1.5, 3.5))
  48.             except Exception as e:
  49.                 print(f"请求异常: {str(e)}")
  50.                 continue
  52.         return all_companies
  54.     def parse_company(self, raw_data):
  55.         return {
  56.             '企业名称': raw_data.get('companyName'),
  57.             '统一社会信用代码': raw_data.get('creditCode'),
  58.             '法定代表人': raw_data.get('legalPersonName'),
  59.             '注册资本': raw_data.get('regCapital'),
  60.             '成立日期': raw_data.get('estabTime'),
  61.             '登记状态': raw_data.get('regStatus'),
  62.             '更新时间': datetime.now().strftime('%Y-%m-%d')
  63.         }
  65.     def save_to_json(self, data_list, filename):
  66.         with open(filename, 'w', encoding='utf-8') as f:
  67.             json.dump(data_list, f, ensure_ascii=False, indent=2)
  69. # 使用示例
  70. if __name__ == '__main__':
  71.     crawler = QCCCrawler()
  72.     companies = crawler.search_company('人工智能', max_pages=2)
  74.     parsed_data = [crawler.parse_company(c) for c in companies]
  75.     crawler.save_to_json(parsed_data, 'qcc_companies.json')
  77.     print(f"共获取{len(parsed_data)}家企业信息")

七、进阶优化方向

  1. 分布式爬取：使用Scrapy-Redis实现多节点协作
  2. 增量更新：通过企业最后更新时间实现增量爬取
  3. 数据清洗：使用Pandas处理缺失值和异常数据
  4. 可视化展示：结合Pyecharts生成行业分布图表

通过系统化的技术实现和合规操作，Python爬虫可高效获取企查查的工商信息，为商业决策提供数据支撑。开发者需在技术实现与法律合规间保持平衡，建立可持续的数据获取体系。

相关文章推荐

发表评论

最热文章

关于作者

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