★ FEATURED ARTICLE 简介本资源是一套完整的本科毕业设计项目实现面向计算机专业高年级学生及Web开发初学者聚焦网络新闻数据的自动化采集、结构化存储与个性化订阅服务全流程实践。项目采用PythonScrapy/Flask为主技术栈涵盖爬虫调度、MongoDB新闻存储、前端订阅界面及推送机制等核心模块具备工程可运行性与教学示范性。压缩包共65个文件含33个Python源码如spider.py、run_server.sh、init_db.py、17张PNG/JPEG系统架构与UML图含爬虫流程、用例图、活动图、Nginx配置等、1份终稿PDF论文及README、LICENSE等工程配套文档整体7.03MB目录层次清晰便于分模块学习与调试。目前已有114人下载学习读者可直接部署运行获取从需求分析、技术选型、数据库设计到前后端集成的全链路参考实现并复用其中代理池IPProxyTool、用户代理管理、API服务启动脚本等实用组件。1. 新闻采集和订阅系统为什么不能只靠“定时爬存数据库”某高校毕业设计选题里“基于网络爬虫的新闻采集和订阅系统”出现频率常年排前三——但真正跑通、能稳定用满一学期、答辩时演示不卡顿的不到三成。问题不在代码写得少而在于多数同学把“爬虫”当成了唯一技术点写个 requests BeautifulSoup 抓首页标题存进 SQLite再加个 Flask 页面展示就以为完成了。结果是第二天新闻源改了 DOM 结构爬虫全挂第三天反爬策略升级IP 被限频第四天用户想按“人工智能”“政策解读”分类订阅发现字段没清洗、关键词没归一、标签体系是空的第五天导师问“怎么保证新闻时效性在 5 分钟内”才发现连时间戳解析都错把“3小时前”当成“2024-03-15 10:00”。这不是代码能力问题是系统级认知断层新闻采集不是单次数据搬运而是带状态感知、可回溯、可分级响应的轻量级信息流管道。本篇就从一个真实可复现的毕业设计落地路径出发讲清楚怎么用最小技术集Python Scrapy Redis Flask SQLite搭出具备增量抓取、去重判重、主题聚类、用户订阅、推送延迟可控这五项核心能力的系统——不堆架构图不画微服务所有模块本地可跑所有配置有据可调所有坑我都替你踩过。2. 用 Scrapy 搭建可中断、可续爬、带指纹去重的新闻采集管道新闻爬虫最常翻车的不是写不出 XPath而是“一跑就崩、一崩就重来”。毕业设计场景下没有运维兜底必须让爬虫自己扛住网络抖动、页面结构微调、目标站临时封 IP 这些现实干扰。Scrapy 是当前 Python 生态中唯一能把“断点续爬请求指纹中间件调度”三者原生捏合的框架比手写 requests 循环高两个维度。2.1 初始化项目并配置持久化爬取状态先创建标准 Scrapy 项目关键不是爬什么而是让每次 run 都知道上次停在哪scrapy startproject news_spider cd news_spider scrapy genspider sina news.sina.com.cn修改settings.py启用JOBDIR这是 Scrapy 原生支持断点续爬的核心# settings.py BOT_NAME news_spider SPIDER_MODULES [news_spider.spiders] NEWSPIDER_MODULE news_spider.spiders # 启用断点续爬Scrapy 会自动保存请求队列、去重指纹、已处理 URL 状态 JOBDIR jobs/news_crawl # 目录名可自定义但必须存在 # 启用请求指纹去重默认开启但显式写出更稳妥 DUPEFILTER_CLASS scrapy.dupefilters.RFPDupeFilter # 指纹生成逻辑可扩展但毕业设计用默认 RFPRequest Fingerprint足够提示JOBDIR目录必须手动创建且不能放在scrapy.cfg同级否则 Git 会误提交。我习惯在项目根目录下建jobs/里面放news_crawl/每次运行前清空该目录即为“全新启动”。2.2 定义新闻条目结构与增量抓取逻辑新闻不是静态网页而是带时间戳、来源、正文摘要的结构化实体。在items.py中明确定义字段强制约束后续清洗逻辑# items.py import scrapy class NewsItem(scrapy.Item): title scrapy.Field() # 标题必填 url scrapy.Field() # 原文链接唯一标识用于去重 source scrapy.Field() # 来源网站如 sina, people publish_time scrapy.Field() # 发布时间ISO 格式字符串如 2024-05-20T08:30:00 content_summary scrapy.Field() # 正文前 200 字用于预览和关键词提取 crawl_time scrapy.Field() # 本系统抓取时间datetime.now().isoformat()关键在spiders/sina_spider.py中实现增量判断只抓取发布时间晚于上一次抓取最大时间的新闻。这里不用数据库查而用 Scrapy 的crawler.stats记录本次抓取的最新时间并在下次启动时读取# spiders/sina_spider.py import scrapy from scrapy import signals from scrapy.crawler import Crawler from news_spider.items import NewsItem from datetime import datetime, timedelta import re class SinaSpider(scrapy.Spider): name sina allowed_domains [news.sina.com.cn] start_urls [https://news.sina.com.cn/] # 在爬虫初始化时加载上次最大发布时间从 stats.json 或文件中读 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 毕业设计简化用文件记录上次最大时间实际可用 Redis self.last_max_time self._load_last_max_time() self.logger.info(fLoaded last max publish time: {self.last_max_time}) def _load_last_max_time(self): try: with open(last_max_time.txt, r) as f: t f.read().strip() return datetime.fromisoformat(t) if t else datetime.now() - timedelta(days7) except: return datetime.now() - timedelta(days7) # 首次运行默认抓 7 天内 def parse(self, response): # 解析列表页提取新闻链接示例为新浪滚动新闻 for sel in response.css(div.listBlk ul li a::attr(href)).getall(): if sel and http in sel: yield scrapy.Request(urlsel, callbackself.parse_article) def parse_article(self, response): item NewsItem() item[url] response.url item[source] sina item[title] response.css(h1.main-title::text).get(default).strip() # 时间解析新浪常见格式有 2024年05月20日 08:30 和 5月20日 08:30 time_text response.css(span.date::text).get(default) item[publish_time] self._parse_sina_time(time_text) # 内容摘要取正文前 200 字过滤 HTML 标签 content .join(response.css(div.article p::text).getall()) item[content_summary] content[:200] if len(content) 200 else content item[crawl_time] datetime.now().isoformat() # 增量过滤只保存发布时间晚于上次最大时间的条目 if item[publish_time] and self.last_max_time: pt datetime.fromisoformat(item[publish_time]) if pt self.last_max_time: yield item # 更新本次抓取的最大时间 if not hasattr(self, _current_max_time) or pt self._current_max_time: self._current_max_time pt def _parse_sina_time(self, text): 解析新浪多种时间格式统一转为 ISO if not text: return None # 匹配 2024年05月20日 08:30 m1 re.match(r(\d{4})年(\d{1,2})月(\d{1,2})日\s(\d{1,2}):(\d{2}), text) if m1: y, m, d, h, minu m1.groups() return f{y}-{int(m):02d}-{int(d):02d}T{int(h):02d}:{minu}:00 # 匹配 5月20日 08:30 → 补全年份假设为今年 m2 re.match(r(\d{1,2})月(\d{1,2})日\s(\d{1,2}):(\d{2}), text) if m2: m, d, h, minu m2.groups() y datetime.now().year return f{y}-{int(m):02d}-{int(d):02d}T{int(h):02d}:{minu}:00 return None def closed(self, reason): 爬虫关闭时保存本次最大时间到文件 if hasattr(self, _current_max_time): with open(last_max_time.txt, w) as f: f.write(self._current_max_time.isoformat()) self.logger.info(fSaved new max time: {self._current_max_time})这段代码的关键价值不在“能爬”而在把时间作为业务主键参与流程控制_load_last_max_time()加载历史边界parse_article()中做实时比较closed()回写新边界。整个过程不依赖外部数据库纯文件 IO适合毕业设计环境。你可能注意到publish_time解析用了正则硬匹配——这正是毕业设计该有的务实不追求覆盖所有网站格式而聚焦 1~2 个主流源如新浪人民网把这 2 个的时间解析写稳比泛泛支持 10 个源但每个都漏时间强十倍。3. 用 Redis 实现跨爬虫、跨进程的去重与任务分发很多同学卡在“为什么我开了两个爬虫实例同一篇新闻存了两遍”——因为Scrapy默认的RFPDupeFilter只在单个爬虫进程内生效一旦你scrapy crawl sina和scrapy crawl people同时跑去重就失效了。毕业设计要体现工程思维就得引入一个中心化去重层。Redis 不是炫技而是解决三个刚需① 多源爬虫共享同一套 URL 指纹库② 用户订阅变更后能主动触发重爬通过 Redis Pub/Sub③ 抓取队列可监控、可人工干预。3.1 配置 Redis 去重中间件Scrapy 支持自定义DUPEFILTER_CLASS我们写一个基于 Redis 的去重器。新建news_spider/dupefilter.py# dupefilter.py import redis import hashlib from scrapy.utils.request import request_fingerprint class RedisDupeFilter: def __init__(self, server, key): self.server server self.key key classmethod def from_settings(cls, settings): # 从 settings.py 读取 Redis 配置 redis_url settings.get(REDIS_URL, redis://localhost:6379/0) key settings.get(DUPEFILTER_KEY, dupefilter:news) server redis.from_url(redis_url) return cls(server, key) def request_seen(self, request): # 生成请求指纹含 URL、method、body、headers 等 fp request_fingerprint(request) # 用 SHA1 缩短长度存入 Redis Set sha1 hashlib.sha1(fp.encode()).hexdigest() added self.server.sadd(self.key, sha1) return added 0 # True 表示已存在去重 def close(self, reason): pass # 不需要清理然后在settings.py中启用它# settings.py # Redis 配置本地开发用默认端口 REDIS_URL redis://localhost:6379/0 DUPEFILTER_KEY dupefilter:news DUPEFILTER_CLASS news_spider.dupefilter.RedisDupeFilter # 同时启用 Redis 调度队列可选但推荐让 Scrapy 任务队列也落盘 SCHEDULER scrapy_redis.scheduler.Scheduler SCHEDULER_PERSIST True # 关闭后队列不清空 SCHEDULER_QUEUE_CLASS scrapy_redis.queue.SpiderPriorityQueue注意scrapy_redis是独立包需pip install scrapy-redis。它依赖redis包也需安装。本地开发用docker run -d --name redis-dev -p 6379:6379 redis启一个容器即可无需复杂配置。3.2 用 Redis List 实现轻量级任务队列与状态广播毕业设计不需要 Kafka但需要一种方式让“用户新增一个订阅关键词”这件事能立刻通知爬虫去抓相关新闻。Redis 的LPUSHBRPOP就够用。我们在news_spider/pipelines.py中加一个管道把新入库的新闻推送到news:queue# pipelines.py import redis import json from scrapy.exceptions import DropItem class RedisPipeline: def __init__(self, redis_url): self.redis redis.from_url(redis_url) classmethod def from_crawler(cls, crawler): redis_url crawler.settings.get(REDIS_URL, redis://localhost:6379/0) return cls(redis_url) def process_item(self, item, spider): # 将 NewsItem 序列化为 JSON 推入 Redis List data { title: item.get(title, ), url: item.get(url, ), source: item.get(source, ), publish_time: item.get(publish_time, ), content_summary: item.get(content_summary, ), crawl_time: item.get(crawl_time, ) } self.redis.lpush(news:queue, json.dumps(data, ensure_asciiFalse)) # 同时推一份到按来源分的频道便于后续按源统计 self.redis.lpush(fnews:source:{item.get(source, unknown)}, json.dumps(data, ensure_asciiFalse)) return item并在settings.py中启用# settings.py ITEM_PIPELINES { news_spider.pipelines.RedisPipeline: 300, }这样每条新闻入库的同时就有一份快照进入news:queue。你的 Flask 订阅服务只需BRPOP news:queue 0阻塞监听拿到新新闻就做匹配、推送、存库——完全解耦且 Redis 天然支持多消费者。4. 避坑毕业设计中最常踩的 5 个“玄学”问题与血泪解法这些不是理论错误而是你在实验室电脑上敲完代码、导出.py文件、打包答辩 PPT 前一定会遇到的真实障碍。我按发生频率排序每条都附可验证的检查命令。4.1 现象爬虫跑着跑着就卡死CPU 占用 100%但没报错原因Scrapy 默认并发请求数CONCURRENT_REQUESTS为 16但某些新闻站如人民网对单 IP 的连接数限制极严大量ConnectionResetError积压导致 Twisted reactor 阻塞。解决在settings.py中显式降低并发CONCURRENT_REQUESTS 2同时增加延时DOWNLOAD_DELAY 3单位秒强制关闭AutoThrottle它在毕业设计小规模场景下反而添乱AUTOTHROTTLE_ENABLED False验证命令scrapy crawl sina -s CONCURRENT_REQUESTS2 -s DOWNLOAD_DELAY34.2 现象publish_time解析出来全是None数据库里时间字段为空原因XPath 或 CSS 选择器写错了但get(default)返回空字符串_parse_sina_time()收到空输入直接返回None没有报错。解决在parse_article()开头加日志self.logger.debug(fRaw time text: {time_text})用scrapy shell https://news.sina.com.cn/...进入交互模式手动执行response.css(span.date::text).get()看真实 HTML 结构血泪经验永远不要信网站文档一定要用scrapy shell抓真实页面源码校验选择器4.3 现象JOBDIR启用后第二次运行还是从头爬原因JOBDIR目录权限不对或路径写错比如写成JOBDIR ./jobs/news_crawl但 Scrapy 工作目录不是项目根目录。解决绝对路径最稳JOBDIR os.path.abspath(jobs/news_crawl)需在settings.py开头import os运行后检查jobs/news_crawl/requests.queue文件是否存在且非空删除jobs/news_crawl后重新运行确认日志出现Resuming crawl from job...4.4 现象Redis 去重生效了但news:queue里一条消息都没有原因RedisPipeline的process_item方法里self.redis.lpush(...)执行失败如 Redis 连接超时但异常被静默吞掉。解决在process_item中加try/except并打 ERROR 日志try: self.redis.lpush(news:queue, json.dumps(data, ensure_asciiFalse)) except Exception as e: self.logger.error(fFailed to push to Redis: {e}) raise DropItem(fRedis push failed: {e})用redis-cli连上去执行LLEN news:queue看长度是否增长4.5 现象Flask 服务启动后访问/subscribe页面空白浏览器控制台报500 Internal Server Error原因SQLite 数据库文件路径写错或db.create_all()时表已存在但字段不匹配引发sqlalchemy.exc.OperationalError。解决Flask 启动加 debug 模式app.run(debugTrue)错误会直接打在终端检查数据库路径是否为绝对路径SQLALCHEMY_DATABASE_URI sqlite:/// os.path.abspath(news.db)删掉旧news.db重新运行 Flask让它自动建表毕业设计阶段数据可丢5. 用 TF-IDF 余弦相似度实现“新闻主题聚类”让订阅不止于关键词毕业设计如果只做到“用户输入‘AI’我就搜标题含‘AI’的新闻”那只是字符串匹配离“订阅系统”差一层认知。真正的订阅是理解用户兴趣的语义边界——比如用户订了“大模型”他其实也想看“LLM”“Transformer”“ChatGPT”但不想看“AI绘画”“机器人”。这就需要轻量级文本聚类。不用 BERT用毕业设计友好的 TF-IDF 余弦相似度20 行代码搞定。5.1 构建新闻向量库从摘要生成 TF-IDF 特征我们不处理全文太重只用content_summary200 字以内做特征。用sklearn的TfidfVectorizer关键是停用词和 ngram_range 的设置# cluster_utils.py from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity import jieba # 中文分词pip install jieba import numpy as np # 自定义中文停用词毕业设计精简版共 37 个高频无意义词 STOPWORDS set([ 的, 了, 在, 是, 我, 有, 和, 就, 不, 人, 都, 一, 一个, 上, 也, 很, 到, 说, 要, 去, 你, 会, 着, 没有, 看, 好, 自己, 这, 那, 他, 她, 它, 们, 为, 以, 及, 等 ]) def chinese_tokenizer(text): 中文分词 去停用词 words jieba.lcut(text) return [w for w in words if w not in STOPWORDS and len(w) 1] # 初始化向量器只用 uni-gram避免稀疏max_features 控制内存 vectorizer TfidfVectorizer( tokenizerchinese_tokenizer, max_features5000, min_df2, # 出现少于 2 次的词直接忽略 max_df0.95 # 出现在 95% 文档里的词如“新闻”也忽略 )5.2 实时计算相似度构建“主题簇”假设有 1000 条已入库新闻我们想找出和某条新闻最相似的 5 篇组成“相关推荐”# 假设 news_list 是从数据库查出的 1000 条 NewsItem 字典列表含 content_summary summaries [n[content_summary] for n in news_list] tfidf_matrix vectorizer.fit_transform(summaries) # 形状 (1000, 5000) # 计算第 0 条新闻与其他所有新闻的相似度 sim_scores cosine_similarity(tfidf_matrix[0:1], tfidf_matrix).flatten() # 获取相似度 top5 的索引排除自己 top_indices np.argsort(sim_scores)[-6:-1][::-1] # 倒序取前 5 print(Related news:) for idx in top_indices: print(f- {news_list[idx][title]} (score: {sim_scores[idx]:.3f}))提示fit_transform会一次性把全部摘要向量化内存占用约 100MB1000 条 × 200 字完全在毕业设计可接受范围。若数据超 5000 条可改用partial_fit分批训练但对毕设没必要。5.3 将聚类结果注入订阅逻辑从“关键词匹配”升级为“语义订阅”在 Flask 的订阅接口中不直接LIKE %keyword%而是用户输入“大模型”先用chinese_tokenizer分词得到[大模型]查找所有content_summary中包含“大模型”的新闻初筛对这些新闻的摘要用已训练好的vectorizer.transform()转成向量计算它们与“大模型”这个词向量的余弦相似度vectorizer.transform([大模型])返回相似度 0.3 的新闻阈值可调这样用户订“大模型”系统会主动把“LLM 技术突破”“Transformer 架构演进”这类语义相近但字面不匹配的新闻也推给他——这才是“订阅”的智能感。6. 把“推送延迟”做成可测量、可调优的硬指标从“大概几分钟”到“P95 90s”答辩时导师最爱问“你们的新闻推送延迟是多少” 如果你回答“大概几分钟”基本等于放弃加分项。毕业设计必须把模糊感受变成可测量的数字。我们用最朴素的方式给每条新闻打上crawl_time和notify_time推送时间用 Redis 存延迟样本最后用 Python 算分位数。6.1 在推送环节埋点记录延迟在 Flask 的推送逻辑中比如用户订阅后后台从news:queue拿到新闻并发送邮件/站内信时# app.py import time import redis import json redis_client redis.from_url(redis://localhost:6379/0) def send_notification(news_item): start_time time.time() # 这里是你的实际推送逻辑发邮件、写数据库、推 WebSocket... # ... your code ... end_time time.time() delay_ms int((end_time - start_time) * 1000) # 将延迟毫秒存入 Redis Sorted Set以时间戳为 score方便按时间窗口查 timestamp int(time.time()) redis_client.zadd(notify:delay, {str(delay_ms): timestamp}) # 同时存入一个 List 用于快速抽样最多存 1000 条 redis_client.lpush(notify:delay:samples, delay_ms) redis_client.ltrim(notify:delay:samples, 0, 999) # 只留最新 1000 条6.2 提供延迟查询接口让答辩演示有数据支撑加一个 Flask 路由返回最近 1000 次推送的 P50/P90/P95 延迟# app.py from flask import jsonify import numpy as np app.route(/api/delay-stats) def delay_stats(): # 从 List 中取最新 1000 条样本 samples redis_client.lrange(notify:delay:samples, 0, -1) if not samples: return jsonify({error: No delay samples yet}) delays [int(x) for x in samples] arr np.array(delays) return jsonify({ count: len(delays), p50_ms: int(np.percentile(arr, 50)), p90_ms: int(np.percentile(arr, 90)), p95_ms: int(np.percentile(arr, 95)), max_ms: int(arr.max()), avg_ms: int(arr.mean()) })部署后访问http://localhost:5000/api/delay-stats你会看到{ count: 1000, p50_ms: 420, p90_ms: 1280, p95_ms: 2150, max_ms: 8900, avg_ms: 876 }这意味着95% 的推送在 2.15 秒内完成。这个数字比“很快”“挺及时”有力一万倍。你甚至可以加一句“当前 P95 延迟为 2150ms通过将CONCURRENT_REQUESTS从 2 调至 4P95 可降至 1800ms但会增加被限频风险——我们在稳定性与速度间做了权衡。”6.3 一个让我后悔没早做的技巧用scrapy check验证爬虫健壮性Scrapy 自带scrapy check命令能静态分析爬虫逻辑是否完备。在sina_spider.py中添加classmethod的check方法# spiders/sina_spider.py class SinaSpider(scrapy.Spider): # ... 其他代码 ... classmethod def check(cls, crawler): Scrapy check 命令调用此方法验证关键路径 errors [] # 检查是否定义了 parse_article 方法 if not hasattr(cls, parse_article): errors.append(Missing parse_article method) # 检查是否用了 yield而非 return返回 Item # Scrapy check 本身会做语法检查此处省略 return errors然后运行scrapy check sina它会报告OK或具体错误。这个动作花不了 2 分钟但能让你在答辩前 1 小时发现parse_article拼错成parse_artical这种低级失误——这种细节导师一眼就能看出你有没有真跑通。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
