新闻采集订阅系统实战：从爬虫到推送的工程化落地

简介：这份毕业设计资料围绕「基于网络爬虫的新闻采集和订阅系统」展开，面向计算机相关专业需要完成毕设的本科生，以及想系统实践爬虫、数据处理与前后端集成的开发者。项目从需求分析、技术选型到系统设计均有覆盖，涉及Scrapy爬虫框架、MongoDB数据存储、订阅推送逻辑与前端展示等模块，适合作为完整课题参考或二次开发基础。资源包共65个文件，以33个Python源码为核心，辅以17张PNG与4张JPEG架构图、流程图和界面截图，另有HTML页面、TXT说明、SH启动脚本及PDF论文终稿等，压缩后约7.03MB，目录按爬虫、订阅展示、API服务等模块划分，结构清晰。目前已有114人学习下载。读者可从中获取可运行的爬虫与订阅系统代码、数据库初始化脚本、论文撰写素材以及系统总体框架、用例图、活动图等设计文档，便于快速理解项目全貌并对照完成自己的毕设实现。

1. 从“能跑”到“能交”：新闻采集订阅系统的真实落地边界

很多同学做毕业设计，选题定了“基于网络爬虫的新闻采集和订阅系统”，第一反应就是打开编辑器写requests.get()，三天后跑通了抓取，却卡在“订阅”两个字上——用户订阅了“人工智能”这个关键词，系统到底该在什么时候、以什么频率、把哪些新闻推给他？这背后其实是一整套从采集调度、去重存储、订阅匹配到推送触达的工程链路，而不是一个爬虫脚本能兜住的。网络爬虫原理决定了你能拿到什么数据，订阅系统设计决定了这些数据怎么变成用户可感知的价值。这篇文章面向正在做这个题目的本科生和刚入行的后端开发者，把我在实际项目中趟过的选型逻辑、参数配置和踩坑记录摊开讲，让你少走弯路，把“能跑”变成“能交”。

2. 采集层怎么搭：从目标站点分析到增量抓取调度

2.1 先定数据源类型，再选采集策略

新闻采集的第一步不是写代码，而是搞清楚你要采的是什么类型的站点。常见的有三类：静态新闻列表页（如传统门户的频道页）、动态渲染页面（内容由 JavaScript 加载）、以及提供 RSS 或开放接口的站点。三类站点的采集策略完全不同，选错了后面全是返工。

静态列表页最省事，用requests+BeautifulSoup或lxml就能解析。动态渲染页面需要上无头浏览器，常见做法是Playwright或Selenium，但资源消耗大，不适合高频调度。有 RSS 的站点优先用 RSS，因为结构稳定、解析成本极低，feedparser一行就能拿到标题、链接、发布时间。

我一般会先花半天时间做站点调研，把每个目标站点的页面结构、更新频率、是否有反爬（验证码、频率限制、UA 检测）记录下来，形成一张采集配置表。这张表直接决定后面的调度策略和解析规则。

2.2 增量抓取的核心：URL 去重与时间戳水位线

全量抓取只适合第一次初始化。之后每次调度都必须是增量的，否则你的数据库会被重复数据撑爆，订阅推送也会变成灾难。增量抓取有两个关键机制：URL 去重和时间戳水位线。

URL 去重最简单的方式是用 Redis 的 Set 或者布隆过滤器。每抓到一个新链接，先查是否已存在，不存在才进入解析队列。布隆过滤器省内存但有误判率，对于新闻采集场景，误判几个链接影响不大，我一般用 Redis Set 就够了，毕竟新闻量级不会大到需要布隆过滤器。

时间戳水位线是另一个维度：记录每个站点上次抓取到的最新文章发布时间，下次只抓比这个时间新的内容。这个方式对列表页按时间倒序排列的站点特别有效。

import redis import hashlib from datetime import datetime r = redis.Redis(host='localhost', port=6379, db=0) def is_new_url(url: str) -> bool: """用 Redis Set 做 URL 去重，返回 True 表示新链接""" url_hash = hashlib.md5(url.encode()).hexdigest() # sadd 返回 1 表示添加成功（新链接），0 表示已存在 return r.sadd('crawled_urls', url_hash) == 1 def get_last_crawl_time(site_id: str) -> str: """获取某站点上次抓取的水位线时间""" ts = r.get(f'watermark:{site_id}') return ts.decode() if ts else '1970-01-01 00:00:00' def update_watermark(site_id: str, latest_time: str): """更新水位线，只允许前进不允许后退""" old = get_last_crawl_time(site_id) if latest_time > old: r.set(f'watermark:{site_id}', latest_time)

这段代码里is_new_url用 MD5 哈希压缩 URL 长度，避免 Redis 里存超长字符串。update_watermark做了单调递增保护，防止因为解析异常导致水位线回退，进而重复抓取大量旧文章。参数上，Redis 的db建议单独分配一个库给爬虫用，不要和业务缓存混在一起，方便清理和监控。

2.3 调度器选型：APScheduler 够用，别上 Celery

毕业设计阶段，调度器用APScheduler完全够。它轻量、纯 Python、支持内存和 Redis 两种 job store，配置起来十分钟搞定。Celery 虽然强大，但需要额外的 broker（RabbitMQ 或 Redis）和 worker 进程管理，对于单机部署的毕设系统来说，复杂度收益比太低。

from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.jobstores.redis import RedisJobStore jobstores = { 'default': RedisJobStore(host='localhost', port=6379, db=1) } scheduler = BackgroundScheduler(jobstores=jobstores) # 每 10 分钟跑一次采集任务 scheduler.add_job( func=run_crawl_task, trigger='interval', minutes=10, id='news_crawl', max_instances=1, # 防止上一次没跑完就启动下一次 coalesce=True # 积压的任务合并成一次执行 ) scheduler.start()

max_instances=1是关键参数。新闻采集任务如果上一次还没跑完（比如网络慢），下一次调度又启动了，会导致同一批 URL 被重复处理。coalesce=True则是在任务积压时只执行一次，避免雪崩。RedisJobStore让调度状态持久化，重启服务后任务不会丢。

3. 订阅系统怎么设计：从关键词匹配到推送触达

3.1 订阅模型：用户、关键词、频次三要素

订阅系统的核心模型不复杂，但设计不好后面全是坑。最基本的三个要素是：用户 ID、订阅关键词（或分类）、推送频次。用户订阅“人工智能”这个关键词，系统需要在每次采集完成后，把新文章标题和摘要跟这个关键词做匹配，命中的文章进入该用户的待推送队列。

关键词匹配有两种做法：精确匹配和分词匹配。精确匹配就是if keyword in article.title，简单但漏报率高，比如用户订阅“AI”，文章标题写的是“人工智能”，就匹配不上。分词匹配用jieba对标题和摘要做分词，然后看关键词是否在分词结果里，召回率更高。

import jieba def match_keywords(article_title: str, article_summary: str, keywords: list) -> list: """返回文章中命中的关键词列表""" text = article_title + ' ' + (article_summary or '') words = set(jieba.cut(text)) matched = [] for kw in keywords: # 精确匹配标题 if kw in article_title: matched.append(kw) # 分词匹配摘要 elif kw in words: matched.append(kw) return matched

这里先做标题精确匹配，再做摘要分词匹配，兼顾准确率和召回率。jieba.cut默认精确模式，对于新闻标题这种短文本足够用。如果关键词本身是英文或中英混合，建议在分词前统一转小写，避免大小写导致的漏匹配。

3.2 推送频次控制：别让用户被消息淹没

用户订阅了关键词，不代表每篇匹配的文章都要立刻推。如果一次采集命中 50 篇文章，全推给用户，第二天他就取消订阅了。推送频次控制是订阅系统能不能留住人的关键。

常见做法是三种模式：实时推送（命中即推）、摘要推送（按频次汇总，如每天早中晚各一次）、以及混合模式（高优先级关键词实时推，普通关键词摘要推）。毕业设计里我建议至少实现摘要推送，用定时任务把待推送队列里的文章按用户聚合，一次性发送。

from collections import defaultdict def build_digest(user_articles: dict) -> dict: """把待推送文章按用户聚合成摘要""" digest = defaultdict(list) for user_id, articles in user_articles.items(): # 按发布时间倒序，取最新的 10 篇 sorted_arts = sorted(articles, key=lambda x: x['publish_time'], reverse=True) digest[user_id] = sorted_arts[:10] return digest

defaultdict省去了初始化空列表的判断。取最新 10 篇是一个经验值，太多用户看不完，太少又显得内容不够。这个数字可以做成用户可配置的，但在毕设阶段写死 10 也说得过去。

3.3 存储选型：MySQL 存元数据，Redis 存队列

新闻文章的元数据（标题、链接、发布时间、来源、摘要）用 MySQL 存，因为需要做条件查询和分页。待推送队列用 Redis 的 List 或 Sorted Set 存，因为读写频繁且不需要持久化保证。用户订阅关系用 MySQL 存，因为数据量小但一致性要求高。

表结构上，至少需要三张表：articles（文章元数据）、subscriptions（用户订阅关系）、push_logs（推送记录，用于去重和排查）。articles表的url字段加唯一索引，配合前面的 Redis 去重形成双保险。

注意：MySQL 的url字段长度建议设为 512 或 768，用utf8mb4字符集。有些新闻站点的 URL 带大量参数，255 长度不够用，插入时会报错。

4. 避坑与排查：新闻采集订阅系统最常见的 5 个翻车现场

4.1 现象：采集任务跑着跑着就卡死，日志停在某一条

原因通常是某个站点的响应超时没有设置，requests.get()默认没有超时，遇到慢站点会一直等。另一个可能是解析逻辑里用了while循环但没有退出条件，比如翻页时下一页链接解析失败导致死循环。

解决：所有网络请求必须加timeout参数，建议(5, 15)元组，分别代表连接超时和读取超时。翻页逻辑加最大页数限制，比如max_pages=50，超过就强制退出并记录警告日志。

4.2 现象：同一篇文章被重复推送多次

原因有两个层面：一是采集层 URL 去重失效，比如 Redis 被清空或者用了不同的哈希算法；二是推送层没有记录推送日志，同一个用户同一篇文章被多次匹配。

解决：采集层用 Redis Set 做去重的同时，MySQL 的articles表url字段加唯一索引，插入时用INSERT IGNORE或ON DUPLICATE KEY UPDATE。推送层在push_logs表里记录(user_id, article_id)唯一索引，推送前先查是否已推过。

4.3 现象：中文关键词匹配不到，英文关键词正常

原因是编码问题。有些站点的响应内容没有在 headers 里声明 charset，requests会默认用 ISO-8859-1 解码，导致中文变成乱码。乱码后的文本跟关键词自然匹配不上。

解决：在requests.get()之后手动设置response.encoding = response.apparent_encoding，或者根据站点实际情况指定response.encoding = 'utf-8'。apparent_encoding用 chardet 库探测编码，准确率较高但速度稍慢，适合在调试阶段用。

4.4 现象：APScheduler 任务在服务器上跑几天后不再触发

原因是BackgroundScheduler在长时间运行后可能因为内存泄漏或线程阻塞导致调度线程挂掉。另一个常见原因是 Redis 连接断开后没有重连机制。

解决：给调度器加一个心跳监控任务，每 5 分钟检查一次调度器状态，如果发现异常就重启调度器。Redis 连接用连接池并设置health_check_interval=30，自动检测并重连。

4.5 现象：推送摘要里文章顺序混乱，用户看到的是旧闻

原因是publish_time字段解析失败，存进去的是空值或默认值，排序时这些文章被排到了前面或后面。不同站点的发布时间格式不一样，有的是2024-01-15 10:30，有的是15 Jan 2024，有的是时间戳。

解决：写一个统一的时间解析函数，用dateutil.parser.parse做兜底解析，解析失败时记录原始字符串并跳过该文章，不要存空值。排序前先过滤掉publish_time为空的数据。

5. 让系统更耐用的两个进阶技巧：代理池与增量摘要去重

5.1 代理池不是必须的，但有了会稳很多

毕业设计阶段，如果目标站点不多、采集频率不高，不用代理池也能跑。但如果你采集的站点超过 10 个，或者某个站点对同一 IP 有频率限制，代理池就是刚需。代理池的核心不是“有很多代理”，而是“能自动剔除失效代理”。

我一般用 Redis 的 Sorted Set 存代理，score 存代理的可用性评分。每次请求成功加 1 分，失败减 2 分，低于阈值就剔除。每次取代理时从高分往低分取，避免随机取到失效代理。

import redis import requests r = redis.Redis(host='localhost', port=6379, db=2) def get_proxy() -> str: """从代理池取评分最高的代理""" proxies = r.zrevrange('proxies', 0, 0, withscores=True) if not proxies: return None return proxies[0][0].decode() def report_proxy(proxy: str, success: bool): """上报代理使用结果，动态调整评分""" if success: r.zincrby('proxies', 1, proxy) else: r.zincrby('proxies', -2, proxy) # 评分低于 -10 的代理直接移除 score = r.zscore('proxies', proxy) if score and score < -10: r.zrem('proxies', proxy)

zrevrange取评分最高的代理，zincrby动态调整评分。失败扣 2 分、成功加 1 分的设计，意味着代理需要保持 2/3 以上的成功率才能留在池子里。这个比例可以根据实际情况调整，但不要设得太宽松，否则失效代理会拖慢整体采集速度。

5.2 摘要去重：让用户不看到重复内容

新闻站点经常转载同一篇文章，标题可能略有不同，但内容高度相似。如果用户订阅的关键词命中了这些转载文章，摘要里会出现多条相似内容，体验很差。解决思路是对摘要做相似度去重。

最简单的方式是用difflib.SequenceMatcher计算标题相似度，超过 0.8 的视为重复，只保留发布时间最早的那条。这个方案不需要额外依赖，性能也够用。

from difflib import SequenceMatcher def deduplicate_articles(articles: list, threshold: float = 0.8) -> list: """基于标题相似度去重，保留发布时间最早的""" articles.sort(key=lambda x: x['publish_time']) unique = [] for art in articles: is_dup = False for u in unique: ratio = SequenceMatcher(None, art['title'], u['title']).ratio() if ratio >= threshold: is_dup = True break if not is_dup: unique.append(art) return unique

threshold=0.8是经验值，低于 0.8 可能漏掉一些转载文章，高于 0.8 又可能误杀不同文章。SequenceMatcher的时间复杂度是 O(n²)，但摘要里文章数量通常不超过 50 条，性能完全没问题。排序后再去重，保证保留的是最早发布的那条，避免用户看到“旧闻新推”。

这套系统我从零搭到能跑通完整链路，大概花了两周业余时间。最大的教训是：不要一开始就追求大而全，先把采集和存储跑通，再加订阅匹配，最后做推送。每加一个模块就压测一次，看看 Redis 内存和 MySQL 连接数有没有异常。我习惯在每次上线新功能前，手动往 Redis 里塞 1000 条测试 URL，跑一轮去重和匹配，确认没有内存泄漏再继续。希望帮到你。

本文还有配套的精品资源，点击获取