简介这是一份基于 Java 实现的新闻爬虫项目覆盖百度新闻与今日头条两个信源支持按关键字批量抓取新闻并写入数据库适合需要采集新闻数据的 Java 开发者、数据分析人员或爬虫初学者参考。压缩包共 20 个文件包含 16 个 Java 源码文件作为主体覆盖 URL 收集、HTTP 请求、HTML 解析、数据入库等爬虫核心环节另配有 XML、properties 配置文件及 README 说明文档便于理解依赖与启动配置。整包仅 19KB代码量精简适合快速阅读和二次开发。目前已有 447 人下载学习。通过该项目读者可直观掌握爬虫的标准工作流程包括请求网页、解析内容、存储数据以及应对反爬机制的常见思路同时可参考其模块拆分方式将其扩展到其他新闻站点或资讯类页面的抓取场景。1. 为什么“按关键字爬所有新闻”听上去简单跑起来却天天翻车按关键字抓取百度新闻和今日头条的所有新闻再落进数据库标题里只有一句话落到工程上是三件事稳定抓取、准确解析、可靠入库。我见过不少这类 .zip 项目包跑到一半就废掉原因往往不是反爬多凶而是重复数据灌了几万条、字段名撞了 SQL 保留字、没做断点导致白跑一整晚。这篇文章给出一套可以直接照做的方案Python requests 抓取BeautifulSoup 加正则解析SQLAlchemy 写 MySQL最后附上最常见 5 个坑和一条自检 SQL。适合已经能跑通 demo、想把关键字爬虫做成能长期维护的数据管道的人。2. 先建表再写爬虫新闻数据模型与 MySQL/SQLite 选型动手写请求之前先把表结构定下来。关键字爬虫和普通爬虫最大的区别在于keyword 必须落库。同一个标题会被不同关键字搜到如果不记录是哪个关键字爬进来的后续按关键字检索数据时直接抓瞎。字段设计也不是随手写几个列就完事URL 唯一性、时间字段、内容字段的长度都要在这一步想清楚。2.1 一张 news 表放下两个平台的公共字段关键字必须落库百度新闻和今日头条的搜索结果是两条数据管道入库后要能合并查询所以表结构按公共字段设计。字段清单大致如下id 自增主键、keyword、title、url、source来源平台、media发布媒体、publish_time、summary、content、crawl_time。其中 url 必须做唯一性约束但直接对 url 建唯一索引在 MySQL 的 utf8mb4 字符集下会踩索引长度限制常见做法是额外存一列 url_hash对 md5 后的 32 位字符串建唯一索引。下面是我一般会用的建表 SQL基本能覆盖这个标题的需求CREATE TABLE IF NOT EXISTS news ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, keyword VARCHAR(64) NOT NULL COMMENT 爬取时使用的关键字, title VARCHAR(512) NOT NULL COMMENT 新闻标题, url VARCHAR(1024) NOT NULL COMMENT 原文链接, url_hash CHAR(32) NOT NULL COMMENT url的md5用于唯一索引, source VARCHAR(32) NOT NULL COMMENT 来源平台: baidu / toutiao, media VARCHAR(128) DEFAULT COMMENT 发布媒体, publish_time DATETIME NULL COMMENT 发布时间解析失败可为空, summary TEXT COMMENT 列表页摘要, content MEDIUMTEXT COMMENT 详情页正文可不抓, crawl_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 入库时间, UNIQUE KEY uq_url_hash (url_hash), KEY idx_keyword_time (keyword, publish_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT新闻数据表;几个参数说明url 存 1024 是因为部分新闻链接带一堆追踪参数长度经常超过 255url_hash 唯一索引替代 url 唯一索引规避 MySQL 5.7 里 utf8mb4 下 varchar 唯一索引超过 768 字节就报错的问题。idx_keyword_time 联合索引是给“按关键字查询最近新闻”这个最核心的检索场景用的。source 字段区分数据来自百度还是今日头条后续排查问题才知道去哪边找原因。2.2 存储选型MySQL、SQLite 怎么选以及数据库连接池参数标题说“存如数据库”没说必须是 MySQL。这个选择直接影响后面所有代码。我的判断标准是数据量和并发单机小规模、数据量在十万级以下、只是自己练手SQLite 完全够要跑多关键字、多进程甚至分布式爬虫或者数据量朝百万级走直接上 MySQL。两者的差异可以用一张表说清楚维度SQLiteMySQL部署复杂度零部署一个文件需要安装服务端并发写入单写锁多进程并发写入会锁库支持并发写入连接池管理数据量上限五十万到一百万条内可用千万级无压力去重能力靠唯一索引效果一样靠唯一索引效果一样分布式爬虫不太合适标准选择如果你决定用 MySQLSQLAlchemy 连接引擎有三个参数必须调否则跑半小时就报连接错误from sqlalchemy import create_engine engine create_engine( mysqlpymysql://root:your_passwordlocalhost:3306/news_db?charsetutf8mb4, pool_size10, # 连接池保持 10 个连接 max_overflow20, # 峰值可再借 20 个连接 pool_pre_pingTrue, # 每次取连接前先探活防止 MySQL 断连后复用死连接 pool_recycle3600 # 连接超过 1 小时强制回收 )pool_size 和 max_overflow 决定并发上限pool_pre_ping 是血泪经验MySQL 默认 wait_timeout 八小时连接池里的连接睡太久会被服务端掐掉没这句配置就会间歇性拿到坏连接。如果你只是个人爬虫、单进程跑这几个参数不至于打满但配上能少踩很多坑。需要说明的是SQLite 不需要连接池一个 connect 到底就行但要注意多进程同时写同一个 SQLite 文件会锁库这也是为什么不建议在分布式爬虫里用 SQLite。3. 用 requests 稳定抓取百度新闻和今日头条搜索页Cookie、重试与频率控制网络爬虫原理不复杂模拟浏览器发请求拿回 HTML 或 JSON再解析成结构化数据。但百度新闻和今日头条的反爬策略不同百度主要靠频率限制今日头条主要靠 Cookie 和验证码。这一章的代码全部围绕“稳定”展开请求头伪装、Cookie 管理、超时重试、频率控制。3.1 百度新闻搜索tnnews 的参数与最小请求代码百度新闻没有独立的搜索域名走的是百度搜索的新闻垂直结果核心参数是 tnnews。请求地址是 https://www.baidu.com/s wd 是关键字pn 是分页偏移量每页通常 10 到 20 条。最小可用代码长这样import requests 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: text/html,application/xhtmlxml, Accept-Language: zh-CN,zh;q0.9, } def fetch_baidu_news(keyword: str, page_index: int): params { tn: news, # 指定新闻垂直搜索 word: keyword, # 搜索关键字 pn: page_index * 10, # 偏移量第 1 页是 0第 2 页是 10 } resp requests.get( https://www.baidu.com/s, paramsparams, headersHEADERS, timeout10 ) resp.raise_for_status() resp.encoding utf-8 return resp.text参数说明requests 的 params 会自动做 URL 编码不用手动 quote 关键字timeout10 是必须的不设超时会导致线程卡死。编码强制指定 utf-8避免百度返回的 HTML 里 meta 声明和实际编码不一致导致乱码。这里返回的是整个 HTML 页面解析逻辑放在下一章。注意百度搜索结果的链接是跳转链接入库前要不要跟随跳转拿真实 URL取决于下游用途需要保留原文地址的话再做一次 HEAD 跟随。3.2 今日头条搜索接口Cookie、Referer 和验证码兜底今日头条的搜索结果页是 JS 渲染的直接抓 HTML 拿不到数据常见做法是调它的搜索接口。接口路径大概长这样https://www.toutiao.com/api/search/content/ 用 POST 提交表单里带 keyword、offset、count、formatjson。这个接口的硬性要求是请求头必须带 Cookie 和 RefererCookie 来自浏览器登录状态Referer 填搜索页地址。def fetch_toutiao_news(keyword: str, offset: int): cookies { xx: 从真实浏览器复制来的 Cookie 值, } headers { User-Agent: HEADERS[User-Agent], Referer: fhttps://www.toutiao.com/search/?keyword{keyword}, Cookie: ; .join(f{k}{v} for k, v in cookies.items()), } data { aid: 24, app_name: toutiao_web, format: json, keyword: keyword, offset: offset, count: 20, cur_tab: 1, search_id: , } resp requests.post( https://www.toutiao.com/api/search/content/, headersheaders, datadata, timeout10 ) if resp.status_code 200: return resp.json() return None代码里的 Cookie 只是一个占位说明正确姿势是在浏览器里打开今日头条搜索页手动过掉一次滑块验证再从开发者工具里复制完整 Cookie 字符串填进来。Cookie 失效后接口不会报 403而是返回 200 但 data 是空数组或验证提示所以解析前必须做一层判断见 4.3 节。不建议写无 Cookie 硬刷的逻辑今日头条的风控对无状态请求特别敏感硬刷大概率二次请求就触发验证。3.3 请求频率与重试退避把“爬所有新闻”控制在安全曲线上“所有新闻”意味着大批量翻页一个关键字翻 50 页就是 50 次请求十个关键字就是 500 次。如果不管频率今天头条风格的站点会直接把 IP 拉进风控名单。我的经验是单请求间隔不低于 1.2 秒并在间隔里加随机抖动不要用固定 sleep。同时给请求套一层指数退避重试应对瞬时网络错误。import random import time from requests.exceptions import RequestException def request_with_retry(func, *args, **kwargs): for attempt in range(3): try: return func(*args, **kwargs) except (RequestException, TimeoutError): wait 2 ** attempt random.uniform(0, 1) time.sleep(wait) return None # 主循环里的调用方式 def crawl_keyword(keyword: str, max_pages: int): for page in range(max_pages): html request_with_retry(fetch_baidu_news, keyword, page) if html: # 解析入库逻辑省略见后续章节 pass time.sleep(random.uniform(1.2, 3.5))指数退避里的 2 的幂次方是标准做法第一次失败等约 2 秒第二次约 4 秒第三次约 8 秒给服务端留出恢复时间。主循环里的随机间隔区间不要随便压缩今日头条和百度对短时间高频请求都敏感。另外解析逻辑如果涉及再请求详情页详情页请求也要走同一个重试与限速逻辑否则列表页没被限制详情页反而先触发风控。注意频率控制不是玄学是爬虫能不能长期跑的分水岭。宁可每小时少抓几百条也别一次把站点请求打满。4. 解析新闻列表BeautifulSoup 选择器优先正则兜底抓下来的百度新闻是一整页 HTML今日头条是一坨 JSON两者解析方式完全不同。这一章的核心思路是“选择器优先正则兜底”结构稳定的部分用 BeautifulSoup 的选择器结构随机或嵌在 JS 里的部分用正则硬抠。4.1 选择器还是正则面对改版时的选型思路BeautifulSoup 的 select 和 find 依赖 HTML 的层级和 class 名只要站点改版就断正则不依赖结构只要文本模式还在就能继续工作。实际新闻站的规律是列表容器的 class 经常变但链接和时间的文本格式相对稳定。所以我一般这么判断页面里能靠 class 稳定命中的区域用选择器像 title 里混着高亮标签、时间戳格式不一致这类情况交给正则清洗。场景推荐方案原因class 名稳定的列表容器BeautifulSoup select代码可读性好层级清晰class 名带随机后缀正则匹配整体块选择器一改版就失效标题里的高亮标签正则去标签文本模式稳定JS 里内嵌的 JSON 数据正则 json.loads不依赖 DOM 渲染4.2 解析百度新闻 HTML从 result 容器里抠标题、来源、时间百度新闻搜索结果里每条新闻通常包在一个 class 为 result 的容器里。标题在 h3 的 a 标签内来源和时间在标题下方的信息行里但不同版式下 class 变化很大。解析代码按“容器 选择器 正则兜底”三层结构写from bs4 import BeautifulSoup import re def parse_baidu_news(html: str, keyword: str): soup BeautifulSoup(html, html.parser) results [] for item in soup.select(div.result): title_tag item.select_one(h3 a) if not title_tag: continue title re.sub(r[^], , str(title_tag)) title title.strip() # 百度结果里的链接是跳转链接先保留原样 url title_tag.get(href, ) # 来源和时间没有稳定 class用正则从容器文本里抓 source_match re.search(r(\S网)\s*$, item.get_text()) media source_match.group(1) if source_match else time_match re.search(r(\d{1,2}小时前|\d{4}年\d{1,2}月\d{1,2}日), item.get_text()) publish_time time_match.group(1) if time_match else results.append({ keyword: keyword, title: title, url: url, media: media, publish_time: publish_time, source: baidu, }) return results这个函数里的正则都是兜底逻辑标题先取出 HTML 再统一去标签是因为百度会给命中的关键字包高亮时间和来源不从固定 class 里取是因为百度新闻改版时最常动的就是这两个位置的样式。publish_time 这里存的是“x小时前”这类相对时间入库前需要转成绝对时间但转的时候拿不到“当前时间”会不准所以更稳的做法是遇到相对时间就用抓取时间减去偏移量遇到绝对时间直接解析。4.3 解析今日头条 JSONdata 数组的提取与空值容错今日头条接口返回的 JSON 里data 是一个数组每个元素包含 title、source、article_url、publish_time、content 等字段。难点不在取字段而在容错data 可能为空、字段可能缺失、title 里可能带 HTML 标签、publish_time 可能是秒级时间戳。代码里要全部兜住import json import re from datetime import datetime, timedelta def parse_toutiao_news(data: dict, keyword: str): results [] raw_items data.get(data) or [] for item in raw_items: if not isinstance(item, dict) or not item.get(title): continue title re.sub(r[^], , item.get(title, )).strip() url item.get(article_url) or item.get(url) or media item.get(source) or item.get(media_name) or publish_raw item.get(publish_time) or 0 publish_time None # 兼容秒级时间戳和毫秒级 if isinstance(publish_raw, (int, float)): ts publish_raw if publish_raw 10_000_000_000 else publish_raw * 1000 publish_time datetime.fromtimestamp(ts / 1000) results.append({ keyword: keyword, title: title, url: url, media: media, publish_time: publish_time, content: item.get(content, ), source: toutiao, }) return results时间戳那行逻辑是关键今日头条返回的时间戳偶尔是秒、偶尔是毫秒判断阈值设成 100 亿大于这个数按毫秒处理否则按秒处理。data 为空数组时直接返回空列表不要报错因为空数组往往意味着 Cookie 失效或触发了验证上层循环靠“返回结果数为 0 就停止翻页”来终止这里设计成容错返回会更安全。解析到这一步数据的 title、url、media、publish_time 都已经规整成表结构需要的类型可以直接进入库模块。5. 写入数据库SQLAlchemy 批量入库、URL 去重与断点续爬抓取和解析都稳定之后最后这道工序决定数据质量。新手最容易在这一步翻车一条条 insert 太慢、重复数据重复入库、跑到一半崩了没法续。这一章用 SQLAlchemy 解决批量写入和去重并用一张进度表实现断点续爬。5.1 建表 DDL 与 ORM 模型唯一索引定生死第 2 章的 DDL 已经把表建好了ORM 模型要和它严格对应。唯一索引 uq_url_hash 是整套去重方案的核心没有它先查后插只是心理安慰因为多进程下两个进程可能同时查到“不存在”然后同时插入。ORM 模型如下from sqlalchemy import ( Column, String, BigInteger, DateTime, Text, Index ) from sqlalchemy.ext.declarative import declarative_base from datetime import datetime Base declarative_base() class News(Base): __tablename__ news id Column(BigInteger, primary_keyTrue, autoincrementTrue) keyword Column(String(64), nullableFalse) title Column(String(512), nullableFalse) url Column(String(1024), nullableFalse) url_hash Column(String(32), nullableFalse, uniqueTrue) source Column(String(32), nullableFalse) media Column(String(128), default) publish_time Column(DateTime, nullableTrue) summary Column(Text, nullableTrue) content Column(Text, nullableTrue) crawl_time Column(DateTime, defaultdatetime.now) __table_args__ ( Index(idx_keyword_time, keyword, publish_time), )代码里的 uniqueTrue 会生成唯一索引保证数据库层面拒绝重复 url_hash。有个细节mysql 表中字段如果撞了“关键字”——这里说的是 SQL 保留字比如 key、order、group——会导致建表和查询都报语法错误所以模型里刻意把关键字字段命名为 keyword而不是 key。虽然 keyword 不算 MySQL 保留字但建表时反引号习惯不要丢。5.2 批量写入先查已存在 URL再批量提交一条一条 add 再 commit 的性能很差新闻爬虫一跑就是几百上千条必须批量提交。正确姿势是先按 url_hash 查出已存在的过滤掉再批量插入。代码如下import hashlib from sqlalchemy.orm import sessionmaker from sqlalchemy.dialects.mysql import insert SessionLocal sessionmaker(bindengine) def build_url_hash(url: str) - str: return hashlib.md5(url.encode(utf-8)).hexdigest() def batch_save_news(items: list[dict]): if not items: return # 第一步过滤已存在的 url_hash hashes [build_url_hash(item[url]) for item in items] with SessionLocal() as session: exist_rows session.query(News.url_hash).filter( News.url_hash.in_(hashes) ).all() exist_set {row[0] for row in exist_rows} new_items [ item for item, h in zip(items, hashes) if h not in exist_set ] # 第二步批量插入 for item in new_items: item[url_hash] build_url_hash(item[url]) session.add(News(**item)) session.commit()这里每个 News 对象单独 add 后统一 commitSQLAlchemy 会把它们合并成一个事务提交速度远快于逐条 commit。exist_set 的过滤在单进程下足够多进程下两个进程同时插入相同 url_hash 时唯一索引会抛 IntegrityError捕获后 rollback 即可数据不会重复。如果用的是 MySQL 方言还可以用 insert().on_duplicate_key_update() 做成幂等写入但先查后插已经能覆盖绝大多数场景代码也更清晰。提示批量入库和去重必须同时存在。先查后插管住流程唯一索引管住并发竞态缺一个都会有脏数据。5.3 断点续爬crawl_state 表记录每个关键字爬到第几页抓取“所有新闻”意味着要跑很久进程随时可能因为网络异常、内存问题、手动中断而挂掉。没有断点续爬每次崩溃都要从第 1 页重新开始。做法是加一张 crawl_state 表记录每个关键字已经抓到了第几页CREATE TABLE crawl_state ( keyword VARCHAR(64) PRIMARY KEY, next_offset INT NOT NULL DEFAULT 0, total_count INT DEFAULT 0, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;主循环从读取进度开始每抓完一页就更新一次class CrawlState(Base): __tablename__ crawl_state keyword Column(String(64), primary_keyTrue) next_offset Column(BigInteger, nullableFalse, default0) total_count Column(BigInteger, default0) updated_at Column(DateTime, defaultdatetime.now) def get_next_offset(keyword: str, session) - int: row session.query(CrawlState).filter_by(keywordkeyword).first() return row.next_offset if row else 0 def update_offset(keyword: str, new_offset: int, session): row session.query(CrawlState).filter_by(keywordkeyword).first() if row: row.next_offset new_offset row.updated_at datetime.now() else: session.add(CrawlState(keywordkeyword, next_offsetnew_offset)) session.commit()每页抓完立刻 commit 进度不要等整个关键字跑完再更新否则中途断电等于没记。断点续爬配合 url_hash 去重还有一个好处就算进度表丢了、从第 1 页重跑已存在的记录会被批量写入模块自动过滤不会产生重复数据。这一套逻辑也天然适合分布式爬虫多进程各领不同关键字共用同一张 news 表去重靠唯一索引进度靠 crawl_state 表全程不需要额外的队列组件。6. 避坑清单5 个翻车现场与一条自检 SQL这个标题方向的坑高度集中下面 5 条是我反复见过的翻车现场每条都按现象、原因、解决的套路写。坑 1字段名撞保留字现象建表 SQL 报语法错误或查询时 order by 报错。原因把字段命名为 key、order、group 这类保留字。keyword 本身不是保留字但 Python 标准库里有个 keyword 模块import keyword 也会撞名。解决字段统一叫 keyword避免直接用 key/order 做列名必要时 SQL 里用反引号包住。坑 2今日头条 Cookie 失效被验证码截住现象请求返回 200但 data 是空数组或者返回页面里写着“安全验证”。原因没有有效登录态服务端把人机校验挡在数据前面。解决浏览器手动过一次滑块验证把完整 Cookie 复制进爬虫发现连续两页 data 为空就停止从浏览器重新复制 Cookie别在无 Cookie 状态下硬刷。坑 3重复数据爆炸现象第二天重跑库里同一篇新闻出现两次甚至更多。原因开始没建唯一索引用“标题是否相同”来去重新闻标题一改版就失效。解决建 url_hash 唯一索引写入前先查已存在的 url_hash双保险兜底。坑 4MySQL 连接耗尽现象跑半小时后报 “Too many connections”。原因循环里反复新建 engine、session 不关闭或者连接池太小。解决全局只建一个 enginesession 用 with 语句包裹设置 pool_size10、max_overflow20、pool_pre_pingTrue。坑 5没做断点崩溃后从头再来现象网络抖动导致进程退出重启后从第 1 页开始前几小时白跑。原因页码存在内存变量里没有持久化。解决用 crawl_state 表记录每个关键字的 next_offset每抓完一页就 commit重跑时从记录的位置继续。跑完一批数据后用下面这条 SQL 做自检能同时确认“所有新闻”真的入库且没有明显重复SELECT keyword, COUNT(*) AS total_cnt, COUNT(DISTINCT url_hash) AS uniq_cnt, COUNT(DISTINCT source) AS source_cnt, MIN(publish_time) AS earliest_time, MAX(publish_time) AS latest_time FROM news GROUP BY keyword;total_cnt 大于 uniq_cnt 说明有重复source_cnt 小于 2 说明某个平台可能整批抓取失败earliest_time 和 latest_time 距离太近说明时间字段解析串位。抽一条 title 去原文站点核对确认解析没有张冠李戴。我自己早年的习惯是一上来就写全量循环被上面五个坑轮番教育后固定套路已经是小样本验证、入库去重、断点续爬、跑完自检四步走。希望帮到你。本文还有配套的精品资源点击获取
简介这是一份基于 Java 实现的新闻爬虫项目覆盖百度新闻与今日头条两个信源支持按关键字批量抓取新闻并写入数据库适合需要采集新闻数据的 Java 开发者、数据分析人员或爬虫初学者参考。压缩包共 20 个文件包含 16 个 Java 源码文件作为主体覆盖 URL 收集、HTTP 请求、HTML 解析、数据入库等爬虫核心环节另配有 XML、properties 配置文件及 README 说明文档便于理解依赖与启动配置。整包仅 19KB代码量精简适合快速阅读和二次开发。目前已有 447 人下载学习。通过该项目读者可直观掌握爬虫的标准工作流程包括请求网页、解析内容、存储数据以及应对反爬机制的常见思路同时可参考其模块拆分方式将其扩展到其他新闻站点或资讯类页面的抓取场景。1. 为什么“按关键字爬所有新闻”听上去简单跑起来却天天翻车按关键字抓取百度新闻和今日头条的所有新闻再落进数据库标题里只有一句话落到工程上是三件事稳定抓取、准确解析、可靠入库。我见过不少这类 .zip 项目包跑到一半就废掉原因往往不是反爬多凶而是重复数据灌了几万条、字段名撞了 SQL 保留字、没做断点导致白跑一整晚。这篇文章给出一套可以直接照做的方案Python requests 抓取BeautifulSoup 加正则解析SQLAlchemy 写 MySQL最后附上最常见 5 个坑和一条自检 SQL。适合已经能跑通 demo、想把关键字爬虫做成能长期维护的数据管道的人。2. 先建表再写爬虫新闻数据模型与 MySQL/SQLite 选型动手写请求之前先把表结构定下来。关键字爬虫和普通爬虫最大的区别在于keyword 必须落库。同一个标题会被不同关键字搜到如果不记录是哪个关键字爬进来的后续按关键字检索数据时直接抓瞎。字段设计也不是随手写几个列就完事URL 唯一性、时间字段、内容字段的长度都要在这一步想清楚。2.1 一张 news 表放下两个平台的公共字段关键字必须落库百度新闻和今日头条的搜索结果是两条数据管道入库后要能合并查询所以表结构按公共字段设计。字段清单大致如下id 自增主键、keyword、title、url、source来源平台、media发布媒体、publish_time、summary、content、crawl_time。其中 url 必须做唯一性约束但直接对 url 建唯一索引在 MySQL 的 utf8mb4 字符集下会踩索引长度限制常见做法是额外存一列 url_hash对 md5 后的 32 位字符串建唯一索引。下面是我一般会用的建表 SQL基本能覆盖这个标题的需求CREATE TABLE IF NOT EXISTS news ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, keyword VARCHAR(64) NOT NULL COMMENT 爬取时使用的关键字, title VARCHAR(512) NOT NULL COMMENT 新闻标题, url VARCHAR(1024) NOT NULL COMMENT 原文链接, url_hash CHAR(32) NOT NULL COMMENT url的md5用于唯一索引, source VARCHAR(32) NOT NULL COMMENT 来源平台: baidu / toutiao, media VARCHAR(128) DEFAULT COMMENT 发布媒体, publish_time DATETIME NULL COMMENT 发布时间解析失败可为空, summary TEXT COMMENT 列表页摘要, content MEDIUMTEXT COMMENT 详情页正文可不抓, crawl_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 入库时间, UNIQUE KEY uq_url_hash (url_hash), KEY idx_keyword_time (keyword, publish_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT新闻数据表;几个参数说明url 存 1024 是因为部分新闻链接带一堆追踪参数长度经常超过 255url_hash 唯一索引替代 url 唯一索引规避 MySQL 5.7 里 utf8mb4 下 varchar 唯一索引超过 768 字节就报错的问题。idx_keyword_time 联合索引是给“按关键字查询最近新闻”这个最核心的检索场景用的。source 字段区分数据来自百度还是今日头条后续排查问题才知道去哪边找原因。2.2 存储选型MySQL、SQLite 怎么选以及数据库连接池参数标题说“存如数据库”没说必须是 MySQL。这个选择直接影响后面所有代码。我的判断标准是数据量和并发单机小规模、数据量在十万级以下、只是自己练手SQLite 完全够要跑多关键字、多进程甚至分布式爬虫或者数据量朝百万级走直接上 MySQL。两者的差异可以用一张表说清楚维度SQLiteMySQL部署复杂度零部署一个文件需要安装服务端并发写入单写锁多进程并发写入会锁库支持并发写入连接池管理数据量上限五十万到一百万条内可用千万级无压力去重能力靠唯一索引效果一样靠唯一索引效果一样分布式爬虫不太合适标准选择如果你决定用 MySQLSQLAlchemy 连接引擎有三个参数必须调否则跑半小时就报连接错误from sqlalchemy import create_engine engine create_engine( mysqlpymysql://root:your_passwordlocalhost:3306/news_db?charsetutf8mb4, pool_size10, # 连接池保持 10 个连接 max_overflow20, # 峰值可再借 20 个连接 pool_pre_pingTrue, # 每次取连接前先探活防止 MySQL 断连后复用死连接 pool_recycle3600 # 连接超过 1 小时强制回收 )pool_size 和 max_overflow 决定并发上限pool_pre_ping 是血泪经验MySQL 默认 wait_timeout 八小时连接池里的连接睡太久会被服务端掐掉没这句配置就会间歇性拿到坏连接。如果你只是个人爬虫、单进程跑这几个参数不至于打满但配上能少踩很多坑。需要说明的是SQLite 不需要连接池一个 connect 到底就行但要注意多进程同时写同一个 SQLite 文件会锁库这也是为什么不建议在分布式爬虫里用 SQLite。3. 用 requests 稳定抓取百度新闻和今日头条搜索页Cookie、重试与频率控制网络爬虫原理不复杂模拟浏览器发请求拿回 HTML 或 JSON再解析成结构化数据。但百度新闻和今日头条的反爬策略不同百度主要靠频率限制今日头条主要靠 Cookie 和验证码。这一章的代码全部围绕“稳定”展开请求头伪装、Cookie 管理、超时重试、频率控制。3.1 百度新闻搜索tnnews 的参数与最小请求代码百度新闻没有独立的搜索域名走的是百度搜索的新闻垂直结果核心参数是 tnnews。请求地址是 https://www.baidu.com/s wd 是关键字pn 是分页偏移量每页通常 10 到 20 条。最小可用代码长这样import requests 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: text/html,application/xhtmlxml, Accept-Language: zh-CN,zh;q0.9, } def fetch_baidu_news(keyword: str, page_index: int): params { tn: news, # 指定新闻垂直搜索 word: keyword, # 搜索关键字 pn: page_index * 10, # 偏移量第 1 页是 0第 2 页是 10 } resp requests.get( https://www.baidu.com/s, paramsparams, headersHEADERS, timeout10 ) resp.raise_for_status() resp.encoding utf-8 return resp.text参数说明requests 的 params 会自动做 URL 编码不用手动 quote 关键字timeout10 是必须的不设超时会导致线程卡死。编码强制指定 utf-8避免百度返回的 HTML 里 meta 声明和实际编码不一致导致乱码。这里返回的是整个 HTML 页面解析逻辑放在下一章。注意百度搜索结果的链接是跳转链接入库前要不要跟随跳转拿真实 URL取决于下游用途需要保留原文地址的话再做一次 HEAD 跟随。3.2 今日头条搜索接口Cookie、Referer 和验证码兜底今日头条的搜索结果页是 JS 渲染的直接抓 HTML 拿不到数据常见做法是调它的搜索接口。接口路径大概长这样https://www.toutiao.com/api/search/content/ 用 POST 提交表单里带 keyword、offset、count、formatjson。这个接口的硬性要求是请求头必须带 Cookie 和 RefererCookie 来自浏览器登录状态Referer 填搜索页地址。def fetch_toutiao_news(keyword: str, offset: int): cookies { xx: 从真实浏览器复制来的 Cookie 值, } headers { User-Agent: HEADERS[User-Agent], Referer: fhttps://www.toutiao.com/search/?keyword{keyword}, Cookie: ; .join(f{k}{v} for k, v in cookies.items()), } data { aid: 24, app_name: toutiao_web, format: json, keyword: keyword, offset: offset, count: 20, cur_tab: 1, search_id: , } resp requests.post( https://www.toutiao.com/api/search/content/, headersheaders, datadata, timeout10 ) if resp.status_code 200: return resp.json() return None代码里的 Cookie 只是一个占位说明正确姿势是在浏览器里打开今日头条搜索页手动过掉一次滑块验证再从开发者工具里复制完整 Cookie 字符串填进来。Cookie 失效后接口不会报 403而是返回 200 但 data 是空数组或验证提示所以解析前必须做一层判断见 4.3 节。不建议写无 Cookie 硬刷的逻辑今日头条的风控对无状态请求特别敏感硬刷大概率二次请求就触发验证。3.3 请求频率与重试退避把“爬所有新闻”控制在安全曲线上“所有新闻”意味着大批量翻页一个关键字翻 50 页就是 50 次请求十个关键字就是 500 次。如果不管频率今天头条风格的站点会直接把 IP 拉进风控名单。我的经验是单请求间隔不低于 1.2 秒并在间隔里加随机抖动不要用固定 sleep。同时给请求套一层指数退避重试应对瞬时网络错误。import random import time from requests.exceptions import RequestException def request_with_retry(func, *args, **kwargs): for attempt in range(3): try: return func(*args, **kwargs) except (RequestException, TimeoutError): wait 2 ** attempt random.uniform(0, 1) time.sleep(wait) return None # 主循环里的调用方式 def crawl_keyword(keyword: str, max_pages: int): for page in range(max_pages): html request_with_retry(fetch_baidu_news, keyword, page) if html: # 解析入库逻辑省略见后续章节 pass time.sleep(random.uniform(1.2, 3.5))指数退避里的 2 的幂次方是标准做法第一次失败等约 2 秒第二次约 4 秒第三次约 8 秒给服务端留出恢复时间。主循环里的随机间隔区间不要随便压缩今日头条和百度对短时间高频请求都敏感。另外解析逻辑如果涉及再请求详情页详情页请求也要走同一个重试与限速逻辑否则列表页没被限制详情页反而先触发风控。注意频率控制不是玄学是爬虫能不能长期跑的分水岭。宁可每小时少抓几百条也别一次把站点请求打满。4. 解析新闻列表BeautifulSoup 选择器优先正则兜底抓下来的百度新闻是一整页 HTML今日头条是一坨 JSON两者解析方式完全不同。这一章的核心思路是“选择器优先正则兜底”结构稳定的部分用 BeautifulSoup 的选择器结构随机或嵌在 JS 里的部分用正则硬抠。4.1 选择器还是正则面对改版时的选型思路BeautifulSoup 的 select 和 find 依赖 HTML 的层级和 class 名只要站点改版就断正则不依赖结构只要文本模式还在就能继续工作。实际新闻站的规律是列表容器的 class 经常变但链接和时间的文本格式相对稳定。所以我一般这么判断页面里能靠 class 稳定命中的区域用选择器像 title 里混着高亮标签、时间戳格式不一致这类情况交给正则清洗。场景推荐方案原因class 名稳定的列表容器BeautifulSoup select代码可读性好层级清晰class 名带随机后缀正则匹配整体块选择器一改版就失效标题里的高亮标签正则去标签文本模式稳定JS 里内嵌的 JSON 数据正则 json.loads不依赖 DOM 渲染4.2 解析百度新闻 HTML从 result 容器里抠标题、来源、时间百度新闻搜索结果里每条新闻通常包在一个 class 为 result 的容器里。标题在 h3 的 a 标签内来源和时间在标题下方的信息行里但不同版式下 class 变化很大。解析代码按“容器 选择器 正则兜底”三层结构写from bs4 import BeautifulSoup import re def parse_baidu_news(html: str, keyword: str): soup BeautifulSoup(html, html.parser) results [] for item in soup.select(div.result): title_tag item.select_one(h3 a) if not title_tag: continue title re.sub(r[^], , str(title_tag)) title title.strip() # 百度结果里的链接是跳转链接先保留原样 url title_tag.get(href, ) # 来源和时间没有稳定 class用正则从容器文本里抓 source_match re.search(r(\S网)\s*$, item.get_text()) media source_match.group(1) if source_match else time_match re.search(r(\d{1,2}小时前|\d{4}年\d{1,2}月\d{1,2}日), item.get_text()) publish_time time_match.group(1) if time_match else results.append({ keyword: keyword, title: title, url: url, media: media, publish_time: publish_time, source: baidu, }) return results这个函数里的正则都是兜底逻辑标题先取出 HTML 再统一去标签是因为百度会给命中的关键字包高亮时间和来源不从固定 class 里取是因为百度新闻改版时最常动的就是这两个位置的样式。publish_time 这里存的是“x小时前”这类相对时间入库前需要转成绝对时间但转的时候拿不到“当前时间”会不准所以更稳的做法是遇到相对时间就用抓取时间减去偏移量遇到绝对时间直接解析。4.3 解析今日头条 JSONdata 数组的提取与空值容错今日头条接口返回的 JSON 里data 是一个数组每个元素包含 title、source、article_url、publish_time、content 等字段。难点不在取字段而在容错data 可能为空、字段可能缺失、title 里可能带 HTML 标签、publish_time 可能是秒级时间戳。代码里要全部兜住import json import re from datetime import datetime, timedelta def parse_toutiao_news(data: dict, keyword: str): results [] raw_items data.get(data) or [] for item in raw_items: if not isinstance(item, dict) or not item.get(title): continue title re.sub(r[^], , item.get(title, )).strip() url item.get(article_url) or item.get(url) or media item.get(source) or item.get(media_name) or publish_raw item.get(publish_time) or 0 publish_time None # 兼容秒级时间戳和毫秒级 if isinstance(publish_raw, (int, float)): ts publish_raw if publish_raw 10_000_000_000 else publish_raw * 1000 publish_time datetime.fromtimestamp(ts / 1000) results.append({ keyword: keyword, title: title, url: url, media: media, publish_time: publish_time, content: item.get(content, ), source: toutiao, }) return results时间戳那行逻辑是关键今日头条返回的时间戳偶尔是秒、偶尔是毫秒判断阈值设成 100 亿大于这个数按毫秒处理否则按秒处理。data 为空数组时直接返回空列表不要报错因为空数组往往意味着 Cookie 失效或触发了验证上层循环靠“返回结果数为 0 就停止翻页”来终止这里设计成容错返回会更安全。解析到这一步数据的 title、url、media、publish_time 都已经规整成表结构需要的类型可以直接进入库模块。5. 写入数据库SQLAlchemy 批量入库、URL 去重与断点续爬抓取和解析都稳定之后最后这道工序决定数据质量。新手最容易在这一步翻车一条条 insert 太慢、重复数据重复入库、跑到一半崩了没法续。这一章用 SQLAlchemy 解决批量写入和去重并用一张进度表实现断点续爬。5.1 建表 DDL 与 ORM 模型唯一索引定生死第 2 章的 DDL 已经把表建好了ORM 模型要和它严格对应。唯一索引 uq_url_hash 是整套去重方案的核心没有它先查后插只是心理安慰因为多进程下两个进程可能同时查到“不存在”然后同时插入。ORM 模型如下from sqlalchemy import ( Column, String, BigInteger, DateTime, Text, Index ) from sqlalchemy.ext.declarative import declarative_base from datetime import datetime Base declarative_base() class News(Base): __tablename__ news id Column(BigInteger, primary_keyTrue, autoincrementTrue) keyword Column(String(64), nullableFalse) title Column(String(512), nullableFalse) url Column(String(1024), nullableFalse) url_hash Column(String(32), nullableFalse, uniqueTrue) source Column(String(32), nullableFalse) media Column(String(128), default) publish_time Column(DateTime, nullableTrue) summary Column(Text, nullableTrue) content Column(Text, nullableTrue) crawl_time Column(DateTime, defaultdatetime.now) __table_args__ ( Index(idx_keyword_time, keyword, publish_time), )代码里的 uniqueTrue 会生成唯一索引保证数据库层面拒绝重复 url_hash。有个细节mysql 表中字段如果撞了“关键字”——这里说的是 SQL 保留字比如 key、order、group——会导致建表和查询都报语法错误所以模型里刻意把关键字字段命名为 keyword而不是 key。虽然 keyword 不算 MySQL 保留字但建表时反引号习惯不要丢。5.2 批量写入先查已存在 URL再批量提交一条一条 add 再 commit 的性能很差新闻爬虫一跑就是几百上千条必须批量提交。正确姿势是先按 url_hash 查出已存在的过滤掉再批量插入。代码如下import hashlib from sqlalchemy.orm import sessionmaker from sqlalchemy.dialects.mysql import insert SessionLocal sessionmaker(bindengine) def build_url_hash(url: str) - str: return hashlib.md5(url.encode(utf-8)).hexdigest() def batch_save_news(items: list[dict]): if not items: return # 第一步过滤已存在的 url_hash hashes [build_url_hash(item[url]) for item in items] with SessionLocal() as session: exist_rows session.query(News.url_hash).filter( News.url_hash.in_(hashes) ).all() exist_set {row[0] for row in exist_rows} new_items [ item for item, h in zip(items, hashes) if h not in exist_set ] # 第二步批量插入 for item in new_items: item[url_hash] build_url_hash(item[url]) session.add(News(**item)) session.commit()这里每个 News 对象单独 add 后统一 commitSQLAlchemy 会把它们合并成一个事务提交速度远快于逐条 commit。exist_set 的过滤在单进程下足够多进程下两个进程同时插入相同 url_hash 时唯一索引会抛 IntegrityError捕获后 rollback 即可数据不会重复。如果用的是 MySQL 方言还可以用 insert().on_duplicate_key_update() 做成幂等写入但先查后插已经能覆盖绝大多数场景代码也更清晰。提示批量入库和去重必须同时存在。先查后插管住流程唯一索引管住并发竞态缺一个都会有脏数据。5.3 断点续爬crawl_state 表记录每个关键字爬到第几页抓取“所有新闻”意味着要跑很久进程随时可能因为网络异常、内存问题、手动中断而挂掉。没有断点续爬每次崩溃都要从第 1 页重新开始。做法是加一张 crawl_state 表记录每个关键字已经抓到了第几页CREATE TABLE crawl_state ( keyword VARCHAR(64) PRIMARY KEY, next_offset INT NOT NULL DEFAULT 0, total_count INT DEFAULT 0, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;主循环从读取进度开始每抓完一页就更新一次class CrawlState(Base): __tablename__ crawl_state keyword Column(String(64), primary_keyTrue) next_offset Column(BigInteger, nullableFalse, default0) total_count Column(BigInteger, default0) updated_at Column(DateTime, defaultdatetime.now) def get_next_offset(keyword: str, session) - int: row session.query(CrawlState).filter_by(keywordkeyword).first() return row.next_offset if row else 0 def update_offset(keyword: str, new_offset: int, session): row session.query(CrawlState).filter_by(keywordkeyword).first() if row: row.next_offset new_offset row.updated_at datetime.now() else: session.add(CrawlState(keywordkeyword, next_offsetnew_offset)) session.commit()每页抓完立刻 commit 进度不要等整个关键字跑完再更新否则中途断电等于没记。断点续爬配合 url_hash 去重还有一个好处就算进度表丢了、从第 1 页重跑已存在的记录会被批量写入模块自动过滤不会产生重复数据。这一套逻辑也天然适合分布式爬虫多进程各领不同关键字共用同一张 news 表去重靠唯一索引进度靠 crawl_state 表全程不需要额外的队列组件。6. 避坑清单5 个翻车现场与一条自检 SQL这个标题方向的坑高度集中下面 5 条是我反复见过的翻车现场每条都按现象、原因、解决的套路写。坑 1字段名撞保留字现象建表 SQL 报语法错误或查询时 order by 报错。原因把字段命名为 key、order、group 这类保留字。keyword 本身不是保留字但 Python 标准库里有个 keyword 模块import keyword 也会撞名。解决字段统一叫 keyword避免直接用 key/order 做列名必要时 SQL 里用反引号包住。坑 2今日头条 Cookie 失效被验证码截住现象请求返回 200但 data 是空数组或者返回页面里写着“安全验证”。原因没有有效登录态服务端把人机校验挡在数据前面。解决浏览器手动过一次滑块验证把完整 Cookie 复制进爬虫发现连续两页 data 为空就停止从浏览器重新复制 Cookie别在无 Cookie 状态下硬刷。坑 3重复数据爆炸现象第二天重跑库里同一篇新闻出现两次甚至更多。原因开始没建唯一索引用“标题是否相同”来去重新闻标题一改版就失效。解决建 url_hash 唯一索引写入前先查已存在的 url_hash双保险兜底。坑 4MySQL 连接耗尽现象跑半小时后报 “Too many connections”。原因循环里反复新建 engine、session 不关闭或者连接池太小。解决全局只建一个 enginesession 用 with 语句包裹设置 pool_size10、max_overflow20、pool_pre_pingTrue。坑 5没做断点崩溃后从头再来现象网络抖动导致进程退出重启后从第 1 页开始前几小时白跑。原因页码存在内存变量里没有持久化。解决用 crawl_state 表记录每个关键字的 next_offset每抓完一页就 commit重跑时从记录的位置继续。跑完一批数据后用下面这条 SQL 做自检能同时确认“所有新闻”真的入库且没有明显重复SELECT keyword, COUNT(*) AS total_cnt, COUNT(DISTINCT url_hash) AS uniq_cnt, COUNT(DISTINCT source) AS source_cnt, MIN(publish_time) AS earliest_time, MAX(publish_time) AS latest_time FROM news GROUP BY keyword;total_cnt 大于 uniq_cnt 说明有重复source_cnt 小于 2 说明某个平台可能整批抓取失败earliest_time 和 latest_time 距离太近说明时间字段解析串位。抽一条 title 去原文站点核对确认解析没有张冠李戴。我自己早年的习惯是一上来就写全量循环被上面五个坑轮番教育后固定套路已经是小样本验证、入库去重、断点续爬、跑完自检四步走。希望帮到你。本文还有配套的精品资源点击获取
