Python爬虫实战：基于Playwright的企业数据采集与反爬策略

1. 项目概述与核心价值

最近在做一个市场分析的项目，需要批量获取一批目标公司的工商信息、股东背景和经营状况。手动去天眼查、企查查这类网站一个个搜，效率低不说，还容易出错。于是，用Python写一个爬虫来自动化这个数据采集过程，就成了一个很自然的选择。这不仅仅是写几行代码那么简单，它背后涉及到反爬策略对抗、数据结构化清洗以及数据应用的完整链条。对于从事数据分析、投资研究、市场调研甚至风控建模的朋友来说，掌握这套方法，能直接从公开信息中挖掘出高价值的商业情报，把人力从重复劳动中解放出来，把精力聚焦在更有价值的分析决策上。

天眼查作为国内领先的商业查询平台，汇聚了海量企业的工商、司法、知识产权等多维度信息。通过爬虫获取这些数据，我们可以进行竞争对手监控、供应链分析、投资标的筛选、风险预警等多种深度分析。但必须清醒认识到，这个过程技术挑战和合规风险并存。技术上，你需要面对动态加载、加密参数、频繁变动的验证机制；合规上，你必须严格遵守robots.txt协议，控制请求频率，仅将数据用于个人学习或合法的分析研究，绝对禁止用于商业售卖、恶意骚扰等非法用途。接下来，我就把自己在实现这个爬虫过程中趟过的路、踩过的坑，以及最终稳定运行的方案，毫无保留地分享出来。

2. 核心思路与关键技术选型

2.1 整体架构设计

一个稳健的企业数据爬虫，不能是简单的requests.get加BeautifulSoup解析。面对天眼查这类防护严密的站点，我们需要一个更系统化的架构。我的核心思路是“模拟真人，层层递进，优雅处理”。

首先，爬虫的核心任务是模拟一个真实用户在浏览器上的操作流程：打开网页、输入搜索词、查看列表、点击进入详情页、提取信息。因此，我们的技术栈必须能完美模拟这一过程。我放弃了传统的requests+BeautifulSoup组合，因为天眼查的大量关键数据是通过JavaScript动态渲染的，直接拿到的HTML是空的。解决方案是使用能执行JavaScript的浏览器自动化工具。

其次，反爬措施是最大的拦路虎。常见的包括：IP封锁、请求头校验、行为验证码、参数加密、接口令牌等。我们的架构必须内置应对这些措施的模块，例如代理IP池、请求头随机化、复杂验证码的识别或绕过策略，以及关键请求参数的逆向工程。

基于以上考虑，我设计的爬虫架构主要包含以下几个模块：

1. 请求与渲染模块：负责加载网页并获取完整HTML。选用Selenium或Playwright驱动无头浏览器。
2. 反爬对抗模块：集成代理IP管理、请求头管理、Cookie持久化、请求频率控制等。
3. 数据解析模块：从渲染后的HTML中，使用XPath或CSS Selector精准定位并提取目标数据。
4. 数据存储模块：将提取的结构化数据保存到文件（如CSV、JSON）或数据库（如SQLite、MySQL）中。
5. 任务调度与监控模块：管理待爬队列、处理异常、记录日志，确保长时间稳定运行。

2.2 关键工具选型解析

• 浏览器自动化工具：Playwright vs Selenium两者都能驱动浏览器。Selenium更老牌，生态成熟。但我最终选择了Playwright，原因有三：一是Playwright的API更现代、简洁，自动等待机制更智能，减少了大量time.sleep的硬编码；二是其性能通常优于Selenium，启动更快；三是它对现代Web技术（如单页应用SPA）的支持更好，内置了拦截网络请求、模拟移动设备等强大功能，这对于分析动态加载的数据接口非常有用。

• 解析库：lxml + parselBeautifulSoup简单易用，但解析速度较慢。对于需要处理成千上万个页面的爬虫，效率至关重要。lxml是一个用C语言编写的解析库，速度极快。parsel库（Scrapy框架的解析组件）构建在lxml之上，提供了更优雅的CSS和XPath选择器语法，错误提示也更友好。因此，我采用parsel进行数据提取。

• 代理IP服务单IP高频请求很快会被封。必须使用代理IP池。可以选择付费的代理服务商（提供高匿、稳定的代理），也可以自建代理池（通过爬取免费代理网站，但需要花费大量精力维护其可用性）。对于企业级数据采集，建议使用付费服务，省心稳定。

• 验证码处理遇到验证码是常态。简单图形验证码可以使用ddddocr、tesseract等OCR库尝试识别。但天眼查的验证码越来越复杂，滑动拼图、点选汉字等行为验证码很难通过纯代码完美破解。更务实的策略是：一、通过降低请求频率、维护会话状态来尽量避免触发验证码；二、当被要求验证时，程序自动暂停并发出告警，转为人工干预处理。切勿尝试破解涉及严重安全风险的验证码系统。

• 数据存储小规模数据（几千家企业）用CSV或JSON文件足够了。如果数据量大、需要关联查询或增量更新，建议使用数据库。SQLite轻量便携，适合桌面应用；MySQL或PostgreSQL更适合服务器端部署。我这里以SQLite为例，方便演示和移植。

注意：合规性提醒在开始编写任何爬虫代码前，请务必仔细阅读天眼查网站的robots.txt文件（通常在网站根目录，如https://www.tianyancha.com/robots.txt），并严格遵守其中关于爬虫抓取频率和禁止抓取目录的规定。我们的代码应设置合理的请求延迟（如每次请求间隔3-5秒以上），模拟人类浏览行为，避免对目标网站服务器造成压力。

3. 环境准备与核心代码实现

3.1 基础环境搭建

首先，我们需要安装必要的Python库。建议使用虚拟环境（如venv或conda）来管理依赖，避免污染全局环境。

# 创建并激活虚拟环境（以venv为例） python -m venv tianyancha_spider source tianyancha_spider/bin/activate # Linux/Mac # tianyancha_spider\Scripts\activate # Windows # 安装核心库 pip install playwright lxml parsel pandas # 安装Playwright的浏览器驱动（Chromium, Firefox, WebKit） playwright install chromium

pandas库用于后续的数据处理和保存为CSV。playwright安装后会下载对应的浏览器。

3.2 爬虫核心类设计与实现

我将爬虫封装成一个类，这样结构更清晰，也便于维护和扩展。

import asyncio import random import time import pandas as pd from datetime import datetime from urllib.parse import quote from playwright.async_api import async_playwright from parsel import Selector import sqlite3 import logging # 配置日志，方便调试和监控 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class TianyanchaSpider: def __init__(self, use_proxy=False, proxy_list=None): """ 初始化爬虫 :param use_proxy: 是否使用代理 :param proxy_list: 代理列表，格式如 ['http://user:pass@ip:port', ...] """ self.use_proxy = use_proxy self.proxy_list = proxy_list or [] self.data_list = [] # 临时存储爬取的数据 # 模拟真实浏览器的请求头 self.headers = { '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-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', } self.base_url = 'https://www.tianyancha.com' async def init_browser(self): """初始化Playwright浏览器上下文，配置代理和用户数据目录以持久化Cookie""" self.playwright = await async_playwright().start() launch_options = { 'headless': True, # 无头模式，不显示浏览器界面 'args': ['--disable-blink-features=AutomationControlled', '--no-sandbox'] } if self.use_proxy and self.proxy_list: proxy = random.choice(self.proxy_list) launch_options['proxy'] = {'server': proxy} logger.info(f"使用代理: {proxy}") self.browser = await self.playwright.chromium.launch(**launch_options) # 使用持久化上下文，可以保存登录状态和Cookie，避免每次重新登录 self.context = await self.browser.new_context( user_agent=self.headers['User-Agent'], viewport={'width': 1920, 'height': 1080} ) self.page = await self.context.new_page() # 设置默认超时和导航超时 self.page.set_default_timeout(60000) self.page.set_default_navigation_timeout(60000) async def search_company(self, keyword): """搜索公司，并返回搜索结果页面的第一页公司链接列表""" search_url = f"{self.base_url}/search?key={quote(keyword)}" logger.info(f"正在搜索: {keyword}") try: await self.page.goto(search_url, wait_until='networkidle') # 等待网络空闲 await self.page.wait_for_timeout(random.uniform(2000, 4000)) # 随机等待，模拟人工 # 检查是否有验证码或访问限制 if await self.page.locator('text=验证码').count() > 0: logger.error("触发验证码，请手动处理或更换IP。") # 这里可以添加告警逻辑，如发送邮件或钉钉消息 return [] # 获取页面内容并用Parsel解析 html = await self.page.content() selector = Selector(text=html) # 使用XPath提取搜索结果中的公司链接 # 注意：天眼查的HTML结构可能变化，此XPath需要根据实际情况调整 company_links = selector.xpath('//a[contains(@class, "index_alink")]/@href').getall() # 过滤出有效的公司详情页链接 company_links = [link for link in company_links if link.startswith('/company/')] # 补全为完整URL full_links = [self.base_url + link for link in company_links[:10]] # 只取前10个结果 logger.info(f"找到 {len(full_links)} 个公司链接") return full_links except Exception as e: logger.error(f"搜索公司 {keyword} 时出错: {e}") return [] async def parse_company_detail(self, url): """解析公司详情页，提取关键信息""" logger.info(f"正在解析: {url}") try: await self.page.goto(url, wait_until='domcontentloaded') # 等待关键信息区域加载 await self.page.wait_for_selector('div.company-header', timeout=10000) await self.page.wait_for_timeout(random.uniform(3000, 5000)) html = await self.page.content() selector = Selector(text=html) company_info = {} # 1. 公司名称 company_info['公司名称'] = selector.xpath('//h1[contains(@class, "name")]/text()').get('').strip() # 2. 统一社会信用代码/注册号 - 通常在一个表格里 # 这里演示一种更健壮的提取方式：先找到包含“统一社会信用代码”的文本节点，再找其兄弟节点或父节点的值 credit_code_label = selector.xpath('//table//td[contains(text(), "统一社会信用代码") or contains(text(), "注册号")]') if credit_code_label: # 假设代码在相邻的td里 company_info['统一社会信用代码'] = credit_code_label.xpath('./following-sibling::td[1]/text()').get('').strip() else: company_info['统一社会信用代码'] = '' # 3. 法定代表人 legal_rep = selector.xpath('//a[contains(@href, "/human/")]/text()').get() company_info['法定代表人'] = legal_rep.strip() if legal_rep else '' # 4. 注册资本 reg_capital = selector.xpath('//div[contains(text(), "注册资本")]/following-sibling::div[1]/text()').get() company_info['注册资本'] = reg_capital.strip() if reg_capital else '' # 5. 成立日期 establish_date = selector.xpath('//div[contains(text(), "成立日期")]/following-sibling::div[1]/text()').get() company_info['成立日期'] = establish_date.strip() if establish_date else '' # 6. 电话和邮箱（可能被保护，需要点击查看） phone = selector.xpath('//div[contains(@class, "contact-sec")]//span[contains(@class, "phone")]/text()').get() company_info['电话'] = phone.strip() if phone else '' email = selector.xpath('//div[contains(@class, "contact-sec")]//span[contains(@class, "email")]/text()').get() company_info['邮箱'] = email.strip() if email else '' # 7. 地址 address = selector.xpath('//div[contains(@class, "address")]/text()').get() company_info['地址'] = address.strip() if address else '' # 8. 经营范围（可能较长） business_scope = selector.xpath('//div[contains(@class, "business-scope")]//span/text()').get() company_info['经营范围'] = business_scope.strip() if business_scope else '' # 9. 公司状态（在营、注销、吊销等） company_status = selector.xpath('//div[contains(@class, "company-status")]/span/text()').get() company_info['公司状态'] = company_status.strip() if company_status else '' logger.info(f"成功解析: {company_info['公司名称']}") return company_info except Exception as e: logger.error(f"解析页面 {url} 时出错: {e}") return None async def run(self, keywords, output_file='companies.csv'): """主运行函数""" await self.init_browser() all_data = [] try: for keyword in keywords: company_links = await self.search_company(keyword) for link in company_links: info = await self.parse_company_detail(link) if info: info['搜索关键词'] = keyword info['爬取时间'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') all_data.append(info) # 关键：随机延迟，避免请求过快 await asyncio.sleep(random.uniform(5, 10)) # 搜索不同关键词之间也延迟 await asyncio.sleep(random.uniform(8, 15)) finally: # 无论是否出错，最后都要关闭浏览器 await self.browser.close() await self.playwright.stop() # 保存数据到CSV if all_data: df = pd.DataFrame(all_data) df.to_csv(output_file, index=False, encoding='utf-8-sig') # utf-8-sig解决Excel打开中文乱码 logger.info(f"数据已保存至 {output_file}, 共 {len(all_data)} 条记录。") else: logger.warning("未爬取到任何数据。") # 异步主函数 async def main(): spider = TianyanchaSpider( use_proxy=False, # 根据实际情况开启 # proxy_list=['http://your_proxy:port'] ) # 要搜索的公司关键词列表 search_keywords = ['科技有限公司', '信息技术有限公司'] await spider.run(search_keywords, 'tianyancha_companies.csv') if __name__ == '__main__': asyncio.run(main())

3.3 代码关键点解析与避坑指南

1. 异步编程：Playwright的API是异步的，使用async/await能显著提升I/O密集型爬虫的效率。主函数需要用asyncio.run()来驱动。

2. 等待策略：page.goto()的wait_until参数至关重要。'networkidle'表示等待网络基本没有活动时再继续，适合静态页面。对于动态加载多的页面，'domcontentloaded'（DOM加载完成）可能更快，但需要后续用wait_for_selector等待特定元素出现。绝对避免无脑使用time.sleep，要用智能等待。

3. XPath定位技巧：天眼查的页面结构复杂且可能变动。编写XPath时：

   • 避免使用绝对路径（如/html/body/div[3]/div[2]/...），一个div顺序变化就会导致失效。
   • 多用属性匹配（contains(@class, ‘xxx’)）、文本匹配（contains(text(), ‘xxx’)）和轴选择（following-sibling::,parent::）来相对定位，这样更健壮。
   • 在浏览器开发者工具中测试：按F12，在Elements面板选中元素，右键Copy->Copy XPath或Copy full XPath可以作为起点，但通常需要简化。

4. 代理IP的使用：代码中预留了代理接口。如果启用，务必确保代理IP是高匿名的，并且有足够的稳定性和速度。免费代理大多不可用，生产环境建议购买可靠的服务。

5. 异常处理与日志：爬虫运行中会遇到各种意外：网络超时、元素未找到、验证码弹出等。完善的try...except和日志记录是调试和监控的基石。日志要记录时间、级别和信息，方便回溯问题。

6. 数据清洗：爬取下来的文本常常包含多余的空格、换行符、不可见字符。在保存前，一定要用.strip()等方法进行清洗。对于“注册资本”这类字段，可能包含“100万元人民币”，后续分析时可能需要将其拆分为数值和单位。

4. 高级策略与稳定性优化

4.1 应对反爬的进阶手段

基础的随机延迟和更换User-Agent只是第一道防线。更高级的反爬策略需要我们更深入地模拟人类。

• Cookie与会话保持：上述代码中，我们使用了browser.new_context()但没有指定storage_state。更优的做法是，先手动用浏览器登录一次天眼查（如果需要登录才能看更多信息），然后通过context.storage_state(path=’state.json’)保存登录状态。下次启动时用browser.new_context(storage_state=’state.json’)恢复，可以维持登录态，避免反复触发登录验证。

• 指纹伪装：现代网站可以通过浏览器指纹（如Canvas、WebGL、字体、屏幕分辨率等）来追踪和识别爬虫。Playwright可以通过context.add_init_script()注入JavaScript来修改或标准化这些指纹特征，增加隐蔽性。

• 模拟鼠标移动和滚动：在关键操作（如点击、翻页）前，让鼠标在页面随机移动一下，并随机滚动页面，可以更好地模拟真人行为。Playwright提供了page.mouse.move()和page.evaluate(‘window.scrollBy(0, 100)’)等方法。

• 分布式爬取：对于海量数据，单机单IP能力有限。可以考虑使用分布式架构，如Scrapy-Redis或Celery，配合多个爬虫节点和庞大的代理IP池同时工作，由中央调度器分配任务。

4.2 数据存储的扩展：使用SQLite数据库

对于需要长期积累、查询或关联的数据，CSV文件显得力不从心。下面演示如何将数据存入SQLite。

import sqlite3 class DataManager: def __init__(self, db_path='tianyancha.db'): self.conn = sqlite3.connect(db_path) self.create_table() def create_table(self): """创建公司信息表""" create_table_sql = ''' CREATE TABLE IF NOT EXISTS companies ( id INTEGER PRIMARY KEY AUTOINCREMENT, 公司名称 TEXT NOT NULL, 统一社会信用代码 TEXT, 法定代表人 TEXT, 注册资本 TEXT, 成立日期 TEXT, 电话 TEXT, 邮箱 TEXT, 地址 TEXT, 经营范围 TEXT, 公司状态 TEXT, 搜索关键词 TEXT, 爬取时间 TIMESTAMP, UNIQUE(公司名称, 统一社会信用代码) -- 防止重复插入 ) ''' self.conn.execute(create_table_sql) self.conn.commit() def insert_company(self, company_info): """插入或更新公司信息""" # 使用 INSERT OR REPLACE 来处理重复（基于唯一约束） sql = ''' INSERT OR REPLACE INTO companies (公司名称, 统一社会信用代码, 法定代表人, 注册资本, 成立日期, 电话, 邮箱, 地址, 经营范围, 公司状态, 搜索关键词, 爬取时间) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''' values = ( company_info.get('公司名称', ''), company_info.get('统一社会信用代码', ''), company_info.get('法定代表人', ''), company_info.get('注册资本', ''), company_info.get('成立日期', ''), company_info.get('电话', ''), company_info.get('邮箱', ''), company_info.get('地址', ''), company_info.get('经营范围', ''), company_info.get('公司状态', ''), company_info.get('搜索关键词', ''), company_info.get('爬取时间', ''), ) try: self.conn.execute(sql, values) self.conn.commit() logger.info(f"数据入库成功: {company_info['公司名称']}") except sqlite3.Error as e: logger.error(f"数据入库失败 {company_info['公司名称']}: {e}") def close(self): self.conn.close() # 在爬虫类中集成 class TianyanchaSpiderWithDB(TianyanchaSpider): def __init__(self, db_manager, **kwargs): super().__init__(**kwargs) self.db_manager = db_manager async def run(self, keywords): # ... 爬取逻辑与父类相同 ... # 在解析到info后，不再存入self.data_list，而是直接入库 info = await self.parse_company_detail(link) if info: info['搜索关键词'] = keyword info['爬取时间'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') self.db_manager.insert_company(info) # 存入数据库

使用数据库的好处是显而易见的：便于去重、支持复杂的查询（例如“查找所有注册资本大于1000万且成立于2020年后的科技公司”）、以及增量更新（只爬取上次之后有变动的信息）。

4.3 绕过动态加载与接口直击

有时，页面上的数据是通过Ajax请求动态加载的，在HTML源码中找不到。这时可以打开浏览器的开发者工具（F12），切换到Network（网络）选项卡，刷新页面或进行交互，观察出现的XHR或Fetch请求。找到返回目标数据的API接口，直接模拟这个请求往往更高效。

例如，天眼查的公司高管列表可能是一个单独的接口。你可以复制这个请求的cURL命令，在Python中用requests库模拟，但需要复制所有的请求头（特别是Authorization、Cookie、X-Token等）和请求参数。这种方式速度快、负载小，但缺点是接口一旦变化，爬虫就需要调整，且参数可能被加密。

如何选择？对于初学者或页面结构不常变的情况，用浏览器自动化（如Playwright）更简单直接，虽然慢但稳定。对于高手或需要极高性能的场景，分析并模拟API请求是终极方案，但需要一定的逆向工程能力。

5. 常见问题排查与实战心得

5.1 高频问题速查表

5.2 实战心得与避坑指南

1. “慢就是快”：在爬虫世界里，急于求成往往导致IP被封、前功尽弃。把请求间隔设置得足够长（比如5-10秒甚至更长），随机化这个间隔，是保证长期稳定运行的最低成本策略。一个能跑一周的慢爬虫，远比一个跑一小时就被封的快爬虫有用。

2. 选择器的维护成本：不要写“一次性”的选择器。尽量使用那些基于语义化、相对稳定的属性来定位元素，比如包含特定文本的标签、具有特定数据属性的元素。定期（例如每周）运行一下爬虫的测试用例，确保核心选择器依然有效。

3. 数据清洗宜早不宜迟：在数据解析的环节，就尽量将其清洗成规整的格式。例如，将“注册资本：1000万元人民币”拆分为数值1000和单位万元，将日期字符串转换为datetime对象。这样存入数据库或文件后，后续分析会非常方便。

4. 做好增量与断点续爬：爬虫可能因为网络、反爬等原因中断。设计时就应该考虑记录爬取进度。可以将成功爬取的URL存入一个“已爬”集合（可以用Redis或数据库），每次启动时先加载这个集合，避免重复爬取。对于列表页，记录上次爬取到的页码或最后一条数据的ID。

5. 法律与道德底线：再次强调，爬取的数据只能用于个人学习、科学研究或合法的内部分析。未经许可，大规模爬取数据用于商业竞争、售卖或个人隐私骚扰，不仅是违反网站服务条款的行为，更可能触犯法律。尊重robots.txt，控制爬取速度，是你作为技术人的基本素养。

这个爬虫项目从简单的需求出发，逐步深入到反爬对抗、效率优化和系统设计，是一个非常好的Python综合实践。它考验的不仅仅是编码能力，更是工程思维、问题解决能力和对网络协议的理解。希望这份详细的拆解和代码，能为你打开网络数据获取的大门，同时也能让你清醒地认识到其中的边界与责任。在实际操作中，多测试、多观察、多思考，你会积累下属于自己的宝贵经验。