我有一个习惯就是会同时盯好几个数据平台的动态。有时候是行业资讯列表更新了有时候是某个商品详情页的参数变了还有时候是热门榜单的排名在跳动。以前靠人工刷新一天几十次累不说还总漏。更让人难受的是平台“什么时候变的、变了什么、趋势如何”完全没有沉淀全凭脑子里的模糊印象。所以两个多星期前我动手写了 PLFM_RADAR。它本质上是一套面向多平台的数据变化雷达系统定时探测目标地址抓取内容后做结构化解析再用相似度算法识别出“到底哪里变了”最后把变化事件、趋势曲线和告警统一呈现在一个极简仪表盘里。从设计到跑通花了两周目前连续运行了十几天稳定捕获了几百条有效变更记录。这篇文章会把 PLFM_RADAR 从需求分析、整体架构、探测器实现、变化检测算法到部署调优完整拆一遍适合正在做信息聚合、数据监控或者单纯想了解“如何感知一个平台动态”的朋友参考。文中的代码都是实际跑过的目录结构和核心思路可以直接抄。1. 我为什么最终决定自建 PLFM_RADAR而不是买现成监控服务1.1 项目要解决的真实痛点PLFM_RADAR 里 PLFM 就是 Platform 的简写RADAR 对应雷达的语义持续扫描、发现目标、报告变化。这里的“平台”范围很宽可以是一个资讯站的列表页可以是一个商品详情页也可以是一个实时热度榜。它们的共性是信息一直在变但变的方式截然不同。我整理了三个实际遇到的场景场景A某商品详情页价格每周调整一次调整时会连带促销文案一起改。我需要知道它什么时候变的、从多少变成多少。场景B一个行业资讯列表每天不定时更新最多一天更新五六次。我关心的是新增了哪几条、哪几条被撤下。场景C一个热门榜单排名每分钟都可能跳动。我不需要逐条告警只想看到半小时级别的趋势曲线。这三个场景如果分别找现成工具会非常割裂。价格监控工具只能盯价格RSS 订阅只能做更新提醒榜单监控基本没有通用方案。把三者统一到一个系统里就是我决定自建的根本原因。换句话说我要的不是“某个字段有没有变”而是“一批平台各自发生了什么样的结构化变化”。1.2 现成监控服务的三个让人不满的地方有人会说GitHub 上现成的监控项目一抓一大把何必自己写。我确实认真看了一圈也试用过几个最终没有直接拿来用原因有三个。第一配置成本不低。大多数现成方案对监控目标的抽象是“URL 关键词”只能告诉你页面上某个字符串是否出现了。但我想监控的是结构化字段比如“价格1999库存有货标题xxx”字段级别的新旧对比在现成工具里基本做不到。第二变化识别太粗糙。很多工具只会告诉你“变了”或“没变”然后丢给你两张整页截图。截图对人工确认有用但没法自动统计也没法喂给下游程序。我想要的是“标题从 A 变成了 B、价格从 1899 变成了 1999”这种精确差异。第三数据不归我。免费工具的数据存在别人服务器上私有化部署版本要么贵要么维护不积极。PLFM_RADAR 的所有数据都落在本地 SQLite 里随时导出、分析、二次处理。这三条每条都踩在我的实际使用场景上所以结论很明确自己写一个把灵活性完全抓在自己手里。同时我也给自己定了三条边界不追求万能只解决我真实会用的场景探测要轻量不能给目标服务造成压力系统要能在一台低配小主机上跑起来。这些边界直接决定了后面的架构选型。2. PLFM_RADAR 的整体设计四层结构各干各的2.1 探测层插件化探测器PLFM_RADAR 的整体结构分成四层探测层、分析层、存储层、展示层另外还有一个贯穿所有层的调度器。探测层是最贴近数据源的地方它的职责非常单纯按照配置去抓取目标内容然后解析成结构化数据。我把它做成了插件化的思路因为不同平台的页面结构差异太大。商品详情页需要提取价格、标题、库存资讯列表需要提取每条新闻的标题、链接、发布时间榜单需要提取排名、名称、热度值。如果所有人都用一套通用解析逻辑代码会变成一团乱麻。插件化之后每个平台对应一个探测器插件插件只负责“如何从这段 HTML 里提取字段”。探测层统一对外暴露一个接口输入是 URL 和解析配置输出是一个结构化的 JSON 对象。这样调度器、分析层完全不用关心具体平台长什么样。模块对应的目录结构大致是这样plfm_radar/ ├── detectors/ # 探测器插件 │ ├── base.py # 插件基类 │ ├── http_detector.py │ ├── detail_page.py # 商品详情页专用 │ ├── list_page.py # 列表页专用 │ └── rank_page.py # 榜单页专用 ├── analyzer/ │ ├── fingerprint.py # SimHash 指纹 │ ├── diff.py # 结构化差异 │ └── noise.py # 误报过滤 ├── storage/ │ ├── db.py # SQLite 读写 │ └── models.py # 数据模型 ├── scheduler/ │ └── scheduler.py # 调度器 ├── web/ │ ├── app.py # 仪表盘后端 │ └── templates/ # 前端页面 └── config.yaml # 全局配置2.2 分析层变化识别与趋势计算分析层是整颗雷达的“大脑”。探测层拿到的是原始快照也就是某个时刻平台的一组字段值但原始快照本身没有意义只有拿它和上一次的快照做对比才能产生“变化”这个信息。变化识别分两步走。第一步是快速判断“整页内容到底变没变”这一步我用 SimHash 指纹来做后面会详细讲。如果指纹相似度极高说明内容基本没动直接跳过。第二步是结构化差异提取也就是对两版 JSON 逐字段比较输出一个统一的差异事件格式类似“字段路径 旧值 新值 变更时间”。趋势计算则是基于时间维度的累积把每次探测得到的数值型字段比如价格、热度值、排名按时间顺序记录下来自然就能画出曲线。这里不需要复杂的数学算法最核心的就是要有一个干净的时序数据表。2.3 存储层快照加事件双写存储层我选了 SQLite没有上 MySQL 或者 PostgreSQL。原因很简单PLFM_RADAR 是单机应用数据量在百万条以内SQLite 完全够用而且零运维、备份就是一个文件。唯一要提前规划的是写入频率——高峰时段探测任务可能几十个并发SQLite 对并发写支持一般所以我用了一个技巧写入全部走单线程队列内存里先聚合再批量写库。存储模型是“快照 事件”双写。快照表保存每次探测的完整结果用于回溯事件表只保存变化的部分用于告警和统计。这两张表是独立的免得每次查告警历史都要去翻几十万条快照。-- 快照表保存每次探测的原始结构化结果 CREATE TABLE snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_id INTEGER NOT NULL, fetched_at DATETIME NOT NULL, content_hash TEXT NOT NULL, payload_json TEXT NOT NULL ); -- 事件表只记录变化 CREATE TABLE change_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_id INTEGER NOT NULL, field_path TEXT NOT NULL, old_value TEXT, new_value TEXT, occurred_at DATETIME NOT NULL ); -- 时序表记录数值型字段随时间的取值 CREATE TABLE time_series ( target_id INTEGER NOT NULL, field_path TEXT NOT NULL, value REAL NOT NULL, captured_at DATETIME NOT NULL );2.4 展示层极简仪表盘展示层我没有做得很花哨一个页面三块区域最新变化事件流、各目标的趋势曲线、告警历史。技术上就是 Python 的 http.server 加原生 HTML 和 Charts.js没有任何重量级框架。之所以这样选是因为这个系统的核心价值在探测和分析展示层只要清晰、能一眼看出发生了什么就够了。最新变化事件流按时间倒序每条事件显示目标名称、字段路径、旧值到新值的箭头以及发生时间。趋势曲线可以切换时间范围从“最近6小时”到“最近7天”。告警历史则单独列出所有触发过通知的变化方便事后追溯。3. 从零手写探测器核心代码与实现细节3.1 探测项的数据结构定义探测器插件的核心数据结构叫DetectItem它描述了一个监控目标的所有信息。字段不能太少否则没法统一调度也不能太多否则插件写起来很累。最终我定为八个字段。# detectors/base.py from dataclasses import dataclass, field from typing import Any, Dict, Optional dataclass class DetectItem: target_id: str # 目标唯一 ID name: str # 目标名称用于展示 url: str # 要探测的地址 plugin: str # 使用的探测器插件名称 interval: int 300 # 探测间隔秒 extractor_config: Dict[str, Any] field(default_factorydict) headers: Dict[str, str] field(default_factorydict) enabled: bool True每个字段都有自己的作用。target_id是全局唯一的所有表都用它做外键。name是给人看的告警消息里会带出来。url是探测地址。plugin决定用哪个探测器插件。interval是该目标的探测频率覆盖全局默认值。extractor_config是插件自定义的解析配置。headers有时必须带因为部分平台对默认 User-Agent 不友好。enabled用于临时停掉某个目标而不用删除配置。3.2 编写一个最简单的 HTTP 探测器所有探测器的基类非常薄只做两件事发起请求、调用子类的解析方法。# detectors/base.py import requests from abc import ABC, abstractmethod from typing import Any, Dict class BaseDetector(ABC): def __init__(self, item: DetectItem): self.item item self.session requests.Session() # 默认模拟常规浏览器的 User-Agent self.session.headers.update({ User-Agent: ( Mozilla/5.0 (Linux; Android 11) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Mobile Safari/537.36 ) }) if item.headers: self.session.headers.update(item.headers) def fetch(self) - Dict[str, Any]: resp self.session.get(self.item.url, timeout15) resp.raise_for_status() return self.parse(resp.text) abstractmethod def parse(self, html: str) - Dict[str, Any]: 子类实现 HTML 到结构化字段的解析 raise NotImplementedErrorfetch方法统一处理网络请求和异常parse由各个插件实现。这样设计有个显而易见的好处后续如果要给所有探测器加代理、加请求重试、加超时控制只需要改基类。第一个完整插件我写的是通用 JSON 探测器。有些平台会提供接口直接返回 JSON 数据这种最省事只要配置好 JSONPath 就能提取字段。# detectors/http_detector.py import json import re from typing import Any, Dict from .base import BaseDetector class HttpJsonDetector(BaseDetector): 通用 JSON 接口探测器从响应中提取字段。 def parse(self, html: str) - Dict[str, Any]: data json.loads(html) result {} # extractor_config 形如 {price: $.data.price, title: $.data.title} for field, jsonpath in self.item.extractor_config.items(): result[field] self._extract_by_path(data, jsonpath) return result def _extract_by_path(self, data: Any, path: str) - Any: # 简化版 JSONPath只支持 $ 开头的点分路径 parts path.lstrip($.).split(.) cur data for part in parts: if isinstance(cur, dict) and part in cur: cur cur[part] else: return None return cur3.3 解析规则从正则到 CSS 选择器JSON 接口虽然省事但现实中大量目标返回的还是 HTML。解析 HTML 我优先推荐 CSS 选择器而不是正则。正则看着灵活但页面结构稍微一变就崩而且写起来难读、难维护。CSS 选择器配合 Python 的 BeautifulSoup既稳定又好写。商品详情页插件就是一个典型例子# detectors/detail_page.py from bs4 import BeautifulSoup from typing import Any, Dict from .base import BaseDetector class DetailPageDetector(BaseDetector): 商品详情页探测器提取标题、价格、库存等字段。 def parse(self, html: str) - Dict[str, Any]: soup BeautifulSoup(html, html.parser) result {} cfg self.item.extractor_config # 标题 sel cfg.get(title_selector, h1.product-title) node soup.select_one(sel) result[title] node.get_text(stripTrue) if node else None # 价格很多页面价格有人民币符号需要清洗 sel cfg.get(price_selector, span.price) node soup.select_one(sel) raw_price node.get_text(stripTrue) if node else None result[price] self._clean_price(raw_price) # 库存状态 sel cfg.get(stock_selector, span.stock-status) node soup.select_one(sel) result[in_stock] node is not None and 缺货 not in node.get_text() return result def _clean_price(self, raw: str | None) - float | None: if not raw: return None # 去掉货币符号和千分位逗号只保留数字和小数点 cleaned re.sub(r[^\d.], , raw) try: return float(cleaned) except ValueError: return None像这种价格清洗逻辑如果没有单独拆出来后面每个页面插件都要重复实现一遍。放在基类或者工具模块里是更合理的归属。这里也建议把页面解析写成一个独立函数方便单测覆盖。4. 变化检测是雷达的“大脑”算法与去重策略4.1 文本指纹用 SimHash 做相似度比较变化检测最忌讳的一件事是页面明明只改了一个时间戳或者一个访问计数值系统却当成一次重大变化来告警。所以先做“整体相似度判断”非常关键。我用 SimHash 作为文本指纹算法。它的特点是两段文本越相似它们的 SimHash 值海明距离就越小。这样我不用精确比较整段 HTML只需要计算指纹然后统计二进制位中不同位的数量。# analyzer/fingerprint.py import jieba from hashlib import md5 class SimHash: def __init__(self, bits: int 64): self.bits bits def _hash_token(self, token: str) - int: # 用 MD5 作为词级别的哈希截断到指定位数 h md5(token.encode(utf-8)).hexdigest() return int(h[:16], 16) ((1 self.bits) - 1) def compute(self, text: str) - int: # 分词统计词频加权求和后归一化为二进制 tokens jieba.lcut(text) weights [0] * self.bits for token in tokens: h self._hash_token(token) for i in range(self.bits): bit (h i) 1 weights[i] 1 if bit else -1 fingerprint 0 for i in range(self.bits): if weights[i] 0: fingerprint | (1 i) return fingerprint def hamming_distance(self, a: int, b: int) - int: return bin(a ^ b).count(1)实际调用时我会把页面文本做简单归一化去掉所有 HTML 标签、去掉空白字符、统一转成小写再送入 SimHash。归一化这一步很重要它能过滤掉因为格式化空行、标签属性顺序变化造成的“伪差异”。判断是否变化的逻辑很直接hamming_distance 3视为无变化大于 3 才进入结构化差异提取。这个阈值是我在真实数据上试出来的64 位指纹下3 以内基本都是排版噪声超过 3 就说明实质内容确实动了。4.2 结构化差异提取检测新增、删除、修改指纹判断只是第一步它告诉你“变了”但没告诉你“哪里变了”。结构化差异提取要做的就是精确到字段的对比。因为我们的快照已经是 JSON 格式差异提取就变成了递归遍历两个字典的工作。# analyzer/diff.py from typing import Any, Dict, List, Tuple def extract_diff(old: Dict[str, Any], new: Dict[str, Any], path: str ) - List[Dict[str, Any]]: 递归比较两个 JSON 对象返回差异事件列表。 events [] all_keys set(old.keys()) | set(new.keys()) for key in all_keys: cur_path f{path}.{key} if path else key old_val old.get(key) new_val new.get(key) if old_val is None and new_val is not None: events.append({ field_path: cur_path, old_value: None, new_value: new_val, type: added }) elif old_val is not None and new_val is None: events.append({ field_path: cur_path, old_value: old_val, new_value: None, type: removed }) elif isinstance(old_val, dict) and isinstance(new_val, dict): events.extend(extract_diff(old_val, new_val, cur_path)) elif old_val ! new_val: events.append({ field_path: cur_path, old_value: old_val, new_value: new_val, type: modified }) return events这段代码的巧妙之处在于递归处理嵌套结构。榜单场景下一个条目本身就是一个字典比如{rank: 1, name: xxx, hot: 1234}如果榜单有 50 个条目整体就是一个大数组。数组不能直接用字典逻辑比较我在实际项目中额外加了一层“列表按主键对齐”的处理先按name或id字段建索引再逐个比较。对于榜单类数据单纯输出“第 3 名热度从 1200 变成 1500”还不够我还需要知道“哪些条目是新进入榜单的”。所以列表比较时分三步找出新增条目、找出消失条目、找出保留下来的条目的字段变化。4.3 误报过滤动态区域与版本号噪声光有算法还不够真实网页里有一类东西几乎每次都会变但对你毫无价值。最典型的是三种页面底部版权年份、网站生成的随机 token、访问统计数字。如果不对这些做过滤告警量会爆炸。我的方案是给每个探测器插件加一个“忽略字段列表”。在结构化差异提取之后凡是field_path命中忽略列表的字段一律丢弃。这个列表在config.yaml里配置不需要改代码# config.yaml 片段 ignore_fields: - footer_copyright - csrf_token - visit_count - debug_info.*debug_info.*里的通配符支持前缀匹配会把所有以debug_info.开头的字段全部忽略。另一个常见噪声是版本号比如“v1.2.3”改成“v1.2.4”这种变化在大多数业务场景里并不重要所以我加了一个开关如果字段值只是按照 SemVer 版本号变化则标记为低优先级不触发告警。经过这三层过滤指纹过滤、忽略字段过滤、版本号过滤实际触发告警的变化量会减少 80% 以上。我两周跑下来每天真正有价值的变更事件大概在 20 到 40 条左右完全在可人工处理的范围。5. 跑起来只是开始部署配置与告警优化5.1 调度频率怎么设置才合理调度器是 PLFM_RADAR 的“心跳”。它的职责不只是定时执行探测任务还要处理优先级、失败重试、超时保护。我踩过的一个坑是不管目标重要程度统一用 5 分钟间隔去探测所有地址。结果是对核心平台更新频率反应太快但对低频平台造成了不必要的请求压力。后来我改成按场景区分目标类型间隔说明实时榜单60 秒排名每分钟都在动需要细粒度曲线商品详情页300 秒价格不会频繁变5 分钟足够捕获关键变化资讯列表600 秒每天更新几次10 分钟一次绰绰有余低频公告3600 秒每小时一次防止漏掉即可除了间隔每次探测还要有超时保护。我的 HTTP 请求统一设 15 秒超时超过就跳过本轮等下个周期再试。连续失败三次后自动停掉该目标避免把所有资源耗在一个无响应的地址上。调度器本身也是一个线程它读取目标配置按下一个执行时间排序循环sleep直到最近的执行时刻。得益于 Python 的heapq实现起来非常简单。5.2 告警压制同一目标 30 分钟内不重复通知告警功能做出来之后我遇到了一个新的麻烦某些目标频繁变化一分钟内连续产生了五六条事件我的 IM 机器人就一直刷屏。刚开始我以为是变化检测太敏感后来发现确实是目标本身在频繁更新。但用户真的需要每一条都立刻知道吗不需要我更希望一个小时内只收到一句汇总“该目标发生了 6 处变化”。所以我在告警模块里加了一个“压制窗口”机制。默认配置是同一个目标在 30 分钟内只发送一次告警如果期间有新的变化合并成一条聚合消息重新计时。首个事件立即发送后续事件进缓冲区30 分钟内如果缓冲区有内容就整体发一条补充消息。# 伪代码告警压制逻辑 class AlertThrottle: def __init__(self, window_seconds: int 1800): self.window window_seconds self.first_alert_at {} self.pending {} def should_send(self, target_id: str, event: dict) - bool: now time.time() if target_id not in self.first_alert_at: self.first_alert_at[target_id] now return True elapsed now - self.first_alert_at[target_id] if elapsed self.window: self.first_alert_at[target_id] now return True # 窗口内先存着返回 False self.pending.setdefault(target_id, []).append(event) return False窗口机制的效果立竿见影告警数量从每天几百条降到三十条左右而信息量几乎没有丢失因为聚合消息里包含了所有变化的字段列表。5.3 资源占用实测与优化我一开始用了一个非常粗暴的实现每个探测目标单独开一个线程所有结果全部写日志没有任何缓存。运行三天后看了下内存吓一跳稳定占用 2GB。排查后发现罪魁祸首是 Requests.Session 连接没复用、每轮探测都新建连接再加上 SQLite 写入时没有开启 WAL 模式频繁进行整库锁操作。优化做了三件事第一全局复用一个requests.Session连接池默认大小是 10足够几十个目标轮询使用。第二SQLite 开启 WAL 模式同时把写入封装成单线程队列批量提交事务。第三给探测结果加了一层 LRU 缓存同一目标在 30 秒内重复探测时直接返回上一个结果跳过网络请求。优化之后常驻内存稳定在 400MB 左右CPU 占用不到 5%一台树莓派级别的设备就能跑得很舒服。6. 真实运行两周后的复盘与改进方向6.1 我实际捕获到了哪些有价值的“平台变化”两周跑下来PLFM_RADAR 捕获的变化里有几类确实体现了这种系统的价值。第一次有价值的捕获是在某商品监测目标上我发现它的价格从 1899 跳到 1599前后只隔了 10 分钟。手动刷新根本不可能捕捉到这个窗口而系统不仅抓到了价格变化还把促销文案的同步修改也记录在案。后来价格回调到 1899趋势曲线完整记录了整个过程。第二次是针对资讯列表的捕获某条目被编辑后重新发布标题变了但链接没变。如果只看 RSS它不会触发更新如果人工刷新也很难注意到标题的细微改动。系统自动做了一个小规模文本差异直接标注了旧标题和新标题的差别。第三次是榜单排名趋势的复盘我用一天的数据画出了几个关键目标的排名变化曲线能明显看出某条目从上午十点开始热度爬升、下午两点达到顶峰、晚上逐渐衰减。这种时间维度上的结论是任何“快照式”观察都无法获得的。6.2 还存在的三个不足作为老老实实的复盘我也得说这个项目还有三个明显不足。第一探测器插件还是偏手写。新增一个目标类型就要写一个插件虽然基类已经简化了很多工作但还不够“零代码”。后续我打算引入通用的 JSON Schema 配置让非技术用户也能通过 YAML 定义解析规则。第二列表比较算法处理超大数组时效率不够高。当榜单超过 500 条时两次探测的全量比较大约耗时 200ms频繁触发会拖慢调度器。下一步准备用分片比较加增量更新的方式优化。第三告警通道目前只接了一种 IM 机器人。我更希望有一个统一的通知抽象层让用户可以自己接邮件、钉钉、Slack 或者企业微信而不是写死在代码里。6.3 我已经在动手的下一步计划PLFM_RADAR 的代码目前已经整理成了可运行的版本下一步有三个明确方向。第一个方向是完善“趋势分析”能力。现在只是把数值型字段存成了时序数据没有做更复杂的分析。我想加一个轻量的异常检测模块比如通过简单的一阶差分自动识别“价格突然上涨 10%”这类事件而不需要用户手动写阈值。第二个方向是做“目标分组”。目前所有目标都是平铺的我想支持分组管理比如把相同类型的目标放到一个组里对组整体生成日报和周报。第三个方向是把插件的配置文件从 YAML 改成更直观的 GUI 编辑。很多非技术的朋友看到 YAML 就头疼但让他们填一个表单式的页面就毫无压力。这个改完PLFM_RADAR 才算真正变成一个任何人都能上手用的工具。最后再分享一点个人的实际体会这种“监控系统”类项目最容易犯的错误是一开始就追求大而全。先明确你真正关心的平台和字段写最少够用的探测器把变化识别和告警做扎实远比你一开始就设计十几种探测器更靠谱。PLFM_RADAR 能两周跑通很大程度上也是因为我在第一天就只允许自己做三种探测器多一个都先记在待办清单里以后再说。这种克制是这个项目能快速跑起来的关键。