1. 项目背景与整体设计思路先说说这个项目解决的是什么痛点。做自然语言处理、舆情分析或者市场调研的时候最头疼的事情之一就是语料从哪来。公开数据集要么太老要么领域不匹配真正符合业务场景的数据往往得自己动手采集。但常规的手工搜索复制粘贴效率太低单关键词采集又容易漏掉大量有价值的信息。这个关键词矩阵驱动的搜索语料库采集引擎本质上就是把搜索词从单一维度扩展到多维组合通过程序化调度批量获取搜索结果自动完成请求发送、页面解析、数据清洗和入库存储的完整链路。适合谁来参考如果你是刚学完Python基础语法、想进阶爬虫实战的初学者这个项目能帮你把requests、解析库、并发控制和数据持久化串成一条线如果你是做数据分析、NLP或者市场研究的朋友这套引擎能帮你快速搭建一个垂直领域的语料采集基础设施即使你已经有爬虫经验关键词矩阵的思维方式和工程化落地细节也值得一看尤其是请求调度和去重策略这两个环节很多人在小规模爬虫里根本不会暴露问题一旦把关键词规模放大到几千上万个整个系统的设计差距就会非常明显。我在设计这个引擎的时候给自己定了几条硬性要求第一代码要足够简单能复用绝不开新坑所有核心逻辑一个main模块加几个工具模块就能跑通第二要能扛住中等规模的采集任务比如一次跑两三千个关键词组合不会因为内存暴涨或者请求积压直接崩掉第三中断后要能断点续采不然跑了三个小时突然断网全部重来是一件非常劝退的事。这三条要求直接决定了后面所有的技术选型和代码结构。2. 关键词矩阵的构建逻辑2.1 为什么单关键词不够用直接采集单个关键词的搜索结果会有两个很明显的问题。第一是覆盖率低比如你做的是新能源汽车相关的舆情语料仅搜这一个词返回的结果大概率是新闻门户和头部垂直媒体的内容那些长尾的论坛讨论、个人博客、地方社区内容很难进入搜索首页但这些恰恰是舆情分析里非常重要的声音来源。第二是语义偏置搜索引擎对同一个词在不同地域、不同账号状态下的返回结果排序是有差别的单一维度搜索容易固化在某一种信息圈层里语料多样性不够。关键词矩阵的思路是把一个搜索需求拆解成多个独立的维度每个维度维护一组候选词最后做维度间的笛卡尔积组合。举个例子新能源汽车这个核心词可以拆成行业细分词纯电动、插电混动、增程式、场景词充电、续航、补贴、保值率、观点词评测、投诉、口碑、深度体验、地域词北京、上海、广州等几个维度组合出来的搜索词数量就是各维度词表的乘积。这样做的好处非常直接语料的覆盖维度从单一话题变成了话题场景地域观点的多维立体结构后续做聚类或者情感分析时产出的标签体系会完整很多。2.2 矩阵结构的工程化表达矩阵在代码层面的实现其实不复杂。我在项目里用一个配置化的方式管理维度词表格式用的JSON因为JSON在Python里就是原生数据结构加载解析不需要额外写解析逻辑。{ core_words: [新能源汽车, 电动车], scene_words: [充电, 续航, 补贴, 保值率], viewpoint_words: [评测, 投诉, 口碑, 深度体验], region_words: [北京, 上海, 广州] }组合的核心函数用Python标准库的itertools.product就能实现这个函数就是专门做笛卡尔积的。加载两个维度以上的词表做全组合几行代码就能搞定from itertools import product def build_keyword_matrix(word_dimensions): 基于多维度词表生成关键词组合 :param word_dimensions: 字典key为维度名value为词列表 :return: 组合后的关键词列表 dim_names list(word_dimensions.keys()) values [word_dimensions[name] for name in dim_names] combinations [] for combo in product(*values): keyword .join(combo) combinations.append(keyword) return combinations用 把多个维度的词拼接起来是因为搜索引擎默认会把空格当作多个独立关键词处理返回的结果同时包含这些词的概率更大。如果你需要更精确的匹配也可以换成引号包裹的短语格式这个取决于目标搜索平台的语法规则。组合数量需要提前评估。以上面的词表为例2个核心词乘以4个场景词乘以4个观点词乘以3个地域词一共是96个搜索词。如果每个词抓取前3页结果每页10条总抓取量就是2880条记录。这个量级对于个人学习或者小型研究项目来说刚刚好。但如果词表维度再增加组合数是呈指数级上涨的所以我在设计时把词表维护做成增量式的用数据库表记录每个关键词的采集状态方便随时调整。2.3 关键词的剪枝与过滤组合出来的关键词不是每条都要采集。有些组合明显没有意义比如新能源汽车 北京 深度体验可能有内容但电动车 上海 补贴 评测这种四个维度词全拼在一个搜索词里返回结果可能会因为词过多而过于稀疏。我在实际运行中总结了一个经验组合词数量控制在2到3个维度词比较合适超过3个词会导致搜索召回率显著下降。所以我在生成的逻辑里加了一个可选参数控制参与组合的维度数量。比如从全部词表中任选2个或者3个维度做组合利用itertools.combinations先选出维度子集再对每个子集做product运算。经过筛选后96条词变成长尾合适的组合每条搜索结果的相关性明显提升。def build_keywords_with_optional_dimensions(word_dimensions, max_dim_count3): from itertools import combinations, product dim_names list(word_dimensions.keys()) result set() for r in range(1, max_dim_count 1): for subset in combinations(dim_names, r): values [word_dimensions[name] for name in subset] for combo in product(*values): result.add( .join(combo)) return list(result)另外还需要对生成的词做一次入库前的规范化处理大小写统一、去掉首尾空格、过滤掉长度异常的词比如超过30个字符的。别小看这批清洗逻辑后续所有去重、统计、查询的可靠性都依赖这一步。3. 采集引擎架构与核心模块实现3.1 整体架构怎么搭我采用的是一个非常经典的分层结构调度层、抓取层、解析层、存储层。调度层负责从关键词库中取出待采集的词按策略分配给抓取线程抓取层负责发送HTTP请求、处理重试、管理Cookie和请求头解析层针对不同平台的返回结果做结构化提取存储层负责把清洗后的数据写入数据库或者文件。这样的分层设计最大的好处是模块之间可以独立替换。比如今天你抓的是百度搜索结果明天要换成搜狗或者360只需要替换解析层对应的函数调度和存储完全不用动。如果采集平台变了请求参数和解析规则调整一下即可。我最早写的爬虫是全部逻辑堆在一个大脚本里改一个页面结构都要小心翼翼生怕动到别的地方。分层的结构在初期看起来多写了一些代码后期改起来真的节省很多时间。目录结构我按照下面的方式组织search_corpus_engine/ ├── config/ │ └── settings.py # 全局配置包括请求间隔、超时、重试次数 ├── core/ │ ├── scheduler.py # 调度器负责任务分发和状态管理 │ ├── fetcher.py # 抓取器封装请求发送和重试 │ ├── parser.py # 解析器处理不同平台的返回内容 │ └── pipeline.py # 存储管道负责数据入库 ├── keywords/ │ └── word_manager.py # 关键词矩阵构建与增量维护 ├── data/ # 存放爬取的原始结果 ├── logs/ # 运行日志 └── main.py # 程序入口3.2 抓取层的关键设计抓取层是整个引擎的地基这里我重点做了几个事情。第一是请求头的随机化。直接用默认的requests User-Agent去访问搜索引擎十有八九会被拦。我在请求头里维护了一个UA池每次请求随机选一个同时带上常见的Accept、Accept-Language、Connection等字段让请求看起来更像真实浏览器发出的。这里分享一个小技巧不一定要用官方推荐的那些Chrome、Firefox UA反而一些冷门浏览器的UA在部分平台上因为特征不明显命中拦截的概率更低。import random import requests 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/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15, Mozilla/5.0 (X11; Linux x86_64; rv:108.0) Gecko/20100101 Firefox/108.0, ] def make_session(): session requests.Session() session.headers.update({ User-Agent: random.choice(USER_AGENTS), Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/avif,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.5, Connection: keep-alive, }) return session第二是请求频率的控制。我使用了一个简单的自适应策略基础间隔随机在0.5到1.5秒之间波动如果连续出现HTTP 4xx或者超时异常间隔自动翻倍等恢复正常后再逐步降回基础值。这个策略相当于给引擎装了一个油门遇到反爬压力会自动减速避免被封IP。第三是重试机制。网络请求失败太常见了超时、连接重置、DNS解析失败都会发生。我给抓取器封装了一个带指数退避的重试函数最多重试3次每次重试间隔为2秒、4秒、8秒。重试时判断异常类型如果是请求被拒绝的4xx状态码不重试直接跳过如果是5xx或者网络层面的异常才触发重试。3.3 调度器与断点续采调度器是保证批量任务稳定的核心。我设计了一个基于待采集队列 已完成集合的调度模型。程序启动时把所有关键词加载进队列同时从数据库中加载已经成功采集的关键词列表把它们标记为已完成。每次从队列弹出任务前先检查是否已经完成过完成过就直接跳过。这样即使程序中途崩溃重启后也能接着跑。已采集状态我放在数据库里维护表结构用最简单的两张表。关键词状态表负责管理关键词的采集进度搜索结果表负责存放采集到的语料明细。CREATE TABLE keyword_status ( id INTEGER PRIMARY KEY AUTOINCREMENT, keyword TEXT UNIQUE NOT NULL, status TEXT DEFAULT pending, page_count INTEGER DEFAULT 0, total_fetched INTEGER DEFAULT 0, last_run_time DATETIME ); CREATE TABLE search_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, keyword TEXT NOT NULL, source_url TEXT NOT NULL, title TEXT, summary TEXT, publish_date TEXT, crawled_at DATETIME DEFAULT CURRENT_TIMESTAMP );调度器用Python标准库的queue.Queue做任务队列消费者线程从队列里取关键词去采集。线程数我一般设为3到5个再配合前面的请求间隔既不会太慢也不会触发反爬。这里有个工程上很实用的优化用队列的task_done()和join()方法控制主流程在所有任务完成后自动退出避免出现主程序提前结束线程还在跑的情况。4. 搜索请求构造与页面解析4.1 搜索接口的参数分析不同搜索引擎的URL结构差异很大。以百度为例搜索接口的核心参数是wd、rn、pn这几个。wd是查询词rn是每页返回条数pn是起始位置的偏移量第一页为0第二页为10。构造URL的代码很简单但有几个容易被忽略的点。关键词里如果包含空格直接拼到URL里是不行的需要用urlencode做编码。Python的urllib.parse.urlencode可以自动处理比手动调quote省心。from urllib.parse import urlencode def build_search_url(base_url, keyword, page_num, page_size10): params { wd: keyword, rn: page_size, pn: page_num * page_size, } return f{base_url}?{urlencode(params)}不同的搜索引擎对页码的参数名定义完全不同。搜狗用的是page360搜索用的是pn但逻辑和百度不同。所以我在解析层做了一个适配器模式针对不同平台注册不同的URL构建函数和解析函数调度层通过配置指定当前抓取哪个平台。4.2 HTML解析策略搜索结果的HTML结构在不同平台间差异非常大但整体上都包含标题链接、摘要文本、来源URL、发布时间这几个核心字段。我最常用的解析工具是lxml配合XPath速度快且语法灵活。相比于BeautifulSouplxml在处理大文档时性能优势明显尤其是在批量解析几百个页面时差别可以感知到。以解析百度搜索结果为例搜索结果容器通常是id为content_left的div里面的每个结果块是带有特定class的div。在写XPath之前我习惯先用浏览器的开发者工具查看目标元素的层级关系。这里分享一个我实际使用的解析函数思路是通用的具体到不同平台只需要修改XPath表达式from lxml import html def parse_baidu_search(html_text): doc html.fromstring(html_text) results [] # 定位搜索结果列表 items doc.xpath(//div[idcontent_left]//div[contains(class,result)]) for item in items: title_node item.xpath(.//h3/a) if not title_node: continue title title_node[0].text_content().strip() link title_node[0].get(href) summary_nodes item.xpath(.//div[contains(class,c-abstract)]) summary summary_nodes[0].text_content().strip() if summary_nodes else if title and link: results.append({ title: title, link: link, summary: summary }) return results这中间有几个细节容易出错。text_content()是把节点下所有文本拼接起来如果页面里有隐藏的样式标签拼接结果会带上多余的空格所以解析后一定要做空白归一化。其次搜索结果的链接很多是经过跳转重定向的直接用href属性拿到的不是最终URL如果需要最终的落地页地址就得额外发一次HEAD请求或者访问一次页面解析跳转逻辑。我在这个项目里保留了原始链接没有做二次跳转解析因为语料分析关注的是内容和标题链接本身作为唯一标识已经够用。4.3 非HTML数据源的处理现在很多搜索引擎的返回结果不再只是纯HTML了。有的是首屏通过JavaScript异步加载接口返回JSON数据有的是整个搜索页就是一个Ajax应用。面对这类平台直接请求HTML页面拿不到有效数据需要分析其背后的XHR接口。我用到的策略是打开浏览器的开发者工具切到Network面板刷新页面观察哪些XHR请求返回了搜索数据然后直接请求这个XHR接口。通常这种接口返回的是格式化的JSON解析起来反而比HTML更方便。JSON解析直接用Python内置的json模块按字段名取值即可。需要注意的是这类接口往往有额外的鉴权参数比如sign、token而且可能是每次请求动态生成的处理起来会复杂一些。我在项目中预留了一个hook函数用于在请求前对参数做动态处理方便应对这类场景。5. 数据清洗与存储落地5.1 清洗流程的几个关键点搜索结果的原始数据比较脏直接入库对后续分析影响很大。我在pipeline里做了一套清洗流程。标题清洗的核心是去除平台附加信息。很多搜索引擎会在标题后面拼接站点来源比如某某网站 - 某某搜索这种格式。如果只是做语料分析来源标识保留下来反而会对词频统计造成干扰。我用正则把这种尾缀剔除掉匹配模式是 - [^-]$把最后一个连字符加中文内容切掉。摘要清洗主要处理空白字符和乱码。直接从HTML里提取的文本可能带有不间断空格(\xa0)、换行符(\n)、还有各种中文全角空格。统一把它们替换成半角空格再把多余的空格压缩掉。还有一个经常被忽略的点时间字段的标准化。有的平台返回3小时前有的返回2024-05-18 10:30这些不统一的格式在后续分析时很难处理。我在解析后设置了一个标准化函数把相对时间转换成绝对时间转换基准取当前采集时间。from datetime import datetime, timedelta import re def normalize_time(raw_time): if not raw_time or not raw_time.strip(): return None raw_time raw_time.strip() now datetime.now() # 处理“x分钟前”“x小时前”等相对时间 m re.search(r(\d)\s*分钟前, raw_time) if m: return (now - timedelta(minutesint(m.group(1)))).strftime(%Y-%m-%d %H:%M:%S) m re.search(r(\d)\s*小时前, raw_time) if m: return (now - timedelta(hoursint(m.group(1)))).strftime(%Y-%m-%d %H:%M:%S) # 处理2024-05-18和2024-05-18 10:30等绝对时间 for fmt in (%Y-%m-%d %H:%M:%S, %Y-%m-%d %H:%M, %Y-%m-%d): try: dt datetime.strptime(raw_time, fmt) return dt.strftime(%Y-%m-%d %H:%M:%S) except ValueError: continue return raw_time5.2 存储方案的取舍对于搜索语料这种规模的数据SQLite是性价比非常高的选择。它的优点是不需要单独部署数据库服务Python标准库自带驱动直接连接文件就能用数据量在几万到几十万条级别时查询性能完全够用。我在项目里默认使用SQLite主要原因就是部署零成本别人拿到代码不需要安装MySQL或者PostgreSQL就能直接跑起来。如果你的语料规模预期会超过百万级或者需要多机并发写入那就应该换成MySQL或者PostgreSQL。切换存储层时只需要保证pipeline里的写入接口保持一致我在pipeline里抽象了一个StorageBackend基类SQLite和MySQL分别实现这个基类主程序里通过配置指定使用哪个后端即可。写入采用批量提交而不是逐条插入这是一个很重要的性能优化点。每条insert执行一次commit在SQLite这种文件数据库上大概需要扫描和刷新磁盘几百次插入就会明显感觉到卡顿。我改成累积到50条或者100条才统一commit一次速度能提升好几倍。具体的批量大小需要根据单条数据体积调整测试下来50到100之间的性价比比较高。5.3 URL级别的去重搜索结果里会出现大量重复URL。同一个新闻可能被多个网站转载同一个网站的文章可能出现在多个关键词的搜索结果中。如果不去重一个链接的重复采集会影响到后续的相似度计算和统计分析。我的去重策略采用双重的方案第一步是对source_url做精确去重URL完全一样的直接丢弃第二步是对标题做MD5归一化去重。为什么要对标题再做一次去重因为很多网站的跟踪参数会导致同一个文章URL不同比如?fromsearch和?fromtimeline这种后缀变化URL精确匹配拦不住但标题是一样的。import hashlib def gen_title_md5(title): normalized re.sub(r\s, , title) return hashlib.md5(normalized.encode(utf-8)).hexdigest()在做标题MD5之前先把空白字符全部去掉避免因为空格差异导致同样的标题算出了不同的MD5值。去重逻辑放在存储层之前新增数据先和库里的MD5索引比对存在则跳过。为了保证比对效率我在title_md5字段上建了唯一索引。6. 常见问题与排查技巧实录6.1 明明有请求结果但解析出来是空的这个问题我遇到过不下十次。典型症状是程序不报错请求也返回了200但解析函数返回的空列表。排查思路一般分三步。第一步把返回的HTML直接保存到本地文件用浏览器打开看内容是否正常。如果浏览器打开也看不到搜索结构说明请求被重定向到了一个验证页面或者安全拦截页这种页面在代码里看不出异常但因为页面结构完全不同导致解析失败。第二步如果浏览器打开正常检查返回的HTML和我预设的XPath是否匹配因为搜索引擎的页面结构会不定期改版改版后class名字变了XPath自然就失效了。第三步检查是不是动态渲染的页面如果搜索结果是通过JavaScript异步加载的初始HTML里根本找不到数据这时候就要走XHR接口的方案。排查时我最常用的是加日志。在每个环节的关键节点打印出前50个字符的响应内容很快就能定位问题出在哪一层。6.2 请求频率不高但IP还是被限制了搜索引擎的反爬策略不只是看请求频率还会看请求的指纹特征。同一个浏览器指纹包括User-Agent、Accept-Language、浏览器头顺序反复出现或者Cookie信息异常都可能触发风控。被限制的典型表现是间歇性出现验证码或者HTTP 302重定向。解决办法有几个方向。一是增加代理池让请求通过不同IP发出这部分我在项目里预留了代理接口但在基础版本里没有启用因为代理池的维护成本比较高。二是在Session里维护稳定的Cookie有时候没有Cookie的请求反而容易被判定为脚本带上一个浏览器访问获得的Cookie能明显降低被限概率。三是拉大随机延时区间把固定的1秒间隔改成1到3秒之间的随机值模拟人类操作习惯。6.3 采集速度太慢能不能多线程并发多线程能提升速度但也带来两个问题一是请求频率同步升高被限概率随之增大二是线程安全处理不好会出现数据错乱或者重复插入。我在设计里加了线程锁保护数据库写入的计数器同时在清洗和存储环节尽量做到纯函数操作不依赖共享的可变状态。有一个实用的经验不要只靠提高并发数来提速更好的方式是优化每个请求的开销。开启requests Session连接复用减少TCP握手的次数解析层优先使用lxml而不是BeautifulSoup因为lxml用的C库解析速度是BeautifulSoup的数倍。这些优化叠加起来即使并发数不变整体吞吐也能提升50%以上。6.4 程序跑几个小时突然异常退出长时间运行的爬虫最怕不稳定。我在工程里加了两道保险。第一道是全局异常捕获最外层主函数用try...except包裹任何未预料的异常都记录到日志文件同时保存当前任务状态然后优雅退出。这样即使程序挂了重启后还能从断点继续。第二道是定期做中间状态落盘每采集成功20个关键词就更新一次数据库中的状态字段而不是等到最后统一更新。这样即便系统崩溃已经完成的关键词不会重复采集。从实际运行的效果来看这套引擎跑3000个关键词组合平均耗时大约40到60分钟能够稳定采集约2万条搜索结果语料。我用的平台返回的数据以标题和摘要为主这对大多数语义分析场景已经足够。如果你需要正文级的数据可以在结果链接的基础上再增加一层正文抓取流程把采集到的URL逐个请求并提取正文内容整体思路不变只需要在pipeline里增加一个正文抓取的步骤。我用这个引擎跑过几个小规模的语料集产出直接用于情绪分析和主题聚类实验数据质量基本能达到可用的水平。对于想快速搭建自己语料库的朋友这个方案是一个挺好的起点顺着这个框架你可以把关键词矩阵换成任何你关注的领域把搜索引擎换成任何你熟悉的信息源把存储层换成任何你习惯的数据库整套系统就能适应完全不同的业务场景。