ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

InfoBridge:RSS聚合翻译自动化流水线,打通外文技术资讯到中文推送的最后一公里

InfoBridge:RSS聚合翻译自动化流水线,打通外文技术资讯到中文推送的最后一公里 如果你也是那种每天必须把十几个技术源刷一遍的人一定能理解我的痛苦订阅源越加越多、外文阅读速度跟不上消息增长、看完一遍转头就忘。我断断续续试过浏览器翻译、网页在线工具、甚至手动复制到翻译框里的笨办法到最后发现这些方式都解决不了本质问题获取信息的流程太碎每一篇都要重新打开网页、复制、翻译、整理。被逼到一定份上我干脆写了一个开源脚本把聚合、翻译、自动化三件事彻底交给程序去跑。这个项目我起名叫 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核心依赖是feedparserRSS解析、requestsHTTP调用、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、statusnew/translating/done/failed、translated_at、retry_count。这张表就是整个流水线的记忆。脚本跑挂了、被kill了、甚至换了台机器只要sqlite文件还在重启后就能从断点继续绝不重复处理已完成的内容。去重我做了两层。第一层是URL去重对链接做归一化处理去掉常见跟踪参数比如?utm_sourcefeed同一篇文章的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(indexidx) 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能自动识别主干内容去掉导航、页脚、侧边栏这些噪音对绝大多数技术博客效果都挺好。但要注意如果源站的结构比较特殊多列布局、大量嵌套divreadability偶尔会误删正文这时候就需要在清洗层加白名单或自定义提取规则。我一般建议先拿三五个主力源测试提取效果再决定默认策略。还有一个极容易被忽略的问题HTML实体。很多RSS摘要里会有lt;这种转义字符如果不清洗直接交给翻译模型模型会把实体当成普通文本翻译输出里全是乱码。所以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: fBearer {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(loaderFileSystemLoader(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}, timeout10) 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 21这里有两个坑必须提前避掉。第一个是crontab的环境变量问题cron执行时不会加载你shell里的~/.bashrc和~/.profile所以API密钥这种环境变量要么写在启动命令前的export里要么在main.py里用dotenv加载.env文件。我推荐后者把.env加入.gitignore防止密钥泄露。第二个坑是日志处理。自动化脚本出问题的时候日志就是唯一线索。我习惯按天切分日志文件保留最近30天超限自动清理。同时日志要输出结构化信息包括entry_hash、状态、耗时这样后续排查时能快速定位到单篇文章的完整生命周期。4.2 云端自动化GitHub Actions定时流水线本地机器不能保证7x24在线或者你不想在本地维护venv和cronGitHub 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/checkoutv4 - name: Setup Python uses: actions/setup-pythonv5 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_sourcefeed同一篇文章每次抓取都生成新URL绕过了URL去重第二源站在UTC零点刷新全文导致内容Hash变化重新触发翻译。应对方法对URL做归一化清洗去掉常见跟踪参数对内容Hash做相似度比较而不是完全相等比较当两段文本相似度超过95%视为同一篇文章直接沿用之前的翻译结果。清洗URL这一步很便宜但能解决大部分重复问题。5.5 定时任务执行时间漂移本地cron和GitHub Actions都可能出现执行时间与预期不符的情况。cron的漂移通常来自时区设置错误Actions的漂移通常来自排队延迟。解决方法是脚本启动时先打印当前时区和计划执行时间方便对照同时允许在配置里设置最迟执行时间超过了就放弃本轮等下一轮调度避免任务堆积。另外不建议把多个耗时很长的任务安排在同一个时间点触发。错开15到30分钟比挤在一起稳得多。提示自动化脚本的价值在于稳定地、无声地跑完日常流程而不是每时每刻都在跑。调度设计的原则很简单任务尽量短、失败尽量快、重试尽量少。写在最后InfoBridge从最初满足个人需求的工具慢慢演变成一个可以横向复用的框架我最大的感悟是把需求拆成采集、清洗、翻译、分发四层之后每一层的替换成本都变得极低。今天你用的是OpenAI兼容接口明天换成其他大模型API只需要改配置和少量请求代码今天推送渠道是群机器人明天想同步到自建知识库只需要新增一个分发模板。这种分层可替换的架构比一开始就把所有逻辑耦合在一个函数里的方案后期省下的力气完全不在一个量级。如果你也想搭一个类似的信息处理流水线我建议从最小可用版本开始先只聚合两三个源、只翻译标题和摘要跑通全流程之后再逐步增加源和功能。先把管道打通再谈优化。
返回列表