先说清楚PanWatch不是某个现成的商业软件是我自己搭的一套“全平台内容监控与预警工具”的名字。Pan取的是全景、泛化的意思Watch是观察——在多个信息源里同时盯住你关心的关键词、竞品动态和异常变化把散落在各处的消息统一清洗、打分、去重之后只把真正值得处理的东西推到面前而不是让你自己满世界去翻。这件事的背景做产品、运营、自媒体的朋友应该都懂总有些变化不是你主动搜就能看到的而是等你发现的时候已经发酵好一阵了。手动开十几个标签页逐个刷既慢又不稳定而且判断标准全凭个人状态。PanWatch要解决的问题很简单——把“我每天主动去看”变成“系统觉得有问题了才叫我”。下面把整个搭建过程拆开讲包括技术选型、采集设计、处理链路、预警分发以及我跑了一个多月踩过的几个坑。这套东西不依赖某个特定平台每个组件都能替换单机就能跑适合想要低成本建立信息监控能力的小团队和个人。1. PanWatch的定位监控的不是账号而是信息流中的变化1.1 人工盯屏为什么必然漏很多问题靠人肉巡视是解决不了的。先说最直接的频率问题。人不可能保持7x24小时持续刷新晚上、周末、假期总会有空档而竞品调价、商品上线、负面内容传播往往就发生在空档里。我自己就有一次凌晨看到某平台出现集中差评第二天早上才被人转告等看到时用户讨论已经散到好几个地方了。其次是信息孤岛。你要盯的内容分散在不同平台每个平台有自己的搜索框、排序规则和推荐算法。今天在A平台按关键词筛一遍明天去B平台看热门榜后天再回A平台看评论区——这套流程不仅累而且没有办法做统一的口径和历史对比。单个平台内部的搜索是在这个平台里发生的但你要处理的问题是跨平台的光这一点人工就做不了。更隐蔽的问题是判断标准不稳定。同一个人状态好的时候觉得某条内容值得关注状态差的时候可能划过去就算了。而实际上你关心的并不是某一条帖子本身而是它的变化趋势昨天没人提今天突然有3个号在讨论价格一直是198今晚悄悄变成178。这些东西靠“刷一刷”很难捕捉到但监控系统可以因为它有阈值、有历史、有固定的计算逻辑。1.2 从“主动翻”到“被动收”的最小闭环所以PanWatch的设计目标从一开始就很明确做一个最小闭环把信息监控变成一条流水线。闭环的四个环节是采集、处理、预警、展示。采集层按配置定期从公开可用的信息源拉取内容统一成一条标准消息。处理层对消息做清洗、去重、相似度合并和热度打分。预警层按规则匹配命中后进入通知队列。展示层一个轻量看板记录历史、趋势、命中情况。这个闭环的核心原则是宁可漏报不可滥报。监控工具如果天天给你推几百条无关内容用户很快就不看了等于整个系统白做。所以规则匹配、热度阈值、静默期这些机制必须从一开始就考虑到而不是等上线了被通知轰炸了再补。这套系统本质上是一个“信息变化检测器”。它不做复杂的舆情分析不做语义理解就做一件事把多个源的变化及时告诉你。技术上一开始也不需要上大数据、离线计算这些重型武器单机加几个开源组件完全够用。2. 技术选型复盘每一层为什么选这个组件2.1 采集、处理、存储、分发四层架构刚开始做的时候很多人容易把项目写成一堆零散脚本一个抓A平台的一个抓B平台的一个发通知的互相之间靠手动触发。这样前两三个源还行源一多就失控了。PanWatch从一开始就按四层来规划采集层只负责拿数据处理层只负责算数据存储层只负责放数据分发层只负责送数据。各层之间通过标准数据结构连接互不干扰。这个分层带来的直接好处是某个源挂了不影响其他源处理的逻辑改了不用动采集的代码换通知渠道也只动分发层。后续想加一个“微博热榜监控”只是新增一个采集器其他管道原样复用。四层的具体交互是采集器产生统一格式的Item含来源、标题、正文、链接、发布时间、原始数据写入Redis队列处理器消费队列完成去重、打分命中的消息写入SQLite通知任务读取命中消息并推送看板读取SQLite做统计展示。2.2 组件的取舍理由选型上我没有追求新潮选了最省心的一套组合。模块选型主要理由开发语言Python 3.11抓取、解析、NLP生态齐全写脚本和写服务都很顺定时调度APScheduler支持cron表达式、持久化任务、可动态增删适合采集任务管理消息缓冲Redis队列处理速度和采集速度天然不匹配队列可以削峰也方便重启后恢复去重缓存Redis Set / 布隆过滤器高频判重必须走内存不能每次都查数据库归档存储SQLite单机写入量不大SQLite足够稳定备份也简单展示层FastAPI 简单模板不需要前端工程化一个服务端页面足够通知渠道企业微信群机器人 / 钉钉机器人 / 邮件Webhook直接可用延迟低不需要额外开发App为什么不用消息队列像Kafka或者RabbitMQ理由很简单数据量没到那个级别。PanWatch单机跑每秒处理几条消息就够了Redis队列足够应对还省了一套运维。为什么不用MySQL因为监控元数据量不大SQLite文件级管理更轻便日常备份直接拷贝文件就行。什么时候需要换等你接入了大量源、单日入库量超过百万条或者需要多机部署时再考虑迁移初期别在这个地方花时间。2.3 单机部署就够用先别上微服务我见过不少朋友做这类工具一上来就规划成“采集服务处理服务前端服务”三个微服务然后考虑Docker、K8s、CI/CD。作为一个内部监控工具这完全是把复杂度前置了。PanWatch目前的形态就是一台普通服务器上的几个常驻进程一个调度进程、一个处理进程、一个Web服务进程外加Redis和SQLite。总共不超过4个进程内存占用大概800MB不到跑一个月都没重启过。这也意味着如果你只是想自己用一台云服务器就能搞定想在本地局域网跑也行没有太多环境依赖。架构演进是有代价的每引入一个组件就多一个需要维护的东西。监控工具的硬指标不是并发能力而是“能不能稳定地持续采集和通知”。把单机版本跑扎实再根据真实瓶颈决定要不要扩展这才是正确顺序。3. 采集层实战别把信息源搞成一堆难以维护的脚本3.1 数据源盘点表先列清楚边界和成本动手写采集器之前先花半天时间把要盯的信息源列出来。这个步骤看起来简单实际上决定了后期大部分工作。我列几个典型的源类型方便参考信息源类型典型例子获取方式建议频率成本与注意点公开RSS新闻站、博客RSS解析15-30分钟最友好解析稳定优先接入官方API电商开放平台授权回调API30-60分钟有配额限制注意频率公开页面目标商品页、专题页页面解析30-120分钟结构可能变需要加解析容错搜索聚合站内搜索结果页查询参数拼接60分钟对频率最敏感容易触发验证Webhook订阅第三方推送被动接收实时需要暴露接收端口配置简单我强烈建议优先接RSS和Webhook这类“平台主动给你”的数据而不是自己写抓取脚本。“别人推给你”和“你去别人那拿”稳定性差一个量级。很多平台虽然有公开页面但结构说变就变脚本经常要跟着调RSS虽然形式老但一旦有基本是最稳的。合规层面也需要说明一下。这些都应该是平台公开提供的接口、RSS或页面抓取时遵循robots协议控制访问频率不突破登录限制、不做批量注册。监控工具是帮你看公开信息不是帮你去破解什么边界这个底线要清楚。3.2 一套统一的抓取任务调度信息源一多最怕的就是调度乱。A源每15分钟一次B源每小时一次C源只在工作时间跑每个频率写一个loop代码很快就乱了。PanWatch的做法是统一用APScheduler管理任务把每个源抽象成一个配置项而不是一段只属于某个脚本的定时逻辑。举个例子核心调度代码大概是这样的from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger scheduler BlockingScheduler() # 每15分钟拉取一次RSS源 scheduler.add_job( fetch_rss_sources, CronTrigger(minute*/15), idrss_job, coalesceTrue, max_instances1, ) # 每小时拉取一次商品价格监控 scheduler.add_job( fetch_price_sources, CronTrigger(minute5), idprice_job, coalesceTrue, max_instances1, ) # 每个工作日上午9点执行一次定时盘点 scheduler.add_job( daily_scan, CronTrigger(day_of_weekmon-fri, hour9, minute0), iddaily_scan, ) scheduler.start()这里有两个参数特别关键coalesceTrue和max_instances1。coalesce的作用是如果某个任务因为网络故障错过了执行时间下一次执行时就不要再补跑中间错过的所有轮次了只跑最新一次防止堆积。max_instances1是防止上一次没跑完下一次又启动同一个源同时跑两个任务这样既浪费资源又容易被目标平台判定为异常访问。对每个源我还加了一个简单的状态标记正常、降级、暂停。如果某个源连续几次抓取失败调度器会自动把它标记为降级把拉取频率从每15分钟降到每2小时连续失败超过6次的源直接暂停并推一条“采集源异常”的消息通知出来。这个机制很重要因为一个挂了8小时的源比一个偶尔失败的源要危险得多不能被静默忽略。3.3 适度并发和错误退避采集层的另一个细节是并发控制。很多人觉得抓取越快越好就上多线程甚至异步并发结果就是很容易触发平台的访问限制。PanWatch的默认策略是保守所有源的总并发不超过3个线程单源内部严格串行。每个请求之间加一个随机延迟范围在1到3秒之间。import time import random import requests def fetch_with_backoff(url, max_retries3): for attempt in range(max_retries): try: time.sleep(random.uniform(1, 3)) resp requests.get(url, timeout10) resp.raise_for_status() return resp.text except requests.RequestException as e: wait 2 ** attempt random.uniform(0, 1) print(f[retry {attempt1}] {url} failed: {e}, wait {wait:.1f}s) time.sleep(wait) return None退避策略用的是指数退避第一次失败等2秒左右第二次等4秒第三次等8秒依此类推。再加上每次请求都有的随机睡眠整体行为在目标平台看来就是“一个普通用户在慢速浏览”而不是“一个程序在刷数据”。我在实际运行中验证过控制住并发以后一个持续跑了两个多月的采集端只发生过一次需要人工处理的验证码情况。这个结果说明绝大多数拦截问题其实是频率问题不是识别问题。4. 处理层增量、去重和热度计算是怎么落到代码里的4.1 用Redis Set加布隆过滤器做“这条内容见过没有”采集层拿到数据之后处理层要做的第一件事就是判断“这条内容我是不是已经见过了”。听起来简单实际上很容易出错因为重复的路径比你想的多得多。同一个事件A平台有、B平台有同一篇文章被转载多个站点同一个商品在不同搜索结果里重复出现网络重试也可能导致同一条消息被推送两次。如果靠数据库读取来判断重复每条消息先查一遍SQLite数据量大了以后性能就是问题。PanWatch的做法是每条内容进入处理层时生成一个唯一ID规则是“标题归一化后的MD5值 链接MD5值 发布时间压缩到小时后的时间戳”。标题归一化就是去掉所有标点符号、全角半角转换、连续空格合并、英文转小写。关键源比如价格监控用Redis Set做精确去重ID存在Set里就跳过。一般源比如新闻RSS、公开讨论用布隆过滤器做近似去重因为偶尔漏掉一条重复内容可以接受但内存占用要小得多。布隆过滤器的原理可以理解成一个多格子的登记簿。每条新内容到了会根据一组哈希函数把登记簿上的几个格子都标记一下判断是否见过时只要检查这几个格子是不是都被标记了——如果有一个没标记那一定没见过如果都标记了也不一定百分百见过但概率足够高可以配合缓存或数据库再确认一次。代码上用Python的redis-py配合一个简单的位图实现就够了import redis import hashlib import math r redis.Redis.from_url(redis://localhost:6379/2) KEY panwatch:bloom:seen def seen(md5_id: str) - bool: # 用7个哈希位误判率控制在很低水平 positions [] for seed in range(7): h int(hashlib.md5(f{seed}:{md5_id}.encode()).hexdigest(), 16) positions.append(h % 100000) existed all(r.getbit(KEY, pos) for pos in positions) if not existed: for pos in positions: r.setbit(KEY, pos, 1) return existed这样做的好处是内存消耗极低。100万个ID用Redis位图只需要100万bit也就是125KB左右的存储这还是算上哈希位置上可能重复的情况。你完全不用担心监控跑一两个月后Redis内存爆炸。4.2 相似文本聚类判断“换了个标题又来”MD5去重只能处理完全相同的文本解决不了“换了个标题但内容基本一样”的转载问题。新闻类内容尤其明显同一个事件A站标题叫“某公司发布新品”B站标题叫“某公司新一代产品正式亮相”正文几乎一模一样。如果不去重你会收到两条几乎重复的预警体验很差也浪费规则命中名额。PanWatch的处理方案是结合MinHash做相似文本检测。原理不复杂先把正文切成若干个shingle滑动窗口分词窗口长度取5个词每个shingle做哈希取其中最小的10个哈希值作为内容的“指纹”。两条内容指纹重合比例高就认为相似。重合度超过75%时合并为同一条消息。def minhash_fingerprint(text: str, num_perm10) - set: tokens text.split() if len(tokens) 6: return set() shingles [] for i in range(len(tokens) - 4): shingle .join(tokens[i:i5]) shingles.append(hashlib.md5(shingle.encode()).hexdigest()) return set(sorted(shingles)[:num_perm]) def is_similar(fp1: set, fp2: set) - bool: if not fp1 or not fp2: return False jaccard len(fp1 fp2) / len(fp1 | fp2) return jaccard 0.75阈值0.75是我跑了几周数据调试出来的。太低会把不相关的文章合并在一起太高又会漏掉大部分转载。你可以先设0.7跑两天把自己觉得“这俩明显是同一件事”的样本捞出来看重合度再微调。4.3 热度评分排序比全量通知更省心去重之后PanWatch会给每条消息计算一个热度分这个分数决定了它有没有资格触发预警。热度分设计的核心思路是不是你命中了关键词就一定要推送而是要看这条内容值不值得打断你。我的评分公式是这样的score 命中词加权分 * 来源权重 新鲜度分 扩散信号分命中词加权分精确命中品牌词的 10命中“降价/涨价”类行为词的 8命中竞品词的 6。来源权重高价值源权重为1.0普通源为0.6低价值源为0.3。新鲜度分发布时间距现在越近越高按20 / (1 log(age_hours 1))递减。扩散信号分阅读量、回复数、转发数等有数值时按比例映射到0到15分。import math def hot_score(item, keyword_hits, source_weight, spread_signals): hit_score sum(10 * w for k, w in keyword_hits.items()) freshness 20 / (1 math.log(item[age_hours] 1, 2)) spread_score min(15, sum(signals.values()) / 10) if signals else 0 return hit_score * source_weight freshness spread_score这个分数的价值在于排序。相同的“降价”命中词出现在一个半小时前刚发布、回复数300的热帖里和出现在一周前、没有互动的静态页面里优先级完全不同。PanWatch的预警规则里有一个min_score字段只有超过这个分数的消息才会进入通知流程。这样你收到的通知基本都是有热度、有时效、真正需要你花费注意力的内容。5. 预警与看板把海量结果压缩成几条可执行动作5.1 规则引擎用配置别用硬编码处理层打完分之后下一步就是规则匹配。最早一版PanWatch把规则直接写死在代码里比如“如果标题包含某品牌词并且热度分大于50就推送”。这样改规则就要改代码、重启服务非常不灵活。后来我把规则抽成了YAML配置文件新规则上线只需要改配置文件再重载不用动一行代码。这里是一个规则配置的例子rules: - name: 竞品新品发布监测 match: type: all keywords: - 竞品品牌名 - 发布 - 新品 exclude_keywords: - 测评课程 - 招商加盟 min_score: 50 expiry_minutes: 120 cooldown_minutes: 1440 channels: - wecom_group - emailexclude_keywords是负向词表用来过滤大量无关内容。比如很多地方“某品牌发布”可能会出现“发布测评课程”这不是你要监控的信息加了负向词后就能直接滤掉。cooldown_minutes是静默期同一条规则命中一次之后24小时内不再重复推送避免被刷屏。5.2 通知渠道优先选Webhook省心又及时通知渠道的选择直接决定这套系统能不能坚持下去。我用过邮件、企业微信群机器人、钉钉群机器人最后稳定下来的方案是“企业微信群机器人为主、邮件兜底”。渠道实时性配置复杂度适合场景企业微信群机器人秒级极低只要一个Webhook地址日常预警主通知渠道钉钉群机器人秒级极低同上团队在钉钉时使用邮件SMTP分钟级中等要配置发件账号日报/周报聚合兜底通知页面站内消息手动查低低频、不重要内容群机器人这条链路实现起来真的只需要把JSON POST到Webhook地址几行代码就能搞定又能做到实时推送到手机是目前性价比最高的通知方式。要注意的是Webhook地址相当于这个群的入口别把它写进代码仓库放在环境变量或单独的配置文件里。5.3 一个查询直接出今天该看的列表为了不让自己在墙内框里迷失在总消息数里PanWatch的看板刻意做得很克制。没有跑马灯式的图表只有三个数据今日新增命中数、今日热榜TOP20、各规则命中统计。页面用FastAPI渲染一个模板就能完成SQLite直接查不需要额外服务。这里有一个我每天都用的查询SELECT title, source, score, rule_name, created_at FROM items WHERE date(created_at) date(now, localtime) AND matched_rule IS NOT NULL ORDER BY score DESC LIMIT 20;这个查询返回的就是今天值得你看的列表。每一条都带着规则名你一眼能知道它是“竞品降价”命中的还是“负面词”命中的不需要点进详情页。PanWatch的理念在这里体现得很清楚工具做得好不好不是看它收集了多少信息而是看它帮你省了多少信息处理的时间。6. 实测中的坑误报、封禁和重启后的增量丢失6.1 误报收敛别让规则太宽PanWatch上线第一周最大的灾难是通知轰炸。最初我把关键词设成“只匹配产品名”心想只要出现产品名就推送肯定不会漏。结果一天下来推了2000多条点进去三分之一是各种搬运号、营销号、用户自发闲聊真正有用的没几条。人一旦被打扰超过一定频次就开始忽略所有通知这个监控系统等于变成了一个高噪音渠道。后来我做了三件事把误报收敛下来规则改成多关键词同时命中而不是单个关键词命中。比如“商品名价格”“商品名修复”“商品名故障”这样能过滤掉大部分无关讨论。加负向词表。把“课程”“招商”“加盟”“下载”“资源”这类高频无关词统一放进去。给每条规则加min_score低于40分的不通知。让低来源权重、低时效的普通内容静默。收敛之后一天的通知量从2000多条降到了30条左右精准度反而上去了。规则宽了不漏但让人不看等于白做规则收紧了可能漏掉极个别边角料但系统整体可用。6.2 采集频率过高的教训有一段时间我贪心想尽快抓到竞品的价格变动把一个商品页源的抓取频率从每30分钟调到了每5分钟。跑了一天没事第二天开始出现验证码第三天这个源彻底被临时封禁。调回30分钟并加了退避之后两天才恢复正常。经验是抓取频率不是越高越好信息源上的内容变化有自己的节奏。商品价格一天改几次就算频繁了新闻源把频率放到15分钟就足够只有实时的行情类才需要高频率。所有抓取任务的默认值都应该是“够用就好”而不是“越快越好”。另外所有源都应该配置Retry-After等待时间。收到418、429这类状态码时就停止该源的一切抓取休眠指定时间再继续。不能像没事人一样继续重试。6.3 Redis重启后重复入库的坑还有一个坑藏得比较深Redis里的布隆过滤器默认不做持久化。有一次服务器重启Redis内存里的所有去重位图全部清空PanWatch处理层并不知道“这些内容已经见过了”结果把过去两天见过的所有消息又处理了一遍。幸好SQLite里有唯一索引兜底不然整个库会被重复数据淹掉。修复方案有三层Redis开启持久化RDB快照重启后能恢复位图数据。SQLite的历史表加UNIQUE索引重复消息入库时直接冲突跳过兜底防重复。布隆过滤器定期把已见ID导出到本地文件Redis清空后可重新加载。这个坑提醒我监控系统里所有缓存性质的东西都要想清楚“缓存丢了会发生什么”。如果后果不能接受那它就不应该只是缓存而应该有一个持久化副本。7. 扩展方向从监控工具变成决策助手7.1 趋势异常检测不只看单条消息还要看整体曲线单条消息的预警只是PanWatch的第一层能力。跑了一个月之后SQLite里积累了上万条历史命中记录这时候你就可以做一件人工根本做不了的事把“昨天新增了30条相关讨论”和“过去30天平均每天只有5条”放在一起比较。明显异常抬升背后往往意味着某个事件在发酵。我目前用一个很简单的移动平均加标准差的检测逻辑计算最近7天的日均命中数再看当天的命中数是否超过“均值2倍标准差”。超过就推一条“声量异常上升”的汇总提醒。这个方法不需要机器学习但已经足够发现大部分有明显的趋势变化。7.2 多订阅与分级通知如果你想把这套系统给团队用最简单的扩展是改成订阅制。每个人关注的关键词不同规则可以按订阅分组。运营盯推广效果客服盯投诉关键词产品盯功能讨论词各看各的规则互不干扰。通知渠道也可以按订阅配置重要规则推群机器人普通规则只进日报。日报、周报自动生成也是可以顺手做的。每天9点把昨天的命中情况聚合一下按“新增命中数”、“命中规则分布”、“热点内容Top10”三个板块通过邮件发出去这个不比单条实时通知累还有人更喜欢这种“不看实时只看总结”的使用方式。7.3 和业务系统联动要说PanWatch最有商业价值的方向我觉得是跟业务系统联动而不只是推消息。消息推给你你还得人工判断、再处理如果系统和工单、表格、内部数据库打通很多环节可以直接自动化。比如检测到某商品页面出现“缺货”状态直接调内部库存接口标记一下检测到差评关键词集中出现自动生成一条客服工单检测到竞品价格下调超过5%自动给负责运营的人发一封摘要邮件。这些实现起来其实就是给预测系统加一个Handler接口命中规则后除了发通知还可以执行一组自定义动作。PanWatch的处理管道本来就是解耦的加动作只需要在规则里加一个actions字段不需要改架构。这套联动能力做扎实之后工具就不再是“你看它的结果”而变成“它直接把结果变成业务动作”。监控工具的意义也就不再是让你少刷几个标签页那么简单了。跑了一个多月PanWatch最大的价值不是技术上的而是它替我把“什么时候该去关注”这个问题从模糊变成了精确。以前我时不时打开平台刷几下纯靠运气和焦虑驱动现在它告诉我今天有30条值得看的内容我花10分钟看完剩下的时间可以专注做别的事。如果你也要搭类似的监控工具我的建议是先把最小闭环跑起来再慢慢调规则别一开始就追求大而全。
全平台内容监控预警工具搭建实战：从采集到通知的完整方案
先说清楚PanWatch不是某个现成的商业软件是我自己搭的一套“全平台内容监控与预警工具”的名字。Pan取的是全景、泛化的意思Watch是观察——在多个信息源里同时盯住你关心的关键词、竞品动态和异常变化把散落在各处的消息统一清洗、打分、去重之后只把真正值得处理的东西推到面前而不是让你自己满世界去翻。这件事的背景做产品、运营、自媒体的朋友应该都懂总有些变化不是你主动搜就能看到的而是等你发现的时候已经发酵好一阵了。手动开十几个标签页逐个刷既慢又不稳定而且判断标准全凭个人状态。PanWatch要解决的问题很简单——把“我每天主动去看”变成“系统觉得有问题了才叫我”。下面把整个搭建过程拆开讲包括技术选型、采集设计、处理链路、预警分发以及我跑了一个多月踩过的几个坑。这套东西不依赖某个特定平台每个组件都能替换单机就能跑适合想要低成本建立信息监控能力的小团队和个人。1. PanWatch的定位监控的不是账号而是信息流中的变化1.1 人工盯屏为什么必然漏很多问题靠人肉巡视是解决不了的。先说最直接的频率问题。人不可能保持7x24小时持续刷新晚上、周末、假期总会有空档而竞品调价、商品上线、负面内容传播往往就发生在空档里。我自己就有一次凌晨看到某平台出现集中差评第二天早上才被人转告等看到时用户讨论已经散到好几个地方了。其次是信息孤岛。你要盯的内容分散在不同平台每个平台有自己的搜索框、排序规则和推荐算法。今天在A平台按关键词筛一遍明天去B平台看热门榜后天再回A平台看评论区——这套流程不仅累而且没有办法做统一的口径和历史对比。单个平台内部的搜索是在这个平台里发生的但你要处理的问题是跨平台的光这一点人工就做不了。更隐蔽的问题是判断标准不稳定。同一个人状态好的时候觉得某条内容值得关注状态差的时候可能划过去就算了。而实际上你关心的并不是某一条帖子本身而是它的变化趋势昨天没人提今天突然有3个号在讨论价格一直是198今晚悄悄变成178。这些东西靠“刷一刷”很难捕捉到但监控系统可以因为它有阈值、有历史、有固定的计算逻辑。1.2 从“主动翻”到“被动收”的最小闭环所以PanWatch的设计目标从一开始就很明确做一个最小闭环把信息监控变成一条流水线。闭环的四个环节是采集、处理、预警、展示。采集层按配置定期从公开可用的信息源拉取内容统一成一条标准消息。处理层对消息做清洗、去重、相似度合并和热度打分。预警层按规则匹配命中后进入通知队列。展示层一个轻量看板记录历史、趋势、命中情况。这个闭环的核心原则是宁可漏报不可滥报。监控工具如果天天给你推几百条无关内容用户很快就不看了等于整个系统白做。所以规则匹配、热度阈值、静默期这些机制必须从一开始就考虑到而不是等上线了被通知轰炸了再补。这套系统本质上是一个“信息变化检测器”。它不做复杂的舆情分析不做语义理解就做一件事把多个源的变化及时告诉你。技术上一开始也不需要上大数据、离线计算这些重型武器单机加几个开源组件完全够用。2. 技术选型复盘每一层为什么选这个组件2.1 采集、处理、存储、分发四层架构刚开始做的时候很多人容易把项目写成一堆零散脚本一个抓A平台的一个抓B平台的一个发通知的互相之间靠手动触发。这样前两三个源还行源一多就失控了。PanWatch从一开始就按四层来规划采集层只负责拿数据处理层只负责算数据存储层只负责放数据分发层只负责送数据。各层之间通过标准数据结构连接互不干扰。这个分层带来的直接好处是某个源挂了不影响其他源处理的逻辑改了不用动采集的代码换通知渠道也只动分发层。后续想加一个“微博热榜监控”只是新增一个采集器其他管道原样复用。四层的具体交互是采集器产生统一格式的Item含来源、标题、正文、链接、发布时间、原始数据写入Redis队列处理器消费队列完成去重、打分命中的消息写入SQLite通知任务读取命中消息并推送看板读取SQLite做统计展示。2.2 组件的取舍理由选型上我没有追求新潮选了最省心的一套组合。模块选型主要理由开发语言Python 3.11抓取、解析、NLP生态齐全写脚本和写服务都很顺定时调度APScheduler支持cron表达式、持久化任务、可动态增删适合采集任务管理消息缓冲Redis队列处理速度和采集速度天然不匹配队列可以削峰也方便重启后恢复去重缓存Redis Set / 布隆过滤器高频判重必须走内存不能每次都查数据库归档存储SQLite单机写入量不大SQLite足够稳定备份也简单展示层FastAPI 简单模板不需要前端工程化一个服务端页面足够通知渠道企业微信群机器人 / 钉钉机器人 / 邮件Webhook直接可用延迟低不需要额外开发App为什么不用消息队列像Kafka或者RabbitMQ理由很简单数据量没到那个级别。PanWatch单机跑每秒处理几条消息就够了Redis队列足够应对还省了一套运维。为什么不用MySQL因为监控元数据量不大SQLite文件级管理更轻便日常备份直接拷贝文件就行。什么时候需要换等你接入了大量源、单日入库量超过百万条或者需要多机部署时再考虑迁移初期别在这个地方花时间。2.3 单机部署就够用先别上微服务我见过不少朋友做这类工具一上来就规划成“采集服务处理服务前端服务”三个微服务然后考虑Docker、K8s、CI/CD。作为一个内部监控工具这完全是把复杂度前置了。PanWatch目前的形态就是一台普通服务器上的几个常驻进程一个调度进程、一个处理进程、一个Web服务进程外加Redis和SQLite。总共不超过4个进程内存占用大概800MB不到跑一个月都没重启过。这也意味着如果你只是想自己用一台云服务器就能搞定想在本地局域网跑也行没有太多环境依赖。架构演进是有代价的每引入一个组件就多一个需要维护的东西。监控工具的硬指标不是并发能力而是“能不能稳定地持续采集和通知”。把单机版本跑扎实再根据真实瓶颈决定要不要扩展这才是正确顺序。3. 采集层实战别把信息源搞成一堆难以维护的脚本3.1 数据源盘点表先列清楚边界和成本动手写采集器之前先花半天时间把要盯的信息源列出来。这个步骤看起来简单实际上决定了后期大部分工作。我列几个典型的源类型方便参考信息源类型典型例子获取方式建议频率成本与注意点公开RSS新闻站、博客RSS解析15-30分钟最友好解析稳定优先接入官方API电商开放平台授权回调API30-60分钟有配额限制注意频率公开页面目标商品页、专题页页面解析30-120分钟结构可能变需要加解析容错搜索聚合站内搜索结果页查询参数拼接60分钟对频率最敏感容易触发验证Webhook订阅第三方推送被动接收实时需要暴露接收端口配置简单我强烈建议优先接RSS和Webhook这类“平台主动给你”的数据而不是自己写抓取脚本。“别人推给你”和“你去别人那拿”稳定性差一个量级。很多平台虽然有公开页面但结构说变就变脚本经常要跟着调RSS虽然形式老但一旦有基本是最稳的。合规层面也需要说明一下。这些都应该是平台公开提供的接口、RSS或页面抓取时遵循robots协议控制访问频率不突破登录限制、不做批量注册。监控工具是帮你看公开信息不是帮你去破解什么边界这个底线要清楚。3.2 一套统一的抓取任务调度信息源一多最怕的就是调度乱。A源每15分钟一次B源每小时一次C源只在工作时间跑每个频率写一个loop代码很快就乱了。PanWatch的做法是统一用APScheduler管理任务把每个源抽象成一个配置项而不是一段只属于某个脚本的定时逻辑。举个例子核心调度代码大概是这样的from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger scheduler BlockingScheduler() # 每15分钟拉取一次RSS源 scheduler.add_job( fetch_rss_sources, CronTrigger(minute*/15), idrss_job, coalesceTrue, max_instances1, ) # 每小时拉取一次商品价格监控 scheduler.add_job( fetch_price_sources, CronTrigger(minute5), idprice_job, coalesceTrue, max_instances1, ) # 每个工作日上午9点执行一次定时盘点 scheduler.add_job( daily_scan, CronTrigger(day_of_weekmon-fri, hour9, minute0), iddaily_scan, ) scheduler.start()这里有两个参数特别关键coalesceTrue和max_instances1。coalesce的作用是如果某个任务因为网络故障错过了执行时间下一次执行时就不要再补跑中间错过的所有轮次了只跑最新一次防止堆积。max_instances1是防止上一次没跑完下一次又启动同一个源同时跑两个任务这样既浪费资源又容易被目标平台判定为异常访问。对每个源我还加了一个简单的状态标记正常、降级、暂停。如果某个源连续几次抓取失败调度器会自动把它标记为降级把拉取频率从每15分钟降到每2小时连续失败超过6次的源直接暂停并推一条“采集源异常”的消息通知出来。这个机制很重要因为一个挂了8小时的源比一个偶尔失败的源要危险得多不能被静默忽略。3.3 适度并发和错误退避采集层的另一个细节是并发控制。很多人觉得抓取越快越好就上多线程甚至异步并发结果就是很容易触发平台的访问限制。PanWatch的默认策略是保守所有源的总并发不超过3个线程单源内部严格串行。每个请求之间加一个随机延迟范围在1到3秒之间。import time import random import requests def fetch_with_backoff(url, max_retries3): for attempt in range(max_retries): try: time.sleep(random.uniform(1, 3)) resp requests.get(url, timeout10) resp.raise_for_status() return resp.text except requests.RequestException as e: wait 2 ** attempt random.uniform(0, 1) print(f[retry {attempt1}] {url} failed: {e}, wait {wait:.1f}s) time.sleep(wait) return None退避策略用的是指数退避第一次失败等2秒左右第二次等4秒第三次等8秒依此类推。再加上每次请求都有的随机睡眠整体行为在目标平台看来就是“一个普通用户在慢速浏览”而不是“一个程序在刷数据”。我在实际运行中验证过控制住并发以后一个持续跑了两个多月的采集端只发生过一次需要人工处理的验证码情况。这个结果说明绝大多数拦截问题其实是频率问题不是识别问题。4. 处理层增量、去重和热度计算是怎么落到代码里的4.1 用Redis Set加布隆过滤器做“这条内容见过没有”采集层拿到数据之后处理层要做的第一件事就是判断“这条内容我是不是已经见过了”。听起来简单实际上很容易出错因为重复的路径比你想的多得多。同一个事件A平台有、B平台有同一篇文章被转载多个站点同一个商品在不同搜索结果里重复出现网络重试也可能导致同一条消息被推送两次。如果靠数据库读取来判断重复每条消息先查一遍SQLite数据量大了以后性能就是问题。PanWatch的做法是每条内容进入处理层时生成一个唯一ID规则是“标题归一化后的MD5值 链接MD5值 发布时间压缩到小时后的时间戳”。标题归一化就是去掉所有标点符号、全角半角转换、连续空格合并、英文转小写。关键源比如价格监控用Redis Set做精确去重ID存在Set里就跳过。一般源比如新闻RSS、公开讨论用布隆过滤器做近似去重因为偶尔漏掉一条重复内容可以接受但内存占用要小得多。布隆过滤器的原理可以理解成一个多格子的登记簿。每条新内容到了会根据一组哈希函数把登记簿上的几个格子都标记一下判断是否见过时只要检查这几个格子是不是都被标记了——如果有一个没标记那一定没见过如果都标记了也不一定百分百见过但概率足够高可以配合缓存或数据库再确认一次。代码上用Python的redis-py配合一个简单的位图实现就够了import redis import hashlib import math r redis.Redis.from_url(redis://localhost:6379/2) KEY panwatch:bloom:seen def seen(md5_id: str) - bool: # 用7个哈希位误判率控制在很低水平 positions [] for seed in range(7): h int(hashlib.md5(f{seed}:{md5_id}.encode()).hexdigest(), 16) positions.append(h % 100000) existed all(r.getbit(KEY, pos) for pos in positions) if not existed: for pos in positions: r.setbit(KEY, pos, 1) return existed这样做的好处是内存消耗极低。100万个ID用Redis位图只需要100万bit也就是125KB左右的存储这还是算上哈希位置上可能重复的情况。你完全不用担心监控跑一两个月后Redis内存爆炸。4.2 相似文本聚类判断“换了个标题又来”MD5去重只能处理完全相同的文本解决不了“换了个标题但内容基本一样”的转载问题。新闻类内容尤其明显同一个事件A站标题叫“某公司发布新品”B站标题叫“某公司新一代产品正式亮相”正文几乎一模一样。如果不去重你会收到两条几乎重复的预警体验很差也浪费规则命中名额。PanWatch的处理方案是结合MinHash做相似文本检测。原理不复杂先把正文切成若干个shingle滑动窗口分词窗口长度取5个词每个shingle做哈希取其中最小的10个哈希值作为内容的“指纹”。两条内容指纹重合比例高就认为相似。重合度超过75%时合并为同一条消息。def minhash_fingerprint(text: str, num_perm10) - set: tokens text.split() if len(tokens) 6: return set() shingles [] for i in range(len(tokens) - 4): shingle .join(tokens[i:i5]) shingles.append(hashlib.md5(shingle.encode()).hexdigest()) return set(sorted(shingles)[:num_perm]) def is_similar(fp1: set, fp2: set) - bool: if not fp1 or not fp2: return False jaccard len(fp1 fp2) / len(fp1 | fp2) return jaccard 0.75阈值0.75是我跑了几周数据调试出来的。太低会把不相关的文章合并在一起太高又会漏掉大部分转载。你可以先设0.7跑两天把自己觉得“这俩明显是同一件事”的样本捞出来看重合度再微调。4.3 热度评分排序比全量通知更省心去重之后PanWatch会给每条消息计算一个热度分这个分数决定了它有没有资格触发预警。热度分设计的核心思路是不是你命中了关键词就一定要推送而是要看这条内容值不值得打断你。我的评分公式是这样的score 命中词加权分 * 来源权重 新鲜度分 扩散信号分命中词加权分精确命中品牌词的 10命中“降价/涨价”类行为词的 8命中竞品词的 6。来源权重高价值源权重为1.0普通源为0.6低价值源为0.3。新鲜度分发布时间距现在越近越高按20 / (1 log(age_hours 1))递减。扩散信号分阅读量、回复数、转发数等有数值时按比例映射到0到15分。import math def hot_score(item, keyword_hits, source_weight, spread_signals): hit_score sum(10 * w for k, w in keyword_hits.items()) freshness 20 / (1 math.log(item[age_hours] 1, 2)) spread_score min(15, sum(signals.values()) / 10) if signals else 0 return hit_score * source_weight freshness spread_score这个分数的价值在于排序。相同的“降价”命中词出现在一个半小时前刚发布、回复数300的热帖里和出现在一周前、没有互动的静态页面里优先级完全不同。PanWatch的预警规则里有一个min_score字段只有超过这个分数的消息才会进入通知流程。这样你收到的通知基本都是有热度、有时效、真正需要你花费注意力的内容。5. 预警与看板把海量结果压缩成几条可执行动作5.1 规则引擎用配置别用硬编码处理层打完分之后下一步就是规则匹配。最早一版PanWatch把规则直接写死在代码里比如“如果标题包含某品牌词并且热度分大于50就推送”。这样改规则就要改代码、重启服务非常不灵活。后来我把规则抽成了YAML配置文件新规则上线只需要改配置文件再重载不用动一行代码。这里是一个规则配置的例子rules: - name: 竞品新品发布监测 match: type: all keywords: - 竞品品牌名 - 发布 - 新品 exclude_keywords: - 测评课程 - 招商加盟 min_score: 50 expiry_minutes: 120 cooldown_minutes: 1440 channels: - wecom_group - emailexclude_keywords是负向词表用来过滤大量无关内容。比如很多地方“某品牌发布”可能会出现“发布测评课程”这不是你要监控的信息加了负向词后就能直接滤掉。cooldown_minutes是静默期同一条规则命中一次之后24小时内不再重复推送避免被刷屏。5.2 通知渠道优先选Webhook省心又及时通知渠道的选择直接决定这套系统能不能坚持下去。我用过邮件、企业微信群机器人、钉钉群机器人最后稳定下来的方案是“企业微信群机器人为主、邮件兜底”。渠道实时性配置复杂度适合场景企业微信群机器人秒级极低只要一个Webhook地址日常预警主通知渠道钉钉群机器人秒级极低同上团队在钉钉时使用邮件SMTP分钟级中等要配置发件账号日报/周报聚合兜底通知页面站内消息手动查低低频、不重要内容群机器人这条链路实现起来真的只需要把JSON POST到Webhook地址几行代码就能搞定又能做到实时推送到手机是目前性价比最高的通知方式。要注意的是Webhook地址相当于这个群的入口别把它写进代码仓库放在环境变量或单独的配置文件里。5.3 一个查询直接出今天该看的列表为了不让自己在墙内框里迷失在总消息数里PanWatch的看板刻意做得很克制。没有跑马灯式的图表只有三个数据今日新增命中数、今日热榜TOP20、各规则命中统计。页面用FastAPI渲染一个模板就能完成SQLite直接查不需要额外服务。这里有一个我每天都用的查询SELECT title, source, score, rule_name, created_at FROM items WHERE date(created_at) date(now, localtime) AND matched_rule IS NOT NULL ORDER BY score DESC LIMIT 20;这个查询返回的就是今天值得你看的列表。每一条都带着规则名你一眼能知道它是“竞品降价”命中的还是“负面词”命中的不需要点进详情页。PanWatch的理念在这里体现得很清楚工具做得好不好不是看它收集了多少信息而是看它帮你省了多少信息处理的时间。6. 实测中的坑误报、封禁和重启后的增量丢失6.1 误报收敛别让规则太宽PanWatch上线第一周最大的灾难是通知轰炸。最初我把关键词设成“只匹配产品名”心想只要出现产品名就推送肯定不会漏。结果一天下来推了2000多条点进去三分之一是各种搬运号、营销号、用户自发闲聊真正有用的没几条。人一旦被打扰超过一定频次就开始忽略所有通知这个监控系统等于变成了一个高噪音渠道。后来我做了三件事把误报收敛下来规则改成多关键词同时命中而不是单个关键词命中。比如“商品名价格”“商品名修复”“商品名故障”这样能过滤掉大部分无关讨论。加负向词表。把“课程”“招商”“加盟”“下载”“资源”这类高频无关词统一放进去。给每条规则加min_score低于40分的不通知。让低来源权重、低时效的普通内容静默。收敛之后一天的通知量从2000多条降到了30条左右精准度反而上去了。规则宽了不漏但让人不看等于白做规则收紧了可能漏掉极个别边角料但系统整体可用。6.2 采集频率过高的教训有一段时间我贪心想尽快抓到竞品的价格变动把一个商品页源的抓取频率从每30分钟调到了每5分钟。跑了一天没事第二天开始出现验证码第三天这个源彻底被临时封禁。调回30分钟并加了退避之后两天才恢复正常。经验是抓取频率不是越高越好信息源上的内容变化有自己的节奏。商品价格一天改几次就算频繁了新闻源把频率放到15分钟就足够只有实时的行情类才需要高频率。所有抓取任务的默认值都应该是“够用就好”而不是“越快越好”。另外所有源都应该配置Retry-After等待时间。收到418、429这类状态码时就停止该源的一切抓取休眠指定时间再继续。不能像没事人一样继续重试。6.3 Redis重启后重复入库的坑还有一个坑藏得比较深Redis里的布隆过滤器默认不做持久化。有一次服务器重启Redis内存里的所有去重位图全部清空PanWatch处理层并不知道“这些内容已经见过了”结果把过去两天见过的所有消息又处理了一遍。幸好SQLite里有唯一索引兜底不然整个库会被重复数据淹掉。修复方案有三层Redis开启持久化RDB快照重启后能恢复位图数据。SQLite的历史表加UNIQUE索引重复消息入库时直接冲突跳过兜底防重复。布隆过滤器定期把已见ID导出到本地文件Redis清空后可重新加载。这个坑提醒我监控系统里所有缓存性质的东西都要想清楚“缓存丢了会发生什么”。如果后果不能接受那它就不应该只是缓存而应该有一个持久化副本。7. 扩展方向从监控工具变成决策助手7.1 趋势异常检测不只看单条消息还要看整体曲线单条消息的预警只是PanWatch的第一层能力。跑了一个月之后SQLite里积累了上万条历史命中记录这时候你就可以做一件人工根本做不了的事把“昨天新增了30条相关讨论”和“过去30天平均每天只有5条”放在一起比较。明显异常抬升背后往往意味着某个事件在发酵。我目前用一个很简单的移动平均加标准差的检测逻辑计算最近7天的日均命中数再看当天的命中数是否超过“均值2倍标准差”。超过就推一条“声量异常上升”的汇总提醒。这个方法不需要机器学习但已经足够发现大部分有明显的趋势变化。7.2 多订阅与分级通知如果你想把这套系统给团队用最简单的扩展是改成订阅制。每个人关注的关键词不同规则可以按订阅分组。运营盯推广效果客服盯投诉关键词产品盯功能讨论词各看各的规则互不干扰。通知渠道也可以按订阅配置重要规则推群机器人普通规则只进日报。日报、周报自动生成也是可以顺手做的。每天9点把昨天的命中情况聚合一下按“新增命中数”、“命中规则分布”、“热点内容Top10”三个板块通过邮件发出去这个不比单条实时通知累还有人更喜欢这种“不看实时只看总结”的使用方式。7.3 和业务系统联动要说PanWatch最有商业价值的方向我觉得是跟业务系统联动而不只是推消息。消息推给你你还得人工判断、再处理如果系统和工单、表格、内部数据库打通很多环节可以直接自动化。比如检测到某商品页面出现“缺货”状态直接调内部库存接口标记一下检测到差评关键词集中出现自动生成一条客服工单检测到竞品价格下调超过5%自动给负责运营的人发一封摘要邮件。这些实现起来其实就是给预测系统加一个Handler接口命中规则后除了发通知还可以执行一组自定义动作。PanWatch的处理管道本来就是解耦的加动作只需要在规则里加一个actions字段不需要改架构。这套联动能力做扎实之后工具就不再是“你看它的结果”而变成“它直接把结果变成业务动作”。监控工具的意义也就不再是让你少刷几个标签页那么简单了。跑了一个多月PanWatch最大的价值不是技术上的而是它替我把“什么时候该去关注”这个问题从模糊变成了精确。以前我时不时打开平台刷几下纯靠运气和焦虑驱动现在它告诉我今天有30条值得看的内容我花10分钟看完剩下的时间可以专注做别的事。如果你也要搭类似的监控工具我的建议是先把最小闭环跑起来再慢慢调规则别一开始就追求大而全。
