从项目名“fox_charon”聊起。这个名字拆开看很有意思fox是狐狸charon是希腊神话里冥河的摆渡人专门把亡灵从河这边送到那边。合在一起可以理解成“一只负责摆渡信息的狐狸”。我给自己的业余项目起这个名字就是想要一个轻量、敏捷、专门做信息采集与调度分发的个人工具。它解决的核心问题是每天面对一堆分散的RSS、网页、API接口、本地笔记人肉整理太耗时固定规则又不够智能于是干脆写了一个能把“碎片化信息”统一收拢、按规则加工、再推送到指定终端的自动化任务代理。这篇文章就把整个项目从命名理念、架构设计到实测踩坑的完整过程拆开来讲适合对自动化工作流、智能Agent、个人知识库建设感兴趣的朋友参考。1. 为什么做这样一个项目从信息过载到“摆渡”思维的转变1.1 项目命名的渊源与定位“fox_charon”这个名字的过程本身就是一个设计决策。我最初想的是一堆类似“data_pipeline”“info_bot”这样直白的名字但后来发现命名不仅影响观感还会反过来约束你对系统的理解。狐狸这个意象强调“灵活、轻巧、不按常理出牌”而charon的“摆渡”意象则点明了系统的核心职责不是生产信息而是传输和转换信息。一个只做搬运与调度的工具如果把自己定位成“平台”或者“框架”很容易越做越重最后变成一个什么都能干但什么都干不利索的大杂烩。把定位收窄到“摆渡人”之后所有模块的设计都有了统一标准能不能让信息更快、更准、更省心地到达它该去的地方。这个项目更适合谁如果你手头有大量订阅源需要定期整理或者你在做个人知识库经常要把网页内容转成结构化笔记再或者你单纯对“如何用代码替代重复性的信息搬运劳动”感兴趣那这个项目的思路和代码就一定值得你看完。它不是那种需要大型团队维护的企业级平台而是一个运行在自己服务器或本机上、可以随时改动的个人工具。1.2 最初要解决的场景问题我自己的痛点很具体每天早上要先刷十几个RSS源把觉得有价值的文章存到稍后读工具里每周还要把分散在不同文档里的资料汇总成一份周报偶尔需要盯某些页面有没有更新有更新就立刻通知我。这些事情用浏览器书签解决不了用现成的IFTTT又觉得规则太死板很多平台提供的自动化只能做“简单判断”没法做“内容级处理”。比如我想把一篇文章的关键段落提取出来再翻译成中文再按标签自动放到对应的知识库目录里——这种多步加工流程现成工具基本做不到。所以“fox_charon”从第一天起就定下了几个硬性目标轻量不引入重框架能用标准库解决就不用第三方依赖。插件化数据源、处理逻辑、输出终端都可以单独开发、单独替换。可观测每一步的输入输出都要有日志出问题要能快速定位。懒人优先配置尽量用简单规则不需要写一堆DSL或配置文件。1.3 方案选型的几个对比与取舍在动手之前我对比过几条技术路线。第一条是直接用现成的自动化平台比如n8n或者Node-RED。这类工具的好处是可视化拖拽上手快但缺点是复杂逻辑写起来很别扭而且一旦流程多了画布上的连线跟蜘蛛网一样维护成本反而高。第二条是用开源爬虫框架比如Scrapy但它偏向于大规模抓取对“小而美”的信息处理场景显得过重而且它的调度模型和任务队列需要额外设计。第三条就是自研一个轻量Agent框架把采集、处理、分发拆成独立模块用消息队列串起来。我最终选了第三条理由也很朴素项目规模不大自研成本可控各模块之间解耦后后续改造成本极低而且这个项目的核心乐趣就在于可控和透明不想被别人的封装卡住脖子。提示如果你是第一次做类似项目不建议一上来就设计特别复杂的插件体系。先把一条最简单的主链路跑通比如“抓取一个RSS → 提取标题和正文 → 推送到飞书群”之后再逐步拆模块。一上来就建抽象层大概率会在前期被各种接口设计折磨到弃坑。2. 核心架构与关键模块设计2.1 整体架构与数据流向“fox_charon”的整体结构可以理解为一条四段式流水线采集端负责从不同数据源把原始数据拉下来规划端负责判断这批数据该怎么处理执行端负责调用具体处理组件输出端负责把结果送出去。这四段通过一个内存队列串起来队列里跑的是统一格式的“Message”对象不管数据从哪里来进入队列之前都会被清洗成同一种结构。数据源 → 采集器(Collector) → 消息队列 → 规划器(Planner) → 执行器(Executor) → 发送器(Sender) → 目标终端单看数据流会觉得简单但真正的设计难点在每一段的“边界”上。采集器只负责拿到原始数据并转换成Message绝不做业务判断规划器只负责根据规则决定“下一步调谁”绝不说“怎么调”执行器只负责干活干完活把结果塞回队列发送器只负责把最终结果推到目标渠道。这种单一职责拆分带来的好处非常明显我后期新增一个Telegram的推送渠道时只写了一个Sender插件完全没动其他模块。2.2 Message对象与统一协议所有模块之间的通信都基于一个自定义的消息协议这是整个系统的血液。协议字段我精简到最少只保留了链路追踪和业务处理必要的信息id消息唯一ID用于日志追踪。source来源标识比如rss.foxera.com。type业务类型比如article、alert、summary。payload实际业务数据JSON格式。meta元信息比如抓取时间、原始URL、重试次数。sink目标终端标识比如feishu、local_md。设计Message时我踩过一个坑一开始把payload设计成“宽松的字典”什么字段都能往里塞。结果就是后期解析逻辑里到处是if author in payload这样的判断代码臭不可闻。后来干脆给常用业务类型定义了固定Schema比如文章类型必须有title、url、content三个字段缺失就直接丢弃并告警。这样虽然前期类型定义多写了一些代码但后期维护省下了大量时间。2.3 采集层插件化数据接入采集层是系统里最“多活”的部分。目前我实现了RSS采集、网页内容提取、API轮询和本地文件监控四类采集器每一类都是一个独立插件。插件化在这里的好处是新数据源的接入不需要改核心代码只要实现一个固定的Collector接口然后注册到配置里就行。以RSS采集器为例核心逻辑并不复杂# collector_rss.py import feedparser from dataclasses import dataclass dataclass class RSSCollector: feed_url: str source_name: str last_entry_id: str def fetch(self) - list[Message]: feed feedparser.parse(self.feed_url) messages [] for entry in feed.entries: if entry.id self.last_entry_id: break messages.append(self.to_message(entry)) if messages: self.last_entry_id messages[0].payload[url] return messages def to_message(self, entry) - Message: payload { title: entry.get(title, ), url: entry.get(link, ), content: entry.get(summary, ), author: entry.get(author, ), } return Message( idmake_uuid(), sourceself.source_name, typearticle, payloadpayload, meta{fetched_at: current_ts()}, )这个采集器有一个值得说的细节用last_entry_id记录最后一次已处理条目的ID下次抓取时一旦遇到这个ID就停止。这个逻辑比单纯按时间戳过滤更可靠因为RSS条目的时间字段格式五花八门直接用字符串比较很容易出错。不过这个方案假设RSS条目顺序是从新到旧绝大多数源都满足极少数乱序源需要额外排序。2.4 规划层规则驱动的任务分配器规划层是整个系统的“大脑”但实际上它并不智能更像一个规则匹配器。我设计了三种决策模式定时模式、触发模式和优先级模式。定时模式最简单比如“每天早上八点处理一遍”触发模式是“当某个采集器产出新消息时立刻处理”优先级模式则是“当队列积压超过阈值时优先处理type为alert的消息”。# planner.py class Planner: def __init__(self, rules: list[Rule]): self.rules rules def decide(self, msg: Message) - list[Task]: tasks [] for rule in self.rules: if rule.match(msg): tasks.append(rule.build_task(msg)) return tasksRule是一个抽象类match方法判断消息是否满足触发条件build_task方法返回具体的处理任务。比如我有一条规则是“如果消息来源是某个特定博客且类型是article则调用摘要生成任务和翻译任务”。这种设计的好处是规则之间互不依赖新增一条规则不影响现有流程。坏处也很明显规则多了以后会互相叠加同一个消息可能被多个规则匹配产生重复处理。解决办法是给Task加一个dedup_key字段在进入执行层之前做一次去重。2.5 执行层与输出层的设计思路执行层是一组具体的处理器比如文本摘要、关键词提取、HTML转Markdown、翻译等。每个处理器接收一个Message处理后返回一个新的Message。处理器可以串联比如“先提取正文再生成摘要最后翻译标题”。串联顺序不是硬编码在代码里的而是由规划层生成的Task携带的步骤列表决定。这样同样一条消息在不同的规则下可以走完全不同的加工路径。输出层负责把最终结果送到目标终端。目前我实现了本地Markdown目录写入和飞书Webhook推送两种Sender。这里有一个很重要的设计约束Sender只负责“传输”不负责“格式化”。所有格式化工作都在执行层完成Sender拿到的数据已经是准备就绪的文本、列表或JSON。这个约束让我在新增推送渠道时非常省心因为不管推到哪数据格式都是统一的只是传输方式不同。3. 从零搭建完整流程的实操记录3.1 技术选型与依赖清单“fox_charon”整体使用Python 3.11开发依赖极少核心依赖只有三个feedparser用于RSS解析requests用于HTTP请求beautifulsoup4用于HTML内容提取。其他像hashlib、json、sqlite3都是用标准库。这样做的原因是依赖越多环境迁移越痛苦。我甚至考虑过用纯标准库实现RSS解析但feedparser对畸形XML的容错做得实在太好自己写解析器得不偿失。项目目录结构是这样的fox_charon/ ├── charon/ │ ├── __init__.py │ ├── message.py # Message结构与协议定义 │ ├── queue.py # 内存队列与去重逻辑 │ ├── planner.py # 规则引擎 │ ├── executor.py # 任务执行器 │ ├── sender.py # 输出发送器基类 │ ├── collectors/ │ │ ├── rss.py │ │ ├── web.py │ │ └── api_poller.py │ ├── processors/ │ │ ├── extractor.py # 正文提取 │ │ ├── summarizer.py # 摘要生成 │ │ └── md_converter.py # HTML转Markdown │ └── senders/ │ ├── local_md.py │ └── feishu.py ├── config/ │ ├── feeds.yaml # 订阅源配置 │ └── rules.yaml # 规则配置 ├── data/ # 输出目录与SQLite数据库 └── main.py # 入口目录结构遵循了“按层分包、按组件分模块”的原则。collectors、processors、senders三个目录天然对应三条扩展线以后加新功能时基本不会出现“不知道该放哪”的情况直接看目录结构就明白了。3.2 关键代码消息队列与去重机制消息队列是整个系统的基础设施。我选用了Python标准库的queue.PriorityQueue但做了一点改造队列元素不只是Message而是(priority, Message)元组。优先级数值越小越先处理这样alert类型的消息可以插队。队列满的时候默认阻塞避免内存无限增长。去重逻辑单独放在一个模块里用一个SQLite表记录消息ID的哈希值。这个方案比直接用内存集合更可靠因为系统重启后内存里的去重记录会丢失SQLite可以持久化保存已处理消息的指纹避免重启后重复处理同一批数据。# queue.py import sqlite3, hashlib from queue import PriorityQueue class DedupStore: def __init__(self, db_path: str): self.conn sqlite3.connect(db_path) self.conn.execute(CREATE TABLE IF NOT EXISTS seen (hash TEXT PRIMARY KEY, ts INTEGER)) def is_dup(self, msg_id: str, source: str) - bool: h hashlib.sha256(f{source}:{msg_id}.encode()).hexdigest() cur self.conn.execute(SELECT 1 FROM seen WHERE hash ?, (h,)) found cur.fetchone() is not None if not found: self.conn.execute(INSERT INTO seen (hash, ts) VALUES (?, ?), (h, int(time.time()))) self.conn.commit() return found实际使用中这个去重表需要定期清理否则会无限膨胀。我加了一个定时任务每天凌晨删除七天前的记录因为消息处理认期的确超过七天的重复也没必要再拦截。3.3 核心流程从RSS抓取到Markdown归档下面这条链路是我用的最多、也最稳定的场景每天抓取五个技术博客的RSS对每篇新文章提取正文、转成Markdown、生成一句话摘要、按博客名分类存入本地目录同时推送一条摘要到飞书群。配置非常简单写在rules.yaml里rules: - name: blog_digest match: source_group: tech_blogs type: article tasks: - processor: html_to_md - processor: summarizer sink: local_md - name: blog_alert match: source_group: tech_blogs type: article tasks: - processor: summarizer sink: feishu我忍不住想强调一下“source_group”的设计。在配置里给多个数据源打组比在规则里一个个列源名称要方便得多。我只需要在feeds.yaml里给每个订阅源加上group: tech_blogs的标记然后规则里写source_group: tech_blogs就能一键匹配整组源。这个设计是后期补上的早期我的规则里写了一长串来源名看起来跟天书一样。3.4 邮件周报的手工执行流程除了定时自动执行我也做了手动触发机制。每周五下午我会跑一条命令生成当周所有处理过文章的汇总Markdown文件附上每篇的标题、链接、摘要和我的即时笔记。这条链路没有走定时器而是靠一个命令行参数触发python main.py --report weeklymain.py会根据参数调用一个特殊的汇总任务从SQLite里捞出最近七天的消息记录按来源分组排序然后调用local_md的周报模式输出。这里的关键点是处理历史记录不能只存在内存里必须持久化。我所有的Message在进入队列时都会先写一份到SQLite的message_log表这样无论实时处理还是事后统计都有据可查。3.5 配置加载与环境变量管理配置文件的加载我采用了“默认值自定义覆盖”的方式。项目内置一份config/default.yaml包含所有模块的默认参数用户自定义的config/config.yaml会覆盖默认值。这样既能保证开箱即用又不会把用户配置搞得太复杂。敏感信息比如Webhook地址、API密钥不放在YAML文件里而是通过环境变量注入。我在配置加载模块里做了简单的模板替换支持${FEISHU_WEBHOOK}这样的写法# config_loader.py import os, re, yaml def load_config(path: str) - dict: raw open(path, encodingutf-8).read() raw re.sub(r\$\{(\w)\}, lambda m: os.environ.get(m.group(1), ), raw) return yaml.safe_load(raw)这个实现虽然简陋但足够用。加密存储、密钥管理这类企业级需求对这个项目来说是过度设计。把密钥放在环境变量里配合systemd的EnvironmentFile已经能覆盖个人使用场景。4. 实测运行效果与调试记录4.1 第一轮实测从零抓取到推送全链路打通系统写完基本功能之后我做了第一轮完整的实测。测试场景是同时加载五个RSS源其中两个源故意配置成无效URL看系统会怎么表现。运行结果是三条正常源共抓取到27篇新文章全部成功提取正文并转成Markdown两条异常源各自抛出了超时和DNS解析错误被重试机制兜住后标记为失败最终推送到飞书的摘要消息延迟大约2.3秒。第一轮跑通后我反而更关注那些“没出问题”但看起来不对的地方。比如有一篇文章的内容提取出来只有两行原因是那个页面用了大量JavaScript渲染beautifulsoup抓到的静态HTML里根本没有正文。这个问题光看日志的“成功”标记是发现不了的必须抽查实际输出内容。后来我加了一个正文长度阈值校验正文少于200字符的提取结果会被标记为“可疑”在飞书消息里加了一个[LOW_CONTENT]标签提醒人工复核。4.2 重复消息问题与去重策略调整运行到第三天我发现一个规律的重复现象某些文章一小时内被处理了两次。查日志发现是RSS源的条目ID不太稳定同一篇文章第一次抓取时的ID和第二次不一样导致去重哈希计算出的指纹不同。RSS的guid标签理应是永久ID但有些源会把不稳定的参数拼进去导致ID漂移。针对这个问题我做了一个两层去重策略。第一层还是基于消息ID的哈希去重第二层基于正文内容的语义指纹去重方法是把正文去掉空格和标点后取前200个字符再计算SHA-256哈希。这个方案对付ID漂移很有效。def content_fingerprint(text: str) - str: compact re.sub(r[\s\W], , text.lower())[:200] return hashlib.sha256(compact.encode()).hexdigest()代价是每次处理需要先截取正文多了几步计算。但对于个人项目这个量级完全不是问题。4.3 时区问题导致的定时任务错乱系统跑了一周后我又发现定时日志的分钟数不对。我在配置里写的是08:00出发定时任务但实际日志显示每天执行的时刻比我预设的晚了八小时。原因很典型服务器的系统时区是UTC而配置解析默认把所有时间按本地时区处理。datetime对象时区信息为naive进行比较时系统会直接按直觉走结果就是每晚八点执行。修复方式有两种一是把所有配置时间统一改成带时区信息的ISO格式比如2025-01-04T08:00:0008:00二是所有内部时间一律用UTC只有展示给用户时才转成本地时区。我选了第二种因为内部处理统一UTC能避免很多类似“周一凌晨跑出的周报落到了上周”的边界问题。4.4 常见问题速查表整理一下我跑这一个月遇到的高频问题做成速查表供大家直接查现象可能原因解决方式某个源持续抓取失败站点封禁或需要User-Agent/ Cookie在采集器里增加请求头配置模拟浏览器访问提取的正文为空目标页面是JS渲染改用非移动版页面或接入渲染服务消息重复处理RSS ID漂移增加正文语义指纹去重定时任务时间不准服务器时区是UTC内部统一用UTC时间展示层再转本地推送消息丢失Webhook网络抖动发送器增加失败重试最多三次指数退避队列阻塞某个处理器执行太慢给每个处理器设置超时优先用requests的timeout参数4.5 关于“日志就是产品的生命线”这条经验调试过程中我的最大体会是日志写得好不好直接决定这个项目能不能持续迭代。我的日志规范是所有消息从采集器出来就打一条INFO内容包含message_id和source进入执行器处理时每步都打一条DEBUG日志记录输入输出大小输出成功打INFO失败打ERROR并附带重试次数。初期为了省事很多日志我跳过了等遇到问题查日志时才发现根本看不出问题出在哪一层。后来花了半天时间把日志补全后面的调试效率直接翻倍。注意所有日志和持久化数据里尽量不要保存API密钥。有一回我的日志模块把完整的Webhook URL打了出来虽然只有我自己看但那之后我再没把任何敏感信息直接打进日志要打也只打脱敏后的后半段。5. 进一步扩展方向与个人体会5.1 可以往哪些方向继续演进“fox_charon”目前的状态是一个运行稳定但功能克制的v1版本。我接下来的计划是把规划层的规则引擎升级为“混合决策”简单的规则匹配继续保留同时接入一个轻量模型让系统能根据内容语义决定是否推送、是否高优处理。比如某篇文章提到了我关注的技术关键词模型可以预测它的重要性评分高分文章直接推送低分文章静默归档。这其实就是一个介于规则系统和智能Agent之间的中间态。另一个想改进的方向是采集器的分布化。现在的所有采集任务都在同一台机器上跑如果订阅源数量到几百个单机IP和带宽都可能成为瓶颈。可以考虑把采集器做成独立的worker进程部署到多台机器上通过一个公共消息队列比如Redis Stream连接。但这会引入额外的中间件个人项目需要谨慎评估复杂度收益比。5.2 实际使用中的几点心得如果你也想做一个类似的信息调度工具我有几个经验可以分享。第一不要追求“全自动”不要一开始就想着AI全自动分类、全自动推荐先把“规则人工抽查”的机制跑稳再逐步引入智能判断。第二配置要比逻辑更引人注意因为配置是一个项目里被改动的频率最高的部分配置写得好运营成本就能降到很低。第三插件接口要尽早稳定下来因为一旦接入的数据源变多再改接口会牵一发而动全身。我个人最大的收获倒不是这套代码本身而是“把信息流当成一条河流来管理”的思路。以前我是在各个信息孤岛之间反复跳转现在我只需要维护一条摆渡线把河那头的信息按时运到我这头来。这只狐狸与摆渡人的组合越用越觉得贴切。