基于Python的新闻采集与订阅系统：爬虫与关键词推送实践

简介：基于Python网络爬虫的新闻采集与订阅系统，是一份答辩评审分达到98分的高分毕业设计项目，主要面向计算机、通信、人工智能、自动化等相关专业的学生、老师或从业者，也适合用于期末课程设计、课程大作业或毕业设计参考。系统完整覆盖新闻爬虫采集、数据存储、订阅管理、接口服务与前端展示等环节，涉及Scrapy爬虫架构与MongoDB存储方案，源码均经过调试测试，可直接运行，学习借鉴价值较高。压缩包内共64个文件，整体约7.04MB，核心包含33个Python脚本、1篇论文终稿PDF、多张系统架构与运行效果示意图、前端页面、启动脚本及多项配置文件，目录按爬虫模块、订阅与展示模块、论文资料等划分清晰，并配有新闻推送活动图、用例图等设计资料，方便理解整体流程。目前已有109人学习下载，基础较好的学习者还可在现有源码上修改采集源、订阅规则等逻辑，完成个性化扩展。

1. 基于Python网络爬虫的新闻采集与订阅系统：先看清毕设的边界

新闻采集与订阅系统，是Python网络爬虫方向最稳的毕设选题之一。它把HTTP请求、HTML解析、增量调度、数据库和消息推送串成完整链路，既有爬虫源码细节，又有能演示的产品形态：用户订阅关键词，系统自动推送匹配新闻。标题里的两个词划清了边界——采集端做内容结构化，订阅端做过滤与触达。

很多同学把这类毕设做成"一个脚本加一张网页"，这是最常见的误区。评阅老师要看的是工程意识：增量更新怎么做、URL怎么去重、爬虫挂了怎么恢复、订阅怎么匹配、拿什么数据证明系统正常。这些问题能写进论文实验章节、撑住答辩追问，才谈得上高分。

下面按"采集→调度→订阅→验证"的路线拆开讲。默认你已掌握Python基础语法并完成python环境安装。主线用requests加BeautifulSoup，数据库用SQLite，调度用APScheduler，兼顾可复现与论文可写。

2. 用 requests + BeautifulSoup 搭建新闻采集模块：请求、解析与正文提取

2.1 新闻站点的页面结构与采集选型

先把网络爬虫原理的核心讲清楚：采集的本质是把HTML结构转成结构化数据。新闻站点页面通常分三层——列表页、详情页链接、详情页正文。正确做法是先抓列表页解析出文章URL，再逐个抓详情页提取正文。如果直接把列表页整页文本存下来，导航、推荐位、页脚全混进来，后面的订阅关键词匹配准确率会立刻崩掉。

选型上，requests负责HTTP请求，BeautifulSoup负责DOM解析，是Python爬虫入门最经典组合。优点是依赖少、出错好定位、源码量适中，论文里每个函数都能对应一段设计说明。只有当目标站点大量使用JS动态渲染时，才需要引入selenium或playwright；毕设场景尽量选静态渲染的新闻站，把精力留给调度和订阅这两层的设计。

from urllib.parse import urljoin import requests from bs4 import BeautifulSoup HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/122.0.0.0 Safari/537.36", "Accept-Language": "zh-CN,zh;q=0.9", } def fetch_list_page(list_url, timeout=10): resp = requests.get(list_url, headers=HEADERS, timeout=timeout) resp.raise_for_status() if not resp.encoding or resp.encoding.lower() == "iso-8859-1": resp.encoding = resp.apparent_encoding soup = BeautifulSoup(resp.text, "html.parser") items = [] for a in soup.select("h2 a, ul.news-list a"): title = a.get_text(strip=True) href = a.get("href") if title and href: items.append({"title": title, "url": urljoin(list_url, href)}) return items

这段代码里有几个参数值得在论文里单独解释。HEADERS里的User-Agent是识别浏览器身份的请求头，不少站点会对空UA的请求做反爬拦截，所以哪怕毕设规模也要带上。timeout=10是连接和读取的超时上限，防止某个慢页面把整个采集流程卡死。raise_for_status()在返回4xx或5xx时直接抛异常，比手动判断status_code干净。resp.apparent_encoding是根据页面字节内容猜测的编码，新闻站经常在meta里声明的charset和实际不一致，这一行能避免中文乱码。select("h2 a, ul.news-list a")是CSS选择器，多个选择器用逗号并列，返回所有符合条件的a标签列表，urljoin把相对地址补全成绝对地址。

提示：采集前先检查目标站点的robots.txt和版权声明，毕设选题应避开有明确禁止采集条款的站点，论文的"合规性说明"章节也建议写这一句。

2.2 最小可运行的详情页采集与入库代码

列表页拿到文章URL后，下一步是抓详情页并抽取正文字段。正文容器在大多数新闻站里是某个带特定class的div，选择器按目标站实际情况调整。抓取后把结构化结果追加写入JSON Lines文件，是最简单的持久化方式，论文里可以说明"先落盘、后入库"的原因：文件写入不会因为数据库连接异常导致整批数据丢失。

import json import re import time def fetch_detail(detail_url, timeout=10): resp = requests.get(detail_url, headers=HEADERS, timeout=timeout) resp.raise_for_status() resp.encoding = resp.apparent_encoding soup = BeautifulSoup(resp.text, "html.parser") content_div = soup.select_one("div.article-content, div.content, article") if content_div is None: return None title = soup.select_one("h1") text = content_div.get_text(separator="\n", strip=True) return { "title": title.get_text(strip=True) if title else "", "url": detail_url, "content": re.sub(r"\n{3,}", "\n\n", text), "publish_time": extract_time(resp.text), # extract_time 见 2.3 小节 } def crawl_and_dump(list_url, out_path="news.jsonl", limit=20): seen = set() with open(out_path, "a", encoding="utf-8") as f: for item in fetch_list_page(list_url): if item["url"] in seen: continue seen.add(item["url"]) detail = fetch_detail(item["url"]) if detail and len(detail["content"]) > 200: f.write(json.dumps(detail, ensure_ascii=False) + "\n") time.sleep(1.5) if len(seen) >= limit: break

采集流程里最容易被忽视的是time.sleep(1.5)。它控制请求频率，既是基本的网络爬虫礼仪，也能避免被目标站限流；论文里可以把它设计成可配置项，答辩时解释"请求频率影响采集速率与被封概率的权衡"。len(detail["content"]) > 200用来过滤空正文或跳转失败的详情页。seen集合做单轮去重，因为同一个列表页里可能多次出现同一篇文章入口。json.dumps的ensure_ascii=False保证中文以明文写入文件，方便排查问题。

2.3 字段清洗与发布时间的常见处理

新闻文本要用于订阅匹配，就必须做两级清洗。第一级是HTML实体和标签清理，BeautifulSoup的get_text已经完成主要工作，但残留的空白符和换行需要正则压缩。第二级是业务字段抽取，发布时间一般藏在meta标签或页面文本里，用正则比用选择器更稳，因为不同页面的meta写法差异太大。

def extract_time(html_text): patterns = [ r"20\d{2}-\d{2}-\d{2}[ T]\d{2}:\d{2}(:\d{2})?", r"20\d{2}年\d{1,2}月\d{1,2}日\s*\d{1,2}:\d{2}", ] for p in patterns: m = re.search(p, html_text) if m: return m.group(0).replace("T", " ") return ""

字段最终以统一格式存库，对比关系如下表。title和url直接来自列表页解析，content来自详情页正文容器，publish_time来自正则抽取后的规范化字符串，source来自站点配置，用于后续按来源聚合统计。

发布时间解析是论文里一个很好写的实验点。可以统计一个站点200条新闻，对比正则方式与meta标签方式各自的命中率，把结果画成表格放进实验章节，这比空写"系统性能良好"有说服力得多。

3. 增量采集与任务调度：定时触发、URL去重与失败重试

3.1 调度方案怎么选：crontab、APScheduler 与 Celery

新闻是持续产生的，采集系统必须有定时触发能力。常见做法有三种：系统级crontab、Python进程内的APScheduler、分布式任务队列Celery。毕设规模用APScheduler最合适，它不用额外部署进程，直接在主程序里注册任务，论文里也好画架构图。Celery虽然更"工业级"，但要引入Redis作为broker，调试成本会明显上升。

APScheduler的代码量很少。BlockingScheduler适合采集脚本独立运行，add_job的interval表示固定间隔，max_instances=1保证上一次任务没跑完时不会开启新实例，这个参数在采集耗时大于调度间隔时非常关键，否则同一时刻会叠着跑两轮任务。

from apscheduler.schedulers.blocking import BlockingScheduler def job_run(): for site in load_site_configs(): crawl_and_dump(site["list_url"], site["name"]) scheduler = BlockingScheduler() scheduler.add_job(job_run, "interval", minutes=30, max_instances=1) scheduler.start()

load_site_configs从配置文件读站点列表，每个站点是一个含list_url和name的字典。interval和minutes组合表示每30分钟触发一次。BlockingScheduler会阻塞主线程，所以进程需要常驻运行；如果部署环境不允许常驻进程，再退回crontab每分钟执行一次入口脚本，让脚本在启动时检查"距上次执行是否超过30分钟"，这是另一种常见但更绕的做法。

3.2 URL去重：从集合到Redis的渐进实现

增量采集的核心是"已经见过的URL不再抓"。最朴素的实现是把URL存进Python的set，程序重启后集合清空，会重复抓全部历史URL，所以持久化去重是必须的。推荐用Redis的SADD集合：判断和写入是原子操作，并发场景下两个进程不会同时抓到同一篇文章。

import redis r = redis.Redis(host="127.0.0.1", port=6379, db=1) def is_duplicate_url(url): # SADD返回1表示插入成功，返回0表示该成员已存在 return r.sadd("news:url_set", url) == 0

sadd的返回值要解释清楚：第一次遇到某URL时返回1，is_duplicate_url返回False，表示"这是新文章，应该采集"；再次遇到时返回0，函数返回True，直接跳过。去重集合的key可以按日期分片，例如news:url_set:20250612，这样论文里能额外写一个数据量为指标的实验——单集合存一年URL会到百万量级，Redis的SADD在这个规模下依然稳定。

提示：如果目标站的文章URL本身带utm_source这类追踪参数，去重前务必先做规范化，否则同一篇文章会被当成两条新闻重复入库。

3.3 失败重试与断点续采

网络请求一定会失败，超时、连不上、被限流都可能在某天半夜触发。失败处理要分两层：单条请求失败就重试几次，超过重试上限后写入失败日志文件；整个任务失败则依靠调度器在下一轮自动恢复。我一般不用复杂的重试库，手写一个带指数退避的重试循环就够。

import time import requests def fetch_with_retry(url, retries=3, backoff=2): for attempt in range(retries): try: return fetch_detail(url) except requests.RequestException as exc: if attempt == retries - 1: with open("failed_urls.log", "a", encoding="utf-8") as f: f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')}\t{url}\t{exc}\n") return None time.sleep(backoff * (attempt + 1)) return None

backoff * (attempt + 1)表示第一次失败等2秒，第二次等4秒，指数退避避免对目标站造成瞬时请求风暴。failed_urls.log是断点续采的锚点：下一轮任务开始前，先读这个文件把URL重新放回待采集队列，采集成功后从日志里删除对应行。这个机制在论文里对应"系统鲁棒性设计"，答辩时很容易被问到，提前把日志格式和恢复流程想清楚。

4. 订阅系统的设计与实现：订阅表、关键词匹配与邮件推送

4.1 新闻与订阅的数据库表设计

订阅系统的数据模型围绕三张表展开：news_article存新闻，subscription存用户订阅关键词，push_log存推送记录防止重复推送。url字段加UNIQUE约束，作为数据库层的第二道去重防线——即使爬虫层的Redis去重遗漏，插入时也会因为唯一约束失败而跳过。publish_time用DATETIME类型，后续按时间过滤订阅结果时能直接走索引。

CREATE TABLE news_article ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, url TEXT NOT NULL UNIQUE, source TEXT, content TEXT, publish_time DATETIME, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE subscription ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, keyword TEXT NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE push_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, article_id INTEGER NOT NULL, channel TEXT DEFAULT 'email', pushed_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE(user_id, article_id) );

三张表的关系和作用可以归结为下表，论文的数据库设计章节直接用它当概述。

push_log的UNIQUE(user_id, article_id)复合唯一约束是"不重复推送"的保险丝。订阅系统跑得越久，同一关键词命中的新闻越可能被重复采集到，推送记录里如果已经存在这对组合，就直接跳过发送。

4.2 关键词匹配：分词前先想清楚边界

订阅匹配最简单可靠的做法是子串匹配：新闻标题和正文拼成一个大文本，用LIKE查询找出所有命中的订阅记录。子串匹配的问题在于关键词"苹果"同时命中"苹果公司"和"苹果手机"，这是答辩里老师喜欢追问的点。回答思路是：系统定位是关键词订阅而非语义推荐，子串匹配保证召回率，精度问题用后续的排除词表机制缓解。

import sqlite3 def find_subscribers_by_keywords(conn, article_text): cur = conn.execute( """ SELECT id, user_id, keyword FROM subscription WHERE ? LIKE '%' || keyword || '%' """, (article_text,), ) return cur.fetchall()

这条SQL用LIKE反向匹配，把每条订阅的关键词拿到新闻文本里找。Python的sqlite3默认每条execute都会自动提交，批量写新闻时建议用事务包裹，数据量大时写入会明显变慢。keyword字段要加CHECK约束限制长度，单个字符的关键词会带来大量无效命中，在入口就把数据质量提上来。

4.3 推送链路：邮件通知的完整实现

推送通道用SMTP邮件最常见，注册一个专用邮箱就能跑通。推送模块要做两件事：组装消息内容，调用smtplib发送并把发送结果写入push_log。触发时机建议放在采集任务收尾时，一次性把本轮所有新增新闻和订阅做匹配，而不是每抓到一条就发一封邮件，后者会让收件箱爆炸。

import smtplib from email.message import EmailMessage def send_digest(subject, body, to_addr, smtp_cfg): msg = EmailMessage() msg["Subject"] = subject msg["From"] = smtp_cfg["from"] msg["To"] = to_addr msg.set_content(body) with smtplib.SMTP_SSL(smtp_cfg["host"], smtp_cfg["port"], timeout=15) as server: server.login(smtp_cfg["user"], smtp_cfg["password"]) server.send_message(msg)

smtp_cfg是包含host、port、user、password、from五个键的配置字典，不要硬编码在源码里。SMTP_SSL使用465端口加密连接，比STARTTLS更省事。timeout=15防止邮件服务器无响应时挂住整个采集进程。发送成功后要在同一事务里写入push_log，邮件发出但日志丢失会造成重复推送；反之发送异常时，把记录标为failed，下一轮任务统一重试。

5. 用日志与数据指标把采集系统讲清楚：论文验证与答辩技巧

答辩最怕"系统能跑"四个字，老师追问"你怎么知道它跑得好"时答不上来。解决方案是从第一天就给采集链路加上结构化日志和统计指标，答辩PPT里直接放真实运行数据的截图，这比任何架构图都有效。

5.1 结构化日志的埋点格式与核心指标

import json import logging logger = logging.getLogger("news") handler = logging.FileHandler("run_stats.jsonl", encoding="utf-8") handler.setFormatter(logging.Formatter("%(asctime)s %(message)s")) logger.addHandler(handler) logger.setLevel(logging.INFO) def log_run_stats(source, new_count, dup_count, fail_count, matched_count, cost_ms): logger.info(json.dumps({ "source": source, "new": new_count, "dup": dup_count, "fail": fail_count, "matched": matched_count, "cost_ms": cost_ms, }, ensure_ascii=False))

每次任务结束输出一行JSON，积累30天后就是一份完整的实验数据。计算三个指标：去重率=dup/(new+dup)，验证去重机制是否生效；失败率=fail/(new+dup+fail)，异常站点会在这一项暴露；订阅命中率=matched/new，反映关键词规则的覆盖质量。答辩时用这组数字配合failed_urls.log里截取的几行真实异常，比任何口头描述都有说服力。

5.2 人工抽样验证与实际呈现

验证采集正确性还有一个单独技巧：把详情页正文和原文站点做抽样对比，随机挑20条新闻，人工确认解析字段是否完整。把人工抽检结果做成表格放进论文附录，标注"20条样本，18条完全正确，2条发布时间缺失"，这种诚实的误差描述比空喊准确率更可信。所有环节跑通后，把run_stats.jsonl里30天的数据直接导出成图表，放进论文实验章节当原始证据。

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