基于Python的财经新闻自动化采集与ETL入库系统

做财经新闻类的数据采集，最怕的不是网站反爬，而是爬到一半发现拿回来的数据乱七八糟，标题里带广告后缀、时间格式五种以上、链接重复率极高。这也是我为什么重新写了NewsPipe_ETL这个项目——一套基于Python爬虫的全球财经新闻自动化采集与结构化入库系统，配上了CSV导出和SQLite持久化存储，把“抓取”这件事从一次性脚本升级成了真正能长期跑的管道。如果你正在做新闻聚合、金融舆情分析、或者只是想搭一套能每天自动更新的财经数据源，这篇文章的完整思路和代码可以直接抄。

这套系统我最满意的地方不是它“能爬多少网站”，而是它把ETL的思路真正落地了：抽取（Extract）、清洗（Transform）、加载（Load）三段式管道的每一层都职责清晰，新增一个新闻源只需要加几行配置，换存储也不需要改采集逻辑。下面我会从设计思路、核心模块、实操流程、踩坑记录四个层面完整拆开讲，文章里涉及的关键代码都能直接跑。

1. 为什么做NewsPipe_ETL：一次多源采集的“返工”教训

1.1 散户式爬虫的三大痛点

大概在项目启动前三个月，我手里已经攒了七八个新闻采集脚本，每个脚本对应一个网站，逻辑高度相似但又各自为政。今天要给A站加一个字段，明天要给B站修一个时间解析，改完A站又担心把B站的逻辑改坏。最痛苦的是数据分析那边要数据时，我得先去跑脚本、再手工把结果拼接成CSV发过去，整个过程毫无工程美感。

这类“散户式爬虫”的通病我总结下来就是三点：

第一，字段结构完全随缘。有的脚本只存了标题和链接，有的存了发布时间但格式是“2024-03-15 09:30”，有的用的是“15 Mar 2024”这种英文格式。真到做统计分析的时候，光是统一时间格式就能耗掉半天。

第二，去重基本靠数据库主键硬扛。如果入库用的是自增主键而没有对文章URL做唯一约束，同一篇新闻被多个源重复抓取时就会产生大量冗余。更麻烦的是有些源会发出相同的文章但URL参数不同，单靠URL去重根本不生效。

第三，爬虫和业务逻辑完全耦合。采集、清洗、入库全写在一个脚本里，看起来简单，但出了问题很难定位。到底是网络请求失败、还是解析规则报错、还是写库失败，全靠print输出肉眼排查，效率极低。

1.2 NewsPipe_ETL想解决什么问题

基于上面这些血泪教训，NewsPipe_ETL在设计之初就定下了几个硬性目标。

首先是“一次采集，处处可用”。采集层只负责把原始网页或者RSS的内容拿回来，所有字段统一成标准结构：标题、正文、链接、来源、发布时间、抓取时间、唯一指纹。这样下游无论是做CSV导出、SQLite入库，还是以后接Elasticsearch或者数据分析框架，都不用再关心原始网站长什么样。

其次是“新增源的成本要压到最低”。我把每个新闻源抽象成一个配置项，里面包含源名称、URL、类型和解析函数。新增一个源只需要写一个解析函数，然后往配置列表里append一条记录。主流程代码完全不用动。

最后是“每一步都可观测、可审计”。每一篇新闻在管道里经历了什么状态，有没有被清洗规则过滤掉、有没有因为重复被丢弃，都要有日志记录。这样即使某天抓到的新闻数量异常，也能快速定位是哪个环节出了问题。

2. 整体架构与管道设计思路

2.1 四阶段管道：采集、清洗、去重、入库

NewsPipe_ETL的架构参考了数据工程里标准的ETL管道，但针对新闻采集场景做了一些轻量化改造，整体分成四个阶段：

• 采集阶段（Extract）：通过HTTP请求获取网页或RSS内容，对返回的HTML用解析器提取结构化字段，对RSS则用feedparser解析标准XML。
• 清洗阶段（Transform）：对字段做标准化处理——时间统一转成ISO 8601字符串，正文去掉HTML标签和广告噪音，标题去除“_官网”之类的站点后缀，空字段给默认值。
• 去重阶段（Deduplicate）：为每篇文章生成一个SHA1指纹，指纹相同则跳过，实现增量采集，避免重复入库。
• 加载阶段（Load）：将标准化后的数据写入SQLite数据库，同时支持按批次导出CSV文件。

这条管道可以类比成一家新闻通讯社的内部流程：记者（采集层）把素材带回来，编辑（清洗层）把稿子改到符合发稿规范，查重系统（去重层）过滤掉一稿多投，最后排版上线（加载层）进入资料库。

一个容易忽略的设计点是：所有阶段之间只通过标准化的字典对象通信。采集阶段输出的是dict，清洗阶段改的也是dict，加载阶段读的还是dict。这样做的好处是每一层都可以单独测试。我在写代码时给每个阶段都加了单元测试的入口，直接喂一段fake数据进去就能验证逻辑是否正确，不依赖真实网络。

2.2 数据模型设计

新闻数据最终落到SQLite里的表结构，我用的是下面这套设计：

CREATE TABLE IF NOT EXISTS news_articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, content TEXT, url TEXT NOT NULL UNIQUE, source TEXT NOT NULL, category TEXT DEFAULT 'general', published_at TEXT, crawled_at TEXT NOT NULL, fingerprint TEXT NOT NULL UNIQUE ); CREATE INDEX IF NOT EXISTS idx_published_at ON news_articles(published_at); CREATE INDEX IF NOT EXISTS idx_source ON news_articles(source);

字段设计上有几个细节值得说明。url加了UNIQUE约束，这是最基础的去重保证，但只依赖它不够，因为有些网站对同一篇文章会给不同URL。所以我又加了fingerprint字段，存的是URL归一化后生成的SHA1哈希，后面去重逻辑会详细讲。published_at统一以字符串形式存ISO 8601格式，比如“2025-03-20T09:30:00Z”。很多人习惯用时间戳存储，我个人建议新闻数据用ISO字符串更直观，导出给业务方看的时候不用再转换。

content字段没有设置长度限制，SQLite本身对TEXT长度基本没有硬上限。source字段用来记录文章来自哪个站点，后续做来源维度的统计非常方便。category字段是新闻分类，当前实现里暂未做自动分类，默认给general，但保留了扩展位。

2.3 为什么选SQLite而不是MySQL或MongoDB

让我先把结论放在前面：对单机版新闻采集系统来说，SQLite是性价比最高的选择，没有之一。我不止一次见人杀鸡用牛刀，上来就部署MySQL，结果搞了一堆账号权限、远程连接、表结构迁移的事，配置时间比写爬虫还长。

SQLite的优势非常契合这个场景。第一，零配置，Python内置sqlite3模块，import之后直接连文件就能建表，不需要单独装服务。第二，单文件存储，整个数据库就是一个news.db文件，备份、拷贝、迁移都极其方便，扔到U盘里拷走就能在另一台机器上继续用。第三，对于每天几千篇新闻的写入量，SQLite的读写性能完全够用，插入操作配合事务批量执行，速度在毫秒级。

对比一下常见存储方案的取舍：

对比下来，SQLite在“个人/小团队新闻采集”这个场景里，在易用性和查询能力之间取得了最好的平衡。等哪天数据量大到SQLite撑不住了，再迁移到PostgreSQL也不迟，因为业务逻辑已经在清洗层解耦了，换存储只改动加载层。

3. 核心模块实现与关键细节

3.1 采集层：多源管理与请求策略

采集层的第一件事是把“源”抽象好。我在项目里维护了一个列表，每个元素是一个dict，包含source_name、url、type、parse_func四个字段：

NEWS_SOURCES = [ { "source_name": "example_finance", "url": "https://example.com/rss/finance.xml", "type": "rss", "parse_func": parse_rss_feed }, { "source_name": "example_market", "url": "https://example.com/news/market", "type": "html", "parse_func": parse_html_page } ]

这里没有做成数据库表管理源，是因为新闻源的规模一般只有几十个，硬编码成配置文件最直观。真正需要动态增删源的时候，改成从外部JSON文件读取配置即可，解析函数还是放在代码里，只是配置和逻辑分离。

请求策略上，我建议用requests.Session而不是裸用requests.get。Session能复用底层的TCP连接，对同一域名连续请求时能显著减少握手开销。另外需要设置合理的UA和超时时间。

import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(): session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (compatible; NewsPipe_ETL/1.0)" }) retry = Retry( total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504] ) adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=10) session.mount("https://", adapter) session.mount("http://", adapter) return session

请求头的UA最好指定一个真实的浏览器UA，但也要在UA里注明自己的爬虫标识，合规性更好。同时要遵守网站的robots.txt，只采集允许公开访问的内容。超时参数必须在每次请求时显式指定，我一般设成connect_timeout=5, read_timeout=10。不设超时的爬虫，遇到一个响应极慢的网站时，整个管道都会被拖死。

3.2 清洗层：时间归一化与正文抽取

清洗层是整个管道里劳动量最大的地方。每家网站的时间格式、正文结构都不同，但我们必须把它们统一成一模一样的输出。

时间解析我封装了一个函数，针对常见的几种格式逐一尝试解析：

from datetime import datetime, timezone import re def normalize_time(raw_value): if not raw_value: return None raw_value = raw_value.strip() # 尝试多种常见格式 patterns = [ "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M", "%d %b %Y %H:%M:%S", "%a, %d %b %Y %H:%M:%S %z" ] for fmt in patterns: try: dt = datetime.strptime(raw_value, fmt) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc).isoformat() except ValueError: continue # 最后的兜底：提取年月日 match = re.search(r"(\d{4})[-/年](\d{1,2})[-/月](\d{1,2})", raw_value) if match: y, m, d = map(int, match.groups()) return datetime(y, m, d, tzinfo=timezone.utc).isoformat() return None

统一存成UTC的标准ISO格式，好处是排序、比较时间范围都方便，且不受时区干扰。如果你需要给本地业务看，可以在导出时再转成东八区，但我建议存储层永远是UTC，这是数据工程的标准实践。

正文抽取方面，HTML页面先用BeautifulSoup解析，把script、style、nav、footer这些噪音节点直接移除，然后在content容器里取文本。如果原始站没有提供正文而是摘要，就把摘要存进content字段，绝不让content为空。

from bs4 import BeautifulSoup def extract_content(html): soup = BeautifulSoup(html, "html.parser") # 移除干扰信息 for tag in soup(["script", "style", "nav", "footer", "aside", "iframe"]): tag.decompose() article = soup.find("article") or soup.find("div", class_=re.compile("content|article|body")) or soup text = article.get_text(separator="\n", strip=True) # 清洗多余空行 text = re.sub(r"\n{3,}", "\n\n", text) return text[:5000]

这里有个实用细节：限制正文长度。有些奇怪页面会把整个网站的内容都塞进content，不截断的话数据库会膨胀得非常快。我设了5000个字符的软上限，对多数财经短文绰绰有余。标题清洗也要做，最常见的噪音是“标题_某某网”，用正则把下划线后缀和竖线分隔的站点名去掉即可。

3.3 去重层：指纹去重与增量更新

去重是新闻管道的灵魂。同一个重大经济新闻，十几家媒体会同时报道，但多数都是转载或改写，URL往往不一样，光靠URL去重是拦不住的。不过在NewsPipe_ETL这个版本里，我不会做语义级去重，而是分两个层级处理。

第一层是URL指纹。URL在生成指纹前先做归一化：去掉锚点#后的部分、去掉utm_source等跟踪参数、把查询参数按key排序拼接。这样同一篇文章即使源站加了不同跟踪参数，也能识别为同一URL。

第二层是内容指纹。对清洗后的标题做去空格、转小写处理后，与URL归一化后的值拼接，计算SHA1。这一层能兜住某些网站用随机参数包装同一篇文章的情况。

import hashlib from urllib.parse import urlparse, parse_qs, urlencode def normalize_url(url): parsed = urlparse(url) # 去掉锚点 path = parsed.path query = parsed.query if query: params = parse_qs(query) # 去掉跟踪参数 for key in ["utm_source", "utm_medium", "utm_campaign", "ref", "spm"]: params.pop(key, None) query = urlencode([(k, v[0]) for k, v in sorted(params.items())]) return f"{parsed.scheme}://{parsed.netloc}{path}" + (f"?{query}" if query else "") def make_fingerprint(title, url): normalized_url = normalize_url(url) raw = f"{title.strip().lower()}|{normalized_url}" return hashlib.sha1(raw.encode("utf-8")).hexdigest()

在入库前，先用fingerprint查一次数据库，已存在就直接跳过，不存在才执行插入。配合SQLite的UNIQUE约束和INSERT OR IGNORE，相当于上了双保险。

增量更新的概念也在这里体现：每次跑管道时，已经入库的文章会被指纹拦下，天然只处理新增内容。所以系统支持每次只抓最近几页/最新RSS条目，少量多次地跑，而不是每次全量抓取。

3.4 存储层：SQLite表结构与CSV导出

加载层负责把标准化的文章字典写入SQLite，同时生成CSV文件。SQLite写入我用的是批量executemany，加上事务控制，避免每篇文章都commit导致的性能浪费。

import sqlite3 def save_to_sqlite(items, db_path="news.db"): conn = sqlite3.connect(db_path) cursor = conn.cursor() try: cursor.executemany(""" INSERT OR IGNORE INTO news_articles (title, content, url, source, category, published_at, crawled_at, fingerprint) VALUES (:title, :content, :url, :source, :category, :published_at, :crawled_at, :fingerprint) """, items) conn.commit() return cursor.rowcount finally: conn.close()

INSERT OR IGNORE配合UNIQUE约束，重复插入不会报错，而是静默跳过。返回的rowcount就是实际新增的数量，这个值可以直接作为监测指标。如果某天跑完全部源新增为0，说明缓存里没有新内容，源可能需要更新解析规则了。

CSV导出要特别注意编码问题。直接用默认编码写出来的CSV，用Excel打开大概率中文乱码。正确做法是用utf-8-sig编码，加BOM头，Excel才认识。

import csv from datetime import datetime def export_to_csv(items, path=None): if path is None: path = f"news_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" fieldnames = ["title", "content", "url", "source", "category", "published_at", "crawled_at"] with open(path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(items) return path

CSV导出我支持两个模式：一是导出当前管道刚抓取到的items，二是从SQLite里按时间范围导出历史数据。前者用于快速分享当日采集结果，后者用于给分析团队出历史数据包。两种模式封装成两个函数，但底层都走同一个字段映射，不会出现导出字段对不上的问题。

4. 完整实操流程与运行效果

4.1 环境准备

在开始跑之前，建议用虚拟环境隔离依赖，避免污染系统Python。我用的是Python 3.10版本，最低建议3.9以上，因为用到了dict合并和内置类型注解的较新语法。

python -m venv .venv source .venv/bin/activate # Windows下是 .venv\Scripts\activate pip install requests beautifulsoup4 feedparser

三个依赖缺一不可：requests负责HTTP请求，beautifulsoup4负责解析HTML页面，feedparser负责解析RSS/Atom源。如果你的新闻源只有RSS，第三个库必装，它能省掉大量XML解析工作。

4.2 主流程串联

整个管道的主入口写成一个run_pipeline函数，顺序调用采集、清洗、去重、入库四个阶段的函数。我用了一个伪代码级的函数来处理每个阶段的衔接：

def run_pipeline(): session = build_session() all_items = [] for source in NEWS_SOURCES: try: raw_data = fetch_source(session, source) items = [normalize_item(item, source) for item in raw_data] all_items.extend(items) except Exception as exc: log("error", f"source {source['source_name']} failed: {exc}") continue # 去重（对新抓取的数据做内存级去重 + 数据库级去重） seen = set() deduped_items = [] for item in all_items: fp = item["fingerprint"] if fp in seen: continue seen.add(fp) deduped_items.append(item) inserted = save_to_sqlite(deduped_items) export_path = export_to_csv(deduped_items) log("info", f"pipeline done, fetched={len(all_items)}, inserted={inserted}, csv={export_path}") return inserted

注意这里fetch_source内部会先看源类型。type为rss的用feedparser解析，type为html的用BeautifulSoup加自研的解析函数。每个源在配置时把parse函数挂进去，主流程完全不需要知道源的具体结构细节。

4.3 运行效果与数据验证

跑完一次管道后，终端日志大概长这样：

[2025-03-20 10:00:01] INFO pipeline started, sources=2 [2025-03-20 10:00:03] INFO example_finance fetched 20 items [2025-03-20 10:00:05] INFO example_market fetched 15 items [2025-03-20 10:00:05] INFO deduplicate removed 7 duplicates [2025-03-20 10:00:05] INFO sqlite inserted 28 rows [2025-03-20 10:00:05] INFO csv exported to news_export_20250320_100005.csv

验证数据的正确性，我一般会跑三条SQL：

-- 查看总数和来源分布 SELECT source, COUNT(*) FROM news_articles GROUP BY source; -- 检查最新抓取的50条 SELECT id, title, source, published_at FROM news_articles ORDER BY crawled_at DESC LIMIT 50; -- 验证重复情况（非0则有问题） SELECT COUNT(*) FROM ( SELECT fingerprint FROM news_articles GROUP BY fingerprint HAVING COUNT(*) > 1 );

第一条SQL确认各源都有数据流入，不会出现某个源静默失效。第三条SQL是去重检查的兜底，正常情况下结果恒为0，如果团队里有人绕过管道直接往表里插数据，这条SQL能第一时间发现。

CSV文件建议用Excel或者任何文本编辑器打开检查一下表头和编码是否正常。如果中文乱码，八成是编码没有用utf-8-sig。

4.4 可扩展的调度方案

管道本身跑通之后，下一步就是让它定时跑起来。最轻量的方案是系统自带的任务计划程序：Linux下的crontab或者macOS下的launchd，Windows下的计划任务。比如每30分钟执行一次：

*/30 * * * * cd /path/to/newspipe_etl && .venv/bin/python main.py >> logs/run.log 2>&1

如果不想依赖系统级定时任务，也可以直接在主函数里用time.sleep循环配合最小间隔。我不太推荐这种方式，因为一旦进程挂了没有任何告警，而且日志管理也比较麻烦。

更健壮的做法是引入调度框架如APScheduler，支持cron表达式和持久化任务存储，但这对当前系统来说是增量优化，先不做部署也能跑得很好。

5. 常见问题与排查技巧实录

5.1 请求被拒、返回403或频繁超时

这是爬虫新手最常遇到的墙。403意味着服务器识别出你不是正常浏览器访问，返回超时可能是目标网站有分布式防护，也可能单纯是你请求频率太高把自己IP封了。

我推荐的排查思路是：

• 先看是不是单个源的问题，把其他源注释掉单独测试，定位复现条件。
• 确认UA是否正常，很多网站拒绝无UA或默认Python-requests UA的请求。
• 降低请求频率，在相邻请求间加0.5到2秒随机延时。财经新闻源通常对低频率访问比较宽容。
• 尽量优先选RSS源而不是HTML页面，RSS是站点主动提供的内容分发方式，被反爬的概率小得多。
• 如果目标站明确在robots.txt里禁止爬取，直接放弃这个源，换其他有授权的渠道。

我在项目里专门为每个源配置了请求间隔参数request_interval，采集池并发为1，单线程串行抓取。虽然慢一点，但胜在稳定，不会被封。

5.2 中文乱码与编码错误

乱码问题经常发生在两个环节：网页响应解码和CSV导出。

网页层面，requests会根据响应头自动判断编码，但有些网站响应头写的charset和实际内容不一致，导致自动检测出错。解决办法是先拿到原始字节，然后用chardet或者直接指定常见编码来解码：

raw = resp.content for encoding in ["utf-8", "gbk", "gb2312", "latin-1"]: try: html = raw.decode(encoding) break except UnicodeDecodeError: continue

CSV导出层面的乱码，基本都是编码没用utf-8-sig导致的。用utf-8-sig导出后，Excel直接双击打开不会乱码。还有一点，如果是在Linux上用less查看，BOM会导致第一行出现一个不可见字符，这是正常现象，不代表文件有问题。

5.3 SQLite并发写入报database is locked

当管道脚本和另一个进程同时打开数据库并且都尝试写入时，SQLite会抛出database is locked。我跑定时任务时遇到过，因为上一次任务还没结束，下一次调度又启动了。

解决方案一个是让调度串行化：如果检测到上一个进程还在运行，这次调度直接跳过。在Python里可以用一个锁文件实现：

import os, sys LOCK_FILE = "/tmp/newspipe_etl.lock" def acquire_lock(): if os.path.exists(LOCK_FILE): pid = open(LOCK_FILE).read().strip() if os.path.exists(f"/proc/{pid}"): print("another instance is running, exit") sys.exit(0) open(LOCK_FILE, "w").write(str(os.getpid()))

另一个方案是开启SQLite的WAL模式。WAL模式显著提升了读性能，也减少了读写互斥。在建库后执行一句PRAGMA journal_mode=WAL，长期运行下来稳定很多。

5.4 时间解析失败导致published_at为空

财经新闻的发布时间格式五花八门，有的精确到秒，有的只有年月日，有的还会带“2小时前”这种相对时间。我在开发时发现，单纯匹配格式列表解决不了相对时间，因为相对时间需要以“当前时间”为基准计算。

对于相对时间，我会做一层额外处理：

def parse_relative_time(raw_value): raw_value = raw_value.strip() now = datetime.now(timezone.utc) patterns = [ (r"(\d+)\s*分钟前", 60), (r"(\d+)\s*小时前", 3600), (r"(\d+)\s*天前", 86400) ] for pattern, seconds in patterns: match = re.search(pattern, raw_value) if match: delta = int(match.group(1)) * seconds return (now - timedelta(seconds=delta)).isoformat() return None

所有解析都失败时，published_at返回None。入库时保留None而不是给一个错误时间，是因为对后续分析来说，“未知时间”和“错误时间”带来的误导程度完全不同。做时间分布统计时可以把None过滤掉，但错误时间会导致个别点异常突出，更难排查。

6. 个人经验与后续扩展方向

6.1 做得对和做得不够的地方

这个项目跑了两个月之后，我复盘过哪些决策对最终效果帮助最大，哪些地方早期设计考虑不足。

做得对的地方集中在分层思路。最开始我并没有刻意做多层架构，只是按照“能抓到就行”的思路写。后来发现只要把清洗逻辑独立成一个函数，新增源的工作量就大幅下降——因为大多数源的差异只在解析函数里，后面的链路完全复用。这让我深刻体会到，爬虫项目的维护成本不在“要写多少抓取代码”，而在“后续要改多少基础代码”。

另一个值得说的决策是数据归档。除了写入SQLite，每天还会自动生成一份CSV存到按日期分好的目录里。数据库万一损坏或者想回溯某天抓到的原始数据，CSV就是退路。这个习惯是从运维同事那学来的，数据管道永远要保留一份“平面备份”。

做得不够的地方也有不少。比如category自动分类始终没有做，现在所有文章默认是general，等到真正要做行业维度统计时，还得回填分类。另外，多语言编码问题只处理了常见编码，遇到一些非中英文的小语种网站，还是有乱码风险。采集源的失败告警也停留在日志层面，没有短信或邮件通知，出问题只能靠定期看日志发现。

6.2 后续可扩展的方向

如果这个系统继续往生产级演进，我计划按下面几个方向迭代。

一是接入可视化看板。把每天采集的文章数量、来源分布、最活跃的发布时间段做成简单图表，这样整个管道是否健康一目了然。SQLite里有全部数据，用现成的BI工具或者写个Streamlit页面都很轻松。

二是标题关键告警。财经新闻场景里，很多人关心特定词语，比如“加息”“降准”“财报暴雷”“油价”这类。在清洗层加一个关键词匹配器，命中就走告警推送，Telegram Bot或者企业微信机器人接一下就行，这比人工整天盯页面效率高得多。

三是把去重从“URL+标题指纹”升级为“正文相似度去重”。不同媒体对同一事件的报道，标题往往不同，但正文第一段重合度很高。可以用编辑距离或者SimHash对标题做相似度计算，过滤掉八股文式的转载稿件。这个改动相对大，要控制好在几百篇文章量级下性能还能接受，但确实值得做。

四是把抓取能力从新闻列表延伸到全文详情页。很多RSS源只给摘要或前几段，完整正文需要点进详情页。可以在清洗层识别content中是否有截断迹象，再对详情页发起二次解析。要注意控制全站请求总量，避免对目标站点造成压力。

每一步迭代都要等前一步稳定跑一段时间再上，不要一次性把功能加完。管道这东西，稳定压倒一切。

最后说点实在的。爬虫和ETL系统的价值不在于代码写得多花哨，而在于它能稳定地一天又一天跑下去，数据不丢、不重、格式不乱。NewsPipe_ETL的核心设计就是把这个目标拆解到每一层的职责里：让抓取只管抓、清洗只管洗、存储只管存。如果你也在搭类似系统，先别急着把源铺得太多太广，把两三个典型源的管道跑稳，再慢慢加源。数据管道最怕的不是起点低，而是还没有跑通就先膨胀，最后到处都是半成品。这套代码我已经在多个项目里复用，方向和设计你可以直接拿去做参照。