1. 项目概述与核心价值最近在做一个市场分析的项目需要批量获取一批目标公司的工商信息、股东背景和经营状况。手动去天眼查、企查查这类网站一个个搜效率低不说还容易出错。于是用Python写一个爬虫来自动化这个数据采集过程就成了一个很自然的选择。这不仅仅是写几行代码那么简单它背后涉及到反爬策略对抗、数据结构化清洗以及数据应用的完整链条。对于从事数据分析、投资研究、市场调研甚至风控建模的朋友来说掌握这套方法能直接从公开信息中挖掘出高价值的商业情报把人力从重复劳动中解放出来把精力聚焦在更有价值的分析决策上。天眼查作为国内领先的商业查询平台汇聚了海量企业的工商、司法、知识产权等多维度信息。通过爬虫获取这些数据我们可以进行竞争对手监控、供应链分析、投资标的筛选、风险预警等多种深度分析。但必须清醒认识到这个过程技术挑战和合规风险并存。技术上你需要面对动态加载、加密参数、频繁变动的验证机制合规上你必须严格遵守robots.txt协议控制请求频率仅将数据用于个人学习或合法的分析研究绝对禁止用于商业售卖、恶意骚扰等非法用途。接下来我就把自己在实现这个爬虫过程中趟过的路、踩过的坑以及最终稳定运行的方案毫无保留地分享出来。2. 核心思路与关键技术选型2.1 整体架构设计一个稳健的企业数据爬虫不能是简单的requests.get加BeautifulSoup解析。面对天眼查这类防护严密的站点我们需要一个更系统化的架构。我的核心思路是“模拟真人层层递进优雅处理”。首先爬虫的核心任务是模拟一个真实用户在浏览器上的操作流程打开网页、输入搜索词、查看列表、点击进入详情页、提取信息。因此我们的技术栈必须能完美模拟这一过程。我放弃了传统的requestsBeautifulSoup组合因为天眼查的大量关键数据是通过JavaScript动态渲染的直接拿到的HTML是空的。解决方案是使用能执行JavaScript的浏览器自动化工具。其次反爬措施是最大的拦路虎。常见的包括IP封锁、请求头校验、行为验证码、参数加密、接口令牌等。我们的架构必须内置应对这些措施的模块例如代理IP池、请求头随机化、复杂验证码的识别或绕过策略以及关键请求参数的逆向工程。基于以上考虑我设计的爬虫架构主要包含以下几个模块请求与渲染模块负责加载网页并获取完整HTML。选用Selenium或Playwright驱动无头浏览器。反爬对抗模块集成代理IP管理、请求头管理、Cookie持久化、请求频率控制等。数据解析模块从渲染后的HTML中使用XPath或CSS Selector精准定位并提取目标数据。数据存储模块将提取的结构化数据保存到文件如CSV、JSON或数据库如SQLite、MySQL中。任务调度与监控模块管理待爬队列、处理异常、记录日志确保长时间稳定运行。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 chromiumpandas库用于后续的数据处理和保存为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(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class TianyanchaSpider: def __init__(self, use_proxyFalse, proxy_listNone): 初始化爬虫 :param use_proxy: 是否使用代理 :param proxy_list: 代理列表格式如 [http://user:passip: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;q0.9,en;q0.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-featuresAutomationControlled, --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_agentself.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_untilnetworkidle) # 等待网络空闲 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(texthtml) # 使用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_untildomcontentloaded) # 等待关键信息区域加载 await self.page.wait_for_selector(div.company-header, timeout10000) await self.page.wait_for_timeout(random.uniform(3000, 5000)) html await self.page.content() selector Selector(texthtml) 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_filecompanies.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, indexFalse, encodingutf-8-sig) # utf-8-sig解决Excel打开中文乱码 logger.info(f数据已保存至 {output_file}, 共 {len(all_data)} 条记录。) else: logger.warning(未爬取到任何数据。) # 异步主函数 async def main(): spider TianyanchaSpider( use_proxyFalse, # 根据实际情况开启 # proxy_list[http://your_proxy:port] ) # 要搜索的公司关键词列表 search_keywords [科技有限公司, 信息技术有限公司] await spider.run(search_keywords, tianyancha_companies.csv) if __name__ __main__: asyncio.run(main())3.3 代码关键点解析与避坑指南异步编程Playwright的API是异步的使用async/await能显著提升I/O密集型爬虫的效率。主函数需要用asyncio.run()来驱动。等待策略page.goto()的wait_until参数至关重要。networkidle表示等待网络基本没有活动时再继续适合静态页面。对于动态加载多的页面domcontentloadedDOM加载完成可能更快但需要后续用wait_for_selector等待特定元素出现。绝对避免无脑使用time.sleep要用智能等待。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可以作为起点但通常需要简化。代理IP的使用代码中预留了代理接口。如果启用务必确保代理IP是高匿名的并且有足够的稳定性和速度。免费代理大多不可用生产环境建议购买可靠的服务。异常处理与日志爬虫运行中会遇到各种意外网络超时、元素未找到、验证码弹出等。完善的try...except和日志记录是调试和监控的基石。日志要记录时间、级别和信息方便回溯问题。数据清洗爬取下来的文本常常包含多余的空格、换行符、不可见字符。在保存前一定要用.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_pathtianyancha.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 高频问题速查表问题现象可能原因排查与解决思路页面加载空白或元素找不到1. 页面未完全加载。2. 网站结构已更新XPath/CSS选择器失效。3. 触发反爬被重定向到验证页。1. 增加等待时间使用wait_for_selector。2. 重新检查并更新选择器路径。3. 检查当前页面URL和内容确认是否被拦截。增加延迟、更换IP/User-Agent。弹出验证码图形、滑块等请求频率过高或行为被识别为非人类。1.首要策略大幅降低请求频率增加随机延迟。2. 维护会话Cookie避免每次请求都是新会话。3. 考虑使用更高质量的代理IP住宅IP。4. 考虑引入第三方打码平台成本与合规性需权衡。返回403 Forbidden或429 Too Many RequestsIP或请求头被识别访问被拒绝或限速。1. 检查并随机化请求头User-Agent, Accept-Language等。2.必须使用代理IP池并确保代理可用。3. 严格遵守robots.txt设置更长的请求间隔。爬取速度非常慢1. 使用无头浏览器本身开销大。2. 网络延迟或代理IP速度慢。3. 等待策略设置过于保守。1. 考虑切换到接口请求模式如果可行。2. 测试代理IP的速度更换优质代理。3. 优化等待条件将networkidle改为domcontentloaded并精准等待必要元素。数据字段错乱或为空解析规则不准确定位到了错误的HTML元素。1. 将爬虫解析到的HTML片段print(selector.get())保存下来与实际浏览器查看的源码对比。2. 使用更精确、更具弹性的XPath或CSS选择器避免依赖不稳定的class或id。异步任务报错Event loop is closed异步事件循环管理不当常见于Windows系统或脚本提前退出。确保主入口使用asyncio.run(main())。如果程序中有多个异步循环确保正确关闭。在Jupyter等环境中运行时需注意环境差异。5.2 实战心得与避坑指南“慢就是快”在爬虫世界里急于求成往往导致IP被封、前功尽弃。把请求间隔设置得足够长比如5-10秒甚至更长随机化这个间隔是保证长期稳定运行的最低成本策略。一个能跑一周的慢爬虫远比一个跑一小时就被封的快爬虫有用。选择器的维护成本不要写“一次性”的选择器。尽量使用那些基于语义化、相对稳定的属性来定位元素比如包含特定文本的标签、具有特定数据属性的元素。定期例如每周运行一下爬虫的测试用例确保核心选择器依然有效。数据清洗宜早不宜迟在数据解析的环节就尽量将其清洗成规整的格式。例如将“注册资本1000万元人民币”拆分为数值1000和单位万元将日期字符串转换为datetime对象。这样存入数据库或文件后后续分析会非常方便。做好增量与断点续爬爬虫可能因为网络、反爬等原因中断。设计时就应该考虑记录爬取进度。可以将成功爬取的URL存入一个“已爬”集合可以用Redis或数据库每次启动时先加载这个集合避免重复爬取。对于列表页记录上次爬取到的页码或最后一条数据的ID。法律与道德底线再次强调爬取的数据只能用于个人学习、科学研究或合法的内部分析。未经许可大规模爬取数据用于商业竞争、售卖或个人隐私骚扰不仅是违反网站服务条款的行为更可能触犯法律。尊重robots.txt控制爬取速度是你作为技术人的基本素养。这个爬虫项目从简单的需求出发逐步深入到反爬对抗、效率优化和系统设计是一个非常好的Python综合实践。它考验的不仅仅是编码能力更是工程思维、问题解决能力和对网络协议的理解。希望这份详细的拆解和代码能为你打开网络数据获取的大门同时也能让你清醒地认识到其中的边界与责任。在实际操作中多测试、多观察、多思考你会积累下属于自己的宝贵经验。
Python爬虫实战：基于Playwright的企业数据采集与反爬策略
1. 项目概述与核心价值最近在做一个市场分析的项目需要批量获取一批目标公司的工商信息、股东背景和经营状况。手动去天眼查、企查查这类网站一个个搜效率低不说还容易出错。于是用Python写一个爬虫来自动化这个数据采集过程就成了一个很自然的选择。这不仅仅是写几行代码那么简单它背后涉及到反爬策略对抗、数据结构化清洗以及数据应用的完整链条。对于从事数据分析、投资研究、市场调研甚至风控建模的朋友来说掌握这套方法能直接从公开信息中挖掘出高价值的商业情报把人力从重复劳动中解放出来把精力聚焦在更有价值的分析决策上。天眼查作为国内领先的商业查询平台汇聚了海量企业的工商、司法、知识产权等多维度信息。通过爬虫获取这些数据我们可以进行竞争对手监控、供应链分析、投资标的筛选、风险预警等多种深度分析。但必须清醒认识到这个过程技术挑战和合规风险并存。技术上你需要面对动态加载、加密参数、频繁变动的验证机制合规上你必须严格遵守robots.txt协议控制请求频率仅将数据用于个人学习或合法的分析研究绝对禁止用于商业售卖、恶意骚扰等非法用途。接下来我就把自己在实现这个爬虫过程中趟过的路、踩过的坑以及最终稳定运行的方案毫无保留地分享出来。2. 核心思路与关键技术选型2.1 整体架构设计一个稳健的企业数据爬虫不能是简单的requests.get加BeautifulSoup解析。面对天眼查这类防护严密的站点我们需要一个更系统化的架构。我的核心思路是“模拟真人层层递进优雅处理”。首先爬虫的核心任务是模拟一个真实用户在浏览器上的操作流程打开网页、输入搜索词、查看列表、点击进入详情页、提取信息。因此我们的技术栈必须能完美模拟这一过程。我放弃了传统的requestsBeautifulSoup组合因为天眼查的大量关键数据是通过JavaScript动态渲染的直接拿到的HTML是空的。解决方案是使用能执行JavaScript的浏览器自动化工具。其次反爬措施是最大的拦路虎。常见的包括IP封锁、请求头校验、行为验证码、参数加密、接口令牌等。我们的架构必须内置应对这些措施的模块例如代理IP池、请求头随机化、复杂验证码的识别或绕过策略以及关键请求参数的逆向工程。基于以上考虑我设计的爬虫架构主要包含以下几个模块请求与渲染模块负责加载网页并获取完整HTML。选用Selenium或Playwright驱动无头浏览器。反爬对抗模块集成代理IP管理、请求头管理、Cookie持久化、请求频率控制等。数据解析模块从渲染后的HTML中使用XPath或CSS Selector精准定位并提取目标数据。数据存储模块将提取的结构化数据保存到文件如CSV、JSON或数据库如SQLite、MySQL中。任务调度与监控模块管理待爬队列、处理异常、记录日志确保长时间稳定运行。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 chromiumpandas库用于后续的数据处理和保存为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(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class TianyanchaSpider: def __init__(self, use_proxyFalse, proxy_listNone): 初始化爬虫 :param use_proxy: 是否使用代理 :param proxy_list: 代理列表格式如 [http://user:passip: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;q0.9,en;q0.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-featuresAutomationControlled, --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_agentself.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_untilnetworkidle) # 等待网络空闲 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(texthtml) # 使用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_untildomcontentloaded) # 等待关键信息区域加载 await self.page.wait_for_selector(div.company-header, timeout10000) await self.page.wait_for_timeout(random.uniform(3000, 5000)) html await self.page.content() selector Selector(texthtml) 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_filecompanies.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, indexFalse, encodingutf-8-sig) # utf-8-sig解决Excel打开中文乱码 logger.info(f数据已保存至 {output_file}, 共 {len(all_data)} 条记录。) else: logger.warning(未爬取到任何数据。) # 异步主函数 async def main(): spider TianyanchaSpider( use_proxyFalse, # 根据实际情况开启 # proxy_list[http://your_proxy:port] ) # 要搜索的公司关键词列表 search_keywords [科技有限公司, 信息技术有限公司] await spider.run(search_keywords, tianyancha_companies.csv) if __name__ __main__: asyncio.run(main())3.3 代码关键点解析与避坑指南异步编程Playwright的API是异步的使用async/await能显著提升I/O密集型爬虫的效率。主函数需要用asyncio.run()来驱动。等待策略page.goto()的wait_until参数至关重要。networkidle表示等待网络基本没有活动时再继续适合静态页面。对于动态加载多的页面domcontentloadedDOM加载完成可能更快但需要后续用wait_for_selector等待特定元素出现。绝对避免无脑使用time.sleep要用智能等待。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可以作为起点但通常需要简化。代理IP的使用代码中预留了代理接口。如果启用务必确保代理IP是高匿名的并且有足够的稳定性和速度。免费代理大多不可用生产环境建议购买可靠的服务。异常处理与日志爬虫运行中会遇到各种意外网络超时、元素未找到、验证码弹出等。完善的try...except和日志记录是调试和监控的基石。日志要记录时间、级别和信息方便回溯问题。数据清洗爬取下来的文本常常包含多余的空格、换行符、不可见字符。在保存前一定要用.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_pathtianyancha.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 高频问题速查表问题现象可能原因排查与解决思路页面加载空白或元素找不到1. 页面未完全加载。2. 网站结构已更新XPath/CSS选择器失效。3. 触发反爬被重定向到验证页。1. 增加等待时间使用wait_for_selector。2. 重新检查并更新选择器路径。3. 检查当前页面URL和内容确认是否被拦截。增加延迟、更换IP/User-Agent。弹出验证码图形、滑块等请求频率过高或行为被识别为非人类。1.首要策略大幅降低请求频率增加随机延迟。2. 维护会话Cookie避免每次请求都是新会话。3. 考虑使用更高质量的代理IP住宅IP。4. 考虑引入第三方打码平台成本与合规性需权衡。返回403 Forbidden或429 Too Many RequestsIP或请求头被识别访问被拒绝或限速。1. 检查并随机化请求头User-Agent, Accept-Language等。2.必须使用代理IP池并确保代理可用。3. 严格遵守robots.txt设置更长的请求间隔。爬取速度非常慢1. 使用无头浏览器本身开销大。2. 网络延迟或代理IP速度慢。3. 等待策略设置过于保守。1. 考虑切换到接口请求模式如果可行。2. 测试代理IP的速度更换优质代理。3. 优化等待条件将networkidle改为domcontentloaded并精准等待必要元素。数据字段错乱或为空解析规则不准确定位到了错误的HTML元素。1. 将爬虫解析到的HTML片段print(selector.get())保存下来与实际浏览器查看的源码对比。2. 使用更精确、更具弹性的XPath或CSS选择器避免依赖不稳定的class或id。异步任务报错Event loop is closed异步事件循环管理不当常见于Windows系统或脚本提前退出。确保主入口使用asyncio.run(main())。如果程序中有多个异步循环确保正确关闭。在Jupyter等环境中运行时需注意环境差异。5.2 实战心得与避坑指南“慢就是快”在爬虫世界里急于求成往往导致IP被封、前功尽弃。把请求间隔设置得足够长比如5-10秒甚至更长随机化这个间隔是保证长期稳定运行的最低成本策略。一个能跑一周的慢爬虫远比一个跑一小时就被封的快爬虫有用。选择器的维护成本不要写“一次性”的选择器。尽量使用那些基于语义化、相对稳定的属性来定位元素比如包含特定文本的标签、具有特定数据属性的元素。定期例如每周运行一下爬虫的测试用例确保核心选择器依然有效。数据清洗宜早不宜迟在数据解析的环节就尽量将其清洗成规整的格式。例如将“注册资本1000万元人民币”拆分为数值1000和单位万元将日期字符串转换为datetime对象。这样存入数据库或文件后后续分析会非常方便。做好增量与断点续爬爬虫可能因为网络、反爬等原因中断。设计时就应该考虑记录爬取进度。可以将成功爬取的URL存入一个“已爬”集合可以用Redis或数据库每次启动时先加载这个集合避免重复爬取。对于列表页记录上次爬取到的页码或最后一条数据的ID。法律与道德底线再次强调爬取的数据只能用于个人学习、科学研究或合法的内部分析。未经许可大规模爬取数据用于商业竞争、售卖或个人隐私骚扰不仅是违反网站服务条款的行为更可能触犯法律。尊重robots.txt控制爬取速度是你作为技术人的基本素养。这个爬虫项目从简单的需求出发逐步深入到反爬对抗、效率优化和系统设计是一个非常好的Python综合实践。它考验的不仅仅是编码能力更是工程思维、问题解决能力和对网络协议的理解。希望这份详细的拆解和代码能为你打开网络数据获取的大门同时也能让你清醒地认识到其中的边界与责任。在实际操作中多测试、多观察、多思考你会积累下属于自己的宝贵经验。