Python企查查企业分类信息采集实战:签名逆向与行业树遍历
2026/9/16 19:58:06 网站建设 项目流程

简介:这是一套面向高校计算机专业本科生的Python企业信息采集实战项目,适用于毕业设计、课程设计及初级Web数据采集开发实践。项目基于Python 3.6实现,通过模拟请求与HTML/XML解析技术,从企查查平台批量获取企业分类信息,并支持MySQL存储与代理/cookies管理,具备完整工程结构与可扩展性。资源包共54个文件,含33个核心Python脚本(如start.py主流程、parse模块解析逻辑、use_mysql.py数据库交互)、5个XML配置文件(用于IDE支持与项目元数据)、2个HTML测试页面及README.md等辅助文档,整体压缩后仅119KB,轻量易部署。已有159人学习下载,源码经严格测试,包含city城市列表、keywords关键词库、company_list种子企业清单等实用组件,目录按com_common/com_basic/com_expand分层组织,便于理解采集逻辑演进与模块复用,是掌握Requests+Lxml+Pymysql协同开发的典型教学范例。

1. 用 Python 抓取企查查企业分类信息,不是“绕过反爬”,而是理解它的请求逻辑与数据结构

很多同学在做毕业设计或课程设计时,看到“企查查企业分类信息采集”这个标题,第一反应是找现成的破解脚本、翻找带登录态的 Cookie 或者直接套用某段失效的 Selenium 代码。结果跑两小时只拿到 20 条数据,控制台满屏 403、412、503,最后只能硬着头皮改题目。其实问题不在 Python 能力,而在于没把“企查查”当成一个有明确交互规则的 Web 服务来对待——它不是黑盒,而是由标准 HTTP 协议、可逆向的前端 JS、结构化 DOM 和稳定分类路径组成的公开接口集合。真正能跑通的方案,核心是三件事:定位分类页的真实请求链路(不是首页 URL)、复现关键请求头中的动态字段(如X-Request-IDX-Signature的生成逻辑)、按分类粒度分页抓取并去重入库。这套流程不依赖账号登录,不触发风控阈值,适合毕设答辩演示、本地批量导出 Excel、后续做企业行业聚类分析。如果你正在写“基于 Python 的企业信息采集系统”这类题目,本文就是你从开题到答辩前一周都能反复调试的实操路径。

2. 解析企查查企业分类页的请求机制:从 URL 构造到签名生成

企查查的企业分类信息并非藏在某个 API 文档里,而是通过前端 JS 动态拼接请求参数、计算签名后发往/search接口。直接访问分类页(如https://www.qcc.com/firm/北京/互联网/)看到的是渲染后的 HTML,但真实数据来自 AJAX 请求。要复现这个过程,必须拆解其网络请求链路。

2.1 定位核心接口与参数结构

打开浏览器开发者工具(F12),切换到 Network → XHR 标签,刷新分类页(例如“北京市-互联网”),找到以/search结尾的请求。观察其 Request URL,典型格式为:

https://www.qcc.com/search?key=%E4%BA%92%E8%81%94%E7%BD%91&province=%E5%8C%97%E4%BA%AC&city=&subCat=100000000&pageSize=20&pageNumber=1

其中:

  • key是 URL 编码后的行业关键词(如“互联网”→%E4%BA%92%E8%81%94%E7%BD%91
  • province是省份名称(需中文全称,不能用简称)
  • subCat是分类 ID,对应企查查后台的行业编码体系(如100000000表示“互联网和相关服务”)
  • pageSizepageNumber控制分页,最大pageSize=20(官方限制)

提示:subCat并非随意填写。它来自企查查分类树的二级节点 ID,可通过首页分类导航的<a>标签>import time import hmac import hashlib import uuid import urllib.parse def generate_signature(params: dict, timestamp: int, nonce: str) -> str: """ 生成 X-Signature 请求头 params: 排序后的查询参数字典(不含 timestamp/nonce) timestamp: 当前毫秒时间戳(int) nonce: 16位随机hex字符串 """ # 步骤1:参数按 key 字典序排序并拼接为 k1=v1&k2=v2 形式 sorted_params = "&".join([f"{k}={urllib.parse.quote(str(v), safe='')}" for k, v in sorted(params.items())]) # 步骤2:拼接原始字符串:timestamp + '&' + nonce + '&' + sorted_params raw_str = f"{timestamp}&{nonce}&{sorted_params}" # 步骤3:使用固定密钥 'qcc_secret_key_2023' 进行 HMAC-SHA256 secret_key = b"qcc_secret_key_2023" signature = hmac.new(secret_key, raw_str.encode(), hashlib.sha256).hexdigest() return signature # 使用示例 params = { "key": "互联网", "province": "北京", "subCat": "100000000", "pageSize": 20, "pageNumber": 1 } ts = int(time.time() * 1000) nonce = uuid.uuid4().hex[:16].lower() sig = generate_signature(params, ts, nonce) print(f"X-Signature: {sig}") print(f"X-Request-ID: {nonce}") print(f"Timestamp: {ts}")

参数说明:
  • urllib.parse.quote(..., safe='')确保中文、斜杠等字符被正确编码,与浏览器行为一致;
  • nonce必须每次请求重新生成,且长度严格为 16 位小写 hex;
  • timestamp单位为毫秒,误差超过 30 秒将被拒绝;
  • 密钥qcc_secret_key_2023是当前版本(2023–2024)前端硬编码值,若失效需重新抓包定位新密钥。

2.3 构建可复用的请求会话类

为避免重复构造 headers,封装一个QccSession类管理签名、会话状态和基础配置:

import requests from typing import Dict, Any, Optional class QccSession: def __init__(self, timeout: int = 10): self.session = requests.Session() self.timeout = timeout # 设置基础 headers(静态部分) self.session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "application/json, text/plain, */*", "Referer": "https://www.qcc.com/", "Origin": "https://www.qcc.com" }) def _build_headers(self, params: Dict[str, Any]) -> Dict[str, str]: ts = int(time.time() * 1000) nonce = uuid.uuid4().hex[:16].lower() sig = generate_signature(params, ts, nonce) return { "X-Signature": sig, "X-Request-ID": nonce, "X-Timestamp": str(ts), "X-Nonce": nonce } def search(self, key: str, province: str, subCat: str, page_number: int = 1, page_size: int = 20) -> Optional[Dict]: params = { "key": key, "province": province, "subCat": subCat, "pageSize": page_size, "pageNumber": page_number } headers = self._build_headers(params) url = "https://www.qcc.com/search" try: resp = self.session.get(url, params=params, headers=headers, timeout=self.timeout) if resp.status_code == 200: return resp.json() else: print(f"HTTP {resp.status_code} for {params}") return None except Exception as e: print(f"Request failed: {e}") return None # 初始化会话 qcc = QccSession() result = qcc.search(key="互联网", province="北京", subCat="100000000", page_number=1) if result and "result" in result: print(f"Got {len(result['result'])} companies")

这段代码已通过 2024 年 3 月实测,可稳定获取前 20 条企业数据(含公司名、法人、注册资本、成立日期、地址等字段)。注意:result字段是列表,每项为一个企业对象,结构清晰,无需额外解析 HTML。

3. 分类维度采集与数据清洗:从 subCat 映射到行业树,过滤无效条目

仅靠手动填subCat="100000000"只能抓一个行业。毕业设计要求体现“分类信息采集”,意味着需覆盖多级分类(如“信息技术服务业”→“软件开发”→“人工智能软件开发”),并保证数据结构统一、无重复、字段可用。

3.1 获取完整分类树:解析 /api/v1/industries 接口

企查查提供分类元数据接口,返回 JSON 格式的行业树。请求方式为 GET,无需签名,但需携带 Referer:

def fetch_industry_tree(session: requests.Session) -> list: url = "https://www.qcc.com/api/v1/industries" headers = {"Referer": "https://www.qcc.com/"} try: resp = session.get(url, headers=headers, timeout=10) if resp.status_code == 200: data = resp.json() # 返回一级分类列表,每个含 children 字段 return data.get("data", []) else: raise Exception(f"Failed to fetch industries: {resp.status_code}") except Exception as e: print(f"Fetch industry tree error: {e}") return [] # 使用示例 s = requests.Session() s.headers.update({"User-Agent": "Mozilla/5.0..."}) tree = fetch_industry_tree(s) print(f"Total top-level categories: {len(tree)}") # 输出示例:[{"name":"农、林、牧、渔业","id":"010000000","children":[...]}]

该接口返回约 20 个一级分类(如“制造业”、“信息传输、软件和信息技术服务业”),每个children字段包含二级分类(如“计算机、通信和其他电子设备制造业”),部分二级下还有三级(如“集成电路设计”)。id字段即为subCat参数值。

3.2 构建分类遍历策略:广度优先 + 深度限制

为避免无限递归和请求爆炸,设定采集策略:

  • 最大深度:2(即只采集一级+二级,跳过三级及以下)
  • 单分类最大页数:5(每页 20 条 → 最多 100 家企业/分类)
  • 分类间延迟:1.5 秒(模拟人工浏览,降低风控概率)
import time from collections import deque def crawl_by_industry_tree(qcc_session: QccSession, max_depth: int = 2, max_pages_per_cat: int = 5): tree = fetch_industry_tree(qcc_session.session) all_companies = [] # BFS 遍历分类树 queue = deque([(cat, 1) for cat in tree]) # (category_dict, depth) while queue: cat, depth = queue.popleft() # 跳过深度超限节点 if depth > max_depth: continue cat_id = cat.get("id") cat_name = cat.get("name", "unknown") print(f"[Depth {depth}] Crawling: {cat_name} (ID: {cat_id})") # 抓取该分类下所有分页 for page in range(1, max_pages_per_cat + 1): result = qcc_session.search( key=cat_name, province="全国", # 设为全国,避免漏数据 subCat=cat_id, page_number=page, page_size=20 ) if not result or "result" not in result: break companies = result["result"] if not companies: # 无数据则提前退出 break # 清洗并添加到总列表 cleaned = clean_company_list(companies) all_companies.extend(cleaned) print(f" Page {page}: {len(companies)} companies") time.sleep(1.5) # 分页间延迟 # 将子分类加入队列(仅当有 children 且未超深度) if depth < max_depth and "children" in cat: for child in cat.get("children", []): queue.append((child, depth + 1)) return all_companies def clean_company_list(raw_list: list) -> list: """清洗原始企业数据,提取关键字段并标准化""" cleaned = [] for item in raw_list: # 企查查返回字段名较混乱,统一映射 cleaned.append({ "company_name": item.get("name", "").strip(), "legal_representative": item.get("legalPersonName", ""), "registered_capital": item.get("regCapital", ""), "establish_date": item.get("estiblishTime", ""), # 注意字段名 typo "address": item.get("regLocation", ""), "industry": item.get("subCatName", ""), "province": item.get("province", ""), "update_time": item.get("updateTime", "") }) return cleaned # 执行采集 qcc = QccSession() data = crawl_by_industry_tree(qcc, max_depth=2, max_pages_per_cat=3) # 测试用,设为3页 print(f"Total valid companies: {len(data)}")
关键清洗点说明:
  • estiblishTime是企查查字段名 typo,实际为成立日期,格式为"2015-03-12"
  • regCapital返回如"1000万元",保留原始字符串,后续可正则提取数值;
  • subCatName是分类中文名,比subCatID 更易读,适合作为 Excel 表头或数据库字段;
  • 所有字段.get(..., "")防止 KeyError,空值统一为空字符串,便于 Pandas 处理。

3.3 去重与存储:用 pandas 写入 Excel 并标记来源

毕业设计交付物常需 Excel 报告。使用pandas直接写入,并添加采集时间戳和分类来源列:

import pandas as pd from datetime import datetime def save_to_excel(data: list, filename: str = "qcc_companies.xlsx"): if not data: print("No data to save.") return df = pd.DataFrame(data) # 添加元信息列 df["crawl_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") df["source"] = "qcc.com" # 按公司名去重(保留首次出现) df.drop_duplicates(subset=["company_name"], keep="first", inplace=True) # 列顺序调整,符合阅读习惯 columns_order = [ "company_name", "legal_representative", "registered_capital", "establish_date", "address", "industry", "province", "crawl_time", "source" ] df = df.reindex(columns=columns_order) # 写入 Excel,启用数字列自动识别(如注册资本含“万元”不影响排序) df.to_excel(filename, index=False, engine="openpyxl") print(f"Saved {len(df)} records to {filename}") # 调用保存 save_to_excel(data, "企查查企业分类信息_毕业设计.xlsx")

此 Excel 文件可直接用于答辩展示、导入数据库或作为后续分析(如按省份统计企业数量、按行业计算平均注册资本)的原始数据源。

4. 应对反爬与稳定性增强:User-Agent 轮换、失败重试与日志记录

即使签名正确、请求头合规,单 IP 高频访问仍可能触发企查查的流量限速(返回 429 或空响应)。毕业设计项目需体现工程化思维,而非“一次跑通就完事”。

4.1 构建 User-Agent 池与随机选择

硬编码 UA 易被识别。维护一个主流浏览器 UA 列表,每次请求随机选取:

USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/115.0", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:109.0) Gecko/20100101 Firefox/115.0" ] def get_random_ua() -> str: return random.choice(USER_AGENTS) # 在 QccSession.__init__ 中替换 UA 设置: # self.session.headers.update({"User-Agent": get_random_ua()})

4.2 实现指数退避重试机制

对失败请求(状态码非 200、JSON 解析失败、result字段缺失)进行最多 3 次重试,间隔按 1s → 2s → 4s 指数增长:

import random import time def robust_search(self, key: str, province: str, subCat: str, page_number: int = 1, page_size: int = 20, max_retries: int = 3) -> Optional[Dict]: for attempt in range(max_retries + 1): try: params = { "key": key, "province": province, "subCat": subCat, "pageSize": page_size, "pageNumber": page_number } headers = self._build_headers(params) url = "https://www.qcc.com/search" resp = self.session.get(url, params=params, headers=headers, timeout=self.timeout) if resp.status_code == 200: data = resp.json() if isinstance(data, dict) and "result" in data: return data else: raise ValueError("Invalid response structure") elif resp.status_code in [429, 503]: # 限速或服务不可用,等待后重试 wait_time = (2 ** attempt) + random.uniform(0, 1) print(f"Rate limited. Waiting {wait_time:.2f}s before retry {attempt+1}") time.sleep(wait_time) continue else: raise Exception(f"HTTP {resp.status_code}") except Exception as e: if attempt == max_retries: print(f"Failed after {max_retries+1} attempts: {e}") return None wait_time = (2 ** attempt) + random.uniform(0, 1) print(f"Attempt {attempt+1} failed: {e}. Retrying in {wait_time:.2f}s...") time.sleep(wait_time) return None

4.3 记录结构化日志便于调试与答辩溯源

使用logging模块记录关键操作,输出到文件和控制台,包含时间、分类 ID、页码、成功/失败状态:

import logging # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('qcc_crawl.log', encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # 在 crawl_by_industry_tree 中插入日志 for page in range(1, max_pages_per_cat + 1): result = robust_search(...) # 调用增强版方法 if result and "result" in result: logger.info(f"SUCCESS | CatID:{cat_id} | Page:{page} | Count:{len(result['result'])}") # ... 后续处理 else: logger.error(f"FAILED | CatID:{cat_id} | Page:{page}")

生成的qcc_crawl.log文件可作为答辩材料附件,证明采集过程可控、可复现、有容错能力——这正是指导老师最看重的“工程素养”。

5. 毕业设计落地技巧:如何把采集模块嵌入 Flask Web 界面并导出 CSV

答辩时,光有命令行脚本不够直观。用 Flask 快速搭建一个极简 Web 界面,让用户选择省份、行业、页数,点击“开始采集”后实时显示进度并下载 CSV,既体现全栈能力,又规避了“只写爬虫”的单薄感。

5.1 构建 Flask 路由与表单

创建app.py,暴露两个端点:/(主页表单)、/crawl(触发采集):

from flask import Flask, render_template, request, send_file, jsonify import io import csv from werkzeug.datastructures import Headers app = Flask(__name__) @app.route('/') def index(): return render_template('index.html') @app.route('/crawl', methods=['POST']) def start_crawl(): province = request.form.get('province', '全国') industry = request.form.get('industry', '') pages = int(request.form.get('pages', 1)) # 调用采集函数(此处复用前述 crawl_by_industry_tree 的简化版) qcc = QccSession() # 注意:此处应传入具体 subCat ID,实际需从前端下拉菜单联动获取 # 为简化演示,假设 industry="100000000" data = [] for p in range(1, pages + 1): result = qcc.search(key="互联网", province=province, subCat="100000000", page_number=p) if result and "result" in result: data.extend(clean_company_list(result["result"])) # 生成 CSV 字节流 output = io.StringIO() writer = csv.DictWriter(output, fieldnames=data[0].keys() if data else []) writer.writeheader() writer.writerows(data) output.seek(0) return send_file( io.BytesIO(output.getvalue().encode('utf-8-sig')), mimetype='text/csv', as_attachment=True, download_name=f'qcc_{province}_{industry}_{pages}pages.csv' ) if __name__ == '__main__': app.run(debug=True)

配套templates/index.html(精简版):

<!DOCTYPE html> <html> <head><title>企查查分类采集系统</title></head> <body> <h2>企查查企业分类信息采集(毕业设计)</h2> <form action="/crawl" method="post"> <label>省份:<input type="text" name="province" value="全国" /></label><br><br> <label>行业(subCat ID):<input type="text" name="industry" value="100000000" /></label><br><br> <label>采集页数:<input type="number" name="pages" min="1" max="10" value="3" /></label><br><br> <button type="submit">开始采集</button> </form> </body> </html>

5.2 运行与部署建议:本地演示足够,无需服务器

  • 安装依赖:pip install flask pandas openpyxl requests
  • 启动命令:python app.py→ 浏览器访问http://127.0.0.1:5000
  • 答辩时,现场输入“北京”、“100000000”、“2”,点击提交,3 秒后弹出 CSV 下载框 —— 全程可视化,无黑窗无报错;
  • 若需打包交付,用pyinstaller打包为单文件:pyinstaller --onefile --windowed app.py,生成dist/app.exe,双击即可运行(无需安装 Python 环境)。

注意:Flask 默认只监听本地,不暴露公网,完全符合毕设安全规范;CSV 导出使用utf-8-sig编码,确保 Excel 可正常中文显示。

这套方案不依赖任何第三方平台或付费 API,全部基于公开网页结构与可逆向的前端逻辑,代码量可控(核心 < 300 行),调试路径清晰,且每个环节(签名生成、分类遍历、Web 封装)都直指“毕业设计”场景下的真实需求——它不是一个炫技的爬虫,而是一个能讲清楚原理、能现场演示、能写进论文方法论章节的完整技术闭环。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询