如果你也是那种每天必须把十几个技术源刷一遍的人,一定能理解我的痛苦:订阅源越加越多、外文阅读速度跟不上消息增长、看完一遍转头就忘。我断断续续试过浏览器翻译、网页在线工具、甚至手动复制到翻译框里的笨办法,到最后发现这些方式都解决不了本质问题:获取信息的流程太碎,每一篇都要重新打开网页、复制、翻译、整理。被逼到一定份上,我干脆写了一个开源脚本,把"聚合、翻译、自动化"三件事彻底交给程序去跑。这个项目我起名叫 InfoBridge,取"信息桥"的意思,核心能力一句话就能讲清:自动拉取RSS订阅源内容,提取正文,调用翻译接口生成中文版本,再按需推送到本地、群机器人或知识库,整个过程无人值守。
这篇文章是整套方案的完整复盘,从需求拆解、架构设计、核心代码,到部署运维、线上问题排查,全部记录在案。适合谁看?第一类,个人技术读者,想提高外文资讯的阅读效率;第二类,团队里负责信息同步、运维自动化的工程师,希望把海外技术动态变成一条稳定的数据流水线;第三类,做内容选题和资讯运营的朋友,需要一个可配置、可扩展的聚合翻译工具。
1. 项目定位与需求拆解
1.1 先搞清楚"聚合翻译自动化"到底要服务谁
动手写代码之前,我花了不少时间琢磨一个问题:这个脚本到底给谁用?同一个工具,服务个人和服务团队,设计方向完全不同。
个人场景下,使用者关注的是"我今天有没有漏掉重要的技术动态"。他们不需要每篇都全文翻译,标题加摘要就能快速判断是否要点开。团队场景下,信息同步负责人关心的是"外文动态如何以最快速度同步给团队",通常需要全文翻译、统一格式、自动进群或进知识库。而资讯运营、技术编辑这类角色,更看重"热点跟踪"和"选题挖掘",他们需要的是标题热度排序、关键词命中提醒,而不是简单的翻译搬运。
InfoBridge最初就是为个人场景设计的,但架构上刻意照顾了后两类需求。核心设计思路是:不替用户做阅读决策,而是把所有"阅读前的准备工作"全部做完——抓取、清洗、翻译、排版、推送,用户只需要在收到推送后决定"这篇文章值不值得读"。这样才能做到对三类使用者的需求通吃。
1.2 为什么把RSS当作聚合底座,而不是直接写爬虫
选了RSS做聚合源,不少朋友第一反应是:RSS都什么年头的东西了,现在谁还用它?但"技术资讯聚合"这个场景恰恰是RSS的主场,理由有三个。
第一,RSS是公开、标准化、无需登录的格式。几乎所有知名技术博客、官方工程团队博客、开源项目Release页面都提供RSS,等于别人已经把内容的结构化做好了,你直接消费元数据就行。第二,RSS返回的信息密度高——标题、链接、摘要、发布时间、标签全都有,程序处理起来非常方便,不需要自己去解析一堆页面结构。第三,RSS对抓取方友好,数据量小、更新频率规律,不会像爬虫那样动不动遇到反爬策略。
直接写爬虫的方案我也评估过,后来放弃了。爬虫方案最大的问题是维护成本不可控:目标站点一改版,选择器就失效;稍微大一点的网站还有各种反爬手段;页面里大量动态渲染内容还得上headless浏览器。对于个人项目来说,这些成本都是不可接受的。RSS不完美,但它用最小的成本覆盖了主流技术资讯源,足够了。
1.3 自动化做到什么程度才算"解放双手"
很多人理解的自动化是"定时跑一下脚本",我觉得这是半自动。真正的自动化应该是一条闭环流水线:自动拉取新增条目、自动判断是否值得翻译、自动执行翻译、自动排版输出、自动推送、自动记录状态。这六个环节只要有一个需要人工介入,就没法叫"解放双手"。
我最开始做的时候,只把"抓取"和"翻译"自动化了,去重和状态记录全没做。结果每次运行都会重复翻以前翻过的文章,API费用直接翻倍。后来补上了去重、缓存、失败重试三个模块,才算真正跑通。所以我的建议是:自动化设计的优先级应该是"去重 → 缓存 → 重试 → 翻译 → 推送",先把地基打稳,再盖楼。
技术选型上,我用的是Python 3.10+,核心依赖是feedparser(RSS解析)、requests(HTTP调用)、BeautifulSoup4和readability-lxml(正文提取)、sqlite3(缓存与状态管理)。选这套组合的原因非常务实:全是成熟稳定的开源库,任何一个环节遇到问题都能搜到现成答案,不用自己在底层协议上从头造轮子。
2. 整体架构与数据流设计
2.1 四层模块:采集、清洗、翻译、分发
整个脚本被拆成了四个互不耦合的模块,数据在模块之间单向流动,这条原则非常重要。
- 采集层:读取config.yaml里的RSS源列表,调feedparser解析,把最新条目输出为标准化字典。
- 清洗层:对标题和正文做HTML实体反转义、标签剥离、代码块占位符替换、无效字符过滤,输出干净的纯文本。
- 翻译层:调用翻译接口,负责分段处理、缓存查询、并发控制、失败重试,输出中文文本。
- 分发层:用Jinja2模板渲染结果,写入本地Markdown文件、发送Webhook,或生成Issue提交到仓库。
四层分层的核心价值在于每一层都能独立测试、独立替换。比如你不想用云翻译API,可以把翻译层换成调用本地模型;不想推送到群机器人,可以把分发层改成写数据库,其他两层完全不用动。这个设计带来的长期收益,远超初期多花的一点拆解时间。
2.2 数据流与状态管理:让脚本"记住"每篇文章
数据流主线如下:RSS源 → 条目列表 → 去重 → 标题/摘要/正文提取 → 翻译 → 渲染 → 输出/推送。看起来不复杂,但真正让流水线稳定运行的关键,是"状态管理"。
我在sqlite里维护了一张state表,字段包括:entry_hash(条目唯一标识)、source_name、link、title、status(new/translating/done/failed)、translated_at、retry_count。这张表就是整个流水线的记忆。脚本跑挂了、被kill了、甚至换了台机器,只要sqlite文件还在,重启后就能从断点继续,绝不重复处理已完成的内容。
去重我做了两层。第一层是URL去重:对链接做归一化处理,去掉常见跟踪参数(比如?utm_source=feed),同一篇文章的URL不会重复处理。第二层是内容Hash去重:有些源会在同一URL上更新正文,或者用不同URL指向同一篇文章,内容Hash能兜住这些情况。两层结合,基本能覆盖90%以上的重复场景。
2.3 为什么做"摘要翻译 + 全文翻译"双档模式
翻译成本是必须考虑的现实问题,尤其当订阅源数量达到十几个、每天新增几十篇文章时,全文翻译的token消耗和API费用是实实在在的账单。我在脚本里做了一个可配置的"翻译决策器",核心逻辑很简单:
- 默认模式:所有新条目先翻译标题和摘要,输出摘要版。
- 触发模式:标题命中你配置的关键词(比如Kubernetes、Rust、AI),或者摘要版被判断为"值得深读",才触发全文翻译。
- 全量模式:所有内容都全文翻译,但不建议长期开启。
这个决策器写起来不复杂,就是在配置里加一个关键词列表和一个阈值开关。但它带来的成本节省非常可观。我用15个订阅源测试过,默认模式下API消耗大概只有全量模式的四成不到,而且阅读效率没有明显下降。很多人一上来就把所有文章全文翻译,最后的结果往往是"翻译了一堆不值得看的",反而被信息淹没。
3. 核心模块的代码实现与关键参数选择
3.1 配置驱动:一切从config.yaml开始
InfoBridge的所有行为都通过YAML配置驱动,不硬编码任何URL、密钥、路径。这样做的好处很直接:改订阅源不需要碰代码,换翻译接口不用改逻辑,换个部署环境复制配置文件就能跑。
sources: - name: "Python 官方博客" url: "https://blog.python.org/feeds/posts/default" - name: "GitHub Blog" url: "https://github.blog/feed/" - name: "Cloudflare Blog" url: "https://blog.cloudflare.com/rss/" translation: provider: "openai_compatible" api_key_env: "TRANSLATE_API_KEY" base_url_env: "TRANSLATE_BASE_URL" target_lang: "zh-CN" model: "gpt-4o-mini" max_tokens: 3000 temperature: 0.3 concurrency: 3 retry_count: 3 storage: output_dir: "output" format: "markdown" push: enabled: true webhook_url_env: "PUSH_WEBHOOK_URL"这里有个参数要特别强调:temperature。翻译任务需要的是严谨、忠实原文,不是创意发挥,所以temperature必须调低,0.2到0.4之间最合适。默认值1.0会带来不必要的随机性,同一个句子两次翻译可能措辞完全不同,这在技术文档里是非常忌讳的。我一开始没注意这个参数,结果同一段话在不同时间翻译风格漂移严重,后来全部调成0.3才稳定下来。
3.2 RSS解析:别只取title,把摘要、发布时间和标签都用起来
feedparser的用法很基础,但很多人只取title和link就完事了,白瞎了这个库的能力。实际项目中,summary、published、tags三个字段都有大用处:摘要是翻译决策器的核心输入,帮助你判断一篇文章是否有全文翻译价值;发布时间用于排序和时效过滤,太久远的文章可以直接跳过;标签可以用于自动分组和关键词命中检查。
import feedparser def fetch_entries(source: dict) -> list: feed = feedparser.parse(source["url"]) entries = [] for item in feed.entries: entry = { "title": item.get("title", ""), "link": item.get("link", ""), "summary": clean_html(item.get("summary", "")), "published_ts": parse_published(item.get("published_parsed")), "tags": [t.get("term") for t in item.get("tags", [])], "source": source["name"], } entries.append(entry) return entriesparse_published这里有个容易踩的坑:published_parsed返回的是结构化时间,但并非所有源都提供。有些源字段名叫published,有些叫updated,格式还各不相同。稳妥的做法是:优先取published_parsed,取不到就退到updated_parsed,再取不到就用当前时间,同时输出warning日志。这样既保证程序不崩,又能暴露数据源的质量问题。
3.3 正文提取:从HTML到干净文本,还要护住代码块
翻译技术文章最忌讳一件事:代码被翻译。代码里的变量名、函数名可以解释,但绝不能改写,更不能把代码块里的内容混入正文翻译。所以我在清洗层设计了一个"代码保护"机制:提取正文时,先把pre和code标签里的内容用占位符替换掉,等翻译完成后再还原。这样无论翻译模型怎么发挥,代码都能原样保留。
from bs4 import BeautifulSoup import re CODE_PLACEHOLDER = "CODEBLOCK_{index}_" def extract_clean_text(html: str) -> str: soup = BeautifulSoup(html, "html.parser") for idx, code in enumerate(soup.find_all(["pre", "code"])): placeholder = CODE_PLACEHOLDER.format(index=idx) code.replace_with(placeholder) # 记录占位符映射 body = soup.get_text("\n") body = html.unescape(body) body = re.sub(r"\n{3,}", "\n\n", body) return body.strip()正文提取本身,我推荐readability-lxml而不是直接用BeautifulSoup全文解析。readability能自动识别主干内容,去掉导航、页脚、侧边栏这些噪音,对绝大多数技术博客效果都挺好。但要注意,如果源站的结构比较特殊(多列布局、大量嵌套div),readability偶尔会误删正文,这时候就需要在清洗层加白名单或自定义提取规则。我一般建议先拿三五个主力源测试提取效果,再决定默认策略。
还有一个极容易被忽略的问题:HTML实体。很多RSS摘要里会有<这种转义字符,如果不清洗直接交给翻译模型,模型会把实体当成普通文本翻译,输出里全是乱码。所以html.unescape这一步必须放在翻译之前,这算是我踩过的比较蠢的坑,但确实影响很大。
3.4 翻译模块:缓存、分段与并发控制
翻译模块是整个脚本的心脏,也是最容易出问题的地方。我把它拆成三个子能力:缓存管理、分段策略、并发调度。
缓存管理用sqlite实现,缓存key由文本内容的MD5计算。同一段文本如果之前翻译成功过,直接读缓存结果,不再调用API。这个设计在重复抓取或源站更新时非常有用,直接从根源上避免了重复付费。
import hashlib import sqlite3 def get_cache_key(text: str) -> str: return hashlib.md5(text.encode("utf-8")).hexdigest() def check_cache(conn, key: str): row = conn.execute( "SELECT translated FROM cache WHERE cache_key = ?", (key,) ).fetchone() return row[0] if row else None def save_cache(conn, key: str, translated: str): conn.execute( "INSERT OR REPLACE INTO cache(cache_key, translated, created_at) VALUES(?,?,datetime('now'))", (key, translated), ) conn.commit()分段策略要处理两个边界:一个是文本超过模型单次最大输入;另一个是分段后译文拼接起来不通顺。我的经验是优先按段落分,而不是按字符数硬切。段落是自然语义单元,模型翻译时上下文完整,拼接时也不会把句子拦腰截断。如果单个段落仍然太长,再按句子边界做二级切割——句号、分号、换行都可以作为切割点,技术文档尤其适合这样处理。
并发控制上,我用线程池把并发数默认设为3。为什么是3不是10?因为大多数翻译API的免费额度或低配套餐并发限制都在个位数,并发过高立刻就会触发429限流。另外并发太高还会互相拖累,单条请求的超时时间会变长,整体吞吐反而不升反降。如果你的API套餐明确支持更高并发,再去调大这个值。
import os import requests from concurrent.futures import ThreadPoolExecutor, as_completed def translate_text(text: str, cfg: dict) -> str: cache = check_cache(conn, get_cache_key(text)) if cache: return cache api_key = os.environ.get(cfg["api_key_env"]) base_url = os.environ.get(cfg["base_url_env"]) resp = requests.post( f"{base_url}/chat/completions", headers={"Authorization": f"Bearer {api_key}"}, json={ "model": cfg["model"], "messages": [ {"role": "system", "content": "你是技术文档翻译助手,要求准确、通顺、保留术语原样。"}, {"role": "user", "content": f"将以下内容翻译为{cfg['target_lang']},不要翻译代码:\n\n{text}"}, ], "temperature": cfg.get("temperature", 0.3), "max_tokens": cfg.get("max_tokens", 3000), }, timeout=(10, 60), ) resp.raise_for_status() result = resp.json()["choices"][0]["message"]["content"] save_cache(conn, get_cache_key(text), result) return result这里要特别提醒requests的timeout设置。我踩过一次很严重的坑:没设置timeout,线程池里所有请求卡在等待状态,整个流水线完全挂掉,日志里没有任何错误信息,只有一堆永久等待的线程。改成timeout=(10, 60)之后,再配合重试逻辑,稳定性立刻提升了一个台阶。网络请求不设超时,等于给自己埋定时炸弹。
3.5 输出与推送:本地Markdown、Webhook、GitHub Issue三选一
翻译结果做完后,总得有个消费出口。InfoBridge内置了三种分发方式,各自对应不同场景:
- 本地Markdown:按日期生成文件,标题带原文链接,方便在Obsidian、语雀这类工具里做二次阅读和归档。
- Webhook推送:对接群机器人、Slack、飞书等渠道,只要填一个Webhook URL,就把翻译好的内容POST过去。
- GitHub Issue:把翻译结果以Issue形式提交到指定仓库,相当于一个带完整内容的"阅读清单",方便追踪和讨论。
Webhook那部分我建议用Jinja2模板渲染消息内容,而不是在代码里硬编码拼接字符串。原因是不同渠道的消息格式差异很大:有的要求Markdown,有的要纯文本,有的要特定JSON结构。用模板隔离这些差异之后,换渠道只需要改一个模板文件,不用动一行代码。
from jinja2 import Environment, FileSystemLoader def render_and_push(entry, translated_content, cfg): env = Environment(loader=FileSystemLoader("templates")) template = env.get_template("default.md.j2") payload = { "title": entry["title"], "link": entry["link"], "source": entry["source"], "content": translated_content, } message = template.render(**payload) webhook_url = os.environ.get(cfg["push"]["webhook_url_env"]) resp = requests.post(webhook_url, json={"text": message}, timeout=10) resp.raise_for_status()推送之前记得加一道结果检查:如果译文里还残留CODEBLOCK_ 占位符,说明代码还原环节出错,这时不能推送,应该直接标记为失败并告警。这个检查逻辑只要十几行代码,但能在真实场景里拦住大量低级事故。我上线初期就是因为没做这个检查,某一次翻译模型擅自改写了占位符,导致整篇推送里的代码位置全都错乱了。
4. 自动化调度与部署落地的实操记录
4.1 本地部署:crontab最小方案
如果你只是自己用,本地部署是最快的路径,三步搞定:
- 克隆项目,创建虚拟环境,安装依赖。
- 复制config.example.yaml为config.yaml,填入订阅源,设置好环境变量。
- 在crontab里追加一行定时执行。
0 8 * * * cd /opt/infobridge && /opt/infobridge/.venv/bin/python main.py >> logs/cron.log 2>&1这里有两个坑必须提前避掉。第一个是crontab的环境变量问题:cron执行时不会加载你shell里的~/.bashrc和~/.profile,所以API密钥这种环境变量要么写在启动命令前的export里,要么在main.py里用dotenv加载.env文件。我推荐后者,把.env加入.gitignore,防止密钥泄露。
第二个坑是日志处理。自动化脚本出问题的时候,日志就是唯一线索。我习惯按天切分日志文件,保留最近30天,超限自动清理。同时日志要输出结构化信息,包括entry_hash、状态、耗时,这样后续排查时能快速定位到单篇文章的完整生命周期。
4.2 云端自动化:GitHub Actions定时流水线
本地机器不能保证7x24在线,或者你不想在本地维护venv和cron,GitHub Actions是更省心的选择。schedule触发器做定时任务,workflow_dispatch支持手动触发,两件事一个文件搞定。
name: daily-translation-sync on: schedule: - cron: "0 2 * * *" workflow_dispatch: jobs: sync: runs-on: ubuntu-latest steps: - name: Checkout uses: actions/checkout@v4 - name: Setup Python uses: actions/setup-python@v5 with: python-version: "3.12" cache: "pip" - name: Install dependencies run: pip install -r requirements.txt - name: Run sync env: TRANSLATE_API_KEY: ${{ secrets.TRANSLATE_API_KEY }} PUSH_WEBHOOK_URL: ${{ secrets.PUSH_WEBHOOK_URL }} run: python main.py用GitHub Actions有几个细节要注意。第一,API密钥必须放secrets,绝对不要直接写在workflow文件里。第二,schedule的cron用的是UTC时间,想要北京时间早上8点执行,就得换算成UTC凌晨0点,也就是cron: "0 0 * * *"。第三,Actions的任务有执行时间上限,如果订阅源很多、翻译内容很大,要注意是否超时,必要时在脚本内部做分批执行。
我实际部署时发现,依赖安装每次都重新跑一遍,耗时将近一分钟。后来给setup-python加上了pip缓存,安装时间从50秒降到8秒左右。对每天跑一次的任务来说,这个优化带来的体感提升很明显。
4.3 状态监控:失败重试与断点续跑
自动化最怕"静默失败"——脚本跑完了,但什么都没输出。可能是真的没有新内容,也可能是中间某个环节静默崩溃了。为了区分这两种情况,我给每次运行加了一个健康检查:执行完主流程后,写一个status.json,记录运行时间、新增条目数、翻译成功数、失败数、总耗时。再用一个看板脚本定期检查这些状态文件,如果连续两天没有新增条目,或者失败率超过阈值,就触发告警。
断点续跑的实现依赖之前说的state表。每处理完一个条目,就把状态更新为done。脚本重启时,只扫描状态不是done的条目。这样即使运行到一半被kill掉,也不会重复翻译已经完成的内容,不会多花一分冤枉钱。
4.4 日常运维:一周检查一次足够了
这个脚本上线之后,我基本上保持一周检查一次的频率。日常维护关注三件事:看看有没有源失效(RSS返回404或者长时间不更新)、看一眼API消耗是否异常、抽查几条翻译质量。如果发现译文有明显的翻译腔或者术语错误,就调整system prompt里的风格描述。
真要说起来,搭建脚本花了不少时间,但长期维护成本非常低。这也是分层架构的好处:只要底层稳定,上层几乎不用动。
5. 常见问题与排查技巧实录
5.1 RSS解析失败或返回空内容
feedparser的兼容性总体不错,但个别源的XML不规范会导致结果为空。排查思路是先看原始XML是否正常,再用feedparser单独测试,重点检查feed.bozo字段。bozo为1说明解析器发现了异常但仍然尝试解析,这时候就要判断是源的格式问题还是字段提取逻辑问题。
如果你的源列表里有长期无法正常解析的源,我的建议是直接从配置里摘掉,而不是花时间修复它。RSS源的选择标准是"稳定优先",一个三天两头抽风的源会拖累整条流水线。
5.2 翻译接口返回429或超时
429限流是最常见的问题,几乎每个用过翻译API的人都遇到过。出现429时,正确的处理方式不是立刻重试,而是等待一段时间再试。我实现了一个简单的指数退避:第一次重试等5秒,第二次10秒,第三次20秒,最多重试3次。如果三次都失败,就把条目标记为pending,等下一轮调度再处理。这样比连续猛烈重试打到接口封禁好得多。
另外,很多API的免费额度对每分钟请求数有严格限制。我的经验是并发数从1开始测,逐级加到3、5,确认不触发限流再保持那个档位。宁可慢一点,也不要被限流打回原形。
5.3 译文里出现未还原的代码占位符
这个问题在3.3提过,这里说下排查方法。先检查清洗层是否对代码块做了占位符替换,再检查翻译层返回的译文里占位符是否还在。如果占位符还在,说明翻译模型把占位符当成普通文本处理甚至改写了,这时候需要在system内容里明确指示:CODEBLOCK_开头的占位符不可翻译。
我还在代码里加了还原失败检测:翻译完成后,对译文做正则扫描,如果发现占位符数量与原文不一致,就判定还原失败,跳过推送并告警。这个保护机制虽然简单,但是能有效阻止问题内容扩散到下游消费端。
5.4 重复翻译导致API账单超标
账单超标通常不是审核数字,而是去重逻辑出了问题。最常见的两个原因:第一,RSS源返回的链接带跟踪参数(如?utm_source=feed),同一篇文章每次抓取都生成新URL,绕过了URL去重;第二,源站在UTC零点刷新全文,导致内容Hash变化,重新触发翻译。
应对方法:对URL做归一化清洗,去掉常见跟踪参数;对内容Hash做相似度比较而不是完全相等比较,当两段文本相似度超过95%,视为同一篇文章,直接沿用之前的翻译结果。清洗URL这一步很便宜,但能解决大部分重复问题。
5.5 定时任务执行时间漂移
本地cron和GitHub Actions都可能出现执行时间与预期不符的情况。cron的漂移通常来自时区设置错误,Actions的漂移通常来自排队延迟。解决方法是:脚本启动时先打印当前时区和计划执行时间,方便对照;同时允许在配置里设置"最迟执行时间",超过了就放弃本轮,等下一轮调度,避免任务堆积。
另外,不建议把多个耗时很长的任务安排在同一个时间点触发。错开15到30分钟,比挤在一起稳得多。
提示:自动化脚本的价值在于稳定地、无声地跑完日常流程,而不是每时每刻都在跑。调度设计的原则很简单:任务尽量短、失败尽量快、重试尽量少。
写在最后
InfoBridge从最初满足个人需求的工具,慢慢演变成一个可以横向复用的框架,我最大的感悟是:把需求拆成采集、清洗、翻译、分发四层之后,每一层的替换成本都变得极低。今天你用的是OpenAI兼容接口,明天换成其他大模型API,只需要改配置和少量请求代码;今天推送渠道是群机器人,明天想同步到自建知识库,只需要新增一个分发模板。这种分层可替换的架构,比一开始就把所有逻辑耦合在一个函数里的方案,后期省下的力气完全不在一个量级。
如果你也想搭一个类似的信息处理流水线,我建议从最小可用版本开始:先只聚合两三个源、只翻译标题和摘要,跑通全流程之后,再逐步增加源和功能。先把管道打通,再谈优化。