Python企查查企业分类信息采集实战:签名逆向与行业树遍历

发布时间:2026/9/16 19:58:13
Python企查查企业分类信息采集实战:签名逆向与行业树遍历 简介这是一套面向高校计算机专业本科生的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分层组织便于理解采集逻辑演进与模块复用是掌握RequestsLxmlPymysql协同开发的典型教学范例。1. 用 Python 抓取企查查企业分类信息不是“绕过反爬”而是理解它的请求逻辑与数据结构很多同学在做毕业设计或课程设计时看到“企查查企业分类信息采集”这个标题第一反应是找现成的破解脚本、翻找带登录态的 Cookie 或者直接套用某段失效的 Selenium 代码。结果跑两小时只拿到 20 条数据控制台满屏 403、412、503最后只能硬着头皮改题目。其实问题不在 Python 能力而在于没把“企查查”当成一个有明确交互规则的 Web 服务来对待——它不是黑盒而是由标准 HTTP 协议、可逆向的前端 JS、结构化 DOM 和稳定分类路径组成的公开接口集合。真正能跑通的方案核心是三件事定位分类页的真实请求链路不是首页 URL、复现关键请求头中的动态字段如X-Request-ID和X-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%91province%E5%8C%97%E4%BA%ACcitysubCat100000000pageSize20pageNumber1其中key是 URL 编码后的行业关键词如“互联网”→%E4%BA%92%E8%81%94%E7%BD%91province是省份名称需中文全称不能用简称subCat是分类 ID对应企查查后台的行业编码体系如100000000表示“互联网和相关服务”pageSize和pageNumber控制分页最大pageSize20官方限制提示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 字典序排序并拼接为 k1v1k2v2 形式 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 bqcc_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(fX-Signature: {sig}) print(fX-Request-ID: {nonce}) print(fTimestamp: {ts})参数说明urllib.parse.quote(..., safe)确保中文、斜杠等字符被正确编码与浏览器行为一致nonce必须每次请求重新生成且长度严格为 16 位小写 hextimestamp单位为毫秒误差超过 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, paramsparams, headersheaders, timeoutself.timeout) if resp.status_code 200: return resp.json() else: print(fHTTP {resp.status_code} for {params}) return None except Exception as e: print(fRequest failed: {e}) return None # 初始化会话 qcc QccSession() result qcc.search(key互联网, province北京, subCat100000000, page_number1) if result and result in result: print(fGot {len(result[result])} companies)这段代码已通过 2024 年 3 月实测可稳定获取前 20 条企业数据含公司名、法人、注册资本、成立日期、地址等字段。注意result字段是列表每项为一个企业对象结构清晰无需额外解析 HTML。3. 分类维度采集与数据清洗从 subCat 映射到行业树过滤无效条目仅靠手动填subCat100000000只能抓一个行业。毕业设计要求体现“分类信息采集”意味着需覆盖多级分类如“信息技术服务业”→“软件开发”→“人工智能软件开发”并保证数据结构统一、无重复、字段可用。3.1 获取完整分类树解析 /api/v1/industries 接口企查查提供分类元数据接口返回 JSON 格式的行业树。请求方式为 GET无需签名但需携带 Refererdef 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, headersheaders, timeout10) if resp.status_code 200: data resp.json() # 返回一级分类列表每个含 children 字段 return data.get(data, []) else: raise Exception(fFailed to fetch industries: {resp.status_code}) except Exception as e: print(fFetch industry tree error: {e}) return [] # 使用示例 s requests.Session() s.headers.update({User-Agent: Mozilla/5.0...}) tree fetch_industry_tree(s) print(fTotal 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( keycat_name, province全国, # 设为全国避免漏数据 subCatcat_id, page_numberpage, page_size20 ) 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_depth2, max_pages_per_cat3) # 测试用设为3页 print(fTotal valid companies: {len(data)})关键清洗点说明estiblishTime是企查查字段名 typo实际为成立日期格式为2015-03-12regCapital返回如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], keepfirst, inplaceTrue) # 列顺序调整符合阅读习惯 columns_order [ company_name, legal_representative, registered_capital, establish_date, address, industry, province, crawl_time, source ] df df.reindex(columnscolumns_order) # 写入 Excel启用数字列自动识别如注册资本含“万元”不影响排序 df.to_excel(filename, indexFalse, engineopenpyxl) print(fSaved {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, paramsparams, headersheaders, timeoutself.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(fRate limited. Waiting {wait_time:.2f}s before retry {attempt1}) time.sleep(wait_time) continue else: raise Exception(fHTTP {resp.status_code}) except Exception as e: if attempt max_retries: print(fFailed after {max_retries1} attempts: {e}) return None wait_time (2 ** attempt) random.uniform(0, 1) print(fAttempt {attempt1} failed: {e}. Retrying in {wait_time:.2f}s...) time.sleep(wait_time) return None4.3 记录结构化日志便于调试与答辩溯源使用logging模块记录关键操作输出到文件和控制台包含时间、分类 ID、页码、成功/失败状态import logging # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(qcc_crawl.log, encodingutf-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(fSUCCESS | CatID:{cat_id} | Page:{page} | Count:{len(result[result])}) # ... 后续处理 else: logger.error(fFAILED | 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实际需从前端下拉菜单联动获取 # 为简化演示假设 industry100000000 data [] for p in range(1, pages 1): result qcc.search(key互联网, provinceprovince, subCat100000000, page_numberp) if result and result in result: data.extend(clean_company_list(result[result])) # 生成 CSV 字节流 output io.StringIO() writer csv.DictWriter(output, fieldnamesdata[0].keys() if data else []) writer.writeheader() writer.writerows(data) output.seek(0) return send_file( io.BytesIO(output.getvalue().encode(utf-8-sig)), mimetypetext/csv, as_attachmentTrue, download_namefqcc_{province}_{industry}_{pages}pages.csv ) if __name__ __main__: app.run(debugTrue)配套templates/index.html精简版!DOCTYPE html html headtitle企查查分类采集系统/title/head body h2企查查企业分类信息采集毕业设计/h2 form action/crawl methodpost label省份input typetext nameprovince value全国 //labelbrbr label行业subCat IDinput typetext nameindustry value100000000 //labelbrbr label采集页数input typenumber namepages min1 max10 value3 //labelbrbr button typesubmit开始采集/button /form /body /html5.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 封装都直指“毕业设计”场景下的真实需求——它不是一个炫技的爬虫而是一个能讲清楚原理、能现场演示、能写进论文方法论章节的完整技术闭环。本文还有配套的精品资源点击获取