★ FEATURED ARTICLE Octop 这个名字起得没什么文化我就想要一个能同时伸出好几只触手去抓信息的工具章鱼正好符合这个意象。项目前前后后折腾了两个月中间推倒重来一次现在稳定跑了快半年。这篇文章把整个项目的来龙去脉、架构取舍和踩过的坑捋一遍给打算自己做信息聚合或者任务调度的朋友一个参考。如果你也遇到过这种场景早上要刷十几个 RSS 源下午要盯几个 API 接口的数据变化晚上还要把当天发现的好内容手动整理到公众号草稿箱、飞书群和邮件列表里那你应该能理解我为什么被逼着写了这个东西。Octop 的核心就三件事采集、清洗、分发。它不存储历史数据不做复杂分析更不搞大而全的平台它只负责把“乱七八糟的多路输入”变成“干净且可用的几条输出”。放在今天来看它就是每个个人开发者桌面下最适合养的一只看门章鱼。1. 为什么会有 Octop被信息源和通知渠道逼出来的项目1.1 需求起点多源信息的无序爆炸我同时维护着一个个人的技术博客、一个公众号还帮团队搭了一套内部告警群。工作中每天需要盯的数据源大概分三类RSS 订阅源、第三方 API、几个需要登录才能看的竞品网页。最早我没有统一方案早上用 RSS 阅读器扫一遍下午用 crontab 写几个脚本抓 API晚上再凭记忆把觉得有意思的链接复制到公众号后台。这个流程最大的问题是漏而且漏得悄无声息。某个 API 参数改了脚本静默失败我可能三天后才发现某个 RSS 源断更了两周我还在傻等。真正让我决定动手的是一件小事那周我漏发了一条行业动态结果被合作方打电话问“怎么没同步”。我翻了一圈日志才发现那条数据其实一直躺在原始接口里只是我的解析规则写错了字段名导致内容被当成无效数据过滤了。手动检查太慢了脚本又没有一个统一的状态出口。那时候我就想要一个系统能让我一眼看出“每个源上一次抓到多少条、失败了多少次、最后一次成功是什么时候”。1.2 为什么没直接用现成工具其实市面上的工具我基本都看了一圈。IFTTT、Zapier 这类在线自动化服务胜在简单但缺点是数据必须经过第三方服务器而且适合“轻触发”不适合我这种要做正文解析去重、还要按业务规则打标签的场景。Huginn 我也部署过它是一个很强大的 Ruby 项目理论上可以搭出无限复杂的 agent但实际维护成本不低Ruby 环境、MySQL、Sidekiq一个版本升级就能折腾半晚上。n8n 的交互式画布很漂亮可我不太愿意为了一个信息聚合工具去学习整套节点编排体系也没法很方便地嵌入我自己的 Python 数据处理代码。所以最后我选择了“造轮子”。Octop 从一开始就不是一个追求通用性的平台它的目标非常窄用 YAML 文件描述“从哪抓、怎么抓、抓到后发给谁”然后把剩下的事情交给一小段稳定的 Python 程序去执行。这不是最优的工程方案但对我这种单人维护、数据源不超过二十个、每天数据量千条以内的场景来说它足够直接、足够透明。1.3 目标边界只做管道不做仓库立项当天我写下三条边界后来每次想加功能都会翻出来对照一下只做“源 - 处理 - 目标”这条管道不负责长期存储和可视化分析每个数据源必须能独立启停、独立看日志不能因为一个源挂了拖垮全部任务所有配置走文本文件方便用 Git 管理换机器时能秒级恢复。这三条边界帮我挡住了很多“顺手再加个功能”的冲动。比如曾经有人建议我加一个 Web 后台来管理数据源我仔细想了想Web 后台会带来登录、权限、前端维护、接口设计等一系列复杂问题而这些并不是我的核心痛点。我的核心痛点是“能不能稳定抓取、准确过滤、按时发送”。只要这个管道是清晰的我就已经赢了 80%。2. 整体架构三个关键取舍让代码瘦了一半2.1 数据流设计一条单向前进的管道Octop 的数据流非常简单从输入到输出总共五步源适配器 - 解析规则 - 过滤器 - 去重指纹 - 输出适配器每个源适配器只负责“拿到原始内容”。RSS 源就返回 feedparser 解析后的 entriesAPI 源就返回 JSON网页源就返回经过 BeautifulSoup 处理的 DOM 对象。拿到原始内容之后统一的解析器会按照配置把标题、正文、链接、发布时间、作者等字段抽出来。过滤器再根据关键词黑名单、日期范围、内容长度做一次粗筛。剩下通过粗筛的内容会进入去重指纹模块如果指纹不存在才允许进入输出适配器。这个设计把每个环节都拆成了“可插拔”的小块。一开始我担心环节多了会不会性能太差实际跑下来完全多虑了单条内容的处理时间平均不到 50 毫秒瓶颈永远在网络请求上而不是这些本地处理逻辑。对我来说这个数据流最大的价值是定位问题时非常快。某条内容没发出去只需要顺着管道看一眼是源没抓到、解析字段为空、过滤器误杀、还是分发报错基本 30 秒内就能找到原因。2.2 为什么用插件加 YAML 规则而不是硬编码第一版 Octop 我是硬编码的每个数据源写一个类类里写死抓取地址、字段名、通知目标。写了六个源以后代码就变得很尴尬大量重复逻辑新增一个源要复制粘贴改一堆东西而且稍不小心就会把上一个源的特殊处理带过来。后来我重构了把“变化的部分”全部抽到配置里。一个标准的数据源配置长这样- id: tech_blog name: 技术博客资讯 enabled: true schedule: 0 8,12,18 * * * source: type: rss url: https://example.com/feed.xml timeout: 15 parse: title: selector: title link: selector: link content: selector: summary strip_html: true published: selector: published filters: min_content_length: 200 keywords_blacklist: - 广告 - 招聘 output: - channel: feishu target: webhook_url_sha256 - channel: mail target: digestexample.com这样做的直接收益是新增一个数据源时我基本不再写 Python 代码只需要复制一段 YAML 改一下地址和解析规则。解析规则里的selector字段同时支持 CSS 选择器、JSONPath 和正则表达式我会在下一章详细讲。更重要的收益是配置文件可以进入 Git 做版本管理每次改完数据源我都能清楚地看到 diff出问题也能回滚到上一个可用版本。2.3 存储选型SQLite 加 Redis 的搭配理由很多朋友问我为什么不用 MySQL 或者 PostgreSQL我的答案很简单单人单机场景下它们带来的运维负担大于收益。Octop 需要持久化的数据只有三类内容指纹、抓取历史、任务调度状态。这些数据量撑死几十万条SQLite 一个文件就能搞定而且无需单独安装、无需配置账号密码、备份时复制文件就行。Retention 策略也很简单保留 30 天超过 30 天的抓取历史定时清理避免文件无限膨胀。Redis 在这里的角色不是最终存储而是“短时状态和高速判断”。我用它做三件事存储内容指纹的布隆过滤器判断重复时 O(1) 完成作为轻量任务队列抓取任务和分发任务之间通过 Redis Stream 解耦做令牌桶限流避免某个源请求太频繁被对方封 IP。这套组合的边界很清楚如果以后数据量涨到百万级或者要多台机器协同SQLite 会变成瓶颈那时再平滑迁移到 PostgreSQL 也不迟。现在这个阶段我用一台 1 核 2G 的云服务器就能跑得很舒服成本几乎可以忽略。3. 核心模块实战从抓取到分发的一整条链路3.1 输入源适配器RSS、API、网页三种接入方式Octop 的源适配器接口定义得很简单每种源实现两个方法fetch()负责拿数据parse()负责把原始数据转换成统一结构。RSS 源是最好写的。Python 生态里的feedparser几乎支持所有常见格式RSS 2.0、Atom、RDF 都能解析。我只需要import feedparser class RssSource: def fetch(self, config): feed feedparser.parse(config[url], request_headers{User-Agent: Octop/0.4.2}) return feed.entries def parse(self, raw_entry): return { title: raw_entry.get(title, ), link: raw_entry.get(link, ), content: raw_entry.get(summary, ), published: raw_entry.get(published, ), }API 源稍微麻烦一点因为有的接口需要签名认证有的需要先登录拿 Token。我的处理方式是给 API 源配置一个预请求模板第一次请求时执行auth配置把返回的 Token 缓存到 Redis过期再刷新。这样既保证了实时性又不会频繁调用登录接口。网页源则是三个源里最脆弱的。一个页面可能今天能解析明天网站改版就不能解析了。所以我给网页源单独加了“健康检查”如果连续两次抓取结果为空立即把源标记为degraded并告警通知我处理。3.2 解析规则用一套 DSL 覆盖所有字段提取场景解析规则是整个 Octop 里最值得说的部分。我最初设计时想了很久到底是支持 XPath、CSS 选择器还是 JSONPath最后决定全部支持但用一套统一的 DSL 来描述。每个字段的提取规则是一个selector加一个methodcss用 CSS 选择器定位元素适合静态网页底层用 BeautifulSoup 的select_one实现xpath复杂 HTML 结构才用比如需要按“包含某个文本”来定位父节点jsonpath针对 JSON API 返回的数据做提取regex对前面提取出的原始字符串做二次清洗比如去掉时间戳或者截取正文摘要。一个复杂字段的配置示例parse: title: selector: article h1 method: css content: selector: //article//div[classpost-body]/p//text() method: xpath join: \n author: selector: $.data.author method: jsonpath clean_title: selector: title method: regex pattern: (.*?) - 站点名$为什么不用单一选择器因为现实世界的数据源参差不齐。RSS 里字段是现成的JSONPath 最合适网页里 CSS 简洁且容错性比 XPath 高而某些页面结构混乱时XPath 更稳最后再用正则收口。这一套组合下来Octop 几乎没有遇到过“完全提取不出来”的页面顶多是选择器写得不严谨导致匹配到了广告区域。3.3 去重策略布隆过滤器加 SimHash 双保险内容去重是信息聚合工具最容易翻车的地方。同一个事件可能早上被源 A 报道下午被源 B 转载标题不一样但正文相似度 90% 以上。如果只比对标题或链接就会出现大量重复通知。Octop 用了两层去重。第一层是精确去重对“标题 正文前 50 个字”计算 SHA-1 指纹同时维护一个 Redis 布隆过滤器。布隆过滤器的好处是查询速度快、内存占用低坏处是可能有极低概率的误判但对我这种场景完全能接受。第二层是相似去重对正文内容计算 SimHash 的 64 位指纹如果两篇文章的 SimHash 汉明距离小于 3就认为它们是高度相似文章只保留最早到达的那一篇。这里有个经验值得分享SimHash 不是对所有内容都适用。比如两篇技术博客谈的是同一个框架但一个讲安装、一个讲调优正文相似度其实很低汉明距离可能超过 10不会误判。但如果是新闻通稿类内容大段文字完全相同SimHash 效果就很好。所以我在配置里加了一个dedup_mode字段可以针对数据源选择exact、simhash或both。新闻源默认开both技术博客默认开exact这样既能拦住转载又不会误伤原创观点类文章。3.4 输出通道飞书、邮件、Webhook 的统一抽象输出模块的设计比采集模块更简单因为发送本质上只有一件事把(title, content, target)传给对应通道的客户端。Octop 内部定义了一个统一的输出接口class OutputChannel: def send(self, message: OutgoingMessage) - bool: raise NotImplementedError飞书、钉钉、企业微信这类群机器人其实都是发一个 Webhook POST 请求区别只是消息体 JSON 的格式。我封装成FeishuChannel、DingTalkChannel底层都用httpx.AsyncClient发送并设置 10 秒超时。邮件通道则用aiosmtplib支持 SMTP_SSL 和 STARTTLS 两种模式配置里填上服务地址、账号、密码就可以把每天汇总的内容整理成一篇纯文本摘要发到指定邮箱。在输出环节我犯过一个很蠢的错最初所有输出都是同步发送一个飞书 Webhook 超时会导致整个调度线程卡住后面的任务全部排队越积越多。后来我把输出改成异步并发并用信号量控制并发数默认最多同时发 5 个请求既不会因为串行拖慢速度也不会因为并发太高触发对方限流。4. 调度与容错能稳定跑几个月的关键细节4.1 用 APScheduler 而不是手写 sleep 循环Octop 的调度核心是 Python 的APScheduler库。最初我图省事直接用time.sleep加while True循环每个源一个线程。跑了两周就发现两个问题一是 Python 的线程调度不是严格精确的有的任务会越跑越偏二是程序重启后没法“补跑”错过的任务。APScheduler 则支持 cron 表达式、持久化 job store还能在任务实例启动时自动计算“从上次运行结束的位置继续”这对我这种偶尔需要重启服务的场景很重要。每个数据源都会根据自身的schedule字段注册成独立 job比如scheduler.add_job( run_source, # 执行函数 triggerCronTrigger.from_crontab(source_config[schedule]), args[source_id], idfsource_{source_id}, replace_existingTrue, misfire_grace_time300, # 错过 5 分钟内的任务仍然执行 )misfire_grace_time是我强烈建议参数化的一个值。它决定了如果调度器因为休眠或重启错过了任务超过多久就不再补跑。我设为 300 秒意思是最多补跑 5 分钟内的任务。这样既不会丢失关键数据源又不会在服务器长时间宕机后一下子把几十个任务全部补跑导致目标网站被请求淹没。4.2 抓取任务与分发任务分离Redis Stream 做缓冲最早的版本里抓取、解析、分发是同步在一个任务里完成的。这样做代码简单但有一个很明显的隐患如果分发通道很慢或者某个通知目标暂时不可用这次抓取就会一直阻塞后续的定时任务全部排队。后来我把它们拆开了中间用 Redis Stream 作为缓冲队列。一条完整内容在抓取解析完成后会XADD进 Redis Stream独立的 worker 进程监听这个 Stream取出消息后做过滤、去重、分发。Redis Stream 天然支持消费组以后我想加第二个 worker 实例只需要增加一个消费者即可不用担心重复消费。这个改动带来的效果非常明显抓取任务永远只关心“我有没有拿到数据”分发任务只关心“我怎么把数据发出去”。两者生命周期完全解耦某一方的波动不会影响另一方。4.3 失败重试与死信让“静默失败”无法发生静默失败是这类工具最可怕的事情。我最初踩过一个大坑某个 API 数据源因为 Token 过期连续三天每天返回 401但程序没有抛异常只是解析出空结果于是 Octop 认为该源今天没有新内容也就不分发。我直到看月度报告才发现这个源三天的产量是 0而正常情况下它每天应该有 20 条左右。现在 Octop 的重试机制是这样的每个源连续失败 3 次后自动进入paused状态不再执行抓取同时触发“告警通知”直接发到团队管理群。进入paused状态后我会收到消息手动检查是接口问题还是配置问题修复后通过命令恢复。重试间隔使用指数退避第一次失败后等 1 分钟第二次 5 分钟第三次 15 分钟。超过三次不再重试等待人工介入。这个策略可能会漏掉某个数据源一个上午的数据但换来的好处是我确信每一次异常都被看见了。4.4 幂等分发防止重试造成重复消息分发环节也有自己的坑。飞书 Webhook 偶尔会返回超时但超时不代表消息没有送达。如果我的程序简单地把这种情况当作失败并重试接收方就会收到两条一模一样的消息。这个问题必须靠“幂等性”解决。Octop 的做法是给每条待分发的内容生成一个全局唯一的event_id组成规则是“内容指纹 目标 channel 目标地址”。第一次发送前把event_id写入 Redis 的SEEN集合并设置 24 小时过期。发送时如果SEEN里已经存在相同的event_id直接跳过。由于 Redis 的写入是原子操作即使我开了多个 worker 同时消费也不会出现两个 worker 同时通过检查的问题。这套机制让我在实际运行中几乎没收到过重复消息而代价只是每次发送多一次 Redis 判断性能损失可以忽略。5. 实测效果与踩坑记录5.1 稳定运行几个月的真实数据Octop 目前稳定运行的配置是数据源数量8 个3 个 RSS、3 个 API、2 个网页输出通道飞书群机器人、SMTP 邮件、通用 Webhook每天抓取原始数据约 700 条去重后有效数据约 120 条消息送达成功率99.5%为什么不是 100%那 0.5% 基本都是对方接口偶发的 5xx 错误重试两次后仍然失败就会进入死信队列等待人工处理。这个成功率配合“失败必告警”的机制对我来说已经完全可用了。值得一提的是700 条原始数据最后只剩 120 条这中间 580 条被过滤掉绝大部分是因为我配置了“相同新闻只保留一条”的相似去重这个比例比我一开始预想的高很多说明多源信息确实存在大量重复。5.2 线上事故Redis 连接池耗尽有一次 Octop 突然开始疯狂报错日志里全是Cannot connect to Redis: Connection pool exhausted。我第一反应是 Redis 负载太高但是redis-cli INFO一看内存、CPU 都很正常连接数却高得离谱。后来定位才发现我在写某个网页源解析函数时from redis import Redis后直接Redis(host127.0.0.1)实例化函数执行完连接没关闭而那个任务每小时跑一次每次创建几十个连接时间一长连接池就被占满了。这个坑的根因就是我一开始没有统一管理 Redis 连接。修复方案很简单定义一个全局的 Redis 连接池所有模块复用同一个实例不再在函数内部创建新的连接。这也让我养成了一个习惯凡是涉及外部资源连接都要统一放在工具层管理并发场景下一个连接池、一个客户端实例是必须遵守的纪律。5.3 解析规则失效空结果不等于没有数据另一个让我印象深刻的坑是某个竞品网站改版原本的正文选择器从div.article-content改成了div.post-content我的解析器没有报错只是提取结果为空。由于过滤器会丢弃空内容这个源在外界看来就是“今天没有更新”。直到我在接收端连续两天没看到它的消息才意识到出了问题。Octop 后来增加了一个“零结果告警”机制如果某个源执行完成后有效内容数量为 0且这个源在过去 7 天里平均每天都有结果就会触发提示“该源可能改版或反爬”。这个机制非常有效因为大部分数据源正常情况下的产量是比较稳定的突然降到 0 大概率是解析规则问题而不是真的没有新内容。空结果有两种可能一是网站真的没更新二是程序坏了。能力上我可以区分但意识上我需要一个自动提醒而不是等自己发现。5.4 内存占用从 200MB 降到 40MB第一版 Octop 运行后占用内存一直在 180MB 到 220MB 之间徘徊对一台 2G 内存的云服务器来说有点奢侈。我做了三处优化减少了 Playwright 实例数量。原来每个需要 JS 渲染的网页源都会启动一个浏览器上下文但 Playwright 本身就非常吃内存。后来我把它改成全局复用同一个 browser 实例用 context 隔离会话内存直接从 120MB 降到 60MB。解析大响应时改用流式读取避免一次性把几百 KB 的 HTML 全部读进内存。移除了 pandas。第一版我为了做简单统计引入了 pandas但实际数据量根本用不着换回纯 Python 的Counter之后内存又降了一截。优化完以后Octop 常驻内存稳定在 40MB 左右这台云服务器甚至还能再跑两三个其他小服务。这也让我意识到个人项目里引入重型依赖前要想清楚它带来的内存和复杂度开销有时候远大于它提供的便利。6. 进阶玩法把 Octop 从信息中转站变成内容预处理器6.1 在分发前接入 LLM 摘要Octop 稳定运行以后我开始不满足于只做转发。很多内容抓回来是一篇长文直接发到群里会刷屏也没人在意。我想让它在分发前自动生成摘要、提取关键词、打上领域标签这样接收方只看摘要就能决定要不要打开原文。我实现了一个可选的llm_summary插件在分发前把正文截断到 4000 字以内通过 OpenAI 兼容接口调用一个本地或云端模型返回结构化的摘要和标签。Prompt 我写得比较固定你是一个内容编辑。请阅读下面的文章输出三行摘要每行不超过50字 再输出3-5个关键词和1个领域标签。 文章内容 {content}这样做有两个意想不到的好处第一它能自动识别出一些规则没覆盖到的低质量内容比如纯 SEO 拼凑的页面摘要会特别空洞第二它让我可以把一些长文自动改写成适合邮件日报的形式。当然接入 LLM 后单条内容的处理时间从几十毫秒变成了几秒钟所以我把调用方式改成异步的并且设置并发上限为 3避免影响主流程。6.2 插件协议标准化让新增数据源变成一件轻事Octop 的插件协议经历了三次调整最终稳定为一个极简的抽象类class SourcePlugin: id name async def fetch(self, config: dict) - list[dict]: ... async def parse(self, raw: dict, config: dict) - ParsedItem | None: ...所有内置数据源都迁移成了插件。这样做的直接好处是如果我想临时加一个数据源只需要写一个新的SourcePlugin子类然后在配置里声明plugin: my_source即可。主程序不再关心这个源是 RSS、API 还是网页它只负责调用插件的fetch和parse把结果送进管道。插件之间完全隔离某个插件抛异常不会影响其他插件运行。这个协议现在也开放给我身边的朋友用了他们不需要理解整个 Octop 的代码结构只需要照着已有插件写一个类就能接入自己的数据源。后面我还想加一个基于setuptools的入口点注册机制让插件能独立打包安装而不是必须把代码拷进项目目录。这样 Octop 才能真正从一个私人工具变成一个小型生态。6.3 几个值得写进下一代版本的设计预期如果下一步要做 Web 控制台我会把配置管理从 YAML 改成数据库存储并保留“导出 YAML”功能。因为 YAML 适合版本管理但不适合多人同时在线编辑。权限上至少要区分“管理员”和“操作员”两种角色操作员只允许启停源、查看日志管理员才能改配置和删除数据。调度方面我打算引入更细粒度的“时段”配置比如某个源只在工作日 9 点到 18 点之间运行周末自动跳过。这些功能单看都不复杂但堆在一起会显著提高代码复杂度所以我会在确有需求时再动手而不是现在就急着把架构做大。我个人现在的体会是一个工具能稳定跑半年靠的不是一开始设计出完美架构而是把“异常可见性”做透。Octop 能活这么久最大的功臣不是那套插件系统也不是去重算法而是“任何异常都会导致告警”“任何源状态都可以随时查看”这两个朴素机制。如果你也想写一个类似的信息聚合工具我建议你先把这两个机制放在最优先的位置其他功能都可以后续再加。
阅读完成 · 觉得有帮助？
