简介基于Python网络爬虫的新闻采集与订阅系统的毕业设计完整资料包面向计算机、通信、人工智能、自动化等相关专业的学生与开发者。项目为个人毕设答辩评分98分代码经调试测试可正常运行既适合新手学习爬虫与Web开发也适合作为课程设计或毕业设计的改造基础。压缩包共64个文件包含33个Python源码文件、17张PNG与4张JPEG架构/流程示意图、2个HTML前端页面、1篇PDF论文终稿以及配置文件、requirements依赖清单和Shell启动脚本等整包约7.04MB。内容涵盖新闻爬虫采集模块、订阅展示模块、MongoDB存储设计、Scrapy框架配置与API服务部署图片素材中可见用例图、新闻推送活动图、系统总体框架图等设计文档便于对照论文理解整体实现。目前已有164人学习下载对想快速搭建新闻聚合类项目或完成毕设的同学有不错的参考价值。1. 基于网络爬虫的新闻采集和订阅系统难点不在爬虫本身把基于网络爬虫的新闻采集和订阅系统这个题目拆开看爬虫抓取只占三成工作量。requests 拉页面、BeautifulSoup 解析、SQLite 存数据一下午就能跑通。源码质量取决于两件事增量去重怎么做订阅匹配怎么保证时效。去重不严谨调度三次后库里满是重复记录匹配写成全表扫描接口延迟立刻暴露。我见过的同类源码里翻车点常在把推送逻辑写死在爬虫线程中一次抓取异常就丢掉一批订阅通知。下面的展开顺序也是我实际搭建时的顺序先定数据模型再写采集层再实现订阅匹配与推送最后给出一套闭环验证方法。适合正在做 Python 毕业设计的学生也适合想了解新闻聚合类项目实现的初级工程师。实现基于 Python 3.10 Flask SQLite依赖只有 requests、BeautifulSoup4、APScheduler源码目录结构可以直接对应论文里的系统设计章节。2. 新闻采集系统的数据模型与源码目录先把三层边界划清2.1 技术选型requests BeautifulSoup 比 Scrapy 更适合毕业设计如果题目要求里没有分布式爬虫增量抓取框架这类字眼我一般会直接选 requests BeautifulSoup4而不是上 Scrapy。Scrapy 是完整框架自带下载中间件、Item Pipeline 和 Twisted 事件循环功能强但调试链路长。写一个只有五六个新闻源、每日抓取量在万级以内的系统用 requests 从发请求到拿到 HTML 只有三行代码出问题用 logging 就能定位这对毕设源码的可读性非常重要。requests 方案对请求头的控制也更直接。新闻站点常见的反爬手段就是校验 User-Agent 和 Refererrequests 每次把 headers 传进去行为完全可预期Scrapy 里要改 UA 需要写中间件虽然不难但很多刚完成 Python 入门、第一次做完整项目的同学会在这里卡住。等场景真变大比如 50 个以上新闻源、需要断点续抓再迁移到 Scrapy 也不迟数据模型按迁移方向预留即可。2.2 三张核心表文章表、订阅表、推送记录表数据模型是整个系统最不该偷懒的部分。新闻采集订阅系统里最少需要三张表文章表存抓取结果订阅表存用户与关键词的绑定关系推送记录表存每次通知的发送结果。第三张表经常被省略但论文里系统可靠性分析和答辩时推送失败怎么办都靠它回答。表名关键字段说明articlesid, source, title, content_text, url, fingerprint, publish_time, created_atfingerprint 建唯一索引承担增量去重subscriptionsid, user_name, keyword, match_mode, status, created_atmatch_mode 预留 exact / contains 两种模式push_logid, subscription_id, article_id, channel, status, error_msg, pushed_atchannel 记 email/webhookstatus 记成功或失败建表 SQL 用 SQLite 写出来就是这样注意 fingerprint 必须加 UNIQUECREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL, title TEXT NOT NULL, content_text TEXT, url TEXT NOT NULL, fingerprint TEXT NOT NULL UNIQUE, publish_time TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS subscriptions ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_name TEXT NOT NULL, keyword TEXT NOT NULL, match_mode TEXT NOT NULL DEFAULT contains, status INTEGER NOT NULL DEFAULT 1, created_at TEXT DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS push_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, subscription_id INTEGER NOT NULL, article_id INTEGER NOT NULL, channel TEXT NOT NULL, status INTEGER NOT NULL DEFAULT 0, error_msg TEXT, pushed_at TEXT );字段说明content_text 存清洗后的正文纯文本排版标签在解析阶段剥掉省磁盘也方便关键词匹配publish_time 存原始新闻时间而不是采集时间按时间排序时旧闻不会被顶到前面push_log 的 error_msg 是排查推送失败的关键线索别嫌脏数据就不记。2.3 源码目录按采集、匹配、推送三层解耦目录结构直接决定论文里系统设计图好不好画。我的习惯是让每个子目录对应一个独立职责爬虫只管产出文章匹配模块只管算命中推送模块只管发通知。news_project/ ├── app.py # Flask 入口注册订阅接口 ├── config.py # 数据库路径、抓取间隔、SMTP 配置 ├── crawler/ │ ├── __init__.py │ ├── fetcher.py # 请求封装与重试 │ ├── parser.py # HTML 解析与字段清洗 │ └── scheduler.py # APScheduler 定时任务 ├── matcher/ │ └── engine.py # 关键词匹配与评分 ├── push/ │ ├── __init__.py │ ├── email_sender.py # SMTP 邮件发送 │ └── webhook.py # HTTP 回调通知 ├── tests/ │ └── test_flow.py └── requirements.txtrequirements.txt 必须锁版本避免答辩演示时环境不一致requests2.31.0 beautifulsoup44.12.3 Flask3.0.0 APScheduler3.10.4 pytest8.0.0在按 Python 安装教程配好的虚拟环境里执行pip install -r requirements.txt版本不一致导致的本地能跑、答辩机跑不了问题基本能避免。提示论文里的系统架构图可以直接按这三个目录分层来画评审最想看到的就是职责边界清晰的模块划分。三层解耦的最大收益是测试可以分层写。推送模块如果依赖爬虫的返回值单元测试就得先启动网络边界划清之后给 matcher 塞一个 dict 结构的文章就能验证匹配逻辑。3. 新闻采集爬虫的实现请求封装、HTML 解析、指纹去重、定时调度3.1 网络爬虫原理与请求封装编码、超时与请求头网络爬虫原理一句话就能讲清模拟浏览器向服务器发起 HTTP 请求拿到响应后用解析库提取结构化字段。新闻采集系统本质上是 Python 爬虫在定时任务下的工程化落地最容易踩的坑不在请求本身而在响应编码。很多新闻站是 GBK 编码requests 默认按 HTTP 头里的 charset 解码遇到头里没写编码的页面中文就变成乱码。用 apparent_encoding 让 requests 从页面内容推断编码是最省事的解法import requests import logging HEADERS { User-Agent: (Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36), Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, } def fetch_page(url: str, timeout: int 10) - str: try: resp requests.get(url, headersHEADERS, timeouttimeout) resp.raise_for_status() resp.encoding resp.apparent_encoding or utf-8 return resp.text except requests.RequestException as exc: logging.warning(fetch failed: %s, error: %s, url, exc) return 参数说明headers 里 User-Agent 必须写完整只写 python-requests/2.31 会被部分站点直接拒绝timeout 同时覆盖连接和读取两个阶段新闻页面图片多、响应慢给 10 秒合理返回空字符串而不是抛异常是为了让上层调度逻辑统计失败率而不是让整个任务中断。3.2 列表页解析与正文清洗解析先用 CSS 选择器圈住新闻列表再逐条提取链接和标题。不同站点的列表结构差异很大通用做法是每个源维护一个解析配置select 用的选择器放在配置里而不是写死在代码中from bs4 import BeautifulSoup def parse_list(html: str, source: str, selector: str) - list[dict]: soup BeautifulSoup(html, html.parser) items [] for a in soup.select(selector): title a.get_text(stripTrue) href a.get(href, ) if not title or not href.startswith(http): continue items.append({source: source, title: title, url: href}) return itemsselector 改成 ul.news li a 或 div.list-item a 就能适配不同站点。详情页解析同理正文区块用更长的选择器圈住先decompose()掉 script 和 style再get_text(separator\n, stripTrue)转纯文本避免正文里混入 JS 代码和广告链接。3.3 指纹去重MD5 组合字段比直接比标题可靠新闻标题经常被编辑加滚动快讯等前缀同一篇稿子在不同时间抓到的标题字符串不完全一样直接用标题去重会漏判或误判。我一般用来源站 标题 发布时间三个字段拼原始串再做 MD5import hashlib def make_fingerprint(source: str, title: str, publish_time: str) - str: raw |.join([ source.strip().lower(), title.strip().lower(), publish_time.strip()[:16], ]) return hashlib.md5(raw.encode(utf-8)).hexdigest()三个维度的去重效果对比如下只用单字段最容易出问题指纹维度重复新闻去重重新发布识别误判风险只用标题好差标题一改就漏标题相同但内容不同会被合并标题 URL中好URL 带跟踪参数时会误判来源 标题 时间好中同一事件多站转载会保留多条写入数据库时配合 INSERT OR IGNORE指纹重复的行直接跳过这是增量抓取的基础INSERT OR IGNORE INTO articles (source, title, url, fingerprint, publish_time) VALUES (?, ?, ?, ?, ?);注意MD5 这里只用于生成去重指纹不承担任何安全职责拼接字段时用 | 做分隔符避免 a|b 和 a |b 产生碰撞。3.4 定时调度APScheduler 的三个必调参数定时任务用 APScheduler 的 BackgroundScheduler比在 while 循环里 sleep 可靠也比 crontab 好演示。有三个参数必须设置明白from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger scheduler BackgroundScheduler() trigger CronTrigger( hour*/1, minute5, timezoneAsia/Shanghai, ) scheduler.add_job( run_all_crawlers, trigger, idnews_crawl_job, replace_existingTrue, misfire_grace_time60, ) scheduler.start()hour*/1 表示每小时执行一次minute5 固定在整点过 5 分钟避开整点大批定时任务挤在同一秒的尖峰misfire_grace_time60 允许任务因资源竞争晚最多 60 秒执行超过就跳过本次避免追任务把爬虫压垮replace_existingTrue 防止重复注册同名任务多线程环境里重复 start 是最常见的报错来源。4. 订阅系统的实现关键词匹配评分、推送渠道与订阅 API4.1 订阅关系建模与增量查询订阅系统的核心关系是用户-关键词-文章的多对多。用户订阅的是关键词而不是某个栏目每次爬虫入库一批新文章后都要把新增文章和所有有效订阅跑一次匹配。这里的关键是只查增量不要每次把全表文章拉出来重算import sqlite3 def get_incremental_articles(db_path: str, last_id: int, limit: int 100) - list[dict]: conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row rows conn.execute( SELECT id, source, title, content_text, publish_time FROM articles WHERE id ? ORDER BY id LIMIT ?, (last_id, limit), ).fetchall() conn.close() return [dict(r) for r in rows]用自增主键 id 做游标记录上次匹配到的最大 id下次只处理大于它的行。这个方案在单机场景比按时间戳增量简单时间戳会遇到同一秒写入多条、边界重复处理的坑id 游标天然严格递增。limit 参数防止单次匹配太多把进程卡住处理完一批再取下一批。4.2 关键词匹配与评分排序匹配逻辑用 Python 的 in 做子串判断就够毕业设计用了不需要上分词和向量检索。但直接输出命中列表太粗糙我给不同位置的命中加分标题命中权重 3正文命中权重 1这样用户收到的第一条永远是直接讲这个关键词的新闻而不是顺带提到的def match_article(article: dict, subscriptions: list[dict]) - list[dict]: hits [] title article[title].lower() text f{title}\n{article.get(content_text, )}.lower() for sub in subscriptions: keyword sub[keyword].lower().strip() if not keyword or keyword not in text: continue score 3 if keyword in title else 1 hits.append({ subscription_id: sub[id], user_name: sub[user_name], article_id: article[id], keyword: sub[keyword], score: score, push_channel: sub.get(push_channel, email), }) hits.sort(keylambda x: x[score], reverseTrue) return hits匹配结果按 score 降序返回推送时把高分条目排在邮件正文最前面。这里有个容易忽略的点keyword 必须 strip 和 lower否则用户输入带空格或大写英文时匹配率明显下降。如果还想支持多关键词可以在订阅表加一个 tag 字段存逗号分隔的关键词组匹配时对每个词分别打分再取最大值。4.3 推送渠道邮件与 Webhook 的选择新闻订阅最常见的通知载体有三种取舍如下渠道实时性实现成本失败处理适合场景SMTP 邮件秒级到分钟级低一个 smtplib 完成可记 error_msg 后重试毕设首选演示效果好Webhook 回调秒级中需要接收方配合依赖对方服务状态对接群机器人、企业内部系统站内消息实时高要写消息表和前端轮询可控性最好需要前端展示消息列表时邮件发送用 smtplib 加 SSL 是最常见做法配置集中在一个 dict 里方便切换邮箱服务商import smtplib from email.mime.text import MIMEText def send_email(to_addr: str, subject: str, body: str, smtp_cfg: dict) - tuple[bool, str]: msg MIMEText(body, plain, utf-8) msg[Subject] subject msg[From] smtp_cfg[from_addr] msg[To] to_addr try: with smtplib.SMTP_SSL(smtp_cfg[host], smtp_cfg[port], timeout15) as srv: srv.login(smtp_cfg[user], smtp_cfg[password]) srv.send_message(msg) return True, except Exception as exc: return False, str(exc)返回的 (bool, error_msg) 元组直接写入 push_log这是论文可靠性设计里失败可追溯最直观的证据。注意 SMTP_SSL 端口一般为 465如果服务商给的是 587就要用srv.starttls()之后再 login两种端口混用是邮件发不出去的第一大原因。4.4 用 Flask 暴露订阅接口系统前台需要一个让用户提交订阅关键词的入口Flask 提供的最小 JSON 接口就够。参数校验必须放在业务逻辑前面——前端可以不做服务端不能不做from flask import Flask, request, jsonify app Flask(__name__) app.post(/api/subscribe) def subscribe(): data request.get_json(silentTrue) or {} user_name (data.get(user_name) or ).strip() keyword (data.get(keyword) or ).strip() if not user_name or len(user_name) 50: return jsonify({error: user_name 必填且不超过 50 字}), 400 if not keyword or len(keyword) 50: return jsonify({error: keyword 必填且不超过 50 字}), 400 insert_subscription(user_name, keyword) return jsonify({ok: True, user_name: user_name, keyword: keyword})get_json(silentTrue) 在请求体不是合法 JSON 时返回 None配合or {}避免接口直接 500。长度限制防止超长关键词拖慢 4.2 的 in 匹配——订阅量上来之后50 字和 500 字关键词的性能差会很明显。insert_subscription 内部就是第 2 章那张 subscriptions 表的 INSERT 语句把 user_name、keyword、match_mode 三个参数填进去。5. 新闻采集订阅系统上线前闭环测试、反爬做法、性能边界5.1 用 pytest 把采集到匹配的链路钉死上线前先跑一条不依赖网络的测试把解析、指纹、匹配三个函数串起来。下面是 tests/test_flow.py 的核心用例from crawler.parser import parse_list from matcher.engine import match_article, make_fingerprint def test_flow(): html ul classnews-listlia hrefhttps://example.com/aAI 大模型落地应用/a/li/ul items parse_list(html, test, ul.news-list li a) assert len(items) 1 article {id: 1, title: items[0][title], content_text: 某地企业开始部署大模型} subs [{id: 1, user_name: stu, keyword: 大模型, push_channel: email}] hits match_article(article, subs) assert len(hits) 1 and hits[0][score] 3 fp make_fingerprint(test, article[title], 2024-01-01 10:00) assert len(fp) 32在项目根目录执行pytest tests/ -v看到 1 passed 就说明采集解析、匹配评分、指纹去重三条核心链路都通。这套用例还能直接写进论文作为系统测试章节的证据。5.2 反爬与合规的三个工程习惯第一抓取频率必须主动限制。每个新闻源请求之间 sleep 1 到 3 秒比被 403 后再设计重试策略省事得多也让系统在面对真实站点时更友好。第二请求异常要区分对待429 和 503 是限流信号退避重试403 多是请求头被识别需要更新 User-Agent 和 Referer。第三存储上只保留标题、摘要和原文链接不做整站镜像采集范围遵守站点公开声明。这三条做到演示和真实部署都不会被动。5.3 SQLite 的性能边界与多线程采集单机几万条文章、几百个订阅时SQLite 完全够用。匹配逻辑在 Python 里做子串判断瓶颈在 Python 循环而不是数据库订阅量到几千以后把匹配改成批量关键词正则或者把文章文本放进内存缓存做集合运算是常见的下一步。采集并发用 ThreadPoolExecutor 控制max_workers 给 4 就好而不是每来一个源就开一个线程from concurrent.futures import ThreadPoolExecutor sources load_sources() with ThreadPoolExecutor(max_workers4) as pool: results pool.map(fetch_and_store, sources)max_workers4 对五六个新闻源的毕业设计已经足够再大要考虑对方站点的连接数限制。抓取失败率高于 5% 时先看日志里是不是集中在某一个源单独把它降频而不是全局调慢订阅匹配慢的先看 push_log 里有没有堆积再决定是加索引还是换缓存方案。本文还有配套的精品资源点击获取 
长河君 · 2026/9/10 22:43:10 · 阅读 1.2k 基于网络爬虫的新闻采集与订阅系统：从爬虫原理到关键词推送的完整实现
简介基于Python网络爬虫的新闻采集与订阅系统的毕业设计完整资料包面向计算机、通信、人工智能、自动化等相关专业的学生与开发者。项目为个人毕设答辩评分98分代码经调试测试可正常运行既适合新手学习爬虫与Web开发也适合作为课程设计或毕业设计的改造基础。压缩包共64个文件包含33个Python源码文件、17张PNG与4张JPEG架构/流程示意图、2个HTML前端页面、1篇PDF论文终稿以及配置文件、requirements依赖清单和Shell启动脚本等整包约7.04MB。内容涵盖新闻爬虫采集模块、订阅展示模块、MongoDB存储设计、Scrapy框架配置与API服务部署图片素材中可见用例图、新闻推送活动图、系统总体框架图等设计文档便于对照论文理解整体实现。目前已有164人学习下载对想快速搭建新闻聚合类项目或完成毕设的同学有不错的参考价值。1. 基于网络爬虫的新闻采集和订阅系统难点不在爬虫本身把基于网络爬虫的新闻采集和订阅系统这个题目拆开看爬虫抓取只占三成工作量。requests 拉页面、BeautifulSoup 解析、SQLite 存数据一下午就能跑通。源码质量取决于两件事增量去重怎么做订阅匹配怎么保证时效。去重不严谨调度三次后库里满是重复记录匹配写成全表扫描接口延迟立刻暴露。我见过的同类源码里翻车点常在把推送逻辑写死在爬虫线程中一次抓取异常就丢掉一批订阅通知。下面的展开顺序也是我实际搭建时的顺序先定数据模型再写采集层再实现订阅匹配与推送最后给出一套闭环验证方法。适合正在做 Python 毕业设计的学生也适合想了解新闻聚合类项目实现的初级工程师。实现基于 Python 3.10 Flask SQLite依赖只有 requests、BeautifulSoup4、APScheduler源码目录结构可以直接对应论文里的系统设计章节。2. 新闻采集系统的数据模型与源码目录先把三层边界划清2.1 技术选型requests BeautifulSoup 比 Scrapy 更适合毕业设计如果题目要求里没有分布式爬虫增量抓取框架这类字眼我一般会直接选 requests BeautifulSoup4而不是上 Scrapy。Scrapy 是完整框架自带下载中间件、Item Pipeline 和 Twisted 事件循环功能强但调试链路长。写一个只有五六个新闻源、每日抓取量在万级以内的系统用 requests 从发请求到拿到 HTML 只有三行代码出问题用 logging 就能定位这对毕设源码的可读性非常重要。requests 方案对请求头的控制也更直接。新闻站点常见的反爬手段就是校验 User-Agent 和 Refererrequests 每次把 headers 传进去行为完全可预期Scrapy 里要改 UA 需要写中间件虽然不难但很多刚完成 Python 入门、第一次做完整项目的同学会在这里卡住。等场景真变大比如 50 个以上新闻源、需要断点续抓再迁移到 Scrapy 也不迟数据模型按迁移方向预留即可。2.2 三张核心表文章表、订阅表、推送记录表数据模型是整个系统最不该偷懒的部分。新闻采集订阅系统里最少需要三张表文章表存抓取结果订阅表存用户与关键词的绑定关系推送记录表存每次通知的发送结果。第三张表经常被省略但论文里系统可靠性分析和答辩时推送失败怎么办都靠它回答。表名关键字段说明articlesid, source, title, content_text, url, fingerprint, publish_time, created_atfingerprint 建唯一索引承担增量去重subscriptionsid, user_name, keyword, match_mode, status, created_atmatch_mode 预留 exact / contains 两种模式push_logid, subscription_id, article_id, channel, status, error_msg, pushed_atchannel 记 email/webhookstatus 记成功或失败建表 SQL 用 SQLite 写出来就是这样注意 fingerprint 必须加 UNIQUECREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL, title TEXT NOT NULL, content_text TEXT, url TEXT NOT NULL, fingerprint TEXT NOT NULL UNIQUE, publish_time TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS subscriptions ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_name TEXT NOT NULL, keyword TEXT NOT NULL, match_mode TEXT NOT NULL DEFAULT contains, status INTEGER NOT NULL DEFAULT 1, created_at TEXT DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS push_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, subscription_id INTEGER NOT NULL, article_id INTEGER NOT NULL, channel TEXT NOT NULL, status INTEGER NOT NULL DEFAULT 0, error_msg TEXT, pushed_at TEXT );字段说明content_text 存清洗后的正文纯文本排版标签在解析阶段剥掉省磁盘也方便关键词匹配publish_time 存原始新闻时间而不是采集时间按时间排序时旧闻不会被顶到前面push_log 的 error_msg 是排查推送失败的关键线索别嫌脏数据就不记。2.3 源码目录按采集、匹配、推送三层解耦目录结构直接决定论文里系统设计图好不好画。我的习惯是让每个子目录对应一个独立职责爬虫只管产出文章匹配模块只管算命中推送模块只管发通知。news_project/ ├── app.py # Flask 入口注册订阅接口 ├── config.py # 数据库路径、抓取间隔、SMTP 配置 ├── crawler/ │ ├── __init__.py │ ├── fetcher.py # 请求封装与重试 │ ├── parser.py # HTML 解析与字段清洗 │ └── scheduler.py # APScheduler 定时任务 ├── matcher/ │ └── engine.py # 关键词匹配与评分 ├── push/ │ ├── __init__.py │ ├── email_sender.py # SMTP 邮件发送 │ └── webhook.py # HTTP 回调通知 ├── tests/ │ └── test_flow.py └── requirements.txtrequirements.txt 必须锁版本避免答辩演示时环境不一致requests2.31.0 beautifulsoup44.12.3 Flask3.0.0 APScheduler3.10.4 pytest8.0.0在按 Python 安装教程配好的虚拟环境里执行pip install -r requirements.txt版本不一致导致的本地能跑、答辩机跑不了问题基本能避免。提示论文里的系统架构图可以直接按这三个目录分层来画评审最想看到的就是职责边界清晰的模块划分。三层解耦的最大收益是测试可以分层写。推送模块如果依赖爬虫的返回值单元测试就得先启动网络边界划清之后给 matcher 塞一个 dict 结构的文章就能验证匹配逻辑。3. 新闻采集爬虫的实现请求封装、HTML 解析、指纹去重、定时调度3.1 网络爬虫原理与请求封装编码、超时与请求头网络爬虫原理一句话就能讲清模拟浏览器向服务器发起 HTTP 请求拿到响应后用解析库提取结构化字段。新闻采集系统本质上是 Python 爬虫在定时任务下的工程化落地最容易踩的坑不在请求本身而在响应编码。很多新闻站是 GBK 编码requests 默认按 HTTP 头里的 charset 解码遇到头里没写编码的页面中文就变成乱码。用 apparent_encoding 让 requests 从页面内容推断编码是最省事的解法import requests import logging HEADERS { User-Agent: (Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36), Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, } def fetch_page(url: str, timeout: int 10) - str: try: resp requests.get(url, headersHEADERS, timeouttimeout) resp.raise_for_status() resp.encoding resp.apparent_encoding or utf-8 return resp.text except requests.RequestException as exc: logging.warning(fetch failed: %s, error: %s, url, exc) return 参数说明headers 里 User-Agent 必须写完整只写 python-requests/2.31 会被部分站点直接拒绝timeout 同时覆盖连接和读取两个阶段新闻页面图片多、响应慢给 10 秒合理返回空字符串而不是抛异常是为了让上层调度逻辑统计失败率而不是让整个任务中断。3.2 列表页解析与正文清洗解析先用 CSS 选择器圈住新闻列表再逐条提取链接和标题。不同站点的列表结构差异很大通用做法是每个源维护一个解析配置select 用的选择器放在配置里而不是写死在代码中from bs4 import BeautifulSoup def parse_list(html: str, source: str, selector: str) - list[dict]: soup BeautifulSoup(html, html.parser) items [] for a in soup.select(selector): title a.get_text(stripTrue) href a.get(href, ) if not title or not href.startswith(http): continue items.append({source: source, title: title, url: href}) return itemsselector 改成 ul.news li a 或 div.list-item a 就能适配不同站点。详情页解析同理正文区块用更长的选择器圈住先decompose()掉 script 和 style再get_text(separator\n, stripTrue)转纯文本避免正文里混入 JS 代码和广告链接。3.3 指纹去重MD5 组合字段比直接比标题可靠新闻标题经常被编辑加滚动快讯等前缀同一篇稿子在不同时间抓到的标题字符串不完全一样直接用标题去重会漏判或误判。我一般用来源站 标题 发布时间三个字段拼原始串再做 MD5import hashlib def make_fingerprint(source: str, title: str, publish_time: str) - str: raw |.join([ source.strip().lower(), title.strip().lower(), publish_time.strip()[:16], ]) return hashlib.md5(raw.encode(utf-8)).hexdigest()三个维度的去重效果对比如下只用单字段最容易出问题指纹维度重复新闻去重重新发布识别误判风险只用标题好差标题一改就漏标题相同但内容不同会被合并标题 URL中好URL 带跟踪参数时会误判来源 标题 时间好中同一事件多站转载会保留多条写入数据库时配合 INSERT OR IGNORE指纹重复的行直接跳过这是增量抓取的基础INSERT OR IGNORE INTO articles (source, title, url, fingerprint, publish_time) VALUES (?, ?, ?, ?, ?);注意MD5 这里只用于生成去重指纹不承担任何安全职责拼接字段时用 | 做分隔符避免 a|b 和 a |b 产生碰撞。3.4 定时调度APScheduler 的三个必调参数定时任务用 APScheduler 的 BackgroundScheduler比在 while 循环里 sleep 可靠也比 crontab 好演示。有三个参数必须设置明白from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger scheduler BackgroundScheduler() trigger CronTrigger( hour*/1, minute5, timezoneAsia/Shanghai, ) scheduler.add_job( run_all_crawlers, trigger, idnews_crawl_job, replace_existingTrue, misfire_grace_time60, ) scheduler.start()hour*/1 表示每小时执行一次minute5 固定在整点过 5 分钟避开整点大批定时任务挤在同一秒的尖峰misfire_grace_time60 允许任务因资源竞争晚最多 60 秒执行超过就跳过本次避免追任务把爬虫压垮replace_existingTrue 防止重复注册同名任务多线程环境里重复 start 是最常见的报错来源。4. 订阅系统的实现关键词匹配评分、推送渠道与订阅 API4.1 订阅关系建模与增量查询订阅系统的核心关系是用户-关键词-文章的多对多。用户订阅的是关键词而不是某个栏目每次爬虫入库一批新文章后都要把新增文章和所有有效订阅跑一次匹配。这里的关键是只查增量不要每次把全表文章拉出来重算import sqlite3 def get_incremental_articles(db_path: str, last_id: int, limit: int 100) - list[dict]: conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row rows conn.execute( SELECT id, source, title, content_text, publish_time FROM articles WHERE id ? ORDER BY id LIMIT ?, (last_id, limit), ).fetchall() conn.close() return [dict(r) for r in rows]用自增主键 id 做游标记录上次匹配到的最大 id下次只处理大于它的行。这个方案在单机场景比按时间戳增量简单时间戳会遇到同一秒写入多条、边界重复处理的坑id 游标天然严格递增。limit 参数防止单次匹配太多把进程卡住处理完一批再取下一批。4.2 关键词匹配与评分排序匹配逻辑用 Python 的 in 做子串判断就够毕业设计用了不需要上分词和向量检索。但直接输出命中列表太粗糙我给不同位置的命中加分标题命中权重 3正文命中权重 1这样用户收到的第一条永远是直接讲这个关键词的新闻而不是顺带提到的def match_article(article: dict, subscriptions: list[dict]) - list[dict]: hits [] title article[title].lower() text f{title}\n{article.get(content_text, )}.lower() for sub in subscriptions: keyword sub[keyword].lower().strip() if not keyword or keyword not in text: continue score 3 if keyword in title else 1 hits.append({ subscription_id: sub[id], user_name: sub[user_name], article_id: article[id], keyword: sub[keyword], score: score, push_channel: sub.get(push_channel, email), }) hits.sort(keylambda x: x[score], reverseTrue) return hits匹配结果按 score 降序返回推送时把高分条目排在邮件正文最前面。这里有个容易忽略的点keyword 必须 strip 和 lower否则用户输入带空格或大写英文时匹配率明显下降。如果还想支持多关键词可以在订阅表加一个 tag 字段存逗号分隔的关键词组匹配时对每个词分别打分再取最大值。4.3 推送渠道邮件与 Webhook 的选择新闻订阅最常见的通知载体有三种取舍如下渠道实时性实现成本失败处理适合场景SMTP 邮件秒级到分钟级低一个 smtplib 完成可记 error_msg 后重试毕设首选演示效果好Webhook 回调秒级中需要接收方配合依赖对方服务状态对接群机器人、企业内部系统站内消息实时高要写消息表和前端轮询可控性最好需要前端展示消息列表时邮件发送用 smtplib 加 SSL 是最常见做法配置集中在一个 dict 里方便切换邮箱服务商import smtplib from email.mime.text import MIMEText def send_email(to_addr: str, subject: str, body: str, smtp_cfg: dict) - tuple[bool, str]: msg MIMEText(body, plain, utf-8) msg[Subject] subject msg[From] smtp_cfg[from_addr] msg[To] to_addr try: with smtplib.SMTP_SSL(smtp_cfg[host], smtp_cfg[port], timeout15) as srv: srv.login(smtp_cfg[user], smtp_cfg[password]) srv.send_message(msg) return True, except Exception as exc: return False, str(exc)返回的 (bool, error_msg) 元组直接写入 push_log这是论文可靠性设计里失败可追溯最直观的证据。注意 SMTP_SSL 端口一般为 465如果服务商给的是 587就要用srv.starttls()之后再 login两种端口混用是邮件发不出去的第一大原因。4.4 用 Flask 暴露订阅接口系统前台需要一个让用户提交订阅关键词的入口Flask 提供的最小 JSON 接口就够。参数校验必须放在业务逻辑前面——前端可以不做服务端不能不做from flask import Flask, request, jsonify app Flask(__name__) app.post(/api/subscribe) def subscribe(): data request.get_json(silentTrue) or {} user_name (data.get(user_name) or ).strip() keyword (data.get(keyword) or ).strip() if not user_name or len(user_name) 50: return jsonify({error: user_name 必填且不超过 50 字}), 400 if not keyword or len(keyword) 50: return jsonify({error: keyword 必填且不超过 50 字}), 400 insert_subscription(user_name, keyword) return jsonify({ok: True, user_name: user_name, keyword: keyword})get_json(silentTrue) 在请求体不是合法 JSON 时返回 None配合or {}避免接口直接 500。长度限制防止超长关键词拖慢 4.2 的 in 匹配——订阅量上来之后50 字和 500 字关键词的性能差会很明显。insert_subscription 内部就是第 2 章那张 subscriptions 表的 INSERT 语句把 user_name、keyword、match_mode 三个参数填进去。5. 新闻采集订阅系统上线前闭环测试、反爬做法、性能边界5.1 用 pytest 把采集到匹配的链路钉死上线前先跑一条不依赖网络的测试把解析、指纹、匹配三个函数串起来。下面是 tests/test_flow.py 的核心用例from crawler.parser import parse_list from matcher.engine import match_article, make_fingerprint def test_flow(): html ul classnews-listlia hrefhttps://example.com/aAI 大模型落地应用/a/li/ul items parse_list(html, test, ul.news-list li a) assert len(items) 1 article {id: 1, title: items[0][title], content_text: 某地企业开始部署大模型} subs [{id: 1, user_name: stu, keyword: 大模型, push_channel: email}] hits match_article(article, subs) assert len(hits) 1 and hits[0][score] 3 fp make_fingerprint(test, article[title], 2024-01-01 10:00) assert len(fp) 32在项目根目录执行pytest tests/ -v看到 1 passed 就说明采集解析、匹配评分、指纹去重三条核心链路都通。这套用例还能直接写进论文作为系统测试章节的证据。5.2 反爬与合规的三个工程习惯第一抓取频率必须主动限制。每个新闻源请求之间 sleep 1 到 3 秒比被 403 后再设计重试策略省事得多也让系统在面对真实站点时更友好。第二请求异常要区分对待429 和 503 是限流信号退避重试403 多是请求头被识别需要更新 User-Agent 和 Referer。第三存储上只保留标题、摘要和原文链接不做整站镜像采集范围遵守站点公开声明。这三条做到演示和真实部署都不会被动。5.3 SQLite 的性能边界与多线程采集单机几万条文章、几百个订阅时SQLite 完全够用。匹配逻辑在 Python 里做子串判断瓶颈在 Python 循环而不是数据库订阅量到几千以后把匹配改成批量关键词正则或者把文章文本放进内存缓存做集合运算是常见的下一步。采集并发用 ThreadPoolExecutor 控制max_workers 给 4 就好而不是每来一个源就开一个线程from concurrent.futures import ThreadPoolExecutor sources load_sources() with ThreadPoolExecutor(max_workers4) as pool: results pool.map(fetch_and_store, sources)max_workers4 对五六个新闻源的毕业设计已经足够再大要考虑对方站点的连接数限制。抓取失败率高于 5% 时先看日志里是不是集中在某一个源单独把它降频而不是全局调慢订阅匹配慢的先看 push_log 里有没有堆积再决定是加索引还是换缓存方案。本文还有配套的精品资源点击获取