PLFM_RADAR 是我去年底开始维护的一个内部项目代号。名字拆开看很简单PLFM 是 Platform 的缩写RADAR 就是雷达。合起来就是一个专门盯平台的“雷达系统”——把分散在多个平台上的关键信息实时聚合、判重、告警让你不用天天手动刷新网页也能在第一时间知道“圈子里发生了什么大事”。干这行的朋友应该秒懂这种痛点你有 N 个平台要盯每个平台的更新频率还不一样靠人肉巡检早晚会漏而且纯手动看一遍半小时就过去了。PLFM_RADAR 就是把这件事自动化、可配置化、可追踪化。这篇文章我打算从整体设计思路、核心模块拆解、实操落地细节、以及我踩过的坑这几块来讲内容偏工程向适合负责信息监控、竞品动态跟踪、舆情预警这类场景的开发者或运维朋友参考。如果你只是偶尔需要盯一两个平台那用脚本就够了但如果你想搭一套能长期跑、能加监控源、能应对数据量增长的体系这篇文章应该能帮你省掉不少弯路。1. 项目概述PLFM_RADAR 到底解决什么问题1.1 从平台监控的痛点说起我先说一下当初为什么会有这个项目。团队的业务依赖好几个外部平台的信息比如竞品上新、行业政策发布、客户的公开动态、技术社区的热点帖子等等。原来大家的做法是上班先刷一遍中午再刷一遍下班前再刷一遍纯靠手动。后来信息源越加越多手动巡检明显跟不上了。我们遇到的最典型的几个问题更新频率不均匀有的平台一天更新几百条有的平台一周更新一条。用一个固定频率去扫要么浪费资源要么漏掉关键内容。重复内容太多同一个新闻被多个渠道转载口径还不太一样人工判断“这条是不是已经看过了”其实挺耗精力。噪音和信号混在一起信息量大之后有价值的内容反而被淹没需要一套机制把“值得关注的变化”从“普通更新”里拎出来。没有留痕谁在什么时间发现了什么信息如果只看聊天记录事后很难追溯。PLFM_RADAR 的目标就是把这些流程变成一条流水线定时扫描、内容采集、规范化处理、去重、判断是否该告警、最后把结果推进告警通道。它本质上不是一个大而全的数据采集平台而是一套“信号处理系统”——重点不在把数据搬回来而在从数据里识别出值得响应的信号。1.2 核心设计思路不只是抓数据而是做“信号处理”“雷达”这个比喻不是随便起的。真实的雷达工作原理是发射探测波、接收回波、过滤杂波、识别目标。PLFM_RADAR 的流程完全对标这个逻辑雷达系统PLFM_RADAR 对应模块发射探测波按策略发起对目标平台的扫描请求接收回波拉取页面或接口返回的原始数据过滤杂波清洗、去重、格式化去掉无关内容识别目标用规则或算法判断哪些信息值得触发告警这个设计思路带来的一个直接好处是分层清晰。采集层、判断层、告警层各自独立哪一块出问题都好排查。另一个好处是扩展性后面想加新的平台源只需要在采集层加一个适配器不用动其他部分。1.3 技术选型与适用场景边界关于技术栈我没有一上来就上大数据全家桶而是按团队现状做了选择。核心是 Python 3.10 Redis SQLite/PostgreSQL。Redis 承担消息队列和去重集合的角色SQLite 用于轻量落库数据量上来之后再换 PostgreSQL 或 ClickHouse。为什么不用像 Scrapy Splash Kafka Spark 那样的重型组合因为 PLFM_RADAR 的实际场景是“准实时”不是“大数据”。我们的大部分监控源都是公开页面或开放接口数据量级在每天几千到十几万条之间这个量级用 Python 的异步框架加上一个轻量消息队列完全够用。重技术栈带来的运维成本和开发复杂度在这种规模下反而是负资产。它适合的场景监控多个信息源的更新做统一汇聚。对更新内容做实时去重和关键词/规则筛选。按重要程度分级告警推送到钉钉、企业微信、邮件等。保留历史数据方便后续做趋势分析。不适合的场景大规模爬取、对抗式采集、超高频实时推送毫秒级。这些属于另外一个领域需要专门的采集系统和流计算框架不是这个项目要解决的问题。2. 核心模块解析从扫描到告警的四层流水线2.1 扫描调度雷达波怎么发扫描调度是整个系统最关键的模块之一直接决定了监控的时效性和成本。我的做法是给每个监控源配置独立的扫描策略而不是用统一的定时器。固定间隔扫描适合更新规律、频率稳定的源比如每小时整点扫一次。抖动间隔扫描在固定间隔基础上加一个随机偏移量避免多个任务在同一时刻集中触发对目标方相对友好。基于活动频率的动态扫描记录每个源的历史平均更新间隔如果最近比较活跃就加大扫描频率反之降频。快慢双频扫描对重要程度高的源用高频率扫其他源用低频率扫。有点像雷达的“扇区扫描”对重点方向反复多看几眼。我实际用的比较多的是“快慢双频 抖动”。比如对重点行业动态源每 5 分钟扫一次对一般技术博客源每 30 分钟扫一次对所有请求都加上 3~5 秒的随机抖动避免整点“抢跑”。扫描调度在代码层面我用了 APScheduler。它支持多种触发器、持久化任务存储还支持错过执行后的补偿逻辑。这套东西已经很成熟了没必要自己造轮子。2.2 采集适配器不同平台的差异化处理不同平台的返回格式差异很大。有的平台提供公开 JSON API有的只有 HTML 页面有的需要通过搜索接口才能拿到最新结果。为了让上层模块不关心这些差异我把采集层封装成了统一的适配器接口。每个适配器对外暴露三件事fetch()返回原始数据。parse()把原始数据转成统一的 Content 对象。normalize()对文本做清洗、标签抽取、内容摘要。举个例子一个平台的 API 返回的字段名是title另一个平台是headline还有的只有 HTML 网页需要自己解析。适配器层把这些问题都消化掉上层拿到的永远是一个标准结构的对象包含标题、正文、链接、发布时间、来源标识、原始标识等几个核心字段。这里我强烈建议原始标识字段一定要保留。就是该内容在原始平台上的唯一 ID。它不仅是去重的重要依据也是将来排查问题、做反向溯源的关键。2.3 判定引擎什么内容值得触发告警采集和解析都完成之后数据被投递到判定引擎。这个模块负责回答一个问题“这条内容要不要推给用户”我的实现是规则引擎和动态阈值相结合。规则引擎的部分比较简单配置关键词列表、正则表达式、重要性权重。命中的关键词越多、权重越高这条内容的“热度分”就越高。比如监控竞品动态时命中“融资”“收购”“发布”这些词分值就高命中“团建”“招聘”这种分值就低。动态阈值的部分稍微复杂。如果只是用固定阈值很容易出现两种情况热点事件爆发时所有内容都超过阈值告警炸了长期风平浪静时阈值设太高导致一条通知都没有。我的做法是对每个监控域维护一个“平均热度基线”阈值随着基线浮动。这个逻辑用一句话讲就是和“平时比”异常突出才告警而不是达到绝对分数就告警。2.4 告警聚合与分发避免变成信息炸弹判定引擎产出告警之后如果直接把每条告警都推出去大概率会被用户屏蔽。我做了两层处理效果好了很多。同类聚合规定时间内同一来源、同一主题的多条告警合并成一条摘要。延迟确认当单条内容的重要度评分极高时立即发送当处于临界值时等 5 分钟如果这段时间内没有更重要的消息再发送。目的是减少打扰。告警分发是通过消息通道插件实现的。每种通道企业微信机器人、钉钉、邮件、Webhook是一个独立插件用一个带重试的发送队列维护。之前踩过一个坑告警服务一重启队列里的消息就丢了。后来把队列换成了 Redis Streams才彻底解决这个问题。Redis Streams 天然支持消息持久化和消费确认对“发送失败要重试”的场景比普通 List 可靠得多。3. 实操环节从零搭起 PLFM_RADAR 的关键步骤3.1 环境准备与基础依赖我假设你已经有一台 Linux 服务器且装了 Python 3.10 和 Redis。如果没有 Redis也可以用内存队列先跑起来但生产环境强烈不建议。基础依赖只有几个httpx异步 HTTP 客户端、APScheduler任务调度、redis消息队列和去重集合、beautifulsoup4HTML 解析。数据库我先用 SQLite一条连接串就能跑。建项目目录时我习惯按模块划分plfm_radar/ ├── app.py # 入口启动调度器和消费任务 ├── config.py # 全局配置 ├── scheduler.py # 扫描调度 ├── sources/ # 各平台适配器 │ ├── example_source.py │ └── base.py ├── pipeline/ │ ├── dedup.py # 去重 │ ├── filter.py # 规则判定 │ └── storage.py # 落库 ├── notify/ │ ├── channel_webhook.py │ └── sender.py └── utils/ ├── fingerprint.py # 内容指纹 └── text.py # 文本清洗3.2 通用字段结构与内容解析每个适配器解析出来的内容统一成ContentItem字段如下字段类型说明source_idstr来源标识比如weibo、rssremote_idstr原始平台上的唯一 IDtitlestr标题summarystr摘要或正文前 200 字urlstr原文链接published_atdatetime发布时间rawdict原始数据留底用为什么强调remote_id因为同一个链接即使 URL 上有不同的跟踪参数remote_id通常是不变的。用remote_id做去重比用 URL 更稳。解析这块的一个建议如果目标平台提供 JSON 接口优先用 JSON。HTML 解析的成本高且脆页面一改版就要重写。我遇到过平台把接口路径改了的情况好在我对解析结果做了断言校验能在第一时间发现字段缺失而不是带着脏数据往下跑。3.3 内容指纹与去重机制的实现即使有了remote_id也不能完全依赖它。因为不同平台之间会互相转载同一篇内容在 A 平台和 B 平台是两条不同的remote_id但内容本身是重复的。PLFM_RADAR 的第二步去重就是对文本做“指纹”比对。我用的算法是 SimHash。SimHash 的原理不复杂把文本分词、赋予权重、做哈希、按位加权求和、最后降维成一个 64 位整数。两个文本的 SimHash 汉明距离小于 3就认为是近似重复内容。去重流程在代码里是这样写的from utils.fingerprint import simhash, hamming_distance class Deduplicator: def __init__(self, redis_client, threshold3): self.redis redis_client self.threshold threshold def is_duplicate(self, content_item): fp simhash(content_item.summary) # 从 Redis 里取最近时间窗口内的指纹做比较 recent_fps self.redis.lrange(fp:recent, 0, -1) for old_fp in recent_fps: if hamming_distance(fp, int(old_fp)) self.threshold: return True # 没有重复就写入 self.redis.lpush(fp:recent, fp) self.redis.ltrim(fp:recent, 0, 1999) return Falsefp:recent这个列表只保留最近 2000 条指纹。为什么不存全量因为时效性强的场景只需要在一段时间内去重太久远的内容就算重复出现在监控里也可以视为“重新进入讨论范围”。这种做法节省内存语义上更贴合实际情况。去重逻辑需要注意一个权衡点阈值设太小比如 1相似内容去不干净设太大比如 8会把主题相同但内容不同的文章误杀。我在实操中调参时会把阈值锁定在 3 到 4 之间并且只对正文摘要做比较标题不参与指纹计算。这样对转载类内容敏感对不同角度的报道又能保留下来。3.4 规则判定与动态阈值判定模块我给它起名叫SignalJudge。它接收去重后的内容输出等级low/normal/high。先看静态规则。每个监控域有一条或多条规则每条规则包含一组关键词和权重RULES { competitor: { keywords: [融资, 收购, 上新, 合作, 发布], weights: [5, 8, 3, 2, 4], } }一条内容的得分等于命中的关键词权重的加权和。比如标题包含“收购”正文又提到“合作”分数就是 8 2 10。再看动态基线。系统定期统计每个监控域最近 7 天的平均得分作为基线。动态阈值的计算逻辑def adaptive_threshold(domain): base get_recent_avg_score(domain) return max(base * 3, base 5)如果今天的实时得分超过了这个自适应阈值说明这条内容比平时明显突出值得告警。这个方法不完美但省去了人工反复调阈值的麻烦。3.5 告警通道接入告警发送统一走notify/sender.py。每个通道都是一个send(message)接口的插件。我的配置里有企业微信机器人和自定义 Webhook两者都是简单 POST JSON 就能完成。这里的一个实操细节告警消息的格式要结构化不能只是一段文字。我会用标题 链接 摘要 得分 来源标识这样的字段组合让接收方能快速判断。消息发出去之后接收方可能有疑问想 相关同事所以还要附上原文链接。发送失败的重试机制我按 3 次来做立即重试、30 秒后重试、5 分钟后重试。如果三次都失败就写错误日志并把这则告警存到专门的失败表里。不能简单丢弃否则重要信息可能无声消失。4. 常见问题与排查实录4.1 告警风暴如何避免收到“信息炸弹”最容易遇到的问题就是告警风暴。某天一个大热点出现所有平台都在报道规则引擎瞬间判断出几十条高分段内容告警一条接一条发出去群里瞬间刷屏。第一次遇到时我们整个工作群直接炸了。后来用三个手段缓解同源聚合同一来源、同一事件在 10 分钟内的多条告警合并成一条汇总标题列出核心关键词和条数。重要度分级high级即时推送normal级进入延迟队列等 5 分钟再发low级只在每日汇总里出现。全局限流同一通道每分钟最多发送 N 条超过的部分进入队列排队。这三个手段叠加后告警量大概下降了 70% 左右而真正重要的信息一条都没漏。4.2 Redis 连接抖动导致任务丢失还有一次系统跑了两周突然有一段时间不再产生告警。查了日志发现大量redis.exceptions.ConnectionError。原因是 Redis 空闲连接被服务器断开但客户端连接池没有及时清理导致后续请求一直拿到无效连接。解决办法有两个一是配置 Redis 客户端的连接池health_check_interval参数二是给消费任务增加自动重连逻辑。这个问题排查的过程花了不少时间因为系统不是立刻报错而是“悄悄变慢、最后卡死”日志里的错误非常不明显。4.3 内容解析中断平台页面改版没有哪个平台会提前通知你它改版了。某天我的解析器突然解析不到标题了系统虽然没报错但产生的ContentItem全是空字段。要是没校验这些脏数据就进流程了。我加了一层“解析结果自检”如果一条内容没有标题、没有链接、也没有原文标识就认为解析失败记录下来并触发告警“解析异常”。这样可以第一时间发现页面结构变了而不是等人工发现漏数据时再回溯排查。4.4 时间处理不一致导致漏检不同平台返回的时间格式五花八门。有的给时间戳有的是 ISO 8601 字符串有的带时区有的不带。我曾经因为没有统一转成 UTC导致“按发布时间排序取最新内容”的逻辑出错漏掉了一些当日更新。这个问题的解法很粗暴但有效所有平台的时间在解析层统一转成datetime对象落库和比较全部用内部统一时区。对外展示时再转成本地时间。一旦时间口径统一后续做增量扫描、时间窗口判断都顺了。5. 数据存储与后续扩展思路5.1 存储模型轻量起步预留迁移空间前面提到落库我用 SQLite。表结构设计得尽量简单contents存储所有解析出来的内容包含上面说的公共字段。signals存储判定为“值得关注”的告警记录。scan_log每次扫描的状态记录包括耗时、数量、异常信息。内容字段上我喜欢加一个is_archived标记。这个字段用于“定期归档”逻辑超过 30 天的历史数据可以搬走不影响主表查询性能。等数据量超过几百万行SQLite 的写锁会成为瓶颈届时可以平滑切到 PostgreSQL。我的建议是尽量把 SQL 用标准写法少用 SQLite 私有特性为迁移留好余地。5.2 从“定时轮询”到“通知驱动”的演进可能目前 PLFM_RADAR 主要是定时扫描也就是“主动去问平台有没有新消息”。但有些平台支持 Webhook 或长连接如果有条件接入可以做到平台主动通知实时性更强也更节省资源。我规划过一个混合模式支持回调的源走事件驱动不支持的源继续走定时扫描。这样既保住了实时性又不会因为强依赖某些平台的接口而让自己陷入被动。之所以没有一开始就全上回调是因为不是所有平台都愿意给你开这朵云而且回调安全认证也是个额外的工作量。5.3 监控成效的分析评估最后建议在系统跑一段时间后做一个简单的效果量化日均扫描任务数、采集到的新内容数、触发告警数、用户手动取消告警数。这几个数字加在一起能反映出来系统的“灵敏度”和“噪音率”。我个人定义的“敏感度”指标是新告警中有多少比例被用户真正点开“噪音率”是被用户忽略或标记为不感兴趣的比例。如果噪音率偏高就去查规则的阈值是不是设低了或者关键词太宽泛。如果敏感度低则反过来调整。这一套小闭环才让系统价值真正体现出来而不是单纯“一直在跑但没人觉得好用”。PLFM_RADAR 能做到现在勉强好用最关键的一点就是我把它当一套“信号处理系统”来设计而不是简单的内容抓取脚本。确保每个模块职责单一扫描、解析、去重、判定、告警各管一段后面无论扩展平台源还是换算法都只是局部改动。如果让我再说一个经验那就是任何一步都别跳过校验和留痕。信息监控这种系统挂了不可怕最可怕的是它挂了你还不知道。有了自检机制、日志留底和告警链路监控这套雷达才能长期稳定运行。