从零构建企业公开数据爬虫：企查查/天眼查基础信息获取实战

一、写在前面：爬虫与法律的边界

在开始任何代码之前，我们必须严肃讨论一个话题：法律风险。

企查查、天眼查等平台虽然展示的是企业工商公开信息（如统一社会信用代码、法定代表人、注册资本、成立日期等），但这些平台通过自身的数据整合、清洗、呈现方式，形成了具有独立知识产权的数据产品。直接大规模、高频次爬取可能面临：

1. 民事侵权风险：违反平台用户协议，可能被以不正当竞争或侵害数据权益为由起诉。

2. 刑事责任红线：若绕过反爬机制（如破解验证码、伪造请求特征等），可能触犯《刑法》第285条“非法获取计算机信息系统数据罪”。

3. 技术封禁后果：IP被封、账号封禁、验证码升级甚至被运营商限速。

✅ 合法替代方案

• 优先使用官方数据源：国家企业信用信息公示系统（http://www.gsxt.gov.cn），该网站数据免费、公开、无需登录即可查询基础信息。

• 使用开放API：天眼查、企查查都提供官方企业信息查询API（一般有免费调用额度）。

• 本教程仅作为技术学习与交流，展示如何通过正常的HTTP请求获取完全公开、无需登录的企业数据，并严格遵守robots.txt及设置合理延时。请勿用于商业用途或对任何平台造成压力。

目录

一、写在前面：爬虫与法律的边界

✅ 合法替代方案

二、目标分析：我们要抓取什么？

三、技术选型（2026年最新栈）

四、环境搭建与项目结构

4.1 创建虚拟环境

4.2 安装依赖

4.3 项目目录结构

五、核心代码实现（逐块解析）

5.1 配置文件 config.py

5.2 异步HTTP客户端 http_client.py

5.3 重试机制 retry.py

5.4 动态渲染客户端 dynamic_loader.py

5.5 数据解析器 parser.py

5.6 主爬虫逻辑 main.py

5.7 日志配置 logger.py

六、反爬策略与避坑指南

6.1 常见反爬机制及对策

6.2 法律与伦理检查清单

七、性能优化与分布式扩展

7.1 性能测试数据（单机异步）

7.2 扩展到分布式

7.3 数据增量更新策略

八、完整代码运行结果示例

九、常见问题与调试技巧

Q1: 请求返回 403 Forbidden

Q2: Playwright启动很慢

Q3: 解析时字段为空

Q4: 如何避免“未登录”限制？

十、替代方案：接入官方企业数据API（推荐）

二、目标分析：我们要抓取什么？

以“国家企业信用信息公示系统”作为练习目标（合规安全），我们需要提取以下字段：

三、技术选型（2026年最新栈）

采用异步+动态渲染混合策略，因为部分现代企业公示系统使用Vue/React前端渲染。

💡 放弃Requests+Selenium旧组合，采用性能更好、资源更省的异步+无头浏览器按需启动。

四、环境搭建与项目结构

4.1 创建虚拟环境

bash

python -m venv enterprise_crawler source enterprise_crawler/bin/activate # Windows: enterprise_crawler\Scripts\activate

4.2 安装依赖

bash

pip install aiohttp beautifulsoup4 playwright pandas loguru tenacity playwright install chromium # 仅备用，默认优先用requests

4.3 项目目录结构

text

enterprise_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.py

python

# 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/..." # 这里替换为你有权调用的合法API

5.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( limit=10, # 总连接数限制 limit_per_host=5, # 单主机并发 ttl_dns_cache=300, ssl=False # 仅测试用，生产应开启SSL验证 ) async def __aenter__(self): self.session = aiohttp.ClientSession( connector=self._connector, headers=self._get_headers(), cookie_jar=aiohttp.CookieJar(unsafe=True) ) 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/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.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, params=params, timeout=REQUEST_TIMEOUT) as resp: if resp.status == 200: # 自动检测编码，部分网站为gbk content = await resp.text(encoding='utf-8', errors='ignore') 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}") raise

5.3 重试机制retry.py

python

# 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_attempts=3): return retry( stop=stop_after_attempt(max_attempts), wait=wait_exponential(multiplier=1, min=2, max=10), retry=retry_if_exception_type((asyncio.TimeoutError, ConnectionError)), before_sleep=lambda 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(headless=True, args=['--disable-blink-features=AutomationControlled']) context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent='Mozilla/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_until="networkidle", timeout=timeout) await page.wait_for_selector(wait_selector, timeout=timeout) 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(strip=True) # 统一社会信用代码: 常出现在“基础信息”卡片 # 通过正则或文本查找更可靠 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 result

5.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_attempts=2) 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 = f"https://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", timeout=20000) # 步骤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_exceptions=True) 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, filename="data/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'], keep='first') df.to_csv(filename, index=False, encoding='utf-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.py

python

# utils/logger.py import sys from loguru import logger def setup_logger(): logger.remove() # 移除默认handler logger.add( sys.stdout, format="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan> - <level>{message}</level>", level="INFO" ) logger.add( "logs/crawler_{time:YYYY-MM-DD}.log", rotation="500 MB", retention="7 days", level="DEBUG", encoding="utf-8" ) logger.info("日志系统初始化完成")

六、反爬策略与避坑指南

6.1 常见反爬机制及对策

6.2 法律与伦理检查清单

✅ 阅读目标网站的robots.txt
✅ 设置请求头中标识爬虫身份（如From: your-email@domain.com）
✅ 控制请求频率，不高于正常用户浏览速度（建议1-2秒/次）
✅ 仅抓取完全公开且无需登录即可访问的信息
✅ 不在商业项目中分发或转售抓取的数据

七、性能优化与分布式扩展

7.1 性能测试数据（单机异步）

在普通VPS（2核4G）上，使用asyncio+aiohttp爬取1000个详情页（假设每个页面响应200ms+解析50ms），并发数5：

• 总耗时 ≈1000 / 5 * 0.25 ≈ 50秒

• 吞吐量 ≈ 20 pages/s（受限于网络IO）

7.2 扩展到分布式

使用Redis Queue或Celery分发任务，代理池使用付费API。架构如下：

text

Master节点 (分发企业名称) -> Redis (任务队列) -> Worker节点 x N (执行异步抓取) -> MongoDB/PostgreSQL存储结果

7.3 数据增量更新策略

企业信息会变更，可记录last_crawl_time，对活跃企业设置30天重新抓取一次。

八、完整代码运行结果示例

执行python main.py后，控制台输出：

text

15: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文件内容示例：

九、常见问题与调试技巧

Q1: 请求返回403 Forbidden

A:增加更真实的请求头（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为例）：

python

import 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, headers=headers, params=params) return resp.json()