1. 项目概述为什么我给自己造了一个“平台雷达”做内容运营和数据分析的朋友应该都有这种体会每天要盯的渠道太多了——内容社区的热榜、竞品的动态、行业关键词的新增讨论、用户对产品的实时反馈。如果全靠人工去刷一个早上就废了如果逐个写爬虫脚本维护成本又感人。这个名为PLFM_RADAR的小项目就是我在这种背景下折腾出来的一个多平台监测告警工具。简单说它像雷达一样定时扫描多个目标平台的公开数据把命中了预设关键词的内容打上信号标记经过去重和评分后推送到你的 IM 工具里。PLFM对应 Platform多平台RADAR就是信号扫描与目标捕捉名字很直白做的事也很直接。这个项目解决的痛点非常具体你不需要每分钟都盯着屏幕也不需要懂复杂的数据分析只要配置好“要盯哪些平台”“要过滤哪些关键词”“告警推到哪”剩下的交给定时任务。它特别适合三类人——做新媒体运营需要追热点的做竞品分析需要跟踪对手动态的以及做独立开发、产品经理想快速感知用户反馈的。不需要多高深的技术底子Python 基础 一点爬虫常识就能上手。我知道很多人一听到“爬虫”“监测”就觉得要写几百行代码其实真不是。PLFM_RADAR 的核心思路是把采集、解析、过滤、告警拆成四个独立模块每个模块都用最朴素的方案实现。整个项目跑下来核心代码也就几百行而且每一部分都可以单独替换——今天想加一个新平台不改其他代码明天想换告警通道也只是改一个函数的事。下面我把整个项目的设计思路、核心代码、踩过的坑完整拆开讲。我尽量用大白话把这套雷达系统的每一个零件都讲透。2. 整体设计与技术选型为什么这样搭而不是那样搭2.1 雷达模型扫描、发现、识别、上报雷达的工作原理大家都有概念不断发射电磁波接收反射信号从杂波中识别目标。PLFM_RADAR 的架构就是模仿这个模型来设计的四个环节一一对应。扫描Scan对应数据采集层。定时去目标平台拉取公开列表数据比如热榜、最新发布、搜索结果页。这里有一个很重要的取舍只做“列表页”采集不做“详情页”深爬。为什么因为列表页信息密度高一次请求能拿到几十条内容的标题、摘要、链接足以完成关键词判断。详情页留到告警之后人工点进去看就行。这样设计既能大幅降低被限流的风险也避免过度抓取带来的法律和技术风险。发现Detect对应信号解析层。把采集回来的 HTML 或 JSON 数据统一转换成标准的“信号对象”。信号对象包含标题、摘要、来源链接、发布时间、平台名称这五个字段。不管底层是哪个平台的接口到了这一层数据格式就是统一的后面的逻辑不用关心平台差异。识别Identify对应关键词过滤层。用配置好的规则集合去匹配信号对象计算命中得分。一个关键词命中加 1 分同义词命中加 0.5 分标题命中加权到 2 分。得分超过阈值的信号就被判定为“目标信号”进入下一步。上报Report对应告警分发层。把目标信号格式化后推送到钉钉群机器人、企业微信机器人或邮件。推送失败的时候还要做本地缓存和重试避免信号丢失。这个四层模型的优势在于边界清晰。我最早写的第一版是把所有逻辑揉在一个脚本里后来想加新平台发现牵一发动全身索性重构成了现在这种流水线结构。实际测试下来加一个新平台平均只需要一小时其中大部分时间花在琢磨这个平台的列表页结构和反爬策略上。2.2 为什么用 Python SQLite APScheduler 这套组合技术选型上我没有追求炫酷就三个考虑生态成熟、上手门槛低、单机够用。Python爬虫生态最完善requests 负责网络请求BeautifulSoup 或 lxml 负责解析完全够用。协程和多线程在这个场景里不是刚需因为扫描频率低一般 5-10 分钟一轮IO 等待反而给了 CPU 喘息空间。SQLite数据量不大一天下来几千条原始信号、几十条告警记录SQLite 一个文件搞定连数据库服务都省了。用sqlite3标准库不需要额外装东西。项目里我把原始信号存一张表、告警记录存一张表两张表用signal_id关联查询和统计都很方便。APScheduler定时任务库支持 cron 表达式。为什么不用系统自带的 crontab因为我想在代码里统一管理调度配置而且 APScheduler 可以做到“任务跑挂了自动重启”crontab 做不到这种进程级守护。这套组合最大的好处是部署极简。目标机器只需要有 Python 3.8pip install apscheduler requests beautifulsoup4三个库然后python plfm_radar.py就能跑。我在自己的云服务器上实测内存占用稳定在 120MB 左右含系统本身负载几乎可以忽略不计。2.3 项目结构和配置驱动的思路配置驱动是 PLFM_RADAR 能用起来的前提。我建了一个config.yaml所有平台开关、关键词规则、告警通道、扫描频率全放在里面。代码本身不硬编码任何平台信息启动时读取配置动态初始化采集任务。这样做的好处是换关键词不需要改代码重启即可生效不同环境测试/生产只需要维护不同的配置文件。监控平台: 平台A: 开关: true 扫描间隔: 300 列表页URL: https://example.com/api/hot 请求头: {...} 平台B: 开关: false 关键词规则: 基础关键词: - 产品名 - 创始人 同义词组: 产品A: [产品A, A产品, productA] 排除词: - 广告 - 招聘 命中阈值: 2 告警通道: 类型: dingtalk Webhook: https://oapi.dingtalk.com/robot/send?access_tokenxxx 关键词标签: #PLFM_RADAR我后面实现的所以逻辑都是围绕这份配置展开的。3. 核心模块拆解扫描层是整颗雷达的“天线”3.1 平台适配器的抽象与实现平台适配器解决的是“不同平台的数据格式不一样”的问题。我的做法是定义一个抽象基类BasePlatform内部约定好三个方法fetch()负责网络请求parse()负责把响应解析成统一信号对象列表run()是调度入口把 fetch 和 parse 串起来。每个新平台写一个子类继承它实现自己的 fetch 和 parse 就行。from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from typing import List, Optional dataclass class Signal: platform: str title: str summary: str url: str published_at: Optional[datetime] None raw: dict field(default_factorydict) class BasePlatform(ABC): def __init__(self, name: str, config: dict): self.name name self.config config self.session self._build_session() def _build_session(self): import requests s requests.Session() s.headers.update(self.config.get(请求头, {})) return s abstractmethod def fetch(self) - List[dict]: 拉取原始数据返回列表每个元素是一条内容的字典 abstractmethod def parse(self, raw_items: List[dict]) - List[Signal]: 把原始数据转换为统一的Signal对象列表 def run(self) - List[Signal]: raw_items self.fetch() signals self.parse(raw_items) # 统一补充平台名称和抓取时间 for sig in signals: sig.platform self.name return signals以某个内容社区为例它的热榜接口返回的是 JSONfetch()只需要带好 cookie 请求对应 URL然后提取data.list数组。parse()里把每条记录的标题、摘要、链接映射到 Signal 即可。整个过程没有任何黑魔法唯一要细心的是字段映射别搞错。注意fetch() 阶段强烈建议设置超时时间我统一用timeout10。有些平台偶发超时如果不设超时整个调度线程会被卡住后续所有平台都跟着排队。3.2 频率控制像雷达一样有节律地扫描雷达不能一直开机狂扫平台接口也一样。我的频率控制策略是“错峰扫描 随机抖动”。错峰扫描的意思是每个平台有独立的扫描间隔避免所有请求在同一秒发出。实现上APScheduler 为每个平台创建一个独立的 jobinterval参数从配置读取。随机抖动是在请求前加一个random.uniform(0.5, 2.0)的延时模拟真实用户的操作节奏能有效降低被封的风险。这里有个细节值得说平台的“列表页”如果是热榜类接口通常更新频率很低一小时甚至半天才变一次扫描间隔设成 600 秒都嫌浪费如果是搜索结果类接口更新可能很快300 秒比较合适。所以配置里的扫描间隔我默认推荐 300-600 秒具体看数据变化速度。不是越频繁越好——频率太高不仅容易被限流而且重复信号会暴涨浪费后续的过滤和去重算力。3.3 数据入库SQLite 表结构设计扫描层产生的信号统一写入 SQLite。我在设计表结构时遵循一个原则原始信号表只做简单清洗存库不做复杂判断判断逻辑交给下一层的识别模块。这样表结构可以保持精简。CREATE TABLE IF NOT EXISTS signals ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, title TEXT NOT NULL, summary TEXT DEFAULT , url TEXT NOT NULL, published_at TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, signal_id INTEGER NOT NULL, keyword_hit TEXT NOT NULL, score REAL NOT NULL, sent_status INTEGER DEFAULT 0, sent_at TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) );两个表的设计意图很明确signals保存所有观测到的信号是全量数据alerts只保存触发告警的信号以及告警的发送状态sent_status0 表示待发送1 表示已发送。为什么要保留全量信号因为有时候你会想复盘“这个关键词今天到底出现了多少次”或者“这个平台的热榜多久更新一次”有全量数据就能做分析。实测下来运行一周大概会产生 2-3 万条信号记录SQLite 完全扛得住。4. 信号识别与过滤别让雷达误报满天飞4.1 关键词命中的评分规则信号识别是整个雷达的“心脏”很多初版雷达系统废就废在误报率太高——什么都报警等于什么都没报。PLFM_RADAR 的过滤逻辑我设计成了一套轻量评分体系规则如下基础关键词命中标题2.0 分基础关键词命中摘要1.0 分同义词组命中标题1.5 分同义词组命中摘要0.8 分命中排除词直接判定为无效信号分数归零阈值默认设为 2.0。为什么是 2.0因为一条内容如果只在摘要里出现了一次泛化关键词1.0 分大概率不是精准目标而如果标题命中基础关键词2.0 分那基本可以确定信号值得关注。这个分数体系我跑了两个月觉得二八定律体现得很明显少数几个精准关键词贡献了 80% 的有效告警泛化词和同义词只是补充覆盖。def match_signal(signal: Signal, rules: dict) - dict: title signal.title.lower() summary signal.summary.lower() hit_keywords [] score 0.0 for kw in rules.get(基础关键词, []): kw_lower kw.lower() if kw_lower in title: score 2.0 hit_keywords.append(kw) elif kw_lower in summary: score 1.0 for group_name, words in rules.get(同义词组, {}).items(): for w in words: w_lower w.lower() if w_lower in title: score 1.5 hit_keywords.append(w) break elif w_lower in summary: score 0.8 hit_keywords.append(w) break for ex in rules.get(排除词, []): if ex.lower() in title or ex.lower() in summary: return {hit: False, score: 0.0, keywords: []} if score rules.get(命中阈值, 2.0): return {hit: True, score: round(score, 1), keywords: hit_keywords} return {hit: False, score: round(score, 1), keywords: hit_keywords}这段代码的逻辑说白了就是小学数学加权打分超过阈值就叮咚。但实际用的效果比我预想的好关键在于规则配置的语感。我建议配置的时候多花点心思在同义词组上比如做产品监测“产品名”和“产品简称”都要放进去否则同一产品在不同内容里的称呼不一致容易漏报。排除词也很重要尤其是“招聘”“广告”“抽奖”这类低质量营销内容排除词一加误报率直线下降。4.2 重复信号与陈腐信号的处理雷达系统最忌讳的一件事同一件事反复报警。比如某个话题上了热榜它可能会在平台的不同榜单里出现三次如果每次都推一条告警你的手机就该炸了。我的方案是双层去重。第一层在信号入库前做根据platform title的哈希值查 SQLite如果已存在就直接丢弃。第二层在告警生成前做记录每个关键词最近 N 条告警的 URL如果新信号与最近 5 条中的任何一条 URL 相同说明是已经在雷达屏幕上显示过的目标跳过发送。def is_duplicate_signal(conn, title, platform): cur conn.execute( SELECT id FROM signals WHERE platform? AND title?, (platform, title) ) return cur.fetchone() is not None时间陈腐的判断也很关键。我在 Signal 里记录published_at如果一条信号发布时间超过 24 小时即便命中了关键词也不再告警。因为雷达的价值在于发现新目标而不是回顾旧新闻。有些平台不返回发布时间这种情况我默认保留信号但会在告警文本里标注“时间未知”让接收方心里有数。重要去重逻辑不能只靠 URL因为同一条内容在不同时间可能被不同 URL 引用比如搜索接口和热榜接口的链接格式不一样。用platform title组合做粗粒度去重用 URL 做细粒度去重两层配合基本能消灭 99% 的重复告警。4.3 误报与漏报的调优经验调优是一个反复迭代的过程。我自己的操作流程是这样的每天看一次告警记录把误报的告警挑出来追溯它命中的关键词和信号内容。如果是关键词太泛就删掉或改成更精确的长尾词如果是同义词组导致误匹配就把它从同义词组里拆出来如果是排除词没覆盖到的场景就补一条排除词。漏报调起来比较麻烦因为漏掉的信号压根不会出现在告警记录里。我的办法是每周抽一个固定时段把原始信号表里评分在 1.0-1.9 之间的“次临界信号”导出来扫一眼看看有没有明显应该被抓到但没被抓到的类型。如果有就在同义词组里补词。这套“告警记录回溯 次临界信号抽检”的双回路调优法是我觉得这个项目里最有价值的经验。它把主观的“感觉不准”变成了客观的、可迭代的反馈循环。5. 实操过程从零搭起一个能用的雷达5.1 环境准备和依赖安装实操部分我按自己机器的环境来讲。我用的是云服务器Ubuntu 22.04Python 3.10。本地开发机如果是 Windows 或 macOS逻辑完全一致只是虚拟环境的创建命令略有不同。mkdir plfm_radar cd plfm_radar python3 -m venv venv source venv/bin/activate pip install requests beautifulsoup4 apscheduler pyyaml四个库各自的分工requests做 HTTP 请求beautifulsoup4处理 HTML 类型的数据源虽然多数平台给 JSON但偶尔会遇到老平台返回 HTMLapscheduler做任务调度pyyaml读取配置文件。再装一个sqlite3的命令行工具用来后面检查数据其实 Python 自带 sqlite3 模块不装也行。5.2 调度器主程序实现主程序的核心是初始化数据库、读取配置、创建各个平台的适配器实例、注册定时 job。我把代码控制在 80 行以内逻辑一目了然import yaml from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger from platform_adapters import BasePlatform, Signal import sqlite3 def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def init_db(conn: sqlite3.Connection): conn.executescript( CREATE TABLE IF NOT EXISTS signals ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, title TEXT NOT NULL, summary TEXT DEFAULT , url TEXT NOT NULL, published_at TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, signal_id INTEGER NOT NULL, keyword_hit TEXT NOT NULL, score REAL NOT NULL, sent_status INTEGER DEFAULT 0, sent_at TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) ); ) conn.commit() def run_platform_task(adapter: BasePlatform, conn: sqlite3.Connection, rules: dict): signals adapter.run() if not signals: return for sig in signals: if is_duplicate_signal(conn, sig.title, sig.platform): continue save_signal(conn, sig) result match_signal(sig, rules) if result[hit]: save_alert(conn, sig, result) def main(): cfg load_config(config.yaml) conn sqlite3.connect(plfm_radar.db, check_same_threadFalse) init_db(conn) scheduler BlockingScheduler() rules cfg[关键词规则] for name, pcfg in cfg[监控平台].items(): if not pcfg.get(开关, False): continue adapter BasePlatform.create_adapter(name, pcfg) # 简单工厂 interval pcfg.get(扫描间隔, 300) scheduler.add_job( run_platform_task, triggerIntervalTrigger(secondsinterval), args[adapter, conn, rules], idfscan_{name}, max_instances1, coalesceTrue ) print(f[PLFM_RADAR] 平台 {name} 已登记扫描间隔 {interval}s) scheduler.start() if __name__ __main__: main()这个主程序里有几个设计细节值得展开解释。max_instances1的含义是如果某个平台的上一次扫描还没跑完比如网络慢导致请求超时下一次定时触发不会并发执行而是跳过。这个参数直接解决了“任务堆积”问题在弱网环境下特别重要。coalesceTrue配合使用意思是把错过的多次触发合并成一次保证调度器不欠债。调度器选了BlockingScheduler它会阻塞主线程适合纯后台任务。如果你还想同时跑一个 Web 管理界面就要换成BackgroundScheduler这是后期扩展要考虑的事。5.3 告警发送模块“叮咚”一下传到手机告警推送我实现了一个极简的notifier.py默认支持钉钉群机器人。钉钉的机器人接口几乎零门槛建一个群添加自定义机器人拿到 Webhook 地址POST 一段 JSON 就行。消息格式我用的是 Markdown 类型因为可以塞链接和加粗阅读体验比纯文本好不少。import requests def send_dingtalk(webhook: str, signal: Signal, result: dict) - bool: title signal.title[:30] ... if len(signal.title) 30 else signal.title text ( f### PLFM_RADAR 信号捕获\n f- **平台**: {signal.platform}\n f- **标题**: [{title}]({signal.url})\n f- **关键词**: {, .join(result[keywords])}\n f- **评分**: {result[score]}\n f- **时间**: {signal.published_at or 未知}\n ) data { msgtype: markdown, markdown: {title: PLFM_RADAR 信号捕获, text: text}, } resp requests.post(webhook, jsondata, timeout10) return resp.json().get(errcode) 0实际使用中我建议在 Webhook URL 后面加一个安全关键词设置钉钉机器人的安全设置里支持自定义关键词。我在配置里给所有告警消息统一加一个#PLFM_RADAR标签这样 Webhook 安全关键词设为PLFM_RADAR就能直接通过校验不用加签逻辑省事很多。如果发送失败网络抖动或机器人限流notifier 返回 False主程序会把这条告警留在alerts表里sent_status0。我再写了一个简单的重试函数每轮扫描结束后检查是否有未发送的告警最多重试三次。三次还失败的就打日志人工处理。这个兜底机制让我在外面断网的情况下回来还能看到漏掉的数据。5.4 守护进程部署让它 7x24 小时待命开发调试用python plfm_radar.py跑在前台没问题但生产环境必须要有守护机制。我用的方案是 systemd service配置如下[Unit] DescriptionPLFM_RADAR Monitor Afternetwork.target [Service] WorkingDirectory/opt/plfm_radar ExecStart/opt/plfm_radar/venv/bin/python /opt/plfm_radar/plfm_radar.py Restartalways RestartSec15 EnvironmentPYTHONUNBUFFERED1 [Install] WantedBymulti-user.targetRestartalways保证进程崩溃后 15 秒自动拉起。EnvironmentPYTHONUNBUFFERED1让 print 输出第一时间刷进日志方便排查。日志查看直接journalctl -u plfm_radar -f不用额外配置 log 文件。这里踩过的一个坑初次部署时忘了给 venv 里的 Python 路径写绝对路径systemd 报找不到命令。其实因为 WorkingDirectory 已经指定了但 ExecStart 里的解释器路径还是必须写绝对路径这是个新手容易卡壳的小地方。6. 常见问题与排查技巧实录6.1 平台接口结构变了解析失败怎么办做爬虫监测最怕的就是平台改版。信号的表现通常是某个平台连续好几轮扫描都是 0 条新数据或者解析报错。我的排查思路分三步走看日志PLFM_RADAR 里每个适配器都有日志输出先确认 fetch 阶段有没有拿到响应。如果连 HTTP 状态码都变了比如 403说明是访问被拒优先查请求头是否需要更新。手工请求用浏览器打开目标列表页看数据结构是否变化。这一步最关键我习惯把响应保存到本地文件翻一翻找新的字段位置。更新适配器改parse()方法里的字段映射重新跑一轮验证。有一个小技巧我在每个适配器的fetch()里对响应内容做了一个 MD5 校验一旦发现页面结构和上次缓存的内容差异过大就输出一条日志提醒“页面结构可能变动”。这相当于给雷达装了一个“结构变化传感器”比被动等报错要主动得多。6.2 告警太多怎么办告警风暴的治理告警风暴是监测系统最常见的过载问题。我遇到过一次某个关键词被一个热点事件带火了一个小时内涌进来 80 多条告警手机震到没电。事后我总结了三个治理手段冷却时间同一关键词的告警发送后设置 30 分钟冷却期冷却期内不重复发送。聚合摘要冷却期内的信号不再逐条发送而是每 15 分钟生成一条聚合告警“近 15 分钟关键词 X 出现 12 次前三条如下...”。动态降频如果某平台最近 1 小时命中率超过 50%自动把该平台的扫描间隔从 300 秒拉长到 900 秒降低信息密度。这三板斧落地后我手机上的告警从日均 40 条降到日均 8 条但关键信息一条都没漏。特别是聚合摘要对于“事件持续发酵”的场景特别有用既能感知热度变化又不会轰炸接收人。6.3 时区与时间同步问题SQLite 里我用datetime(now, localtime)存本地时间Signal 里的published_at则由各平台接口返回。不同平台返回的时间格式五花八门有的是 Unix 时间戳有的是 ISO 8601 字符串带Z后缀有的直接在字符串里带“小时前”。我在解析层做了一个统一转换函数把所有时间都转成YYYY-MM-DD HH:MM:SS的本地时间字符串再存入数据库。为什么这么在意时间因为雷达的价值有一部分在于“时效排序”。如果某个平台的时间字段解析错了可能导致 24 小时陈腐判断失效要么漏掉新信号要么报警陈旧内容。建议大家在适配器里单独写一个normalize_time()函数针对一个平台的多种时间格式做充分测试这是投入产出比很高的一件事。6.4 常见问题速查表现象可能原因排查与解决某平台连续 0 条信号接口改版、请求被拒查日志手工访问列表页更新适配器告警发送失败Webhook 失效、网络波动检查钉钉机器人是否被移除检查消息中是否包含安全关键词重复告警去重逻辑未生效检查 signals 表是否有重复记录升级去重键为 platformtitle进程假死某个平台请求卡死确保所有请求设 timeout开启 Restartalways数据库文件变大信号积累过多定期清理 30 天前的旧数据或重建表内存持续上涨信号对象堆积检查 run_platform_task 里是否有未释放的循环引用改用局部变量7. 扩展思路从“雷达”到“情报网”PLFM_RADAR 目前做的是“单点扫描 关键词告警”但它的架构天然支持往两个方向扩展我在这里给出后续可延伸的路径。第一个方向是“多维信号关联”。现在的信号都是孤立的下一步可以给信号增加实体识别能力从标题和摘要里提取公司名、人名、产品名然后做关联聚合。比如“A 公司宣布与 B 公司合作”这条信号在雷达里会关联到 A 和 B 两个实体。这样当你在监测 A 的时候能顺带感知 B 的动态。第二个方向是“趋势分析与热度曲线”。已经在数据库里积累了全量信号天然可以做时间序列分析。统计各个关键词在时间轴上的出现频率配合简单的滑动平均算法就能画出热度曲线。这个曲线用来判断话题生命周期、竞品动作频率比单纯看告警消息有价值得多。结合我自己的使用场景我在想下一步给它接入一个简单的 Web 控制台用 FastAPI 暴露接口展示当日信号量、关键词命中分布、告警统计这些指标。虽然不做复杂的前端但一个简单的页面对日常巡检还是很有帮助的。如果你有兴趣这个方向同样可以从 SQLite 数据表出发用 100 行代码搭出一个可行的雏形。我在维护 PLFM_RADAR 这几个月里的最大体会是监测类工具的价值不在于技术多深而在于反馈回路是否顺畅。你设了关键词、设了告警、设了推送然后每天根据告警结果反过来调整规则这套闭环一旦转起来它就从一个“工具”变成你工作流里的一部分。雷达不会替你思考但它能让你在需要思考的时候拥有足够新鲜的数据。
1. 项目概述为什么我给自己造了一个“平台雷达”做内容运营和数据分析的朋友应该都有这种体会每天要盯的渠道太多了——内容社区的热榜、竞品的动态、行业关键词的新增讨论、用户对产品的实时反馈。如果全靠人工去刷一个早上就废了如果逐个写爬虫脚本维护成本又感人。这个名为PLFM_RADAR的小项目就是我在这种背景下折腾出来的一个多平台监测告警工具。简单说它像雷达一样定时扫描多个目标平台的公开数据把命中了预设关键词的内容打上信号标记经过去重和评分后推送到你的 IM 工具里。PLFM对应 Platform多平台RADAR就是信号扫描与目标捕捉名字很直白做的事也很直接。这个项目解决的痛点非常具体你不需要每分钟都盯着屏幕也不需要懂复杂的数据分析只要配置好“要盯哪些平台”“要过滤哪些关键词”“告警推到哪”剩下的交给定时任务。它特别适合三类人——做新媒体运营需要追热点的做竞品分析需要跟踪对手动态的以及做独立开发、产品经理想快速感知用户反馈的。不需要多高深的技术底子Python 基础 一点爬虫常识就能上手。我知道很多人一听到“爬虫”“监测”就觉得要写几百行代码其实真不是。PLFM_RADAR 的核心思路是把采集、解析、过滤、告警拆成四个独立模块每个模块都用最朴素的方案实现。整个项目跑下来核心代码也就几百行而且每一部分都可以单独替换——今天想加一个新平台不改其他代码明天想换告警通道也只是改一个函数的事。下面我把整个项目的设计思路、核心代码、踩过的坑完整拆开讲。我尽量用大白话把这套雷达系统的每一个零件都讲透。2. 整体设计与技术选型为什么这样搭而不是那样搭2.1 雷达模型扫描、发现、识别、上报雷达的工作原理大家都有概念不断发射电磁波接收反射信号从杂波中识别目标。PLFM_RADAR 的架构就是模仿这个模型来设计的四个环节一一对应。扫描Scan对应数据采集层。定时去目标平台拉取公开列表数据比如热榜、最新发布、搜索结果页。这里有一个很重要的取舍只做“列表页”采集不做“详情页”深爬。为什么因为列表页信息密度高一次请求能拿到几十条内容的标题、摘要、链接足以完成关键词判断。详情页留到告警之后人工点进去看就行。这样设计既能大幅降低被限流的风险也避免过度抓取带来的法律和技术风险。发现Detect对应信号解析层。把采集回来的 HTML 或 JSON 数据统一转换成标准的“信号对象”。信号对象包含标题、摘要、来源链接、发布时间、平台名称这五个字段。不管底层是哪个平台的接口到了这一层数据格式就是统一的后面的逻辑不用关心平台差异。识别Identify对应关键词过滤层。用配置好的规则集合去匹配信号对象计算命中得分。一个关键词命中加 1 分同义词命中加 0.5 分标题命中加权到 2 分。得分超过阈值的信号就被判定为“目标信号”进入下一步。上报Report对应告警分发层。把目标信号格式化后推送到钉钉群机器人、企业微信机器人或邮件。推送失败的时候还要做本地缓存和重试避免信号丢失。这个四层模型的优势在于边界清晰。我最早写的第一版是把所有逻辑揉在一个脚本里后来想加新平台发现牵一发动全身索性重构成了现在这种流水线结构。实际测试下来加一个新平台平均只需要一小时其中大部分时间花在琢磨这个平台的列表页结构和反爬策略上。2.2 为什么用 Python SQLite APScheduler 这套组合技术选型上我没有追求炫酷就三个考虑生态成熟、上手门槛低、单机够用。Python爬虫生态最完善requests 负责网络请求BeautifulSoup 或 lxml 负责解析完全够用。协程和多线程在这个场景里不是刚需因为扫描频率低一般 5-10 分钟一轮IO 等待反而给了 CPU 喘息空间。SQLite数据量不大一天下来几千条原始信号、几十条告警记录SQLite 一个文件搞定连数据库服务都省了。用sqlite3标准库不需要额外装东西。项目里我把原始信号存一张表、告警记录存一张表两张表用signal_id关联查询和统计都很方便。APScheduler定时任务库支持 cron 表达式。为什么不用系统自带的 crontab因为我想在代码里统一管理调度配置而且 APScheduler 可以做到“任务跑挂了自动重启”crontab 做不到这种进程级守护。这套组合最大的好处是部署极简。目标机器只需要有 Python 3.8pip install apscheduler requests beautifulsoup4三个库然后python plfm_radar.py就能跑。我在自己的云服务器上实测内存占用稳定在 120MB 左右含系统本身负载几乎可以忽略不计。2.3 项目结构和配置驱动的思路配置驱动是 PLFM_RADAR 能用起来的前提。我建了一个config.yaml所有平台开关、关键词规则、告警通道、扫描频率全放在里面。代码本身不硬编码任何平台信息启动时读取配置动态初始化采集任务。这样做的好处是换关键词不需要改代码重启即可生效不同环境测试/生产只需要维护不同的配置文件。监控平台: 平台A: 开关: true 扫描间隔: 300 列表页URL: https://example.com/api/hot 请求头: {...} 平台B: 开关: false 关键词规则: 基础关键词: - 产品名 - 创始人 同义词组: 产品A: [产品A, A产品, productA] 排除词: - 广告 - 招聘 命中阈值: 2 告警通道: 类型: dingtalk Webhook: https://oapi.dingtalk.com/robot/send?access_tokenxxx 关键词标签: #PLFM_RADAR我后面实现的所以逻辑都是围绕这份配置展开的。3. 核心模块拆解扫描层是整颗雷达的“天线”3.1 平台适配器的抽象与实现平台适配器解决的是“不同平台的数据格式不一样”的问题。我的做法是定义一个抽象基类BasePlatform内部约定好三个方法fetch()负责网络请求parse()负责把响应解析成统一信号对象列表run()是调度入口把 fetch 和 parse 串起来。每个新平台写一个子类继承它实现自己的 fetch 和 parse 就行。from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from typing import List, Optional dataclass class Signal: platform: str title: str summary: str url: str published_at: Optional[datetime] None raw: dict field(default_factorydict) class BasePlatform(ABC): def __init__(self, name: str, config: dict): self.name name self.config config self.session self._build_session() def _build_session(self): import requests s requests.Session() s.headers.update(self.config.get(请求头, {})) return s abstractmethod def fetch(self) - List[dict]: 拉取原始数据返回列表每个元素是一条内容的字典 abstractmethod def parse(self, raw_items: List[dict]) - List[Signal]: 把原始数据转换为统一的Signal对象列表 def run(self) - List[Signal]: raw_items self.fetch() signals self.parse(raw_items) # 统一补充平台名称和抓取时间 for sig in signals: sig.platform self.name return signals以某个内容社区为例它的热榜接口返回的是 JSONfetch()只需要带好 cookie 请求对应 URL然后提取data.list数组。parse()里把每条记录的标题、摘要、链接映射到 Signal 即可。整个过程没有任何黑魔法唯一要细心的是字段映射别搞错。注意fetch() 阶段强烈建议设置超时时间我统一用timeout10。有些平台偶发超时如果不设超时整个调度线程会被卡住后续所有平台都跟着排队。3.2 频率控制像雷达一样有节律地扫描雷达不能一直开机狂扫平台接口也一样。我的频率控制策略是“错峰扫描 随机抖动”。错峰扫描的意思是每个平台有独立的扫描间隔避免所有请求在同一秒发出。实现上APScheduler 为每个平台创建一个独立的 jobinterval参数从配置读取。随机抖动是在请求前加一个random.uniform(0.5, 2.0)的延时模拟真实用户的操作节奏能有效降低被封的风险。这里有个细节值得说平台的“列表页”如果是热榜类接口通常更新频率很低一小时甚至半天才变一次扫描间隔设成 600 秒都嫌浪费如果是搜索结果类接口更新可能很快300 秒比较合适。所以配置里的扫描间隔我默认推荐 300-600 秒具体看数据变化速度。不是越频繁越好——频率太高不仅容易被限流而且重复信号会暴涨浪费后续的过滤和去重算力。3.3 数据入库SQLite 表结构设计扫描层产生的信号统一写入 SQLite。我在设计表结构时遵循一个原则原始信号表只做简单清洗存库不做复杂判断判断逻辑交给下一层的识别模块。这样表结构可以保持精简。CREATE TABLE IF NOT EXISTS signals ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, title TEXT NOT NULL, summary TEXT DEFAULT , url TEXT NOT NULL, published_at TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, signal_id INTEGER NOT NULL, keyword_hit TEXT NOT NULL, score REAL NOT NULL, sent_status INTEGER DEFAULT 0, sent_at TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) );两个表的设计意图很明确signals保存所有观测到的信号是全量数据alerts只保存触发告警的信号以及告警的发送状态sent_status0 表示待发送1 表示已发送。为什么要保留全量信号因为有时候你会想复盘“这个关键词今天到底出现了多少次”或者“这个平台的热榜多久更新一次”有全量数据就能做分析。实测下来运行一周大概会产生 2-3 万条信号记录SQLite 完全扛得住。4. 信号识别与过滤别让雷达误报满天飞4.1 关键词命中的评分规则信号识别是整个雷达的“心脏”很多初版雷达系统废就废在误报率太高——什么都报警等于什么都没报。PLFM_RADAR 的过滤逻辑我设计成了一套轻量评分体系规则如下基础关键词命中标题2.0 分基础关键词命中摘要1.0 分同义词组命中标题1.5 分同义词组命中摘要0.8 分命中排除词直接判定为无效信号分数归零阈值默认设为 2.0。为什么是 2.0因为一条内容如果只在摘要里出现了一次泛化关键词1.0 分大概率不是精准目标而如果标题命中基础关键词2.0 分那基本可以确定信号值得关注。这个分数体系我跑了两个月觉得二八定律体现得很明显少数几个精准关键词贡献了 80% 的有效告警泛化词和同义词只是补充覆盖。def match_signal(signal: Signal, rules: dict) - dict: title signal.title.lower() summary signal.summary.lower() hit_keywords [] score 0.0 for kw in rules.get(基础关键词, []): kw_lower kw.lower() if kw_lower in title: score 2.0 hit_keywords.append(kw) elif kw_lower in summary: score 1.0 for group_name, words in rules.get(同义词组, {}).items(): for w in words: w_lower w.lower() if w_lower in title: score 1.5 hit_keywords.append(w) break elif w_lower in summary: score 0.8 hit_keywords.append(w) break for ex in rules.get(排除词, []): if ex.lower() in title or ex.lower() in summary: return {hit: False, score: 0.0, keywords: []} if score rules.get(命中阈值, 2.0): return {hit: True, score: round(score, 1), keywords: hit_keywords} return {hit: False, score: round(score, 1), keywords: hit_keywords}这段代码的逻辑说白了就是小学数学加权打分超过阈值就叮咚。但实际用的效果比我预想的好关键在于规则配置的语感。我建议配置的时候多花点心思在同义词组上比如做产品监测“产品名”和“产品简称”都要放进去否则同一产品在不同内容里的称呼不一致容易漏报。排除词也很重要尤其是“招聘”“广告”“抽奖”这类低质量营销内容排除词一加误报率直线下降。4.2 重复信号与陈腐信号的处理雷达系统最忌讳的一件事同一件事反复报警。比如某个话题上了热榜它可能会在平台的不同榜单里出现三次如果每次都推一条告警你的手机就该炸了。我的方案是双层去重。第一层在信号入库前做根据platform title的哈希值查 SQLite如果已存在就直接丢弃。第二层在告警生成前做记录每个关键词最近 N 条告警的 URL如果新信号与最近 5 条中的任何一条 URL 相同说明是已经在雷达屏幕上显示过的目标跳过发送。def is_duplicate_signal(conn, title, platform): cur conn.execute( SELECT id FROM signals WHERE platform? AND title?, (platform, title) ) return cur.fetchone() is not None时间陈腐的判断也很关键。我在 Signal 里记录published_at如果一条信号发布时间超过 24 小时即便命中了关键词也不再告警。因为雷达的价值在于发现新目标而不是回顾旧新闻。有些平台不返回发布时间这种情况我默认保留信号但会在告警文本里标注“时间未知”让接收方心里有数。重要去重逻辑不能只靠 URL因为同一条内容在不同时间可能被不同 URL 引用比如搜索接口和热榜接口的链接格式不一样。用platform title组合做粗粒度去重用 URL 做细粒度去重两层配合基本能消灭 99% 的重复告警。4.3 误报与漏报的调优经验调优是一个反复迭代的过程。我自己的操作流程是这样的每天看一次告警记录把误报的告警挑出来追溯它命中的关键词和信号内容。如果是关键词太泛就删掉或改成更精确的长尾词如果是同义词组导致误匹配就把它从同义词组里拆出来如果是排除词没覆盖到的场景就补一条排除词。漏报调起来比较麻烦因为漏掉的信号压根不会出现在告警记录里。我的办法是每周抽一个固定时段把原始信号表里评分在 1.0-1.9 之间的“次临界信号”导出来扫一眼看看有没有明显应该被抓到但没被抓到的类型。如果有就在同义词组里补词。这套“告警记录回溯 次临界信号抽检”的双回路调优法是我觉得这个项目里最有价值的经验。它把主观的“感觉不准”变成了客观的、可迭代的反馈循环。5. 实操过程从零搭起一个能用的雷达5.1 环境准备和依赖安装实操部分我按自己机器的环境来讲。我用的是云服务器Ubuntu 22.04Python 3.10。本地开发机如果是 Windows 或 macOS逻辑完全一致只是虚拟环境的创建命令略有不同。mkdir plfm_radar cd plfm_radar python3 -m venv venv source venv/bin/activate pip install requests beautifulsoup4 apscheduler pyyaml四个库各自的分工requests做 HTTP 请求beautifulsoup4处理 HTML 类型的数据源虽然多数平台给 JSON但偶尔会遇到老平台返回 HTMLapscheduler做任务调度pyyaml读取配置文件。再装一个sqlite3的命令行工具用来后面检查数据其实 Python 自带 sqlite3 模块不装也行。5.2 调度器主程序实现主程序的核心是初始化数据库、读取配置、创建各个平台的适配器实例、注册定时 job。我把代码控制在 80 行以内逻辑一目了然import yaml from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger from platform_adapters import BasePlatform, Signal import sqlite3 def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def init_db(conn: sqlite3.Connection): conn.executescript( CREATE TABLE IF NOT EXISTS signals ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, title TEXT NOT NULL, summary TEXT DEFAULT , url TEXT NOT NULL, published_at TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, signal_id INTEGER NOT NULL, keyword_hit TEXT NOT NULL, score REAL NOT NULL, sent_status INTEGER DEFAULT 0, sent_at TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) ); ) conn.commit() def run_platform_task(adapter: BasePlatform, conn: sqlite3.Connection, rules: dict): signals adapter.run() if not signals: return for sig in signals: if is_duplicate_signal(conn, sig.title, sig.platform): continue save_signal(conn, sig) result match_signal(sig, rules) if result[hit]: save_alert(conn, sig, result) def main(): cfg load_config(config.yaml) conn sqlite3.connect(plfm_radar.db, check_same_threadFalse) init_db(conn) scheduler BlockingScheduler() rules cfg[关键词规则] for name, pcfg in cfg[监控平台].items(): if not pcfg.get(开关, False): continue adapter BasePlatform.create_adapter(name, pcfg) # 简单工厂 interval pcfg.get(扫描间隔, 300) scheduler.add_job( run_platform_task, triggerIntervalTrigger(secondsinterval), args[adapter, conn, rules], idfscan_{name}, max_instances1, coalesceTrue ) print(f[PLFM_RADAR] 平台 {name} 已登记扫描间隔 {interval}s) scheduler.start() if __name__ __main__: main()这个主程序里有几个设计细节值得展开解释。max_instances1的含义是如果某个平台的上一次扫描还没跑完比如网络慢导致请求超时下一次定时触发不会并发执行而是跳过。这个参数直接解决了“任务堆积”问题在弱网环境下特别重要。coalesceTrue配合使用意思是把错过的多次触发合并成一次保证调度器不欠债。调度器选了BlockingScheduler它会阻塞主线程适合纯后台任务。如果你还想同时跑一个 Web 管理界面就要换成BackgroundScheduler这是后期扩展要考虑的事。5.3 告警发送模块“叮咚”一下传到手机告警推送我实现了一个极简的notifier.py默认支持钉钉群机器人。钉钉的机器人接口几乎零门槛建一个群添加自定义机器人拿到 Webhook 地址POST 一段 JSON 就行。消息格式我用的是 Markdown 类型因为可以塞链接和加粗阅读体验比纯文本好不少。import requests def send_dingtalk(webhook: str, signal: Signal, result: dict) - bool: title signal.title[:30] ... if len(signal.title) 30 else signal.title text ( f### PLFM_RADAR 信号捕获\n f- **平台**: {signal.platform}\n f- **标题**: [{title}]({signal.url})\n f- **关键词**: {, .join(result[keywords])}\n f- **评分**: {result[score]}\n f- **时间**: {signal.published_at or 未知}\n ) data { msgtype: markdown, markdown: {title: PLFM_RADAR 信号捕获, text: text}, } resp requests.post(webhook, jsondata, timeout10) return resp.json().get(errcode) 0实际使用中我建议在 Webhook URL 后面加一个安全关键词设置钉钉机器人的安全设置里支持自定义关键词。我在配置里给所有告警消息统一加一个#PLFM_RADAR标签这样 Webhook 安全关键词设为PLFM_RADAR就能直接通过校验不用加签逻辑省事很多。如果发送失败网络抖动或机器人限流notifier 返回 False主程序会把这条告警留在alerts表里sent_status0。我再写了一个简单的重试函数每轮扫描结束后检查是否有未发送的告警最多重试三次。三次还失败的就打日志人工处理。这个兜底机制让我在外面断网的情况下回来还能看到漏掉的数据。5.4 守护进程部署让它 7x24 小时待命开发调试用python plfm_radar.py跑在前台没问题但生产环境必须要有守护机制。我用的方案是 systemd service配置如下[Unit] DescriptionPLFM_RADAR Monitor Afternetwork.target [Service] WorkingDirectory/opt/plfm_radar ExecStart/opt/plfm_radar/venv/bin/python /opt/plfm_radar/plfm_radar.py Restartalways RestartSec15 EnvironmentPYTHONUNBUFFERED1 [Install] WantedBymulti-user.targetRestartalways保证进程崩溃后 15 秒自动拉起。EnvironmentPYTHONUNBUFFERED1让 print 输出第一时间刷进日志方便排查。日志查看直接journalctl -u plfm_radar -f不用额外配置 log 文件。这里踩过的一个坑初次部署时忘了给 venv 里的 Python 路径写绝对路径systemd 报找不到命令。其实因为 WorkingDirectory 已经指定了但 ExecStart 里的解释器路径还是必须写绝对路径这是个新手容易卡壳的小地方。6. 常见问题与排查技巧实录6.1 平台接口结构变了解析失败怎么办做爬虫监测最怕的就是平台改版。信号的表现通常是某个平台连续好几轮扫描都是 0 条新数据或者解析报错。我的排查思路分三步走看日志PLFM_RADAR 里每个适配器都有日志输出先确认 fetch 阶段有没有拿到响应。如果连 HTTP 状态码都变了比如 403说明是访问被拒优先查请求头是否需要更新。手工请求用浏览器打开目标列表页看数据结构是否变化。这一步最关键我习惯把响应保存到本地文件翻一翻找新的字段位置。更新适配器改parse()方法里的字段映射重新跑一轮验证。有一个小技巧我在每个适配器的fetch()里对响应内容做了一个 MD5 校验一旦发现页面结构和上次缓存的内容差异过大就输出一条日志提醒“页面结构可能变动”。这相当于给雷达装了一个“结构变化传感器”比被动等报错要主动得多。6.2 告警太多怎么办告警风暴的治理告警风暴是监测系统最常见的过载问题。我遇到过一次某个关键词被一个热点事件带火了一个小时内涌进来 80 多条告警手机震到没电。事后我总结了三个治理手段冷却时间同一关键词的告警发送后设置 30 分钟冷却期冷却期内不重复发送。聚合摘要冷却期内的信号不再逐条发送而是每 15 分钟生成一条聚合告警“近 15 分钟关键词 X 出现 12 次前三条如下...”。动态降频如果某平台最近 1 小时命中率超过 50%自动把该平台的扫描间隔从 300 秒拉长到 900 秒降低信息密度。这三板斧落地后我手机上的告警从日均 40 条降到日均 8 条但关键信息一条都没漏。特别是聚合摘要对于“事件持续发酵”的场景特别有用既能感知热度变化又不会轰炸接收人。6.3 时区与时间同步问题SQLite 里我用datetime(now, localtime)存本地时间Signal 里的published_at则由各平台接口返回。不同平台返回的时间格式五花八门有的是 Unix 时间戳有的是 ISO 8601 字符串带Z后缀有的直接在字符串里带“小时前”。我在解析层做了一个统一转换函数把所有时间都转成YYYY-MM-DD HH:MM:SS的本地时间字符串再存入数据库。为什么这么在意时间因为雷达的价值有一部分在于“时效排序”。如果某个平台的时间字段解析错了可能导致 24 小时陈腐判断失效要么漏掉新信号要么报警陈旧内容。建议大家在适配器里单独写一个normalize_time()函数针对一个平台的多种时间格式做充分测试这是投入产出比很高的一件事。6.4 常见问题速查表现象可能原因排查与解决某平台连续 0 条信号接口改版、请求被拒查日志手工访问列表页更新适配器告警发送失败Webhook 失效、网络波动检查钉钉机器人是否被移除检查消息中是否包含安全关键词重复告警去重逻辑未生效检查 signals 表是否有重复记录升级去重键为 platformtitle进程假死某个平台请求卡死确保所有请求设 timeout开启 Restartalways数据库文件变大信号积累过多定期清理 30 天前的旧数据或重建表内存持续上涨信号对象堆积检查 run_platform_task 里是否有未释放的循环引用改用局部变量7. 扩展思路从“雷达”到“情报网”PLFM_RADAR 目前做的是“单点扫描 关键词告警”但它的架构天然支持往两个方向扩展我在这里给出后续可延伸的路径。第一个方向是“多维信号关联”。现在的信号都是孤立的下一步可以给信号增加实体识别能力从标题和摘要里提取公司名、人名、产品名然后做关联聚合。比如“A 公司宣布与 B 公司合作”这条信号在雷达里会关联到 A 和 B 两个实体。这样当你在监测 A 的时候能顺带感知 B 的动态。第二个方向是“趋势分析与热度曲线”。已经在数据库里积累了全量信号天然可以做时间序列分析。统计各个关键词在时间轴上的出现频率配合简单的滑动平均算法就能画出热度曲线。这个曲线用来判断话题生命周期、竞品动作频率比单纯看告警消息有价值得多。结合我自己的使用场景我在想下一步给它接入一个简单的 Web 控制台用 FastAPI 暴露接口展示当日信号量、关键词命中分布、告警统计这些指标。虽然不做复杂的前端但一个简单的页面对日常巡检还是很有帮助的。如果你有兴趣这个方向同样可以从 SQLite 数据表出发用 100 行代码搭出一个可行的雏形。我在维护 PLFM_RADAR 这几个月里的最大体会是监测类工具的价值不在于技术多深而在于反馈回路是否顺畅。你设了关键词、设了告警、设了推送然后每天根据告警结果反过来调整规则这套闭环一旦转起来它就从一个“工具”变成你工作流里的一部分。雷达不会替你思考但它能让你在需要思考的时候拥有足够新鲜的数据。
