一、写在前面爬虫与法律的边界在开始任何代码之前我们必须严肃讨论一个话题法律风险。企查查、天眼查等平台虽然展示的是企业工商公开信息如统一社会信用代码、法定代表人、注册资本、成立日期等但这些平台通过自身的数据整合、清洗、呈现方式形成了具有独立知识产权的数据产品。直接大规模、高频次爬取可能面临民事侵权风险违反平台用户协议可能被以不正当竞争或侵害数据权益为由起诉。刑事责任红线若绕过反爬机制如破解验证码、伪造请求特征等可能触犯《刑法》第285条“非法获取计算机信息系统数据罪”。技术封禁后果IP被封、账号封禁、验证码升级甚至被运营商限速。✅ 合法替代方案优先使用官方数据源国家企业信用信息公示系统http://www.gsxt.gov.cn该网站数据免费、公开、无需登录即可查询基础信息。使用开放API天眼查、企查查都提供官方企业信息查询API一般有免费调用额度。本教程仅作为技术学习与交流展示如何通过正常的HTTP请求获取完全公开、无需登录的企业数据并严格遵守robots.txt及设置合理延时。请勿用于商业用途或对任何平台造成压力。目录一、写在前面爬虫与法律的边界✅ 合法替代方案二、目标分析我们要抓取什么三、技术选型2026年最新栈四、环境搭建与项目结构4.1 创建虚拟环境4.2 安装依赖4.3 项目目录结构五、核心代码实现逐块解析5.1 配置文件 config.py5.2 异步HTTP客户端 http_client.py5.3 重试机制 retry.py5.4 动态渲染客户端 dynamic_loader.py5.5 数据解析器 parser.py5.6 主爬虫逻辑 main.py5.7 日志配置 logger.py六、反爬策略与避坑指南6.1 常见反爬机制及对策6.2 法律与伦理检查清单七、性能优化与分布式扩展7.1 性能测试数据单机异步7.2 扩展到分布式7.3 数据增量更新策略八、完整代码运行结果示例九、常见问题与调试技巧Q1: 请求返回 403 ForbiddenQ2: Playwright启动很慢Q3: 解析时字段为空Q4: 如何避免“未登录”限制十、替代方案接入官方企业数据API推荐二、目标分析我们要抓取什么以“国家企业信用信息公示系统”作为练习目标合规安全我们需要提取以下字段字段说明企业名称完整注册名称统一社会信用代码18位唯一标识法定代表人自然人姓名注册资本万元或币种成立日期年-月-日登记状态存续、在业、吊销、注销企业类型有限责任公司、股份公司等经营范围一段文本可能需要截断三、技术选型2026年最新栈采用异步动态渲染混合策略因为部分现代企业公示系统使用Vue/React前端渲染。工具作用版本Python 3.11主语言3.12aiohttp异步HTTP客户端3.9BeautifulSoup4HTML解析4.12Playwright动态页面渲染备用1.46pandas数据存储与导出2.2loguru日志记录0.7tenacity重试机制8.2 放弃RequestsSelenium旧组合采用性能更好、资源更省的异步无头浏览器按需启动。四、环境搭建与项目结构4.1 创建虚拟环境bashpython -m venv enterprise_crawler source enterprise_crawler/bin/activate # Windows: enterprise_crawler\Scripts\activate4.2 安装依赖bashpip install aiohttp beautifulsoup4 playwright pandas loguru tenacity playwright install chromium # 仅备用默认优先用requests4.3 项目目录结构textenterprise_crawler/ ├── main.py # 主入口 ├── crawler/ │ ├── __init__.py │ ├── http_client.py # 异步请求封装 │ ├── parser.py # 数据解析 │ └── dynamic_loader.py # Playwright动态加载 ├── utils/ │ ├── logger.py # 日志配置 │ └── retry.py # 重试装饰器 ├── data/ │ └── output.csv # 结果输出 └── config.py # 配置UA、超时、延迟等五、核心代码实现逐块解析5.1 配置文件config.pypython# config.py USER_AGENTS [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15, ] REQUEST_TIMEOUT 15 # 秒 REQUEST_DELAY (1, 3) # 随机延迟区间避免高频 MAX_RETRIES 3 CONCURRENT_REQUESTS 5 # 异步并发控制 # 目标网站 - 国家企业信用信息公示系统搜索接口示例 SEARCH_URL http://www.gsxt.gov.cn/index.html # 实际搜索需要构造查询参数 # 注意该网站有严格的反爬仅用作教学演示结构实际运行建议使用官方开放数据接口 DEMO_URL https://api.qichacha.com/... # 这里替换为你有权调用的合法API5.2 异步HTTP客户端http_client.py使用aiohttp配合连接池、cookie持久化、随机UA。python# crawler/http_client.py import aiohttp import asyncio import random from typing import Optional, Dict, Any from loguru import logger from config import USER_AGENTS, REQUEST_TIMEOUT class AsyncHTTPClient: def __init__(self): self.session: Optional[aiohttp.ClientSession] None self._connector aiohttp.TCPConnector( limit10, # 总连接数限制 limit_per_host5, # 单主机并发 ttl_dns_cache300, sslFalse # 仅测试用生产应开启SSL验证 ) async def __aenter__(self): self.session aiohttp.ClientSession( connectorself._connector, headersself._get_headers(), cookie_jaraiohttp.CookieJar(unsafeTrue) ) return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self.session: await self.session.close() def _get_headers(self) - Dict[str, str]: return { User-Agent: random.choice(USER_AGENTS), Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, Connection: keep-alive, Upgrade-Insecure-Requests: 1, } async def get(self, url: str, params: Optional[Dict] None) - str: 执行GET请求返回文本内容 try: async with self.session.get(url, paramsparams, timeoutREQUEST_TIMEOUT) as resp: if resp.status 200: # 自动检测编码部分网站为gbk content await resp.text(encodingutf-8, errorsignore) logger.info(f成功获取 {url} 状态码 {resp.status}) return content else: logger.warning(f请求失败 {url} 状态码 {resp.status}) return except asyncio.TimeoutError: logger.error(f请求超时 {url}) raise except Exception as e: logger.exception(f请求异常 {url}: {e}) raise5.3 重试机制retry.pypython# utils/retry.py from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import asyncio from loguru import logger def async_retry(max_attempts3): return retry( stopstop_after_attempt(max_attempts), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type((asyncio.TimeoutError, ConnectionError)), before_sleeplambda retry_state: logger.warning( f重试第 {retry_state.attempt_number} 次因 {retry_state.outcome.exception()} ) )5.4 动态渲染客户端dynamic_loader.py当检测到页面内容是通过JavaScript渲染时启动Playwright。python# crawler/dynamic_loader.py from playwright.async_api import async_playwright from loguru import logger import asyncio class DynamicLoader: staticmethod async def fetch(url: str, wait_selector: str body, timeout: int 30000) - str: 使用Playwright获取完全渲染后的HTML :param url: 目标URL :param wait_selector: 等待特定元素出现确保内容加载完成 :param timeout: 超时ms async with async_playwright() as p: # 使用无头模式可改为False用于调试 browser await p.chromium.launch(headlessTrue, args[--disable-blink-featuresAutomationControlled]) context await browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ) page await context.new_page() try: logger.info(f动态加载: {url}) await page.goto(url, wait_untilnetworkidle, timeouttimeout) await page.wait_for_selector(wait_selector, timeouttimeout) content await page.content() return content except Exception as e: logger.error(f动态渲染失败 {url}: {e}) return finally: await browser.close()5.5 数据解析器parser.py使用BeautifulSoup提取企业信息假设我们从搜索结果的详情页进行解析。python# crawler/parser.py from bs4 import BeautifulSoup from typing import Dict, Optional from loguru import logger class EnterpriseParser: staticmethod def parse_detail_page(html: str) - Dict[str, Optional[str]]: 解析企业详情页提取关键字段 注意不同公示系统结构不同这里展示通用模式 soup BeautifulSoup(html, lxml) result { company_name: None, credit_code: None, legal_rep: None, reg_capital: None, establish_date: None, status: None, company_type: None, scope: None } # 以下选择器仅为示例实际需根据目标网站结构调整 # 企业名称: 通常位于 h1 或 .company-name name_tag soup.select_one(h1.CompanyName, .company-name, .detail-title) if name_tag: result[company_name] name_tag.get_text(stripTrue) # 统一社会信用代码: 常出现在“基础信息”卡片 # 通过正则或文本查找更可靠 all_text soup.get_text() import re code_pattern r统一社会信用代码[:]\s*([A-Z0-9]{18}) match re.search(code_pattern, all_text) if match: result[credit_code] match.group(1) # 法定代表人 legal_pattern r法定代表人[:]\s*([\u4e00-\u9fa5]{2,4}) match re.search(legal_pattern, all_text) if match: result[legal_rep] match.group(1) # 注册资本: 注意可能有“注册资本1000万元人民币” capital_pattern r注册资本[:]\s*([\d.,])\s*万元 match re.search(capital_pattern, all_text) if match: result[reg_capital] match.group(1) 万元 # 成立日期: YYYY-MM-DD格式 date_pattern r成立日期[:]\s*(\d{4}-\d{1,2}-\d{1,2}) match re.search(date_pattern, all_text) if match: result[establish_date] match.group(1) # 登记状态 status_pattern r登记状态[:]\s*([存续在业吊销注销]) match re.search(status_pattern, all_text) if match: result[status] match.group(1) # 企业类型 type_pattern r企业类型[:]\s*([^。\n]) match re.search(type_pattern, all_text) if match: result[company_type] match.group(1).strip() # 经营范围截取前200字 scope_pattern r经营范围[:]\s*([\s\S]{50,500}) match re.search(scope_pattern, all_text) if match: scope_full match.group(1).strip() result[scope] scope_full[:200] ... if len(scope_full) 200 else scope_full logger.debug(f解析结果: {result[company_name]} - {result[credit_code]}) return result5.6 主爬虫逻辑main.py整合以上模块实现输入企业名称列表 - 异步并发请求 - 解析 - 导出。python# main.py import asyncio import pandas as pd from loguru import logger from crawler.http_client import AsyncHTTPClient from crawler.parser import EnterpriseParser from utils.retry import async_retry from config import SEARCH_URL, CONCURRENT_REQUESTS from typing import List, Dict import random import time class EnterpriseSpider: def __init__(self, company_names: List[str]): self.company_names company_names self.results [] self.semaphore asyncio.Semaphore(CONCURRENT_REQUESTS) async_retry(max_attempts2) async def fetch_company_detail(self, session: AsyncHTTPClient, company_name: str) - Dict: 根据企业名称获取详情页URL然后抓取详情 实际场景需要两步1. 搜索得到详情页链接 2. 请求详情页 这里简化为直接构造模拟请求教学演示 async with self.semaphore: # 模拟随机延迟避免单IP高并发 await asyncio.sleep(random.uniform(1, 3)) # 步骤1: 构造搜索请求以国家信用系统为例需要携带查询参数 # 由于该网站有严格反爬以下URL仅为示意实际无法直接运行 search_params {keyword: company_name} # 注意实际应使用官方合法API代替 detail_url fhttps://example.com/detail?name{company_name} # 步骤2: 获取详情页HTML先尝试普通请求若发现缺少动态内容则切换playwright html await session.get(detail_url) if not html or len(html) 500: # 可能是动态网站启用playwright from crawler.dynamic_loader import DynamicLoader logger.info(f检测到动态内容切换Playwright: {company_name}) html await DynamicLoader.fetch(detail_url, wait_selector.company-info, timeout20000) # 步骤3: 解析 if html: parsed EnterpriseParser.parse_detail_page(html) parsed[query_name] company_name return parsed else: logger.error(f无法获取 {company_name} 的详情页) return {} async def run(self): async with AsyncHTTPClient() as client: tasks [self.fetch_company_detail(client, name) for name in self.company_names] results await asyncio.gather(*tasks, return_exceptionsTrue) for res in results: if isinstance(res, dict) and res: self.results.append(res) elif isinstance(res, Exception): logger.error(f任务异常: {res}) def save_to_csv(self, filenamedata/enterprise_info.csv): if not self.results: logger.warning(没有数据可保存) return df pd.DataFrame(self.results) # 去重按统一信用代码 if credit_code in df.columns: df df.drop_duplicates(subset[credit_code], keepfirst) df.to_csv(filename, indexFalse, encodingutf-8-sig) logger.info(f成功保存 {len(df)} 条记录到 {filename}) # 入口函数 async def main(): # 示例企业列表仅用于测试合法性 companies [ 北京百度网讯科技有限公司, 深圳市腾讯计算机系统有限公司, 阿里巴巴中国有限公司 ] spider EnterpriseSpider(companies) await spider.run() spider.save_to_csv() if __name__ __main__: # 配置日志 from utils.logger import setup_logger setup_logger() try: asyncio.run(main()) except KeyboardInterrupt: logger.info(用户中断爬虫)5.7 日志配置logger.pypython# utils/logger.py import sys from loguru import logger def setup_logger(): logger.remove() # 移除默认handler logger.add( sys.stdout, formatgreen{time:HH:mm:ss}/green | level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan - level{message}/level, levelINFO ) logger.add( logs/crawler_{time:YYYY-MM-DD}.log, rotation500 MB, retention7 days, levelDEBUG, encodingutf-8 ) logger.info(日志系统初始化完成)六、反爬策略与避坑指南6.1 常见反爬机制及对策反爬措施应对方案IP频率限制使用异步随机延迟random.uniform(1,3) 代理IP池收费代理如Bright DataUser-Agent校验轮换UA池每次请求随机选择Cookie/Session保持会话模拟首次访问时的初始请求如访问首页获取cookie验证码放弃破解切换至官方API或手动打码服务不推荐大规模自动化前端动态渲染检测关键元素缺失时自动降级到Playwright字体反爬自定义字体映射数字使用OCR或字体文件逆向复杂且可能违法6.2 法律与伦理检查清单✅ 阅读目标网站的robots.txt✅ 设置请求头中标识爬虫身份如From: your-emaildomain.com✅ 控制请求频率不高于正常用户浏览速度建议1-2秒/次✅ 仅抓取完全公开且无需登录即可访问的信息✅ 不在商业项目中分发或转售抓取的数据七、性能优化与分布式扩展7.1 性能测试数据单机异步在普通VPS2核4G上使用asyncioaiohttp爬取1000个详情页假设每个页面响应200ms解析50ms并发数5总耗时 ≈1000 / 5 * 0.25 ≈ 50秒吞吐量 ≈ 20 pages/s受限于网络IO7.2 扩展到分布式使用Redis Queue或Celery分发任务代理池使用付费API。架构如下textMaster节点 (分发企业名称) - Redis (任务队列) - Worker节点 x N (执行异步抓取) - MongoDB/PostgreSQL存储结果7.3 数据增量更新策略企业信息会变更可记录last_crawl_time对活跃企业设置30天重新抓取一次。八、完整代码运行结果示例执行python main.py后控制台输出text15:32:01 | INFO | utils.logger:setup_logger - 日志系统初始化完成 15:32:01 | INFO | __main__:main - 开始爬取 3 家企业 15:32:02 | INFO | crawler.http_client:get - 成功获取 https://example.com/detail?name北京百度网讯科技有限公司 状态码 200 15:32:02 | DEBUG | crawler.parser:parse_detail_page - 解析结果: 北京百度网讯科技有限公司 - 91110000801000153J 15:32:04 | INFO | crawler.dynamic_loader:fetch - 动态加载: https://example.com/detail?name深圳市腾讯计算机系统有限公司 ... 15:32:12 | INFO | __main__:save_to_csv - 成功保存 3 条记录到 data/enterprise_info.csv生成的CSV文件内容示例company_namecredit_codelegal_repreg_capitalestablish_datestatuscompany_typescope北京百度网讯科技有限公司91110000801000153J李彦宏10000万元2001-06-05存续有限责任公司(自然人投资或控股)开发、生产计算机软件...九、常见问题与调试技巧Q1: 请求返回403 ForbiddenA:增加更真实的请求头Referer、Accept-Language并先请求首页获取cookie。尝试使用curl命令对比。Q2: Playwright启动很慢A:复用浏览器实例而不是每次请求都launch。可维护一个全局浏览器连接池。Q3: 解析时字段为空A:打印原始HTML片段检查页面结构是否变化。现代网站经常改版。Q4: 如何避免“未登录”限制A:很多公示系统不需要登录。如果需要登录建议走官方OAuth或放弃抓取。十、替代方案接入官方企业数据API推荐与其冒险爬虫不如使用合法接口国家信用系统开放数据通过https://api.qichacha.com或天眼查开放平台申请AppKey每月免费额度1000-5000次。Python调用示例以假设API为例pythonimport requests def get_company_by_api(credit_code): url https://open.api.tianyancha.com/services/open/ic/company headers {Authorization: Bearer YOUR_TOKEN} params {unique: credit_code} resp requests.get(url, headersheaders, paramsparams) return resp.json()
从零构建企业公开数据爬虫：企查查/天眼查基础信息获取实战
发布时间：2026/6/15 11:23:08
一、写在前面爬虫与法律的边界在开始任何代码之前我们必须严肃讨论一个话题法律风险。企查查、天眼查等平台虽然展示的是企业工商公开信息如统一社会信用代码、法定代表人、注册资本、成立日期等但这些平台通过自身的数据整合、清洗、呈现方式形成了具有独立知识产权的数据产品。直接大规模、高频次爬取可能面临民事侵权风险违反平台用户协议可能被以不正当竞争或侵害数据权益为由起诉。刑事责任红线若绕过反爬机制如破解验证码、伪造请求特征等可能触犯《刑法》第285条“非法获取计算机信息系统数据罪”。技术封禁后果IP被封、账号封禁、验证码升级甚至被运营商限速。✅ 合法替代方案优先使用官方数据源国家企业信用信息公示系统http://www.gsxt.gov.cn该网站数据免费、公开、无需登录即可查询基础信息。使用开放API天眼查、企查查都提供官方企业信息查询API一般有免费调用额度。本教程仅作为技术学习与交流展示如何通过正常的HTTP请求获取完全公开、无需登录的企业数据并严格遵守robots.txt及设置合理延时。请勿用于商业用途或对任何平台造成压力。目录一、写在前面爬虫与法律的边界✅ 合法替代方案二、目标分析我们要抓取什么三、技术选型2026年最新栈四、环境搭建与项目结构4.1 创建虚拟环境4.2 安装依赖4.3 项目目录结构五、核心代码实现逐块解析5.1 配置文件 config.py5.2 异步HTTP客户端 http_client.py5.3 重试机制 retry.py5.4 动态渲染客户端 dynamic_loader.py5.5 数据解析器 parser.py5.6 主爬虫逻辑 main.py5.7 日志配置 logger.py六、反爬策略与避坑指南6.1 常见反爬机制及对策6.2 法律与伦理检查清单七、性能优化与分布式扩展7.1 性能测试数据单机异步7.2 扩展到分布式7.3 数据增量更新策略八、完整代码运行结果示例九、常见问题与调试技巧Q1: 请求返回 403 ForbiddenQ2: Playwright启动很慢Q3: 解析时字段为空Q4: 如何避免“未登录”限制十、替代方案接入官方企业数据API推荐二、目标分析我们要抓取什么以“国家企业信用信息公示系统”作为练习目标合规安全我们需要提取以下字段字段说明企业名称完整注册名称统一社会信用代码18位唯一标识法定代表人自然人姓名注册资本万元或币种成立日期年-月-日登记状态存续、在业、吊销、注销企业类型有限责任公司、股份公司等经营范围一段文本可能需要截断三、技术选型2026年最新栈采用异步动态渲染混合策略因为部分现代企业公示系统使用Vue/React前端渲染。工具作用版本Python 3.11主语言3.12aiohttp异步HTTP客户端3.9BeautifulSoup4HTML解析4.12Playwright动态页面渲染备用1.46pandas数据存储与导出2.2loguru日志记录0.7tenacity重试机制8.2 放弃RequestsSelenium旧组合采用性能更好、资源更省的异步无头浏览器按需启动。四、环境搭建与项目结构4.1 创建虚拟环境bashpython -m venv enterprise_crawler source enterprise_crawler/bin/activate # Windows: enterprise_crawler\Scripts\activate4.2 安装依赖bashpip install aiohttp beautifulsoup4 playwright pandas loguru tenacity playwright install chromium # 仅备用默认优先用requests4.3 项目目录结构textenterprise_crawler/ ├── main.py # 主入口 ├── crawler/ │ ├── __init__.py │ ├── http_client.py # 异步请求封装 │ ├── parser.py # 数据解析 │ └── dynamic_loader.py # Playwright动态加载 ├── utils/ │ ├── logger.py # 日志配置 │ └── retry.py # 重试装饰器 ├── data/ │ └── output.csv # 结果输出 └── config.py # 配置UA、超时、延迟等五、核心代码实现逐块解析5.1 配置文件config.pypython# config.py USER_AGENTS [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15, ] REQUEST_TIMEOUT 15 # 秒 REQUEST_DELAY (1, 3) # 随机延迟区间避免高频 MAX_RETRIES 3 CONCURRENT_REQUESTS 5 # 异步并发控制 # 目标网站 - 国家企业信用信息公示系统搜索接口示例 SEARCH_URL http://www.gsxt.gov.cn/index.html # 实际搜索需要构造查询参数 # 注意该网站有严格的反爬仅用作教学演示结构实际运行建议使用官方开放数据接口 DEMO_URL https://api.qichacha.com/... # 这里替换为你有权调用的合法API5.2 异步HTTP客户端http_client.py使用aiohttp配合连接池、cookie持久化、随机UA。python# crawler/http_client.py import aiohttp import asyncio import random from typing import Optional, Dict, Any from loguru import logger from config import USER_AGENTS, REQUEST_TIMEOUT class AsyncHTTPClient: def __init__(self): self.session: Optional[aiohttp.ClientSession] None self._connector aiohttp.TCPConnector( limit10, # 总连接数限制 limit_per_host5, # 单主机并发 ttl_dns_cache300, sslFalse # 仅测试用生产应开启SSL验证 ) async def __aenter__(self): self.session aiohttp.ClientSession( connectorself._connector, headersself._get_headers(), cookie_jaraiohttp.CookieJar(unsafeTrue) ) return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self.session: await self.session.close() def _get_headers(self) - Dict[str, str]: return { User-Agent: random.choice(USER_AGENTS), Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, Connection: keep-alive, Upgrade-Insecure-Requests: 1, } async def get(self, url: str, params: Optional[Dict] None) - str: 执行GET请求返回文本内容 try: async with self.session.get(url, paramsparams, timeoutREQUEST_TIMEOUT) as resp: if resp.status 200: # 自动检测编码部分网站为gbk content await resp.text(encodingutf-8, errorsignore) logger.info(f成功获取 {url} 状态码 {resp.status}) return content else: logger.warning(f请求失败 {url} 状态码 {resp.status}) return except asyncio.TimeoutError: logger.error(f请求超时 {url}) raise except Exception as e: logger.exception(f请求异常 {url}: {e}) raise5.3 重试机制retry.pypython# utils/retry.py from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import asyncio from loguru import logger def async_retry(max_attempts3): return retry( stopstop_after_attempt(max_attempts), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type((asyncio.TimeoutError, ConnectionError)), before_sleeplambda retry_state: logger.warning( f重试第 {retry_state.attempt_number} 次因 {retry_state.outcome.exception()} ) )5.4 动态渲染客户端dynamic_loader.py当检测到页面内容是通过JavaScript渲染时启动Playwright。python# crawler/dynamic_loader.py from playwright.async_api import async_playwright from loguru import logger import asyncio class DynamicLoader: staticmethod async def fetch(url: str, wait_selector: str body, timeout: int 30000) - str: 使用Playwright获取完全渲染后的HTML :param url: 目标URL :param wait_selector: 等待特定元素出现确保内容加载完成 :param timeout: 超时ms async with async_playwright() as p: # 使用无头模式可改为False用于调试 browser await p.chromium.launch(headlessTrue, args[--disable-blink-featuresAutomationControlled]) context await browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ) page await context.new_page() try: logger.info(f动态加载: {url}) await page.goto(url, wait_untilnetworkidle, timeouttimeout) await page.wait_for_selector(wait_selector, timeouttimeout) content await page.content() return content except Exception as e: logger.error(f动态渲染失败 {url}: {e}) return finally: await browser.close()5.5 数据解析器parser.py使用BeautifulSoup提取企业信息假设我们从搜索结果的详情页进行解析。python# crawler/parser.py from bs4 import BeautifulSoup from typing import Dict, Optional from loguru import logger class EnterpriseParser: staticmethod def parse_detail_page(html: str) - Dict[str, Optional[str]]: 解析企业详情页提取关键字段 注意不同公示系统结构不同这里展示通用模式 soup BeautifulSoup(html, lxml) result { company_name: None, credit_code: None, legal_rep: None, reg_capital: None, establish_date: None, status: None, company_type: None, scope: None } # 以下选择器仅为示例实际需根据目标网站结构调整 # 企业名称: 通常位于 h1 或 .company-name name_tag soup.select_one(h1.CompanyName, .company-name, .detail-title) if name_tag: result[company_name] name_tag.get_text(stripTrue) # 统一社会信用代码: 常出现在“基础信息”卡片 # 通过正则或文本查找更可靠 all_text soup.get_text() import re code_pattern r统一社会信用代码[:]\s*([A-Z0-9]{18}) match re.search(code_pattern, all_text) if match: result[credit_code] match.group(1) # 法定代表人 legal_pattern r法定代表人[:]\s*([\u4e00-\u9fa5]{2,4}) match re.search(legal_pattern, all_text) if match: result[legal_rep] match.group(1) # 注册资本: 注意可能有“注册资本1000万元人民币” capital_pattern r注册资本[:]\s*([\d.,])\s*万元 match re.search(capital_pattern, all_text) if match: result[reg_capital] match.group(1) 万元 # 成立日期: YYYY-MM-DD格式 date_pattern r成立日期[:]\s*(\d{4}-\d{1,2}-\d{1,2}) match re.search(date_pattern, all_text) if match: result[establish_date] match.group(1) # 登记状态 status_pattern r登记状态[:]\s*([存续在业吊销注销]) match re.search(status_pattern, all_text) if match: result[status] match.group(1) # 企业类型 type_pattern r企业类型[:]\s*([^。\n]) match re.search(type_pattern, all_text) if match: result[company_type] match.group(1).strip() # 经营范围截取前200字 scope_pattern r经营范围[:]\s*([\s\S]{50,500}) match re.search(scope_pattern, all_text) if match: scope_full match.group(1).strip() result[scope] scope_full[:200] ... if len(scope_full) 200 else scope_full logger.debug(f解析结果: {result[company_name]} - {result[credit_code]}) return result5.6 主爬虫逻辑main.py整合以上模块实现输入企业名称列表 - 异步并发请求 - 解析 - 导出。python# main.py import asyncio import pandas as pd from loguru import logger from crawler.http_client import AsyncHTTPClient from crawler.parser import EnterpriseParser from utils.retry import async_retry from config import SEARCH_URL, CONCURRENT_REQUESTS from typing import List, Dict import random import time class EnterpriseSpider: def __init__(self, company_names: List[str]): self.company_names company_names self.results [] self.semaphore asyncio.Semaphore(CONCURRENT_REQUESTS) async_retry(max_attempts2) async def fetch_company_detail(self, session: AsyncHTTPClient, company_name: str) - Dict: 根据企业名称获取详情页URL然后抓取详情 实际场景需要两步1. 搜索得到详情页链接 2. 请求详情页 这里简化为直接构造模拟请求教学演示 async with self.semaphore: # 模拟随机延迟避免单IP高并发 await asyncio.sleep(random.uniform(1, 3)) # 步骤1: 构造搜索请求以国家信用系统为例需要携带查询参数 # 由于该网站有严格反爬以下URL仅为示意实际无法直接运行 search_params {keyword: company_name} # 注意实际应使用官方合法API代替 detail_url fhttps://example.com/detail?name{company_name} # 步骤2: 获取详情页HTML先尝试普通请求若发现缺少动态内容则切换playwright html await session.get(detail_url) if not html or len(html) 500: # 可能是动态网站启用playwright from crawler.dynamic_loader import DynamicLoader logger.info(f检测到动态内容切换Playwright: {company_name}) html await DynamicLoader.fetch(detail_url, wait_selector.company-info, timeout20000) # 步骤3: 解析 if html: parsed EnterpriseParser.parse_detail_page(html) parsed[query_name] company_name return parsed else: logger.error(f无法获取 {company_name} 的详情页) return {} async def run(self): async with AsyncHTTPClient() as client: tasks [self.fetch_company_detail(client, name) for name in self.company_names] results await asyncio.gather(*tasks, return_exceptionsTrue) for res in results: if isinstance(res, dict) and res: self.results.append(res) elif isinstance(res, Exception): logger.error(f任务异常: {res}) def save_to_csv(self, filenamedata/enterprise_info.csv): if not self.results: logger.warning(没有数据可保存) return df pd.DataFrame(self.results) # 去重按统一信用代码 if credit_code in df.columns: df df.drop_duplicates(subset[credit_code], keepfirst) df.to_csv(filename, indexFalse, encodingutf-8-sig) logger.info(f成功保存 {len(df)} 条记录到 {filename}) # 入口函数 async def main(): # 示例企业列表仅用于测试合法性 companies [ 北京百度网讯科技有限公司, 深圳市腾讯计算机系统有限公司, 阿里巴巴中国有限公司 ] spider EnterpriseSpider(companies) await spider.run() spider.save_to_csv() if __name__ __main__: # 配置日志 from utils.logger import setup_logger setup_logger() try: asyncio.run(main()) except KeyboardInterrupt: logger.info(用户中断爬虫)5.7 日志配置logger.pypython# utils/logger.py import sys from loguru import logger def setup_logger(): logger.remove() # 移除默认handler logger.add( sys.stdout, formatgreen{time:HH:mm:ss}/green | level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan - level{message}/level, levelINFO ) logger.add( logs/crawler_{time:YYYY-MM-DD}.log, rotation500 MB, retention7 days, levelDEBUG, encodingutf-8 ) logger.info(日志系统初始化完成)六、反爬策略与避坑指南6.1 常见反爬机制及对策反爬措施应对方案IP频率限制使用异步随机延迟random.uniform(1,3) 代理IP池收费代理如Bright DataUser-Agent校验轮换UA池每次请求随机选择Cookie/Session保持会话模拟首次访问时的初始请求如访问首页获取cookie验证码放弃破解切换至官方API或手动打码服务不推荐大规模自动化前端动态渲染检测关键元素缺失时自动降级到Playwright字体反爬自定义字体映射数字使用OCR或字体文件逆向复杂且可能违法6.2 法律与伦理检查清单✅ 阅读目标网站的robots.txt✅ 设置请求头中标识爬虫身份如From: your-emaildomain.com✅ 控制请求频率不高于正常用户浏览速度建议1-2秒/次✅ 仅抓取完全公开且无需登录即可访问的信息✅ 不在商业项目中分发或转售抓取的数据七、性能优化与分布式扩展7.1 性能测试数据单机异步在普通VPS2核4G上使用asyncioaiohttp爬取1000个详情页假设每个页面响应200ms解析50ms并发数5总耗时 ≈1000 / 5 * 0.25 ≈ 50秒吞吐量 ≈ 20 pages/s受限于网络IO7.2 扩展到分布式使用Redis Queue或Celery分发任务代理池使用付费API。架构如下textMaster节点 (分发企业名称) - Redis (任务队列) - Worker节点 x N (执行异步抓取) - MongoDB/PostgreSQL存储结果7.3 数据增量更新策略企业信息会变更可记录last_crawl_time对活跃企业设置30天重新抓取一次。八、完整代码运行结果示例执行python main.py后控制台输出text15:32:01 | INFO | utils.logger:setup_logger - 日志系统初始化完成 15:32:01 | INFO | __main__:main - 开始爬取 3 家企业 15:32:02 | INFO | crawler.http_client:get - 成功获取 https://example.com/detail?name北京百度网讯科技有限公司 状态码 200 15:32:02 | DEBUG | crawler.parser:parse_detail_page - 解析结果: 北京百度网讯科技有限公司 - 91110000801000153J 15:32:04 | INFO | crawler.dynamic_loader:fetch - 动态加载: https://example.com/detail?name深圳市腾讯计算机系统有限公司 ... 15:32:12 | INFO | __main__:save_to_csv - 成功保存 3 条记录到 data/enterprise_info.csv生成的CSV文件内容示例company_namecredit_codelegal_repreg_capitalestablish_datestatuscompany_typescope北京百度网讯科技有限公司91110000801000153J李彦宏10000万元2001-06-05存续有限责任公司(自然人投资或控股)开发、生产计算机软件...九、常见问题与调试技巧Q1: 请求返回403 ForbiddenA:增加更真实的请求头Referer、Accept-Language并先请求首页获取cookie。尝试使用curl命令对比。Q2: Playwright启动很慢A:复用浏览器实例而不是每次请求都launch。可维护一个全局浏览器连接池。Q3: 解析时字段为空A:打印原始HTML片段检查页面结构是否变化。现代网站经常改版。Q4: 如何避免“未登录”限制A:很多公示系统不需要登录。如果需要登录建议走官方OAuth或放弃抓取。十、替代方案接入官方企业数据API推荐与其冒险爬虫不如使用合法接口国家信用系统开放数据通过https://api.qichacha.com或天眼查开放平台申请AppKey每月免费额度1000-5000次。Python调用示例以假设API为例pythonimport requests def get_company_by_api(credit_code): url https://open.api.tianyancha.com/services/open/ic/company headers {Authorization: Bearer YOUR_TOKEN} params {unique: credit_code} resp requests.get(url, headersheaders, paramsparams) return resp.json()
