企业官网建设流程全解析

从项目名“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( id=make_uuid(), source=self.source_name, type="article", payload=payload, 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 tasks

Rule是一个抽象类，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 weekly

main.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, encoding="utf-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:00+08:00"；二是所有内部时间一律用UTC，只有展示给用户时才转成本地时区。我选了第二种，因为内部处理统一UTC，能避免很多类似“周一凌晨跑出的周报落到了上周”的边界问题。

4.4 常见问题速查表

整理一下我跑这一个月遇到的高频问题，做成速查表供大家直接查：

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全自动分类、全自动推荐，先把“规则+人工抽查”的机制跑稳，再逐步引入智能判断。第二，配置要比逻辑更引人注意，因为配置是一个项目里被改动的频率最高的部分，配置写得好，运营成本就能降到很低。第三，插件接口要尽早稳定下来，因为一旦接入的数据源变多，再改接口会牵一发而动全身。

我个人最大的收获倒不是这套代码本身，而是“把信息流当成一条河流来管理”的思路。以前我是在各个信息孤岛之间反复跳转，现在我只需要维护一条摆渡线，把河那头的信息按时运到我这头来。这只狐狸与摆渡人的组合，越用越觉得贴切。