简介基于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;q0.9, } def fetch_list_page(list_url, timeout10): resp requests.get(list_url, headersHEADERS, timeouttimeout) 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(stripTrue) href a.get(href) if title and href: items.append({title: title, url: urljoin(list_url, href)}) return items这段代码里有几个参数值得在论文里单独解释。HEADERS里的User-Agent是识别浏览器身份的请求头不少站点会对空UA的请求做反爬拦截所以哪怕毕设规模也要带上。timeout10是连接和读取的超时上限防止某个慢页面把整个采集流程卡死。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, timeout10): resp requests.get(detail_url, headersHEADERS, timeouttimeout) 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, stripTrue) return { title: title.get_text(stripTrue) 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_pathnews.jsonl, limit20): seen set() with open(out_path, a, encodingutf-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_asciiFalse) \n) time.sleep(1.5) if len(seen) limit: break采集流程里最容易被忽视的是time.sleep(1.5)。它控制请求频率既是基本的网络爬虫礼仪也能避免被目标站限流论文里可以把它设计成可配置项答辩时解释请求频率影响采集速率与被封概率的权衡。len(detail[content]) 200用来过滤空正文或跳转失败的详情页。seen集合做单轮去重因为同一个列表页里可能多次出现同一篇文章入口。json.dumps的ensure_asciiFalse保证中文以明文写入文件方便排查问题。2.3 字段清洗与发布时间的常见处理新闻文本要用于订阅匹配就必须做两级清洗。第一级是HTML实体和标签清理BeautifulSoup的get_text已经完成主要工作但残留的空白符和换行需要正则压缩。第二级是业务字段抽取发布时间一般藏在meta标签或页面文本里用正则比用选择器更稳因为不同页面的meta写法差异太大。def extract_time(html_text): patterns [ r20\d{2}-\d{2}-\d{2}[ T]\d{2}:\d{2}(:\d{2})?, r20\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来自站点配置用于后续按来源聚合统计。字段来源处理方式title列表页a标签文本strip去首尾空白url列表页hrefurljoin补全为绝对地址content详情页正文容器正则压缩连续空行publish_time详情页HTML文本正则匹配后统一为YYYY-MM-DD HH:MM格式source站点配置常量直接写入发布时间解析是论文里一个很好写的实验点。可以统计一个站点200条新闻对比正则方式与meta标签方式各自的命中率把结果画成表格放进实验章节这比空写系统性能良好有说服力得多。3. 增量采集与任务调度定时触发、URL去重与失败重试3.1 调度方案怎么选crontab、APScheduler 与 Celery新闻是持续产生的采集系统必须有定时触发能力。常见做法有三种系统级crontab、Python进程内的APScheduler、分布式任务队列Celery。毕设规模用APScheduler最合适它不用额外部署进程直接在主程序里注册任务论文里也好画架构图。Celery虽然更工业级但要引入Redis作为broker调试成本会明显上升。方案触发方式适合场景主要缺点系统crontab操作系统定时执行命令单机简单任务无任务状态管理失败难追踪APScheduler进程内调度器单机多任务的毕设/小项目进程退出调度即停止Celery Beat独立beat进程worker分布式多worker场景依赖broker部署链路长APScheduler的代码量很少。BlockingScheduler适合采集脚本独立运行add_job的interval表示固定间隔max_instances1保证上一次任务没跑完时不会开启新实例这个参数在采集耗时大于调度间隔时非常关键否则同一时刻会叠着跑两轮任务。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, minutes30, max_instances1) 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(host127.0.0.1, port6379, db1) def is_duplicate_url(url): # SADD返回1表示插入成功返回0表示该成员已存在 return r.sadd(news:url_set, url) 0sadd的返回值要解释清楚第一次遇到某URL时返回1is_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, retries3, backoff2): 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, encodingutf-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 Nonebackoff * (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) );三张表的关系和作用可以归结为下表论文的数据库设计章节直接用它当概述。表名作用关键约束news_article存储清洗后的新闻正文url UNIQUEsubscription用户与关键词的映射keyword长度2push_log推送留痕防止重复推送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], timeout15) 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更省事。timeout15防止邮件服务器无响应时挂住整个采集进程。发送成功后要在同一事务里写入push_log邮件发出但日志丢失会造成重复推送反之发送异常时把记录标为failed下一轮任务统一重试。5. 用日志与数据指标把采集系统讲清楚论文验证与答辩技巧答辩最怕系统能跑四个字老师追问你怎么知道它跑得好时答不上来。解决方案是从第一天就给采集链路加上结构化日志和统计指标答辩PPT里直接放真实运行数据的截图这比任何架构图都有效。5.1 结构化日志的埋点格式与核心指标import json import logging logger logging.getLogger(news) handler logging.FileHandler(run_stats.jsonl, encodingutf-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_asciiFalse))每次任务结束输出一行JSON积累30天后就是一份完整的实验数据。计算三个指标去重率dup/(newdup)验证去重机制是否生效失败率fail/(newdupfail)异常站点会在这一项暴露订阅命中率matched/new反映关键词规则的覆盖质量。答辩时用这组数字配合failed_urls.log里截取的几行真实异常比任何口头描述都有说服力。5.2 人工抽样验证与实际呈现验证采集正确性还有一个单独技巧把详情页正文和原文站点做抽样对比随机挑20条新闻人工确认解析字段是否完整。把人工抽检结果做成表格放进论文附录标注20条样本18条完全正确2条发布时间缺失这种诚实的误差描述比空喊准确率更可信。所有环节跑通后把run_stats.jsonl里30天的数据直接导出成图表放进论文实验章节当原始证据。本文还有配套的精品资源点击获取 基于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;q0.9, } def fetch_list_page(list_url, timeout10): resp requests.get(list_url, headersHEADERS, timeouttimeout) 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(stripTrue) href a.get(href) if title and href: items.append({title: title, url: urljoin(list_url, href)}) return items这段代码里有几个参数值得在论文里单独解释。HEADERS里的User-Agent是识别浏览器身份的请求头不少站点会对空UA的请求做反爬拦截所以哪怕毕设规模也要带上。timeout10是连接和读取的超时上限防止某个慢页面把整个采集流程卡死。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, timeout10): resp requests.get(detail_url, headersHEADERS, timeouttimeout) 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, stripTrue) return { title: title.get_text(stripTrue) 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_pathnews.jsonl, limit20): seen set() with open(out_path, a, encodingutf-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_asciiFalse) \n) time.sleep(1.5) if len(seen) limit: break采集流程里最容易被忽视的是time.sleep(1.5)。它控制请求频率既是基本的网络爬虫礼仪也能避免被目标站限流论文里可以把它设计成可配置项答辩时解释请求频率影响采集速率与被封概率的权衡。len(detail[content]) 200用来过滤空正文或跳转失败的详情页。seen集合做单轮去重因为同一个列表页里可能多次出现同一篇文章入口。json.dumps的ensure_asciiFalse保证中文以明文写入文件方便排查问题。2.3 字段清洗与发布时间的常见处理新闻文本要用于订阅匹配就必须做两级清洗。第一级是HTML实体和标签清理BeautifulSoup的get_text已经完成主要工作但残留的空白符和换行需要正则压缩。第二级是业务字段抽取发布时间一般藏在meta标签或页面文本里用正则比用选择器更稳因为不同页面的meta写法差异太大。def extract_time(html_text): patterns [ r20\d{2}-\d{2}-\d{2}[ T]\d{2}:\d{2}(:\d{2})?, r20\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来自站点配置用于后续按来源聚合统计。字段来源处理方式title列表页a标签文本strip去首尾空白url列表页hrefurljoin补全为绝对地址content详情页正文容器正则压缩连续空行publish_time详情页HTML文本正则匹配后统一为YYYY-MM-DD HH:MM格式source站点配置常量直接写入发布时间解析是论文里一个很好写的实验点。可以统计一个站点200条新闻对比正则方式与meta标签方式各自的命中率把结果画成表格放进实验章节这比空写系统性能良好有说服力得多。3. 增量采集与任务调度定时触发、URL去重与失败重试3.1 调度方案怎么选crontab、APScheduler 与 Celery新闻是持续产生的采集系统必须有定时触发能力。常见做法有三种系统级crontab、Python进程内的APScheduler、分布式任务队列Celery。毕设规模用APScheduler最合适它不用额外部署进程直接在主程序里注册任务论文里也好画架构图。Celery虽然更工业级但要引入Redis作为broker调试成本会明显上升。方案触发方式适合场景主要缺点系统crontab操作系统定时执行命令单机简单任务无任务状态管理失败难追踪APScheduler进程内调度器单机多任务的毕设/小项目进程退出调度即停止Celery Beat独立beat进程worker分布式多worker场景依赖broker部署链路长APScheduler的代码量很少。BlockingScheduler适合采集脚本独立运行add_job的interval表示固定间隔max_instances1保证上一次任务没跑完时不会开启新实例这个参数在采集耗时大于调度间隔时非常关键否则同一时刻会叠着跑两轮任务。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, minutes30, max_instances1) 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(host127.0.0.1, port6379, db1) def is_duplicate_url(url): # SADD返回1表示插入成功返回0表示该成员已存在 return r.sadd(news:url_set, url) 0sadd的返回值要解释清楚第一次遇到某URL时返回1is_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, retries3, backoff2): 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, encodingutf-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 Nonebackoff * (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) );三张表的关系和作用可以归结为下表论文的数据库设计章节直接用它当概述。表名作用关键约束news_article存储清洗后的新闻正文url UNIQUEsubscription用户与关键词的映射keyword长度2push_log推送留痕防止重复推送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], timeout15) 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更省事。timeout15防止邮件服务器无响应时挂住整个采集进程。发送成功后要在同一事务里写入push_log邮件发出但日志丢失会造成重复推送反之发送异常时把记录标为failed下一轮任务统一重试。5. 用日志与数据指标把采集系统讲清楚论文验证与答辩技巧答辩最怕系统能跑四个字老师追问你怎么知道它跑得好时答不上来。解决方案是从第一天就给采集链路加上结构化日志和统计指标答辩PPT里直接放真实运行数据的截图这比任何架构图都有效。5.1 结构化日志的埋点格式与核心指标import json import logging logger logging.getLogger(news) handler logging.FileHandler(run_stats.jsonl, encodingutf-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_asciiFalse))每次任务结束输出一行JSON积累30天后就是一份完整的实验数据。计算三个指标去重率dup/(newdup)验证去重机制是否生效失败率fail/(newdupfail)异常站点会在这一项暴露订阅命中率matched/new反映关键词规则的覆盖质量。答辩时用这组数字配合failed_urls.log里截取的几行真实异常比任何口头描述都有说服力。5.2 人工抽样验证与实际呈现验证采集正确性还有一个单独技巧把详情页正文和原文站点做抽样对比随机挑20条新闻人工确认解析字段是否完整。把人工抽检结果做成表格放进论文附录标注20条样本18条完全正确2条发布时间缺失这种诚实的误差描述比空喊准确率更可信。所有环节跑通后把run_stats.jsonl里30天的数据直接导出成图表放进论文实验章节当原始证据。本文还有配套的精品资源点击获取