Python爬虫实战：央视新闻国内版块数据采集与结构化存储

1. 项目缘起与整体设计思路做数据采集这行十来年我接过不少需求但“爬取央视新闻国内版块”这个活儿第一次听到时还是让我多琢磨了一会儿。原因很简单新闻类站点和电商、论坛、招聘网站完全不是一个路数。它的页面结构相对规整但内容更新频率极高栏目层级多而且反爬策略往往不是靠复杂的登录验证而是靠访问频率、请求头特征和页面动态渲染来做拦截。换句话说你如果拿爬商品详情页那套思路直接套过来大概率跑不了几百条就会被限流。这个项目的核心目标很明确把央视新闻国内版块下各个子栏目的新闻条目抓下来包括标题、发布时间、来源、正文链接如果进一步扩展还可以把正文内容也一并落库。它解决的是什么问题对于做舆情分析、内容聚合、选题研究、媒体监测的团队来说手动复制粘贴显然不现实而市面上通用的采集工具往往对新闻站点的适配不够精细抓下来的数据要么字段缺失要么时间格式混乱要么去重逻辑一塌糊涂。所以自己动手写一套针对性的采集方案反而是最稳妥的路子。适合谁来参考我认为有三类人第一类是有一定编程基础、想练手数据采集的开发者这个项目结构清晰、难度适中非常适合作为第一个完整的采集项目第二类是做内容运营或数据分析的朋友需要定期获取新闻数据但不想依赖第三方服务第三类是做技术调研的同行想看看新闻类站点采集和普通站点采集在策略上到底有什么区别。整体设计上我采用的是“列表页解析 详情页按需抓取 本地结构化存储”的三段式架构。为什么不直接上分布式爬虫框架因为央视新闻国内版块的数据量级并没有大到需要多节点协同的程度单机加合理的并发控制完全够用。而且新闻站点的反爬机制通常对高频请求非常敏感分布式反而容易触发风控。所以我选择用轻量级的请求库配合解析库加上一套自己写的调度和去重逻辑既可控又容易调试。在技术选型上请求层我用的是requests原因是它足够成熟配合Session对象可以很方便地维持会话和复用连接。解析层用的是BeautifulSoup配合lxml解析器虽然有人推荐parsel或直接上Scrapy但对于这种页面结构相对固定的新闻站点BeautifulSoup的容错性和可读性更胜一筹。存储层我选了SQLite轻量、无需额外部署适合中小规模的数据落地后期要迁移到MySQL或PostgreSQL也很方便。如果你需要做全文检索可以在存储之后再加一层Elasticsearch但那是后话。还有一个关键设计决策列表页和详情页分离处理。列表页只负责抓取新闻条目的元信息标题、链接、时间、来源详情页则按需抓取正文。这样做的好处是当你只需要做标题级别的分析时完全不用碰详情页采集效率能提升好几倍。而且详情页的请求可以单独控制频率避免因为正文抓取拖慢整体进度。注意新闻站点的页面结构会不定期调整所以解析逻辑一定要写得足够健壮不能把某个class名写死到无法维护的程度。我的做法是把所有选择器集中配置在一个字典里页面改版时只需要改配置不用动核心代码。2. 核心细节解析与实操要点2.1 目标页面结构分析与字段定义在动手写代码之前我花了大概半小时把国内版块的列表页和详情页结构摸了一遍。国内版块通常是一个聚合页面下面会分多个子栏目比如时政、社会、经济、科技等。每个子栏目有自己的列表页列表页上每条新闻一般包含这几个字段标题、缩略图可选、发布时间、来源、详情页链接。有些条目还会带一个简短的摘要。这里有个细节值得注意发布时间格式不统一。有的条目显示“2024-06-15 14:30”有的只显示“14:30”还有的显示“昨天”“前天”这种相对时间。如果你直接存字符串后期做时间排序和分析时会非常痛苦。所以我在解析阶段就做了一层时间归一化处理把所有时间统一转换成YYYY-MM-DD HH:MM:SS格式。对于相对时间我会结合当前日期做推算虽然不能保证百分之百精确但误差通常在一分钟以内对于新闻分析来说完全够用。字段定义我整理成了下面这张表方便你在建库和写解析逻辑时对照字段名类型说明是否必填title文本新闻标题是url文本详情页链接是publish_time时间归一化后的发布时间是source文本新闻来源否category文本所属子栏目是summary文本摘要否content文本正文内容否crawl_time时间采集时间是提示url字段建议加唯一索引这是去重的第一道防线。crawl_time虽然看起来不起眼但在排查数据问题时非常有用能帮你判断某条数据是什么时候抓进来的。2.2 请求头与访问频率控制策略新闻站点对请求头的检查比一般站点严格。我实测下来如果只带一个默认的User-Agent前几十个请求可能没问题但很快就会遇到返回空内容或者直接拒绝的情况。所以请求头至少要包含以下几项User-Agent建议准备一组常见的浏览器 UA随机轮换使用。Referer设置为国内版块的首页地址模拟从首页点进去的正常行为。Accept设置为text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8。Accept-Language设置为zh-CN,zh;q0.9。访问频率这块我的经验是列表页请求间隔控制在 2 到 4 秒之间详情页请求间隔控制在 1 到 2 秒之间。这个频率听起来很慢但新闻站点的更新速度并没有你想象的那么快国内版块一天新增的条目通常也就几百条用这个频率完全能在合理时间内抓完。如果你追求更高的效率可以适当缩短间隔但建议不要低于 1 秒否则很容易触发风控。另外我强烈建议使用Session对象来复用 TCP 连接。这不仅能让请求看起来更像正常用户的连续访问还能减少握手开销提升整体效率。下面是我常用的请求头配置代码片段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/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36, Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/115.0, ] def build_session(): session requests.Session() session.headers.update({ User-Agent: random.choice(USER_AGENTS), Referer: https://news.example.com/domestic/, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9, }) return session注意UA 不要频繁切换同一个 Session 内最好保持一致。频繁切换 UA 反而会让站点觉得异常因为正常用户不会在几分钟内换好几个浏览器。2.3 列表页解析的关键技巧列表页解析的核心任务是提取每一条新闻的链接和元信息。这里最容易踩的坑是选择器写得太死。比如你写div.news-list ul li a一旦页面结构调整整个解析就崩了。我的做法是用“语义化选择器 多级回退”的策略。先尝试用最精确的选择器如果匹配不到再降级到更宽泛的选择器。具体来说我会先定位到新闻列表的容器然后在容器内遍历每一条新闻。对于标题和链接优先找a标签的href和文本内容对于时间优先找带有时间特征的span或em标签。如果某个字段解析失败不会让整个条目丢弃而是把该字段置空保证数据不丢失。还有一个细节列表页可能存在分页或滚动加载。央视新闻国内版块的分页通常是 URL 参数控制比如?page2。你可以通过分析分页链接的规律来构造请求。如果遇到滚动加载那就需要分析接口请求找到返回 JSON 数据的接口地址直接请求接口比解析 HTML 更高效。2.4 详情页正文提取的容错方案详情页的正文提取是另一个难点。新闻正文通常包裹在div或article标签里但不同子栏目的模板可能略有差异。我的做法是先用一组候选选择器去匹配正文容器匹配到之后再对容器内的p标签做遍历过滤掉空行、广告、版权声明等无关内容。这里有个经验正文中经常夹杂图片说明、编辑署名、责任编辑等信息。这些内容要不要保留取决于你的用途。如果做全文检索建议保留如果做纯文本分析可以过滤掉。我一般会保留但在存储时加一个标记字段方便后期筛选。另外详情页的编码问题也要注意。有些页面返回的编码声明和实际编码不一致导致中文乱码。我的处理方式是先让requests自动推断编码如果推断结果不可靠就手动指定为utf-8或gbk然后检查解码后的文本是否包含乱码字符。3. 实操过程与核心环节实现3.1 环境准备与依赖安装这个项目对环境的依赖非常轻量Python 3.8 以上版本即可。我建议用虚拟环境来管理依赖避免和系统里的其他项目冲突。下面是完整的依赖清单pip install requests2.31.0 pip install beautifulsoup44.12.2 pip install lxml4.9.3如果你打算把数据存到SQLitePython 标准库自带sqlite3不需要额外安装。如果你需要更强大的数据处理能力可以再加一个pandas但这不是必须的。提示lxml在某些系统上安装时可能需要编译工具如果遇到问题可以改用html.parser作为解析器虽然速度慢一点但兼容性更好。3.2 数据库表结构设计与初始化存储层我用的是SQLite表结构设计如下CREATE TABLE IF NOT EXISTS news ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, url TEXT NOT NULL UNIQUE, publish_time TEXT, source TEXT, category TEXT, summary TEXT, content TEXT, crawl_time TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_publish_time ON news(publish_time); CREATE INDEX IF NOT EXISTS idx_category ON news(category);这里有几个设计考量url加唯一索引是为了去重避免重复抓取同一条新闻publish_time和category加普通索引是为了加速查询后期做时间范围筛选或栏目筛选时会明显更快crawl_time不加索引因为它主要用于排查问题查询频率不高。3.3 列表页采集与分页处理列表页采集的流程可以拆成四步构造请求、解析页面、提取字段、翻页。我以国内版块某个子栏目为例假设它的列表页 URL 是https://news.example.com/domestic/politics/分页参数是page。import time import requests from bs4 import BeautifulSoup def fetch_list_page(session, base_url, page): url f{base_url}?page{page} try: resp session.get(url, timeout10) resp.raise_for_status() resp.encoding resp.apparent_encoding return resp.text except requests.RequestException as e: print(f请求失败: {url}, 错误: {e}) return None def parse_list_page(html, category): soup BeautifulSoup(html, lxml) items [] # 多级回退选择器 container soup.select_one(div.news-list) or soup.select_one(ul.news-list) or soup for li in container.select(li): a_tag li.select_one(a) if not a_tag or not a_tag.get(href): continue title a_tag.get_text(stripTrue) url a_tag[href] if not url.startswith(http): url https://news.example.com url time_tag li.select_one(span.time) or li.select_one(em.time) publish_time time_tag.get_text(stripTrue) if time_tag else source_tag li.select_one(span.source) or li.select_one(em.source) source source_tag.get_text(stripTrue) if source_tag else items.append({ title: title, url: url, publish_time: publish_time, source: source, category: category, }) return items翻页逻辑我一般设置一个最大页数限制比如 50 页避免因为分页参数异常导致无限循环。每翻一页sleep2 到 4 秒模拟正常浏览节奏。3.4 详情页正文抓取与时间归一化详情页抓取是在列表页数据入库之后进行的。我会先从数据库里查出所有content为空的记录然后逐条抓取正文。这样做的好处是如果中途中断下次可以接着上次的进度继续不用从头再来。import re from datetime import datetime, timedelta def normalize_time(raw_time): raw_time raw_time.strip() now datetime.now() # 处理 14:30 这种只有时间的格式 if re.match(r^\d{1,2}:\d{2}$, raw_time): hour, minute map(int, raw_time.split(:)) dt now.replace(hourhour, minuteminute, second0, microsecond0) if dt now: dt - timedelta(days1) return dt.strftime(%Y-%m-%d %H:%M:%S) # 处理 昨天 14:30 这种格式 if raw_time.startswith(昨天): time_part raw_time.replace(昨天, ).strip() if re.match(r^\d{1,2}:\d{2}$, time_part): hour, minute map(int, time_part.split(:)) dt (now - timedelta(days1)).replace(hourhour, minuteminute, second0, microsecond0) return dt.strftime(%Y-%m-%d %H:%M:%S) # 处理标准格式 for fmt in (%Y-%m-%d %H:%M:%S, %Y-%m-%d %H:%M, %Y-%m-%d): try: return datetime.strptime(raw_time, fmt).strftime(%Y-%m-%d %H:%M:%S) except ValueError: continue return raw_time正文提取部分我会先定位正文容器然后遍历其中的段落标签过滤掉空段落和明显的广告文本。这里有个小技巧正文段落通常长度较长而广告和版权声明往往很短。所以我会设置一个最小长度阈值比如 10 个字符低于这个长度的段落直接跳过。3.5 数据入库与去重逻辑入库时我用的是INSERT OR IGNORE配合url字段的唯一索引天然实现去重。如果某条新闻已经存在数据库会自动忽略不会报错。这样即使重复抓取同一个列表页也不会产生重复数据。import sqlite3 def save_news(items): conn sqlite3.connect(news.db) cursor conn.cursor() for item in items: cursor.execute( INSERT OR IGNORE INTO news (title, url, publish_time, source, category, crawl_time) VALUES (?, ?, ?, ?, ?, ?) , ( item[title], item[url], normalize_time(item[publish_time]), item[source], item[category], datetime.now().strftime(%Y-%m-%d %H:%M:%S), )) conn.commit() conn.close()注意INSERT OR IGNORE虽然方便但它不会更新已有记录。如果你希望每次抓取都更新crawl_time那就需要改用INSERT OR REPLACE但这样会丢失自增id。我的建议是保持INSERT OR IGNORE如果需要更新时间单独写一条UPDATE语句。4. 常见问题与排查技巧实录4.1 请求被拒绝或返回空内容这是最常见的问题表现是resp.status_code返回 403 或 200 但内容为空。排查思路如下现象可能原因解决方法403 Forbidden请求头缺失或 UA 被识别补全 Referer、Accept 等头部更换 UA200 但内容为空触发频率限制增大请求间隔降低并发200 但返回验证页站点要求验证检查是否需要 Cookie适当增加访问间隔连接超时网络不稳定或站点限流增加 timeout重试 2 到 3 次我的经验是遇到 403 先别急着换 IP先检查请求头。很多时候补上Referer就能解决。如果补了还不行再考虑降低频率。4.2 解析结果为空或字段缺失解析失败通常有两个原因选择器写错了或者页面结构变了。排查时我会先把 HTML 保存到本地然后用BeautifulSoup在本地反复调试选择器。这样比在线调试快得多也不会因为反复请求触发风控。另一个常见问题是编码错误导致中文乱码。如果你看到类似æ–°é—»这样的字符说明编码推断错了。解决办法是手动指定编码或者用resp.content.decode(utf-8, errorsignore)来强制解码。4.3 时间格式混乱导致排序异常前面提到过新闻站点的时间格式不统一。如果你不做归一化直接存字符串那么按时间排序时会出现“14:30”排在“2024-06-15”前面的情况。我的建议是在入库前就完成归一化数据库里只存标准格式。对于无法识别的时间可以存原始字符串但加一个标记字段方便后期人工修正。4.4 数据重复与漏抓数据重复通常是因为列表页翻页时出现了重复条目或者详情页被多次抓取。解决办法就是前面说的url唯一索引加INSERT OR IGNORE。漏抓则可能是因为分页逻辑有误或者某些条目被选择器过滤掉了。我一般会在采集完成后做一个统计对比列表页显示的条目总数和实际入库数量如果差距较大就说明有漏抓需要检查选择器。提示建议每次采集完成后输出一份简单的统计报告包括本次抓取条数、新增条数、跳过条数、失败条数。这样能快速判断采集是否正常。4.5 正文提取不完整或包含无关内容正文提取不完整通常是因为正文容器选择器匹配到了错误的div。解决办法是打印出候选容器的文本长度选择最长的那个作为正文容器。如果正文中包含了广告或版权声明可以通过关键词过滤比如包含“责任编辑”“版权声明”“免责声明”的段落直接跳过。还有一个容易被忽略的问题正文中的图片和视频。如果你只需要文本那直接忽略img和video标签即可。如果需要保留图片链接可以在解析时单独提取src属性存到一个附加字段里。5. 采集策略优化与扩展思路5.1 增量采集与定时任务新闻站点的内容是持续更新的所以没必要每次都全量抓取。我的做法是记录上次采集的时间戳每次只抓取比这个时间戳更新的条目。具体实现上可以在列表页解析时判断发布时间如果早于上次采集时间就停止翻页。这样能大幅减少请求量也能降低触发风控的概率。定时任务可以用cron或schedule库来实现。我一般设置为每小时跑一次每次只抓最新的一两页。这样既能保证数据及时性又不会给站点造成压力。5.2 数据清洗与结构化增强抓下来的原始数据往往需要进一步清洗。比如标题中可能包含多余的空格、换行符正文中可能包含 HTML 实体字符。我一般会用正则表达式做一轮清洗把nbsp;、amp;这类实体替换成正常字符把连续的空格合并成一个。如果要做更深入的分析还可以对正文做分词和关键词提取。这部分可以用jieba做中文分词然后用TF-IDF或TextRank提取关键词。不过这就超出了采集本身的范围属于数据应用的范畴了。5.3 存储方案升级与查询优化SQLite适合中小规模数据但如果你的数据量超过百万条查询性能会明显下降。这时候可以考虑迁移到MySQL或PostgreSQL。迁移时只需要改一下连接配置和建表语句核心逻辑不用动。如果要做全文检索可以在存储之后再加一层Elasticsearch。把新闻标题和正文导入ES就能实现毫秒级的全文搜索。不过ES的部署和维护成本较高建议在数据量确实很大、检索需求确实很强时再考虑。5.4 合规采集的边界与注意事项做数据采集合规是底线。我的原则是只采集公开可见的内容不绕过任何登录或验证机制不采集个人隐私信息不将数据用于商业转售。采集频率要控制在合理范围内不对目标站点造成明显压力。如果站点提供了robots.txt建议先看一下相关约定虽然它不具备法律强制力但尊重站点意愿是基本的行业礼仪。另外采集到的新闻内容版权归原站点所有如果要在自己的产品中展示需要获得授权。如果只是做内部研究或分析一般问题不大但也要注意不要大规模复制传播。6. 实操心得与避坑经验这个项目我前后迭代了三个版本踩过的坑不算少这里挑几个最有代表性的分享一下。第一个坑是过度依赖单一选择器。第一版代码里我把列表页的选择器写成了div.list ul li a结果站点改版后整个采集直接挂掉。后来我改成了多级回退策略虽然代码复杂了一点但稳定性提升非常明显。这个教训让我明白采集代码的健壮性比简洁性更重要。第二个坑是忽略时间归一化。第一版直接把时间字符串存进数据库后来做时间范围查询时发现排序完全乱套。补上归一化逻辑后查询效率和分析准确性都上了一个台阶。所以我现在做任何采集项目时间字段一定是重点处理对象。第三个坑是没有做增量采集。早期版本每次都是全量抓取跑一次要十几分钟而且经常触发风控。后来改成增量采集每次只抓最新内容跑一次不到一分钟风控问题也基本消失了。这个优化带来的收益远超我的预期。最后一个心得是日志要写清楚。采集过程中难免会遇到各种异常如果日志只记录“请求失败”排查起来会很痛苦。我现在的做法是记录完整的 URL、状态码、错误信息和时间戳这样出问题时能快速定位。日志不用写得太花哨但关键信息一个都不能少。如果你也在做类似的新闻采集项目我的建议是先把列表页跑通确保能稳定拿到标题和链接然后再逐步加上详情页抓取、时间归一化、增量采集这些功能。不要一上来就追求大而全小步快跑、逐步迭代反而更容易做出稳定可靠的东西。