先说个场景。我做内容运营和技术情报这块儿有些年头了每天打开浏览器的第一件事就是去各个平台看一眼热榜、关键词搜索结果、同行有没有发新东西。一开始只有两三个信息源手动刷还撑得住后来源一多问题立刻来了——页面加载慢、信息重复、漏检率高半小时折腾下来手里全是零散截图根本凑不出一份干净的情报。PLFM_RADAR 就是在这种状态下被逼出来的。PLFM_RADAR 全称是 Platform Radar翻译过来就是“平台雷达”。它不是某个商业软件而是我自己写的一套小工具对多个平台的公开信息做定时扫描、统一聚合、去重、规则过滤再按需把重要变化推送到钉钉、飞书或者邮箱。运行了大半年它帮我省下的时间不是一星半点更重要的是那些本来会被我刷漏的信息现在基本都能在一个看板上被捞回来。这篇文章就把这个项目的设计思路、核心代码、踩坑经历都摊开说说适合正在做信息监控、竞品观察、热点追踪的朋友参考。1. 为什么我动手做了个“平台雷达”1.1 信息过载带来的真实痛点先还原一下没有雷达之前的工作流。每天我需要盯的信息源包括几个内容社区的热榜、行业资讯站的 RSS、竞品官方的动态页、以及若干关键词在站内搜索的返回结果。每个源有自己的更新节奏有的五分钟就刷屏有的一天才更新两三次有的是结构化接口有的只能靠肉眼翻页面。手动巡检最大的问题不是“看得慢”而是“无法持续”。人总有走神的时候午休回来、开会两小时可能正好错过了某个重要的信息窗口。等再打开页面那条内容已经被刷下去了。信息监控这件事靠人工盯是注定有漏的需要有一个替身在固定的时间点替你去“看一眼”并且把不符合要求的大部分内容过滤掉只把值得人工介入的那部分挑出来。另外就是重复。同一篇内容经常在热榜、搜索、分类页里出现三次标题不一样、链接不一样人工很难判断“这个是不是我已经看过的”。于是好不容易攒下来的阅读时间又被浪费在重复确认上。1.2 PLFM_RADAR 的定位与使用场景PLFM_RADAR 不是爬虫框架也不是舆情系统它的定位是一个“聚合 规则判断”的小型中间层。它每天定时去各个平台拿到原始数据转成统一格式判断哪些是新的哪些是重复的然后跑一遍用户自定义的规则决定要不要通知你。我实际用得最多的场景有三个热点追踪。比如我长期关注“开源大模型”“本地部署”“智能体”这几个方向凡是命中关键词的新内容都会被收进实时列表如果热度冲进前几还会单独推送提醒。竞品动态监控。竞品的官方公告、版本发布、社区新帖只要来源支持 RSS 或公开页面就能纳入监控。竞品一有新动作我甚至会比公司群里的转发快几分钟。数据源异常发现。这是个附带收益。每个源都有挂掉的时候接口改版、抓取失败雷达会在连续失败后主动告警我不用等到第二天才发现某个源已经失效三天。适合用这套思路的人不仅仅是对技术感兴趣的程序员。内容创作者、自媒体运营、产品经理、做市场调研的人只要每天需要从多个平台收集公开信息都可以借鉴这里的思路搭一套轻量版本。1.3 和现成工具对比为什么还要自己写市面上的选择其实不少。采购一套商业舆情系统功能是强大但价格贵、部署重而且很多时候数据源和规则是锁死的你没法按自己的业务习惯去定制。RSS 阅读器类工具能解决“订阅聚合”的问题但只解决了“看哪里”没有解决“看什么值得看”它不会帮你过滤、去重、按规则告警。至于开源爬虫框架专注解决的是采集效率抓回来的数据放哪儿、怎么变成决策还需要自己搭一套。我选择自己写核心原因就两条第一需求足够个人化规则、通知渠道、数据源想改就改不存在等待厂商提需求的周期第二数据是自持的所有内容落在自己的库里后续想怎么分析都有地基。代价也很明确就是需要自己维护但作为长期使用的基础设施这个成本是值得的。2. 整体链路一个扫描任务从数据源到告警要走几步2.1 四层结构采集、处理、存储、通知在设计 PLFM_RADAR 之前我先把一条扫描任务的完整路径画了出来。从触发到最终有产出一共经历四层采集。调度器到了时间点去数据源拉取原始数据。这里可能是调一个公开 API可能是解析一条 RSS也可能是一个公开页面的内容。处理。把不同平台乱七八糟的字段统一成一套标准结构去掉无用的噪声归档或丢弃重复内容。存储。把清洗后的数据写入本地库作为后续分析的底座同时记录去重标记。通知。新数据跑一遍规则如果符合告警条件就构建一条消息推到聊天机器人或邮箱。用一句话概括定时触发 - 采集原始内容 - 标准化 - 去重入库 - 规则匹配 - 告警通知。这个链路看起来简单但它把一个原本只能靠人工执行的重复过程变成了可以无人值守的循环。我在设计的时候一直提醒自己每一层都要只做一件事越短越容易排查。比如“采集层”只管拿数据不管清洗。如果你把清洗逻辑和采集逻辑写在一个函数里第一周会觉得爽第二周开始改数据源的时候就会后悔。2.2 技术选型的取舍优先考虑可维护性PLFM_RADAR 这个项目我选用的技术栈相当朴素Python 3.10Requests 做 HTTP 请求feedparser 解析 RSSAPScheduler 做定时调度SQLite 起步做存储后期换 PostgreSQL 也不会伤筋动骨通知端直接发 Webhook 和 SMTP 邮件。Web 端用一个 FastAPI 提供的简易看板。为什么是 Python而不是 Go 或者 Node原因很简单这个项目的大部分工作是字符串清洗、RSS 解析、规则匹配Python 的库生态可以帮我省掉大量重复代码而且调试效率高。Go 的并发能力强但这里根本没有高并发需求不值得额外增加心智负担。调度器我选了 APScheduler 而非系统的 cron。原因是我希望任务的调度信息也能随项目走在代码里可配置、可修改而不是散落在系统 crontab 里。APScheduler 在单体小项目里非常合适支持 Cron 表达式和 interval 间隔两种触发方式最基本的用法足够覆盖需求。这里也要吐槽一个点。有些人一开始就上 Celery、上 Redis、上消息队列对于维护一个小型信息监控工具来说这是明显的过度设计。机器上没有消息中间件服务之间不需要异步解耦一个进程里的调度器完全可以搞定。先跑起来等到并发源超过几十个、任务之间出现明显依赖的时候再去引入这些东西那时你已经知道问题到底在哪里了。2.3 核心数据结构统一内容项模型所有数据源的内容进入系统后都会被转成一个统一的 RadarItem。之所以要统一是因为不统一的代价会在规则引擎和前端展示阶段集中爆发。如果每个源的数据结构都不一样那么每加一个源规则引擎和看板都得跟着改一遍这是最典型的失控路径。我用一个 dataclass 来定义这个核心模型from dataclasses import dataclass, field dataclass class RadarItem: source_id: str # 数据源唯一标识例如 community_a item_id: str # 原始条目ID平台自带的唯一标识 title: str # 标题 url: str # 原文链接 content: str # 摘要或正文截断 published_at: str # 发布时间统一存 UTC ISO8601 字符串 fetched_at: str # 抓取时间 extra: dict field(default_factorydict) # 各平台特有字段比如热度分、排名对应的存储层核心表至少需要两张内容表和去重表。内容表用来存数据去重表用source_id item_id作为联合主键保证同一条原始记录只入库一次。CREATE TABLE seen_items ( source_id TEXT NOT NULL, item_id TEXT NOT NULL, first_seen_at TIMESTAMP NOT NULL, PRIMARY KEY (source_id, item_id) );这个模型是整个项目的地基后面所有规则、所有展示都建立在这套字段之上。有一段时间我想给某个源单独加字段直接在 RadarItem 里堆属性结果模型越来越臃肿。后来改成把平台特有数据一律丢进 extra 字典规则引擎按需读取模型又恢复了清爽。这个教训很值得记住核心模型尽量保持稳定变化的细节放到扩展字段里。3. 数据源接入官方API、RSS、公开页面怎么排优先级3.1 官方API是首选但要注意频率配额接入任何数据源之前先做优先级判断有官方 API 的优先用官方 API没有的看有没有公开 RSS都没有才考虑解析公开页面。这个顺序不是技术洁癖而是维护成本的排列。官方 API 的好处很明显字段结构化、有明确的文档、更新及时。但它也有自己的问题最典型的是频率配额和增量游标。我曾经配置完一个官方 API 后直接按分钟级频率去轮询结果半天后收到限流通知请求被拒。后来老老实实把频率降到平台允许的范围并开启增量更新。增量更新的思路是平台返回一个since_id或cursor参数我每次只拉“上次游标之后”的新内容。这样每次请求的数据量最小、对服务端的压力也最小限流概率自然降低。游标的保存位置很简单数据源配置表里加一个last_cursor字段就行。接入 API 时还有个容易踩的坑不要一上来就把所有字段全量拉回来。很多新内容的正文很长存储和清洗成本都很高但你的规则可能只需要标题和链接。我建议只取最小必要字段正文等内容真正命中规则之后再按需获取或者保留摘要即可。3.2 RSS解析库的使用与常见字段映射没有官方 API 的源RSS 往往是最好的替代方案。它在数据量上虽然不如 API 丰富但胜在稳定和轻量。feedparser 是解析 RSS 的标配库用起来很简单import feedparser feeds feedparser.parse(https://example.com/rss.xml) for entry in feeds.entries: title (entry.get(title) or ).strip() link (entry.get(link) or ).strip() published entry.get(published) or entry.get(updated) summary (entry.get(summary) or )[:200]这里有个非常关键的字段映射问题。不同平台的 RSS 里发布时间字段的命名差异很大有的叫published有的叫pubDate有的只有updated有的干脆一个时间字段都没有。我在设计清洗层时会做一个兜底链published优先其次updated都没有就把发布时间标记为“未知”而不是随口用当前抓取时间塞进去。为什么不愿意默认用抓取时间因为我后面有条规则是“只看最近24小时的新内容”如果把历史老内容的发布时间都错标成抓取时间它们就会被错误地当成新内容推送导致告警失真。宁可缺一个时间信息也不能伪造一个错误的时间。3.3 公开页面解析的底线原则与降级方案当某个源既没有 API 也没有 RSS只能解析公开页面时我的原则是先找页面里内嵌的 JSON 数据再考虑解析 HTML 结构。现在的很多前端页面首屏数据都是以window.__INITIAL_STATE__或类似形式直接塞在 HTML 里的。用正则或字符串截取把这段 JSON 抠出来比解析一堆 CSS 类名要稳得多。页面上展示的数字、标题、链接往往就是这份 JSON 的子集。你直接拿结构化数据没必要去猜 HTML 结构。如果实在只能解析静态 HTML那就必须接受一个现实页面改版时你的解析器大概率会失效。为了减少损失我会把解析器注册成带版本号的模板一旦连续解析失败立刻触发“数据源异常”告警而不是让错误悄无声息地堆积。降级方案也很重要这个源暂时没抓到不代表下次也要放弃跳过本次任务、记录日志、下一轮继续尝试。无论哪种公开页面解析都要守住底线遵守 robots.txt 的约束、控制访问频率、不要试图绕过访问限制。做信息雷达的目的是让自己省时间不是给自己惹麻烦。4. 调度与去重雷达转动时的“不重不漏”问题4.1 多源任务的调度配置与错峰设计调度器是整个雷达的“心跳”。我用 APScheduler 给每个数据源单独注册任务因为不同源的更新节奏完全不同。热榜类源每五分钟扫一次资讯类 RSS 每十五分钟扫一次官方公告类源一小时扫一次就够。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger scheduler BlockingScheduler() scheduler.add_job( scrape_source_a, CronTrigger(minute*/5), idsource_a, max_instances1, ) scheduler.add_job( scrape_source_b, CronTrigger(minute*/15), idsource_b, max_instances1, )这里有两个容易被忽略的细节。第一个是max_instances1它的作用是防止上一次任务还没跑完下一次触发又来了导致同一源并发抓取。加了之后如果任务超时或阻塞后续触发会自动跳过保证同一时间同一源只有一个任务在跑。第二个是错峰设计。同一个源如果需要抓多个关键词不要在同一秒全部发出请求。我在代码里对每次请求加了一个随机延迟import random import time # 同一数据源每次请求之间增加 0.8~2.5 秒随机间隔避免触发限流 time.sleep(random.uniform(0.8, 2.5))这个随机抖动不是玄学。信息平台的反爬机制里常见手段是统计“单位时间内的请求数”请求间隔过于规律反而容易被识别为自动化程序。加上随机延迟后请求分布更接近自然人行为限流风险也会小很多。4.2 指纹去重与重复内容合并去重是整个雷达最核心的工程问题之一分两层处理。第一层是“原始唯一键去重”用前面提到的(source_id, item_id)联合主键。每个平台在生成内容时基本都有自己的一套 ID 体系直接用平台的 ID 是最可靠的判重方式。但现实是有些源并不能提供稳定的内容 ID比如纯 HTML 页面解析出来的数据。这时我会退而求其次计算一条内容指纹import hashlib def make_fingerprint(source_id: str, title: str, url: str) - str: raw f{source_id}|{title.strip()}|{url.strip()}.lower() return hashlib.md5(raw.encode(utf-8)).hexdigest()这层指纹保证的是“完全相同的内容不会再次入库”。第二层是“相似内容合并”专门处理“同一篇内容换个标题、换个入口又出现”的情况。最典型的场景是某个内容既出现在平台热榜又出现在关键词搜索结果里而且标题有细微差异比如加了“热”字、或者“首发”两个字。标题一变上面的指纹就失效了。我的处理方法是不对标题做硬性相等而是先做归一化再比较相似度。归一化指的是去除多余空格、统一全半角、大小写折叠。进阶一点对归一化后的标题用SequenceMatcher或 simhash 做相似度判断相似度超过阈值的就认为是同一篇内容直接丢进去重表。这个逻辑不能太激进否则会把真正的不同内容误合并所以我只在“标题长度超过15个字且相似度超过0.9”时才启用。4.3 增量更新与失败重试增量更新是为了避免每次抓全量数据。很多内容平台的接口或页面默认会返回最近几十条内容如果每次都把这几十条全量清洗入库数据量很快会失控。正确姿势是记录一个游标或时间点下一次只处理比游标更新的内容。在代码里这个游标可以直接存在数据源配置表里每次成功抓取后更新。失败重试我采用了指数退避策略某源第一次失败后等1分钟再试再失败等5分钟、15分钟连续失败超过5次就把该源标记为“暂停”并触发一条告警。直接看代码更清晰retry_delay_minutes [1, 5, 15, 30, 60] def handle_failure(source_id: str, fail_count: int): if fail_count len(retry_delay_minutes): delay retry_delay_minutes[fail_count - 1] # 安排 delay 分钟后重新执行 scheduler.add_job( scrape_source, date, args[source_id], run_datedatetime.now() timedelta(minutesdelay), ) else: # 连续失败过多暂停该源并告警 notify_webhook(f数据源 {source_id} 连续失败已自动暂停)重试策略的核心是“别在同一时间点上反复撞墙”。如果没有退避机制一个源出了故障你会看到日志里全是同一时刻的重复错误而对定位问题毫无帮助。5. 告警规则从“我看到”到“它通知我”5.1 规则引擎的三层设计雷达真正的价值体现在规则引擎上。我把规则分成三层从简单到复杂依次是关键词规则、热度阈值规则、趋势突变规则。关键词规则是基础。每一条规则由三个部分组成必须包含的词AND关系、任意包含的词OR关系、排除词NOT关系。比如我想监控国产模型动态会设置必须包含“大模型”任意包含“开源”“发布”“上线”排除“广告”和“招聘”。执行顺序是先排除再必含再任含。热度阈值规则解决的是“关键词命中很多但我想只看重要的”。有些平台会给每条内容一个热度分或排名雷达可以拿到。当某条内容进入平台热榜前20或者热度分超过一个阈值即使它没有命中关键词也应该被推送出来因为它代表“大众正在关注的东西”。趋势突变规则是我后期才加上去的也是最出效果的一层。它的逻辑很简单同一个关键词在最近30分钟内的出现频率相比过去7天的同期均值如果突然增长了300%就说明这个话题正在发酵。这套逻辑最早用得并不顺利原因后面会讲但一旦过了基线期它的效率非常高能抓到很多关键词规则完全覆盖不到的新话题。5.2 通知渠道与告警模板通知渠道我接了三种企业微信群机器人、钉钉群机器人、邮件 SMTP。所有渠道都走同一套接口抽象在配置里切换即可。不要小看这个抽象因为团队的通知偏好经常变比如今天用钉钉明天可能就切到飞书改了适配器不影响核心逻辑。一条合格的告警消息至少应该包含五个字段触发时间、内容来源、标题、原文链接、命中原因。命中原因一定要写清楚否则你收到推送后还得去猜“为什么这条消息会推给我”。我之前有一条规则配得太宽收到的告警自己也看不懂最后查了日志才发现是关键词里的某个字太常见了。curl -X POST https://example.com/webhook \ -H Content-Type: application/json \ -d { msgtype: markdown, markdown: { content: 【PLFM_RADAR 告警】\n- 来源某社区热榜\n- 标题开源大模型最新进展\n- 发布时间2025-01-01 12:00:00\n- 命中规则关键词「开源」 热度排名前10\n- 链接https://example.com/article/123 } }邮件渠道你同样用 SMTP 发模板就一行纯文本加链接别整花活。群消息的目的就是快速扫一眼、判断要不要点开模板越简洁越好。5.3 防告警轰炸去重与抑制窗口告警功能上线后的第一个事故就是告警轰炸。有一天某个平台的数据异常一次抓取把几十条历史内容当新内容导入了结果规则引擎瞬间匹配出二十多条告警手机被通知刷了一屏。问题的根源不在规则而在告警层缺少保护。我加了两个机制。第一个是告警去重表以“规则ID 内容指纹”做联合主键同一内容命中同一规则最多只推送一次。就算它被重复导入、反复匹配也只发一次通知。第二个是抑制窗口。同一个“规则ID”在5分钟时间窗口内只发摘要不再逐条推送。这样即使出现批量命中收件人最多收到一条形如“该规则在5分钟内新增12条匹配点击查看”的聚合消息。如果真有必要逐条看去数据库或看板里翻就行。我还加了一个保险丝单条规则单日最多告警50条超过之后当天自动静默次日零点重置。这是为了防止规则本身配错、没人及时发现的情况下重复轰炸一整天。6. 上线运行后的踩坑实录6.1 时间字段格式混乱新内容被排到了历史里上线第一周我就发现一个诡异现象某条标题里写着“今日发布”的内容在列表里一直排在很靠后的位置像是一个几年前的旧闻。查了原始数据才发现这个源返回的时间字符串格式是GMT标准里的老式写法和我库里的 ISO8601 格式默认排序不兼容清洗层没有正确转换导致它被解析失败后回退成了“未知时间”再被我兜底逻辑当成很久以前的数据。这个坑让我意识到时间解析是整个清洗层里最不能含糊的部分。真实世界里各平台的时间格式五花八门Wed, 01 Jan 2025 12:00:00 GMT、2025-01-01T12:00:0008:00、2025-01-01 12:00:00还有纯时间戳。现在我的清洗函数统一走dateutil.parser加一个解析失败兜底from dateutil import parser as date_parser from datetime import datetime, timezone def parse_to_utc(value: str): if not value: return None try: # 自动识别常见时间格式统一转成 UTC 感知时间 dt date_parser.parse(value) return dt.astimezone(timezone.utc).isoformat() except Exception: return None # 解析失败宁可返回 None不伪造时间条件反射式的“抓取时间兜底”看起来聪明其实最容易掩盖问题。我后来把逻辑改成解析失败返回 None由上层决定是否忽略这条内容。宁可少一条也不收一条错的。6.2 标题重复入库热榜和搜索接口的同一篇文章另一个常见问题同一篇内容在平台的“热榜接口”和“搜索接口”里同时出现但两个接口返回的链接并不一样——一个是直达链接一个是带搜索参数的重定向链接。我的item_id也不相同于是同一篇文章被当成两条新内容存了进去。去重表只解决“完全相同的指纹”解决不了“语义上相同的内容”。后来我加了一道标题归一化二次去重链路先把标题去掉所有空白、统一全半角、低大小写再计算近似指纹。运行一段时间后发现很多重复条目都是这一层过滤掉的。这个逻辑放到规则匹配之前效果最好因为规则引擎不用反复处理那些已经被认定重复的数据。6.3 限流与IP风险随机抖动与访问间隔做过采集的人都懂“429 Too Many Requests”这个状态码。我第一次大规模接入多个热榜源时就撞上了。原因是我把任务触发时间都设置在整点结果每个小时的第一分钟十几个任务同时发请求某些平台直接不客气地返回限流。解决方式就两招一是错峰任务启动时间手工错开不集中在整点二是在每次请求之间加随机延迟。代码层面的随机抖动前面给出过不再重复。我想补充的是限流发生后怎么处理比怎么预防更重要。我建议先读取响应头里的Retry-After字段如果服务端明确告诉你“等多久再试”就严格按照那个时间来。没有这个字段的才轮到自己算退避。对待平台的反爬限制我的态度是别硬刚。降低频率、减少字段、找更规范的数据源通常都可以解决。如果一个源怎么都绕不过限制那就换个思路比如它有没有官方的 RSS、有没有第三方官方的分发渠道而不是天天与反爬机制博弈。6.4 规则误报与基线数据预热趋势突变规则刚上线时头一天给我发了60多条告警全是不该发的。原因是我的历史数据表是空的第一轮全量抓取把平台最近几十条内容全部灌了进来然后趋势算法拿“眼前这几十条”和“几乎为零的历史基线”对比自然触发了一个爆发式的增长率。现在我的解决方案是“基线预热期”任何新接入的数据源至少静默运行三天不开启趋势突变告警只默默积累历史数据。三天之后基线稳定了再打开趋势规则。关键词规则和热度阈值规则可以即时生效因为它们的判断不依赖历史。另外趋势算法里我加了一个最小样本量要求比如过去7天同期均值至少要大于5条否则增长率再高也不告警防止拿小样本做草率结论。7. 下一步规划从个人工具到团队基础设施7.1 数据源适配器插件化目前我新增一个数据源还需要改一部分采集代码。下一步计划是把数据源抽成标准的 Adapter 接口每个源实现fetch()和normalize()两个基础方法然后用一个注册表管理。新增一个源只需要在配置目录里加一个 Python 文件写上解析逻辑即可主程序完全不用动。这对团队协作很重要成员各自维护自己负责的数据源互不影响合并代码时也基本没有冲突。7.2 从关键词命中到语义判断关键词规则说到底还是字面匹配它识别不了“没提到关键词但明显在讲同一个话题”的内容。我打算引入轻量文本向量对标题和摘要计算 embedding然后和一批种子文本做相似度匹配。这样“本地部署大模型”这条内容即使没出现“开源大模型”也有可能通过语义相似度被捞回来。这里我不建议每个人一上来就搞大模型。先在关键词规则之上加一层基于 TF-IDF 的相似文本召回成本极低效果也不错。词向量和语义模型放到后面等规则判不过来、关键词列表膨胀到几十上百个词的时候再换也不迟。7.3 可视化看板与趋势洞察雷达的最终形态应该是一个让人“只做决策、不做搬运”的工具。我想在看板上展示三件事各来源的新增内容趋势曲线、命中规则的告警排行、以及每个关键词在时间轴上的热度脉冲。这其实就是把自己的小库变成一张可交互的情报地图。技术选型上用自建 FastAPI 加一点前端图表库就行不需要引入重型 BI 系统。数据仍存在本地库后续可以做周报、月报的自动导出把这段时间监控到的异常、热点、竞品动态汇总成一份文档。信息雷达带来的不只是当下降噪还为长期分析积累了一笔有价值的数据资产。做这个项目大半年我最深的体会是不要一开始就追求完美架构先把“定时抓取 - 去重 - 告警”这个最小闭环跑起来然后让它在真实使用中暴露问题再逐步修正。整个系统最值得优先投资的不是调度器不是告警模板而是前面那层数据标准化。它对所有后续功能的影响是全局的这一层稳了后面的规则、看板、统计都会顺很多。还有一个小技巧日志里所有时间字段统一记录为 UTC ISO8601别用本地时间。很多排查工作摸不着头脑最后都发现是各台的时区混在了一起。时区统一这件事看起来小实际省掉我大量烧脑时间。如果你也在做类似的信息监控工具建议从第一天起就搞定它。
先说个场景。我做内容运营和技术情报这块儿有些年头了每天打开浏览器的第一件事就是去各个平台看一眼热榜、关键词搜索结果、同行有没有发新东西。一开始只有两三个信息源手动刷还撑得住后来源一多问题立刻来了——页面加载慢、信息重复、漏检率高半小时折腾下来手里全是零散截图根本凑不出一份干净的情报。PLFM_RADAR 就是在这种状态下被逼出来的。PLFM_RADAR 全称是 Platform Radar翻译过来就是“平台雷达”。它不是某个商业软件而是我自己写的一套小工具对多个平台的公开信息做定时扫描、统一聚合、去重、规则过滤再按需把重要变化推送到钉钉、飞书或者邮箱。运行了大半年它帮我省下的时间不是一星半点更重要的是那些本来会被我刷漏的信息现在基本都能在一个看板上被捞回来。这篇文章就把这个项目的设计思路、核心代码、踩坑经历都摊开说说适合正在做信息监控、竞品观察、热点追踪的朋友参考。1. 为什么我动手做了个“平台雷达”1.1 信息过载带来的真实痛点先还原一下没有雷达之前的工作流。每天我需要盯的信息源包括几个内容社区的热榜、行业资讯站的 RSS、竞品官方的动态页、以及若干关键词在站内搜索的返回结果。每个源有自己的更新节奏有的五分钟就刷屏有的一天才更新两三次有的是结构化接口有的只能靠肉眼翻页面。手动巡检最大的问题不是“看得慢”而是“无法持续”。人总有走神的时候午休回来、开会两小时可能正好错过了某个重要的信息窗口。等再打开页面那条内容已经被刷下去了。信息监控这件事靠人工盯是注定有漏的需要有一个替身在固定的时间点替你去“看一眼”并且把不符合要求的大部分内容过滤掉只把值得人工介入的那部分挑出来。另外就是重复。同一篇内容经常在热榜、搜索、分类页里出现三次标题不一样、链接不一样人工很难判断“这个是不是我已经看过的”。于是好不容易攒下来的阅读时间又被浪费在重复确认上。1.2 PLFM_RADAR 的定位与使用场景PLFM_RADAR 不是爬虫框架也不是舆情系统它的定位是一个“聚合 规则判断”的小型中间层。它每天定时去各个平台拿到原始数据转成统一格式判断哪些是新的哪些是重复的然后跑一遍用户自定义的规则决定要不要通知你。我实际用得最多的场景有三个热点追踪。比如我长期关注“开源大模型”“本地部署”“智能体”这几个方向凡是命中关键词的新内容都会被收进实时列表如果热度冲进前几还会单独推送提醒。竞品动态监控。竞品的官方公告、版本发布、社区新帖只要来源支持 RSS 或公开页面就能纳入监控。竞品一有新动作我甚至会比公司群里的转发快几分钟。数据源异常发现。这是个附带收益。每个源都有挂掉的时候接口改版、抓取失败雷达会在连续失败后主动告警我不用等到第二天才发现某个源已经失效三天。适合用这套思路的人不仅仅是对技术感兴趣的程序员。内容创作者、自媒体运营、产品经理、做市场调研的人只要每天需要从多个平台收集公开信息都可以借鉴这里的思路搭一套轻量版本。1.3 和现成工具对比为什么还要自己写市面上的选择其实不少。采购一套商业舆情系统功能是强大但价格贵、部署重而且很多时候数据源和规则是锁死的你没法按自己的业务习惯去定制。RSS 阅读器类工具能解决“订阅聚合”的问题但只解决了“看哪里”没有解决“看什么值得看”它不会帮你过滤、去重、按规则告警。至于开源爬虫框架专注解决的是采集效率抓回来的数据放哪儿、怎么变成决策还需要自己搭一套。我选择自己写核心原因就两条第一需求足够个人化规则、通知渠道、数据源想改就改不存在等待厂商提需求的周期第二数据是自持的所有内容落在自己的库里后续想怎么分析都有地基。代价也很明确就是需要自己维护但作为长期使用的基础设施这个成本是值得的。2. 整体链路一个扫描任务从数据源到告警要走几步2.1 四层结构采集、处理、存储、通知在设计 PLFM_RADAR 之前我先把一条扫描任务的完整路径画了出来。从触发到最终有产出一共经历四层采集。调度器到了时间点去数据源拉取原始数据。这里可能是调一个公开 API可能是解析一条 RSS也可能是一个公开页面的内容。处理。把不同平台乱七八糟的字段统一成一套标准结构去掉无用的噪声归档或丢弃重复内容。存储。把清洗后的数据写入本地库作为后续分析的底座同时记录去重标记。通知。新数据跑一遍规则如果符合告警条件就构建一条消息推到聊天机器人或邮箱。用一句话概括定时触发 - 采集原始内容 - 标准化 - 去重入库 - 规则匹配 - 告警通知。这个链路看起来简单但它把一个原本只能靠人工执行的重复过程变成了可以无人值守的循环。我在设计的时候一直提醒自己每一层都要只做一件事越短越容易排查。比如“采集层”只管拿数据不管清洗。如果你把清洗逻辑和采集逻辑写在一个函数里第一周会觉得爽第二周开始改数据源的时候就会后悔。2.2 技术选型的取舍优先考虑可维护性PLFM_RADAR 这个项目我选用的技术栈相当朴素Python 3.10Requests 做 HTTP 请求feedparser 解析 RSSAPScheduler 做定时调度SQLite 起步做存储后期换 PostgreSQL 也不会伤筋动骨通知端直接发 Webhook 和 SMTP 邮件。Web 端用一个 FastAPI 提供的简易看板。为什么是 Python而不是 Go 或者 Node原因很简单这个项目的大部分工作是字符串清洗、RSS 解析、规则匹配Python 的库生态可以帮我省掉大量重复代码而且调试效率高。Go 的并发能力强但这里根本没有高并发需求不值得额外增加心智负担。调度器我选了 APScheduler 而非系统的 cron。原因是我希望任务的调度信息也能随项目走在代码里可配置、可修改而不是散落在系统 crontab 里。APScheduler 在单体小项目里非常合适支持 Cron 表达式和 interval 间隔两种触发方式最基本的用法足够覆盖需求。这里也要吐槽一个点。有些人一开始就上 Celery、上 Redis、上消息队列对于维护一个小型信息监控工具来说这是明显的过度设计。机器上没有消息中间件服务之间不需要异步解耦一个进程里的调度器完全可以搞定。先跑起来等到并发源超过几十个、任务之间出现明显依赖的时候再去引入这些东西那时你已经知道问题到底在哪里了。2.3 核心数据结构统一内容项模型所有数据源的内容进入系统后都会被转成一个统一的 RadarItem。之所以要统一是因为不统一的代价会在规则引擎和前端展示阶段集中爆发。如果每个源的数据结构都不一样那么每加一个源规则引擎和看板都得跟着改一遍这是最典型的失控路径。我用一个 dataclass 来定义这个核心模型from dataclasses import dataclass, field dataclass class RadarItem: source_id: str # 数据源唯一标识例如 community_a item_id: str # 原始条目ID平台自带的唯一标识 title: str # 标题 url: str # 原文链接 content: str # 摘要或正文截断 published_at: str # 发布时间统一存 UTC ISO8601 字符串 fetched_at: str # 抓取时间 extra: dict field(default_factorydict) # 各平台特有字段比如热度分、排名对应的存储层核心表至少需要两张内容表和去重表。内容表用来存数据去重表用source_id item_id作为联合主键保证同一条原始记录只入库一次。CREATE TABLE seen_items ( source_id TEXT NOT NULL, item_id TEXT NOT NULL, first_seen_at TIMESTAMP NOT NULL, PRIMARY KEY (source_id, item_id) );这个模型是整个项目的地基后面所有规则、所有展示都建立在这套字段之上。有一段时间我想给某个源单独加字段直接在 RadarItem 里堆属性结果模型越来越臃肿。后来改成把平台特有数据一律丢进 extra 字典规则引擎按需读取模型又恢复了清爽。这个教训很值得记住核心模型尽量保持稳定变化的细节放到扩展字段里。3. 数据源接入官方API、RSS、公开页面怎么排优先级3.1 官方API是首选但要注意频率配额接入任何数据源之前先做优先级判断有官方 API 的优先用官方 API没有的看有没有公开 RSS都没有才考虑解析公开页面。这个顺序不是技术洁癖而是维护成本的排列。官方 API 的好处很明显字段结构化、有明确的文档、更新及时。但它也有自己的问题最典型的是频率配额和增量游标。我曾经配置完一个官方 API 后直接按分钟级频率去轮询结果半天后收到限流通知请求被拒。后来老老实实把频率降到平台允许的范围并开启增量更新。增量更新的思路是平台返回一个since_id或cursor参数我每次只拉“上次游标之后”的新内容。这样每次请求的数据量最小、对服务端的压力也最小限流概率自然降低。游标的保存位置很简单数据源配置表里加一个last_cursor字段就行。接入 API 时还有个容易踩的坑不要一上来就把所有字段全量拉回来。很多新内容的正文很长存储和清洗成本都很高但你的规则可能只需要标题和链接。我建议只取最小必要字段正文等内容真正命中规则之后再按需获取或者保留摘要即可。3.2 RSS解析库的使用与常见字段映射没有官方 API 的源RSS 往往是最好的替代方案。它在数据量上虽然不如 API 丰富但胜在稳定和轻量。feedparser 是解析 RSS 的标配库用起来很简单import feedparser feeds feedparser.parse(https://example.com/rss.xml) for entry in feeds.entries: title (entry.get(title) or ).strip() link (entry.get(link) or ).strip() published entry.get(published) or entry.get(updated) summary (entry.get(summary) or )[:200]这里有个非常关键的字段映射问题。不同平台的 RSS 里发布时间字段的命名差异很大有的叫published有的叫pubDate有的只有updated有的干脆一个时间字段都没有。我在设计清洗层时会做一个兜底链published优先其次updated都没有就把发布时间标记为“未知”而不是随口用当前抓取时间塞进去。为什么不愿意默认用抓取时间因为我后面有条规则是“只看最近24小时的新内容”如果把历史老内容的发布时间都错标成抓取时间它们就会被错误地当成新内容推送导致告警失真。宁可缺一个时间信息也不能伪造一个错误的时间。3.3 公开页面解析的底线原则与降级方案当某个源既没有 API 也没有 RSS只能解析公开页面时我的原则是先找页面里内嵌的 JSON 数据再考虑解析 HTML 结构。现在的很多前端页面首屏数据都是以window.__INITIAL_STATE__或类似形式直接塞在 HTML 里的。用正则或字符串截取把这段 JSON 抠出来比解析一堆 CSS 类名要稳得多。页面上展示的数字、标题、链接往往就是这份 JSON 的子集。你直接拿结构化数据没必要去猜 HTML 结构。如果实在只能解析静态 HTML那就必须接受一个现实页面改版时你的解析器大概率会失效。为了减少损失我会把解析器注册成带版本号的模板一旦连续解析失败立刻触发“数据源异常”告警而不是让错误悄无声息地堆积。降级方案也很重要这个源暂时没抓到不代表下次也要放弃跳过本次任务、记录日志、下一轮继续尝试。无论哪种公开页面解析都要守住底线遵守 robots.txt 的约束、控制访问频率、不要试图绕过访问限制。做信息雷达的目的是让自己省时间不是给自己惹麻烦。4. 调度与去重雷达转动时的“不重不漏”问题4.1 多源任务的调度配置与错峰设计调度器是整个雷达的“心跳”。我用 APScheduler 给每个数据源单独注册任务因为不同源的更新节奏完全不同。热榜类源每五分钟扫一次资讯类 RSS 每十五分钟扫一次官方公告类源一小时扫一次就够。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger scheduler BlockingScheduler() scheduler.add_job( scrape_source_a, CronTrigger(minute*/5), idsource_a, max_instances1, ) scheduler.add_job( scrape_source_b, CronTrigger(minute*/15), idsource_b, max_instances1, )这里有两个容易被忽略的细节。第一个是max_instances1它的作用是防止上一次任务还没跑完下一次触发又来了导致同一源并发抓取。加了之后如果任务超时或阻塞后续触发会自动跳过保证同一时间同一源只有一个任务在跑。第二个是错峰设计。同一个源如果需要抓多个关键词不要在同一秒全部发出请求。我在代码里对每次请求加了一个随机延迟import random import time # 同一数据源每次请求之间增加 0.8~2.5 秒随机间隔避免触发限流 time.sleep(random.uniform(0.8, 2.5))这个随机抖动不是玄学。信息平台的反爬机制里常见手段是统计“单位时间内的请求数”请求间隔过于规律反而容易被识别为自动化程序。加上随机延迟后请求分布更接近自然人行为限流风险也会小很多。4.2 指纹去重与重复内容合并去重是整个雷达最核心的工程问题之一分两层处理。第一层是“原始唯一键去重”用前面提到的(source_id, item_id)联合主键。每个平台在生成内容时基本都有自己的一套 ID 体系直接用平台的 ID 是最可靠的判重方式。但现实是有些源并不能提供稳定的内容 ID比如纯 HTML 页面解析出来的数据。这时我会退而求其次计算一条内容指纹import hashlib def make_fingerprint(source_id: str, title: str, url: str) - str: raw f{source_id}|{title.strip()}|{url.strip()}.lower() return hashlib.md5(raw.encode(utf-8)).hexdigest()这层指纹保证的是“完全相同的内容不会再次入库”。第二层是“相似内容合并”专门处理“同一篇内容换个标题、换个入口又出现”的情况。最典型的场景是某个内容既出现在平台热榜又出现在关键词搜索结果里而且标题有细微差异比如加了“热”字、或者“首发”两个字。标题一变上面的指纹就失效了。我的处理方法是不对标题做硬性相等而是先做归一化再比较相似度。归一化指的是去除多余空格、统一全半角、大小写折叠。进阶一点对归一化后的标题用SequenceMatcher或 simhash 做相似度判断相似度超过阈值的就认为是同一篇内容直接丢进去重表。这个逻辑不能太激进否则会把真正的不同内容误合并所以我只在“标题长度超过15个字且相似度超过0.9”时才启用。4.3 增量更新与失败重试增量更新是为了避免每次抓全量数据。很多内容平台的接口或页面默认会返回最近几十条内容如果每次都把这几十条全量清洗入库数据量很快会失控。正确姿势是记录一个游标或时间点下一次只处理比游标更新的内容。在代码里这个游标可以直接存在数据源配置表里每次成功抓取后更新。失败重试我采用了指数退避策略某源第一次失败后等1分钟再试再失败等5分钟、15分钟连续失败超过5次就把该源标记为“暂停”并触发一条告警。直接看代码更清晰retry_delay_minutes [1, 5, 15, 30, 60] def handle_failure(source_id: str, fail_count: int): if fail_count len(retry_delay_minutes): delay retry_delay_minutes[fail_count - 1] # 安排 delay 分钟后重新执行 scheduler.add_job( scrape_source, date, args[source_id], run_datedatetime.now() timedelta(minutesdelay), ) else: # 连续失败过多暂停该源并告警 notify_webhook(f数据源 {source_id} 连续失败已自动暂停)重试策略的核心是“别在同一时间点上反复撞墙”。如果没有退避机制一个源出了故障你会看到日志里全是同一时刻的重复错误而对定位问题毫无帮助。5. 告警规则从“我看到”到“它通知我”5.1 规则引擎的三层设计雷达真正的价值体现在规则引擎上。我把规则分成三层从简单到复杂依次是关键词规则、热度阈值规则、趋势突变规则。关键词规则是基础。每一条规则由三个部分组成必须包含的词AND关系、任意包含的词OR关系、排除词NOT关系。比如我想监控国产模型动态会设置必须包含“大模型”任意包含“开源”“发布”“上线”排除“广告”和“招聘”。执行顺序是先排除再必含再任含。热度阈值规则解决的是“关键词命中很多但我想只看重要的”。有些平台会给每条内容一个热度分或排名雷达可以拿到。当某条内容进入平台热榜前20或者热度分超过一个阈值即使它没有命中关键词也应该被推送出来因为它代表“大众正在关注的东西”。趋势突变规则是我后期才加上去的也是最出效果的一层。它的逻辑很简单同一个关键词在最近30分钟内的出现频率相比过去7天的同期均值如果突然增长了300%就说明这个话题正在发酵。这套逻辑最早用得并不顺利原因后面会讲但一旦过了基线期它的效率非常高能抓到很多关键词规则完全覆盖不到的新话题。5.2 通知渠道与告警模板通知渠道我接了三种企业微信群机器人、钉钉群机器人、邮件 SMTP。所有渠道都走同一套接口抽象在配置里切换即可。不要小看这个抽象因为团队的通知偏好经常变比如今天用钉钉明天可能就切到飞书改了适配器不影响核心逻辑。一条合格的告警消息至少应该包含五个字段触发时间、内容来源、标题、原文链接、命中原因。命中原因一定要写清楚否则你收到推送后还得去猜“为什么这条消息会推给我”。我之前有一条规则配得太宽收到的告警自己也看不懂最后查了日志才发现是关键词里的某个字太常见了。curl -X POST https://example.com/webhook \ -H Content-Type: application/json \ -d { msgtype: markdown, markdown: { content: 【PLFM_RADAR 告警】\n- 来源某社区热榜\n- 标题开源大模型最新进展\n- 发布时间2025-01-01 12:00:00\n- 命中规则关键词「开源」 热度排名前10\n- 链接https://example.com/article/123 } }邮件渠道你同样用 SMTP 发模板就一行纯文本加链接别整花活。群消息的目的就是快速扫一眼、判断要不要点开模板越简洁越好。5.3 防告警轰炸去重与抑制窗口告警功能上线后的第一个事故就是告警轰炸。有一天某个平台的数据异常一次抓取把几十条历史内容当新内容导入了结果规则引擎瞬间匹配出二十多条告警手机被通知刷了一屏。问题的根源不在规则而在告警层缺少保护。我加了两个机制。第一个是告警去重表以“规则ID 内容指纹”做联合主键同一内容命中同一规则最多只推送一次。就算它被重复导入、反复匹配也只发一次通知。第二个是抑制窗口。同一个“规则ID”在5分钟时间窗口内只发摘要不再逐条推送。这样即使出现批量命中收件人最多收到一条形如“该规则在5分钟内新增12条匹配点击查看”的聚合消息。如果真有必要逐条看去数据库或看板里翻就行。我还加了一个保险丝单条规则单日最多告警50条超过之后当天自动静默次日零点重置。这是为了防止规则本身配错、没人及时发现的情况下重复轰炸一整天。6. 上线运行后的踩坑实录6.1 时间字段格式混乱新内容被排到了历史里上线第一周我就发现一个诡异现象某条标题里写着“今日发布”的内容在列表里一直排在很靠后的位置像是一个几年前的旧闻。查了原始数据才发现这个源返回的时间字符串格式是GMT标准里的老式写法和我库里的 ISO8601 格式默认排序不兼容清洗层没有正确转换导致它被解析失败后回退成了“未知时间”再被我兜底逻辑当成很久以前的数据。这个坑让我意识到时间解析是整个清洗层里最不能含糊的部分。真实世界里各平台的时间格式五花八门Wed, 01 Jan 2025 12:00:00 GMT、2025-01-01T12:00:0008:00、2025-01-01 12:00:00还有纯时间戳。现在我的清洗函数统一走dateutil.parser加一个解析失败兜底from dateutil import parser as date_parser from datetime import datetime, timezone def parse_to_utc(value: str): if not value: return None try: # 自动识别常见时间格式统一转成 UTC 感知时间 dt date_parser.parse(value) return dt.astimezone(timezone.utc).isoformat() except Exception: return None # 解析失败宁可返回 None不伪造时间条件反射式的“抓取时间兜底”看起来聪明其实最容易掩盖问题。我后来把逻辑改成解析失败返回 None由上层决定是否忽略这条内容。宁可少一条也不收一条错的。6.2 标题重复入库热榜和搜索接口的同一篇文章另一个常见问题同一篇内容在平台的“热榜接口”和“搜索接口”里同时出现但两个接口返回的链接并不一样——一个是直达链接一个是带搜索参数的重定向链接。我的item_id也不相同于是同一篇文章被当成两条新内容存了进去。去重表只解决“完全相同的指纹”解决不了“语义上相同的内容”。后来我加了一道标题归一化二次去重链路先把标题去掉所有空白、统一全半角、低大小写再计算近似指纹。运行一段时间后发现很多重复条目都是这一层过滤掉的。这个逻辑放到规则匹配之前效果最好因为规则引擎不用反复处理那些已经被认定重复的数据。6.3 限流与IP风险随机抖动与访问间隔做过采集的人都懂“429 Too Many Requests”这个状态码。我第一次大规模接入多个热榜源时就撞上了。原因是我把任务触发时间都设置在整点结果每个小时的第一分钟十几个任务同时发请求某些平台直接不客气地返回限流。解决方式就两招一是错峰任务启动时间手工错开不集中在整点二是在每次请求之间加随机延迟。代码层面的随机抖动前面给出过不再重复。我想补充的是限流发生后怎么处理比怎么预防更重要。我建议先读取响应头里的Retry-After字段如果服务端明确告诉你“等多久再试”就严格按照那个时间来。没有这个字段的才轮到自己算退避。对待平台的反爬限制我的态度是别硬刚。降低频率、减少字段、找更规范的数据源通常都可以解决。如果一个源怎么都绕不过限制那就换个思路比如它有没有官方的 RSS、有没有第三方官方的分发渠道而不是天天与反爬机制博弈。6.4 规则误报与基线数据预热趋势突变规则刚上线时头一天给我发了60多条告警全是不该发的。原因是我的历史数据表是空的第一轮全量抓取把平台最近几十条内容全部灌了进来然后趋势算法拿“眼前这几十条”和“几乎为零的历史基线”对比自然触发了一个爆发式的增长率。现在我的解决方案是“基线预热期”任何新接入的数据源至少静默运行三天不开启趋势突变告警只默默积累历史数据。三天之后基线稳定了再打开趋势规则。关键词规则和热度阈值规则可以即时生效因为它们的判断不依赖历史。另外趋势算法里我加了一个最小样本量要求比如过去7天同期均值至少要大于5条否则增长率再高也不告警防止拿小样本做草率结论。7. 下一步规划从个人工具到团队基础设施7.1 数据源适配器插件化目前我新增一个数据源还需要改一部分采集代码。下一步计划是把数据源抽成标准的 Adapter 接口每个源实现fetch()和normalize()两个基础方法然后用一个注册表管理。新增一个源只需要在配置目录里加一个 Python 文件写上解析逻辑即可主程序完全不用动。这对团队协作很重要成员各自维护自己负责的数据源互不影响合并代码时也基本没有冲突。7.2 从关键词命中到语义判断关键词规则说到底还是字面匹配它识别不了“没提到关键词但明显在讲同一个话题”的内容。我打算引入轻量文本向量对标题和摘要计算 embedding然后和一批种子文本做相似度匹配。这样“本地部署大模型”这条内容即使没出现“开源大模型”也有可能通过语义相似度被捞回来。这里我不建议每个人一上来就搞大模型。先在关键词规则之上加一层基于 TF-IDF 的相似文本召回成本极低效果也不错。词向量和语义模型放到后面等规则判不过来、关键词列表膨胀到几十上百个词的时候再换也不迟。7.3 可视化看板与趋势洞察雷达的最终形态应该是一个让人“只做决策、不做搬运”的工具。我想在看板上展示三件事各来源的新增内容趋势曲线、命中规则的告警排行、以及每个关键词在时间轴上的热度脉冲。这其实就是把自己的小库变成一张可交互的情报地图。技术选型上用自建 FastAPI 加一点前端图表库就行不需要引入重型 BI 系统。数据仍存在本地库后续可以做周报、月报的自动导出把这段时间监控到的异常、热点、竞品动态汇总成一份文档。信息雷达带来的不只是当下降噪还为长期分析积累了一笔有价值的数据资产。做这个项目大半年我最深的体会是不要一开始就追求完美架构先把“定时抓取 - 去重 - 告警”这个最小闭环跑起来然后让它在真实使用中暴露问题再逐步修正。整个系统最值得优先投资的不是调度器不是告警模板而是前面那层数据标准化。它对所有后续功能的影响是全局的这一层稳了后面的规则、看板、统计都会顺很多。还有一个小技巧日志里所有时间字段统一记录为 UTC ISO8601别用本地时间。很多排查工作摸不着头脑最后都发现是各台的时区混在了一起。时区统一这件事看起来小实际省掉我大量烧脑时间。如果你也在做类似的信息监控工具建议从第一天起就搞定它。
