拓冰网站定制 
2. 从零搭建PLFM_RADAR一个平台雷达监测系统的完整复盘这个项目的起源有点意思。我一直在做数据采集和竞品分析相关的工作每天手动打开十几个网页查看价格、榜单、库存这些信息重复劳动严重而且经常错过重要的变化节点。后来一个做电商运营的朋友吐槽说他们监控竞品价格全靠人工截图一天八小时都盯不过来。当时我就萌生了一个念头能不能做一个通用的“雷达系统”让它像雷达扫描空域一样自动扫描目标平台上的公开数据发现异常变化就主动报告于是就有了PLFM_RADAR这个项目。PLFM可以理解为Platform的缩写RADAR则是借用了雷达的原理——主动发射探测信号、接收回波、识别目标。落到实际系统中就是主动采集目标平台的公开页面数据解析、比对、判定变化再按规则推送通知。它解决的核心问题不是“爬数据”本身而是“如何持续、稳定、低干扰地监测一个平台的变化”。这篇文章我会把这个项目的完整思路、技术选型、代码实现和踩坑记录都写出来适合正在做数据采集、竞品监控、舆情预警或者任何需要“盯平台”的开发者参考。无论你是想监控电商价格、追踪内容榜单还是监测库存状态这套思路都可以直接套用。3. 整体设计与思路拆解在动手写代码之前我先想清楚了一个问题市面上已经有很多采集框架、爬虫工具为什么还要自己写一个雷达系统后来我给了自己一个答案——我要的不是“爬一次”而是“持续盯”。这决定了整个项目的架构方向。3.1 雷达系统与普通爬虫的本质区别传统的爬虫项目核心目标是“怎么把数据拿下来”解析规则、处理分页、绕开反爬、存储结果。工作流是“请求 - 解析 - 存储”跑完一次就结束。PLFM_RADAR的定位完全不同。它的核心目标是“怎么在长时间运行中稳定地发现变化”什么时候该去扫描、扫描哪些页面、数据和上次比有什么不同、这种不同是否值得通知。它的工作流是“调度 - 采集 - 比对 - 判定 - 通知”五个环节缺一不可。这个区别直接决定了几个关键设计决策采集频率不是越高越好而是匹配目标平台的数据更新规律。价格类数据可以半小时扫一次榜单类数据一天扫一次就够扫得太频繁反而容易被风控。原始数据必须历史留存因为“变化”是一个相对概念没有昨天的快照就判断不了今天的变化。通知必须分级不是所有变化都值得打扰人。价格变动5%和变动0.5%重要程度完全不同。3.2 方案选型为什么是Python Playwright技术选型上我对比过几个方案最终用了Python Playwright配合APScheduler做调度SQLite做存储。选型逻辑是这样的候选方案优势劣势结论Scrapy Requests性能高、异步采集无法处理JavaScript渲染的页面部分平台适用Selenium Chrome生态成熟、资料多资源占用高、速度慢可用但重Playwright ChromiumAPI友好、自动等待、多浏览器支持相对年轻、资料略少最终选择选择Playwright最关键的原因有两个。第一现在绝大多数平台的前端页面都是JavaScript渲染的Price字段、库存状态这些关键信息往往在XHR接口返回后再填充到DOM里用Requests直接请求HTML根本拿不到真实数据。第二Playwright的自动等待机制非常省心——它在元素出现之前会自动轮询等待无需手动写sleep这对需要稳定长时间运行的雷达系统来说太重要了。3.3 系统架构的三大模块划分我把整个系统拆成了三个独立模块它们之间通过数据库和消息队列解耦互不阻塞扫描模块Scanner负责按照调度策略访问目标页面提取结构化数据。这个模块是可插拔的针对不同平台写不同的解析器。雷达判定模块Detector负责把新扫描的数据和历史快照进行比对根据预设的阈值规则判定是否需要触发通知。这个模块是整个系统的大脑。通知模块Notifier负责把检测到的变化通过飞书机器人、邮件或钉钉推送出去。通知带上下文——变化前是什么、变化后是什么、变化幅度多大。这三个模块跑在同一个进程里但逻辑完全隔离。后面如果数据量大了可以把扫描模块单独拆成Worker节点横向扩容。4. 核心细节解析与实操要点系统架构确定之后真正的工作量其实都在细节里。这一章我把每个核心环节的实现要点和埋过的坑都写清楚这些经验都是常规文档里不会写的。4.1 页面解析不要每次都从头写选择器我做了一个通用解析器基类内置了常用的抽取方法CSS选择器提取、XPath提取、正则提取、JSON路径提取。每个平台只需要继承基类声明自己的字段映射规则即可。from typing import Any, Dict, List from parsel import Selector class BasePageParser: 页面解析器基类封装常用的抽取方法 def __init__(self, html: str): self.selector Selector(texthtml) def css(self, rule: str) - str: CSS选择器抽取返回第一个匹配的文本 result self.selector.css(rule).get() return result.strip() if result else def xpath(self, rule: str) - str: XPath抽取返回第一个匹配的文本 result self.selector.xpath(rule).get() return result.strip() if result else def regex(self, pattern: str, text: str) - str: 正则抽取返回第一个匹配组 import re match re.search(pattern, text) return match.group(1) if match else def parse(self) - Dict[str, Any]: 子类实现具体字段解析逻辑 raise NotImplementedError具体的商品页面解析器可以这样写class ProductParser(BasePageParser): 商品详情页解析器 def parse(self) - Dict[str, Any]: data { title: self.css(h1.product-title::text), price: self.css(.price-now::text), stock: self.css(.stock-status::attr(data-stock)), sales: self.regex(r已售(\d)件, self.css(.sales-count::text)), } # 价格需要清洗成float方便后面比对 raw_price data[price].replace(¥, ).replace(,, ) data[price] float(raw_price) if raw_price else 0.0 return data提示解析规则一定要做“空值兜底”。页面改版或者字段缺失时解析器返回空字符串而不是抛异常导致整个扫描任务中断。我在parse方法里把所有抽取结果都做了strip和空值判断宁可返回空字段也不能让雷达卡死在一个坏数据上。4.2 数据快照增量存储与版本比对雷达系统的存储层设计是个容易被忽略但要命的地方。我一开始只存了最新一条数据后来发现根本没法追溯“从什么时候开始变的”。后来改成快照表结构CREATE TABLE snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_key TEXT NOT NULL, -- 监测目标唯一标识如商品ID source_url TEXT NOT NULL, -- 原始页面URL data_json TEXT NOT NULL, -- 解析后的完整字段快照(JSON) checksum TEXT NOT NULL, -- 字段校验和用于快速判断是否变化 scanned_at DATETIME NOT NULL -- 扫描时间 ); CREATE INDEX idx_snapshots_target_time ON snapshots(target_key, scanned_at DESC);每次扫描时做的事情很简单按target_key查出最近一次快照的checksum和完整数据。计算新数据的checksum如果一致说明没有变化直接跳过。如果不一致逐字段比对生成差异列表插入新快照触发通知流程。checksum我用了hashlib.md5把data_json序列化后取摘要。注意对data_json做sort_keysTrue处理保证字段顺序变化不会导致误判。4.3 变化判定策略怎么减少误报雷达系统最怕的就是误报——通知轰炸几次之后用户就会把所有通知都屏蔽掉。我在判定模块里设计了几个过滤层第一层是“绝对变化”判断。价格从100变成101虽然变了但对业务没有实质影响。我设置了一个幅度阈值比如5%只有当变化幅度超过阈值时才进入下一层判断。第二层是“连续确认”机制。单次扫描发现变化不立即通知而是连续扫描两次都确认新值才触发。这个设计的初衷就是排除页面异常抖动、反爬返回假数据的情况。我踩过这个坑某平台在遇到异常流量时会随机把价格显示为9999如果不做连续确认系统一天能报警几十次。第三层是“静默期”控制。同一个监测目标的通知间隔最少30分钟。如果一个商品的价格在半小时内反复横跳只通知最开始的变化后续的等稳定了再说。def should_notify(self, current: dict, previous: dict, config: dict) - bool: 判定是否需要触发通知 :param current: 本次扫描的数据 :param previous: 上次扫描的数据 :param config: 监测配置包含阈值、静默期等 # 第一层字段差异检测 diffs self.diff_fields(current, previous) if not diffs: return False # 第二层幅度过滤 for diff in diffs: if diff.field price and config.get(price_threshold): change_ratio abs(diff.new_value - diff.old_value) / diff.old_value if change_ratio config[price_threshold]: return False # 第三层连续确认需要查最近两次扫描结果 if not self.is_confirmed(current): return False # 第四层静默期检查 if self.in_silence_period(current.target_key, config.get(silence_minutes, 30)): return False return True4.4 通知模块带着上下文的才叫通知通知不是简单发一句话“价格变了”而是要把“雷达判定”的完整信息带出去。我的通知模板是这样的[PLFM_RADAR] 监测目标变化通知 目标商品ID 883423 链接https://example.com/item/883423 变化字段 - 价格¥99.00 - ¥129.00涨幅 30.30% - 库存有货 - 无货 最近变化时间2025-01-15 14:32:07 历史记录最近5次扫描价格 [99.00, 99.00, 129.00, 129.00, 129.00]这样的通知用户不用打开页面就知道发生了什么而且有历史趋势可以判断是持续变化还是偶发抖动。飞书机器人的Webhook接入非常方便构造一个POST JSON请求就行import requests def send_feishu(webhook_url: str, content: str) - None: 发送飞书机器人通知 payload { msg_type: text, content: {text: content} } response requests.post(webhook_url, jsonpayload, timeout10) response.raise_for_status()5. 实操过程与核心环节实现这一章我按实际搭建的顺序把整个项目从0到1的过程记录下来。读者照着操作应该能在半天内跑起来一个可用的雷达系统。5.1 环境准备与项目结构我的环境是Python 3.11 Playwright APScheduler SQLite。Python版本至少3.9以上因为代码里用到了现代类型注解语法。# 创建虚拟环境 python -m venv venv source venv/bin/activate # 安装依赖 pip install playwright apscheduler requests parsel playwright install chromium项目结构我习惯这样组织模块边界清晰plfm_radar/ ├── scanner/ # 扫描模块 │ ├── base.py # 基类与通用逻辑 │ ├── product.py # 商品页面解析器 │ └── runner.py # 扫描调度执行器 ├── detector/ # 雷达判定模块 │ ├── comparator.py # 字段比对 │ └── rules.py # 阈值与规则引擎 ├── notifier/ # 通知模块 │ ├── feishu.py # 飞书通知 │ ├── email.py # 邮件通知 │ └── console.py # 控制台输出调试用 ├── storage/ # 数据层 │ ├── database.py # SQLite封装 │ └── models.py # 数据模型 ├── config/ # 配置管理 │ └── settings.py # 全局配置 └── main.py # 入口5.2 编写调度器如何优雅地定时执行调度我用的是APScheduler的BackgroundScheduler它支持cron表达式和固定间隔两种模式。我的实际经验是不同目标用不同的调度频率全部统一一个频率反而会浪费资源或者漏掉关键变化。from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger from apscheduler.triggers.cron import CronTrigger def setup_scheduler(scanner_runner): 初始化调度器注册所有扫描任务 scheduler BackgroundScheduler(timezoneAsia/Shanghai) # 商品价格监测每30分钟一次 scheduler.add_job( scanner_runner.run_target_group, triggerIntervalTrigger(minutes30), args[product_prices], idscan_product_prices, max_instances1, # 防止上一次未完成就启动下一次 coalesceTrue, # 积压的任务合并执行 misfire_grace_time3600 # 错过执行时间1小时内的任务不丢弃 ) # 榜单追踪每天上午9点执行 scheduler.add_job( scanner_runner.run_target_group, triggerCronTrigger(hour9, minute0), args[daily_rankings], idscan_daily_rankings ) return scheduler这里有几个参数必须说明一下。max_instances1是防止扫描任务本身运行太慢前一次还没跑完下一次又启动了导致同一目标并发扫描。coalesceTrue的意思是如果有积压的调度任务合并成一次执行而不是逐个补齐。这两个参数对长时间稳定运行至关重要不加的话系统跑一段时间就会越来越卡。5.3 实现扫描RunnerPlaywright页面自动化的完整流程Runner是实际执行页面访问和解析的组件。核心逻辑是启动浏览器上下文 - 访问页面 - 等待关键元素出现 - 提取HTML - 交给解析器 - 存储快照 - 关闭页面。from playwright.sync_api import sync_playwright import time class ScanRunner: 扫描执行器负责页面访问、提取、存储 def __init__(self, parser_class, storage): self.parser_class parser_class self.storage storage def scan_url(self, target_key: str, url: str, wait_selector: str None) - dict: 执行单个URL的扫描 with sync_playwright() as p: # 启动浏览器 browser p.chromium.launch( headlessTrue, args[ --no-sandbox, --disable-blink-featuresAutomationControlled, --disable-dev-shm-usage ] ) context browser.new_context( user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, viewport{width: 1280, height: 800} ) page context.new_page() try: # 访问页面并等待关键元素 page.goto(url, timeout60000, wait_untildomcontentloaded) # 如果指定了等待选择器等待它出现 if wait_selector: page.wait_for_selector(wait_selector, timeout15000) else: # 兜底等待让JavaScript完成渲染 time.sleep(3) # 提取页面HTML html page.content() # 解析数据 parser self.parser_class(html) data parser.parse() # 补全元信息 data[_target_key] target_key data[_source_url] url data[_scanned_at] time.strftime(%Y-%m-%d %H:%M:%S) # 存入快照 checksum self.storage.save_snapshot(target_key, url, data) return {target_key: target_key, data: data, checksum: checksum} except Exception as e: # 记录异常但不要中断整个调度 print(f[ERROR] 扫描失败 {target_key}: {e}) self.storage.log_error(target_key, url, str(e)) return {target_key: target_key, error: str(e)} finally: # 保证浏览器释放 context.close() browser.close()注意Playwright在headless模式下的浏览器指纹特征相对明显容易被平台识别。我加了--disable-blink-featuresAutomationControlled参数同时用了一个真实浏览器的User-Agent。这两个设置能解决部分基本的反爬检测但如果平台使用了更高级的指纹识别就需要引入stealth插件或者使用更复杂的应对策略。采集时务必遵守目标平台的robots协议和法律法规只采集合法公开数据。5.4 实现数据比对与变化检测比对模块是整个雷达系统的核心。我封装了一个通用Comparator类支持数字、字符串、列表三种字段类型的差异检测class FieldComparator: 字段比对器 staticmethod def diff(old_data: dict, new_data: dict) - list: 比对两组数据返回字段差异列表 每个差异项包含字段名、旧值、新值、变化类型 diffs [] all_keys set(old_data.keys()) | set(new_data.keys()) for key in all_keys: old_val old_data.get(key, ) new_val new_data.get(key, ) # 完全相等跳过 if old_val new_val: continue # 判断变化类型 if old_val : change_type 新增 elif new_val : change_type 消失 else: change_type 修改 diffs.append({ field: key, old_value: old_val, new_value: new_val, change_type: change_type }) return diffs这里需要注意的是不同类型字段的“相等”判断逻辑不同。字符串要strip再比数字要round保留两位再比列表要排序后再比。否则会出现明明没变化但checksum不一致的情况白白浪费通知额度。5.5 完整流程串起来main入口把调度器、Runner、Comparator和Notifier在main.py里组装起来整个系统就能跑了import logging from scanner.runner import ScanRunner from scanner.product import ProductParser from detector.comparator import FieldComparator from detector.rules import NotificationRules from notifier.console import ConsoleNotifier from notifier.feishu import FeishuNotifier from storage.database import SQLiteStorage from config.settings import TARGET_GROUPS def main(): # 初始化各模块 storage SQLiteStorage(plfm_radar.db) comparator FieldComparator() rules NotificationRules(threshold0.05, silence_minutes30) notifier FeishuNotifier(webhook_urlhttps://open.feishu.cn/open-apis/bot/v2/hook/xxxx) # 组装扫描Runner runner ScanRunner( parser_classProductParser, storagestorage, comparatorcomparator, rulesrules, notifiernotifier ) # 从配置读取监测目标列表 targets storage.load_targets() for group_name, urls in TARGET_GROUPS.items(): for target_key, url in urls.items(): runner.register_target(target_key, url, groupgroup_name) # 启动调度器 scheduler setup_scheduler(runner) scheduler.start() # 保持主线程运行 logging.info(PLFM_RADAR 已启动正在等待扫描任务...) try: while True: time.sleep(60) except KeyboardInterrupt: scheduler.shutdown() logging.info(系统已停止) if __name__ __main__: main()6. 常见问题与排查技巧实录系统跑起来之后真正的挑战才刚刚开始。我在实际运行的头两周里遇到了不少问题这里把最典型的几个整理成速查表都是真实踩过的坑。6.1 问题速查表症状可能原因排查方法解决方案扫描任务不定期中断浏览器进程崩溃查看日志中是否有Playwright异常增加失败重试机制单次失败不停止调度数据全是空的页面选择器失效手动打开页面检查DOM结构页面改版导致选择器过时需要更新解析规则价格数据出现9999平台反爬机制触发了假数据对比多个时间点的历史快照增加连续确认机制多次一致才通知通知发送失败Webhook失效或网络异常检查飞书机器人配置增加重试和备用通知渠道邮件扫描速度越来越慢历史数据堆积、数据库膨胀检查数据库大小定期归档旧快照只保留最近30天数据频繁被平台限流扫描频率过高查看HTTP返回码降低频率、增加随机延迟、使用代理池6.2 我踩过最深的坑页面结构突然改版某平台在没有任何预告的情况下改版了前端框架所有CSS类名全部换新。我的解析器一夜之间全部失效监测系统变成了瞎子。当时是凌晨两点等我早上看到通知的时候一晚上扫描的全部是空数据。这次教训让我做了三个改进第一解析器加了“自检”功能。每次解析后检查关键字段是否为空如果连续3次都为空自动触发一个“解析器异常”的紧急通知提醒人工介入。第二所有解析器升级为“多路径抽取”。同一个字段同时配置CSS选择器和XPath或者配置正则表达式作为兜底只要有一条路径能抽出值就算成功。第三建立了一个“页面结构指纹”机制。每次扫描记录页面中某些稳定元素的特征比如标题标签的内容、某些数据属性的值。当指纹和之前明显不同时说明页面结构变了自动降级处理并通知管理员。6.3 关于反爬策略的经验之谈这里必须说明采集行为一定要在合规前提下进行。我说的反爬应对指的是应对常见的频率限制和基础防护而不是突破平台的安全机制。实际项目中我的做法是扫描频率保持在温和水平不对目标平台造成压力每次请求之间加入随机延迟2-5秒模拟人类浏览节奏所有数据只用于内部分析不做任何公开传播或商业转售有能力就优先使用平台提供的公开API页面采集只是补充方案一个很现实的情况是如果你把扫描频率控制在一小时一次绝大多数平台的普通页面都不会触发任何风控。真正的问题往往出在“盲目加速”上——采集速度越快被封的风险越大数据的可靠性反而下降。雷达系统要的是持续稳定的情报流不是一次性的大规模数据落地。6.4 监控与自愈机制最后一个重大的设计经验是雷达系统本身也需要监控。如果一个监测任务已经连续失败6小时应该有人知道这件事而不是等到用户发现数据不对才来排查。我加了一个“心跳”机制系统每完成一轮扫描就向一个状态页面写入心跳时间戳。外部监控比如一个简单的定时请求检查心跳是否正常更新如果超过15分钟没有更新就触发告警。同时扫描任务内部的失败重试逻辑也是必须的。我用的是指数退避策略——第一次失败等1分钟重试第二次等2分钟第三次等4分钟最多重试5次。这样做既给了临时网络问题恢复的时间又不会因为持续重试给平台造成额外压力。7. 场景扩展与更多应用可能PLFM_RADAR虽然是围绕“平台监测”这个场景做的但拆开来看它的核心能力其实是“定时扫描公开页面 - 提取结构化字段 - 比对历史快照 - 分级通知”。这个能力可以平移应用到很多场景。7.1 电商价格与库存监测这是最直接的应用。注册一批竞品商品链接设置好价格波动阈值系统就能在你睡觉的时候盯住竞品调价动作。配合之前说的历史快照表还能画出竞品的价格走势曲线对定价决策很有参考价值。库存状态监测也很好用——某个热销商品突然有货了第一时间收到通知抢单的时间窗口能缩短到分钟级。7.2 内容平台榜单追踪我后来把解析器扩展到了内容平台的榜单页面比如热播剧排名、音乐榜单、App Store下载榜等。每天固定时间扫描排行榜生成榜单变化报告“新进榜单”和“排名下滑”都会触发通知。这对做内容运营和热点追踪的人来说等于省去了人工盯屏的时间。7.3 词条监控与舆情预警如果把扫描对象从商品页换成搜索页把提取字段从价格换成“结果数量”雷达系统就变成了一个简单的舆情预警工具。比如监测某个关键词在搜索页的排位变化、相关文章数量的增长速度。虽然不如专业的舆情系统全面但对个人和中小团队来说性价比很高。7.4 资源与服务状态巡检我还用过这套逻辑监测自己部署的API服务的可用性——定时访问健康检查端点提取状态码和响应时间配合通知模块做异常告警。相比商业监控工具自建这套系统的优势在于完全可控、逻辑透明、没有按量计费。8. 最后的经验沉淀这个项目从构思到跑稳定前后花了大概一周时间。我最深的体会是做一个监测系统难点不在于页面解析写得多漂亮而在于“稳定”二字——稳定的调度、稳定的数据、稳定的通知、稳定的自我运行。为此牺牲掉的首版功能集中在“过度采集”上。如果你打算复刻这个项目我给三个最实际的建议第一上线第一周先跑“影子模式”——系统正常扫描、正常存储、正常判定但通知只发到自己私人的测试群。观察几天数据质量确认无误后再接入正式通知渠道。我一开始就接入了正式通知结果误报轰炸了同事两天体验非常尴尬。第二监测目标从3个以内起步跑通全流程后再逐步添加。目标越多解析规则维护成本越高页面改版的影响面也越大。小而精起步比大而全更稳健。第三数据库备份要自动化。SQLite虽然轻量但历史快照文件是核心资产。我用了一个简单的定时任务每天凌晨把数据库文件复制到备份目录保留最近7天的副本。这个操作成本极低但能避免数据丢失时的灭顶之灾——毕竟雷达的历史快照就是它存在的全部意义。最后说一句心里话PLFM_RADAR这套东西技术门槛并不高每一个模块单独拎出来都是别人写过无数遍的常规操作。但当它们组合在一起形成了一个“持续盯守、主动报告”的系统时价值的提升是指数级的。自动化并不是什么神秘高深的事它就藏在把这些常规操作耐心地、严谨地串联起来的过程里。希望这篇复盘对你有所启发。 从零搭建PLFM_RADAR：平台雷达监测系统完整复盘
2. 从零搭建PLFM_RADAR一个平台雷达监测系统的完整复盘这个项目的起源有点意思。我一直在做数据采集和竞品分析相关的工作每天手动打开十几个网页查看价格、榜单、库存这些信息重复劳动严重而且经常错过重要的变化节点。后来一个做电商运营的朋友吐槽说他们监控竞品价格全靠人工截图一天八小时都盯不过来。当时我就萌生了一个念头能不能做一个通用的“雷达系统”让它像雷达扫描空域一样自动扫描目标平台上的公开数据发现异常变化就主动报告于是就有了PLFM_RADAR这个项目。PLFM可以理解为Platform的缩写RADAR则是借用了雷达的原理——主动发射探测信号、接收回波、识别目标。落到实际系统中就是主动采集目标平台的公开页面数据解析、比对、判定变化再按规则推送通知。它解决的核心问题不是“爬数据”本身而是“如何持续、稳定、低干扰地监测一个平台的变化”。这篇文章我会把这个项目的完整思路、技术选型、代码实现和踩坑记录都写出来适合正在做数据采集、竞品监控、舆情预警或者任何需要“盯平台”的开发者参考。无论你是想监控电商价格、追踪内容榜单还是监测库存状态这套思路都可以直接套用。3. 整体设计与思路拆解在动手写代码之前我先想清楚了一个问题市面上已经有很多采集框架、爬虫工具为什么还要自己写一个雷达系统后来我给了自己一个答案——我要的不是“爬一次”而是“持续盯”。这决定了整个项目的架构方向。3.1 雷达系统与普通爬虫的本质区别传统的爬虫项目核心目标是“怎么把数据拿下来”解析规则、处理分页、绕开反爬、存储结果。工作流是“请求 - 解析 - 存储”跑完一次就结束。PLFM_RADAR的定位完全不同。它的核心目标是“怎么在长时间运行中稳定地发现变化”什么时候该去扫描、扫描哪些页面、数据和上次比有什么不同、这种不同是否值得通知。它的工作流是“调度 - 采集 - 比对 - 判定 - 通知”五个环节缺一不可。这个区别直接决定了几个关键设计决策采集频率不是越高越好而是匹配目标平台的数据更新规律。价格类数据可以半小时扫一次榜单类数据一天扫一次就够扫得太频繁反而容易被风控。原始数据必须历史留存因为“变化”是一个相对概念没有昨天的快照就判断不了今天的变化。通知必须分级不是所有变化都值得打扰人。价格变动5%和变动0.5%重要程度完全不同。3.2 方案选型为什么是Python Playwright技术选型上我对比过几个方案最终用了Python Playwright配合APScheduler做调度SQLite做存储。选型逻辑是这样的候选方案优势劣势结论Scrapy Requests性能高、异步采集无法处理JavaScript渲染的页面部分平台适用Selenium Chrome生态成熟、资料多资源占用高、速度慢可用但重Playwright ChromiumAPI友好、自动等待、多浏览器支持相对年轻、资料略少最终选择选择Playwright最关键的原因有两个。第一现在绝大多数平台的前端页面都是JavaScript渲染的Price字段、库存状态这些关键信息往往在XHR接口返回后再填充到DOM里用Requests直接请求HTML根本拿不到真实数据。第二Playwright的自动等待机制非常省心——它在元素出现之前会自动轮询等待无需手动写sleep这对需要稳定长时间运行的雷达系统来说太重要了。3.3 系统架构的三大模块划分我把整个系统拆成了三个独立模块它们之间通过数据库和消息队列解耦互不阻塞扫描模块Scanner负责按照调度策略访问目标页面提取结构化数据。这个模块是可插拔的针对不同平台写不同的解析器。雷达判定模块Detector负责把新扫描的数据和历史快照进行比对根据预设的阈值规则判定是否需要触发通知。这个模块是整个系统的大脑。通知模块Notifier负责把检测到的变化通过飞书机器人、邮件或钉钉推送出去。通知带上下文——变化前是什么、变化后是什么、变化幅度多大。这三个模块跑在同一个进程里但逻辑完全隔离。后面如果数据量大了可以把扫描模块单独拆成Worker节点横向扩容。4. 核心细节解析与实操要点系统架构确定之后真正的工作量其实都在细节里。这一章我把每个核心环节的实现要点和埋过的坑都写清楚这些经验都是常规文档里不会写的。4.1 页面解析不要每次都从头写选择器我做了一个通用解析器基类内置了常用的抽取方法CSS选择器提取、XPath提取、正则提取、JSON路径提取。每个平台只需要继承基类声明自己的字段映射规则即可。from typing import Any, Dict, List from parsel import Selector class BasePageParser: 页面解析器基类封装常用的抽取方法 def __init__(self, html: str): self.selector Selector(texthtml) def css(self, rule: str) - str: CSS选择器抽取返回第一个匹配的文本 result self.selector.css(rule).get() return result.strip() if result else def xpath(self, rule: str) - str: XPath抽取返回第一个匹配的文本 result self.selector.xpath(rule).get() return result.strip() if result else def regex(self, pattern: str, text: str) - str: 正则抽取返回第一个匹配组 import re match re.search(pattern, text) return match.group(1) if match else def parse(self) - Dict[str, Any]: 子类实现具体字段解析逻辑 raise NotImplementedError具体的商品页面解析器可以这样写class ProductParser(BasePageParser): 商品详情页解析器 def parse(self) - Dict[str, Any]: data { title: self.css(h1.product-title::text), price: self.css(.price-now::text), stock: self.css(.stock-status::attr(data-stock)), sales: self.regex(r已售(\d)件, self.css(.sales-count::text)), } # 价格需要清洗成float方便后面比对 raw_price data[price].replace(¥, ).replace(,, ) data[price] float(raw_price) if raw_price else 0.0 return data提示解析规则一定要做“空值兜底”。页面改版或者字段缺失时解析器返回空字符串而不是抛异常导致整个扫描任务中断。我在parse方法里把所有抽取结果都做了strip和空值判断宁可返回空字段也不能让雷达卡死在一个坏数据上。4.2 数据快照增量存储与版本比对雷达系统的存储层设计是个容易被忽略但要命的地方。我一开始只存了最新一条数据后来发现根本没法追溯“从什么时候开始变的”。后来改成快照表结构CREATE TABLE snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_key TEXT NOT NULL, -- 监测目标唯一标识如商品ID source_url TEXT NOT NULL, -- 原始页面URL data_json TEXT NOT NULL, -- 解析后的完整字段快照(JSON) checksum TEXT NOT NULL, -- 字段校验和用于快速判断是否变化 scanned_at DATETIME NOT NULL -- 扫描时间 ); CREATE INDEX idx_snapshots_target_time ON snapshots(target_key, scanned_at DESC);每次扫描时做的事情很简单按target_key查出最近一次快照的checksum和完整数据。计算新数据的checksum如果一致说明没有变化直接跳过。如果不一致逐字段比对生成差异列表插入新快照触发通知流程。checksum我用了hashlib.md5把data_json序列化后取摘要。注意对data_json做sort_keysTrue处理保证字段顺序变化不会导致误判。4.3 变化判定策略怎么减少误报雷达系统最怕的就是误报——通知轰炸几次之后用户就会把所有通知都屏蔽掉。我在判定模块里设计了几个过滤层第一层是“绝对变化”判断。价格从100变成101虽然变了但对业务没有实质影响。我设置了一个幅度阈值比如5%只有当变化幅度超过阈值时才进入下一层判断。第二层是“连续确认”机制。单次扫描发现变化不立即通知而是连续扫描两次都确认新值才触发。这个设计的初衷就是排除页面异常抖动、反爬返回假数据的情况。我踩过这个坑某平台在遇到异常流量时会随机把价格显示为9999如果不做连续确认系统一天能报警几十次。第三层是“静默期”控制。同一个监测目标的通知间隔最少30分钟。如果一个商品的价格在半小时内反复横跳只通知最开始的变化后续的等稳定了再说。def should_notify(self, current: dict, previous: dict, config: dict) - bool: 判定是否需要触发通知 :param current: 本次扫描的数据 :param previous: 上次扫描的数据 :param config: 监测配置包含阈值、静默期等 # 第一层字段差异检测 diffs self.diff_fields(current, previous) if not diffs: return False # 第二层幅度过滤 for diff in diffs: if diff.field price and config.get(price_threshold): change_ratio abs(diff.new_value - diff.old_value) / diff.old_value if change_ratio config[price_threshold]: return False # 第三层连续确认需要查最近两次扫描结果 if not self.is_confirmed(current): return False # 第四层静默期检查 if self.in_silence_period(current.target_key, config.get(silence_minutes, 30)): return False return True4.4 通知模块带着上下文的才叫通知通知不是简单发一句话“价格变了”而是要把“雷达判定”的完整信息带出去。我的通知模板是这样的[PLFM_RADAR] 监测目标变化通知 目标商品ID 883423 链接https://example.com/item/883423 变化字段 - 价格¥99.00 - ¥129.00涨幅 30.30% - 库存有货 - 无货 最近变化时间2025-01-15 14:32:07 历史记录最近5次扫描价格 [99.00, 99.00, 129.00, 129.00, 129.00]这样的通知用户不用打开页面就知道发生了什么而且有历史趋势可以判断是持续变化还是偶发抖动。飞书机器人的Webhook接入非常方便构造一个POST JSON请求就行import requests def send_feishu(webhook_url: str, content: str) - None: 发送飞书机器人通知 payload { msg_type: text, content: {text: content} } response requests.post(webhook_url, jsonpayload, timeout10) response.raise_for_status()5. 实操过程与核心环节实现这一章我按实际搭建的顺序把整个项目从0到1的过程记录下来。读者照着操作应该能在半天内跑起来一个可用的雷达系统。5.1 环境准备与项目结构我的环境是Python 3.11 Playwright APScheduler SQLite。Python版本至少3.9以上因为代码里用到了现代类型注解语法。# 创建虚拟环境 python -m venv venv source venv/bin/activate # 安装依赖 pip install playwright apscheduler requests parsel playwright install chromium项目结构我习惯这样组织模块边界清晰plfm_radar/ ├── scanner/ # 扫描模块 │ ├── base.py # 基类与通用逻辑 │ ├── product.py # 商品页面解析器 │ └── runner.py # 扫描调度执行器 ├── detector/ # 雷达判定模块 │ ├── comparator.py # 字段比对 │ └── rules.py # 阈值与规则引擎 ├── notifier/ # 通知模块 │ ├── feishu.py # 飞书通知 │ ├── email.py # 邮件通知 │ └── console.py # 控制台输出调试用 ├── storage/ # 数据层 │ ├── database.py # SQLite封装 │ └── models.py # 数据模型 ├── config/ # 配置管理 │ └── settings.py # 全局配置 └── main.py # 入口5.2 编写调度器如何优雅地定时执行调度我用的是APScheduler的BackgroundScheduler它支持cron表达式和固定间隔两种模式。我的实际经验是不同目标用不同的调度频率全部统一一个频率反而会浪费资源或者漏掉关键变化。from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger from apscheduler.triggers.cron import CronTrigger def setup_scheduler(scanner_runner): 初始化调度器注册所有扫描任务 scheduler BackgroundScheduler(timezoneAsia/Shanghai) # 商品价格监测每30分钟一次 scheduler.add_job( scanner_runner.run_target_group, triggerIntervalTrigger(minutes30), args[product_prices], idscan_product_prices, max_instances1, # 防止上一次未完成就启动下一次 coalesceTrue, # 积压的任务合并执行 misfire_grace_time3600 # 错过执行时间1小时内的任务不丢弃 ) # 榜单追踪每天上午9点执行 scheduler.add_job( scanner_runner.run_target_group, triggerCronTrigger(hour9, minute0), args[daily_rankings], idscan_daily_rankings ) return scheduler这里有几个参数必须说明一下。max_instances1是防止扫描任务本身运行太慢前一次还没跑完下一次又启动了导致同一目标并发扫描。coalesceTrue的意思是如果有积压的调度任务合并成一次执行而不是逐个补齐。这两个参数对长时间稳定运行至关重要不加的话系统跑一段时间就会越来越卡。5.3 实现扫描RunnerPlaywright页面自动化的完整流程Runner是实际执行页面访问和解析的组件。核心逻辑是启动浏览器上下文 - 访问页面 - 等待关键元素出现 - 提取HTML - 交给解析器 - 存储快照 - 关闭页面。from playwright.sync_api import sync_playwright import time class ScanRunner: 扫描执行器负责页面访问、提取、存储 def __init__(self, parser_class, storage): self.parser_class parser_class self.storage storage def scan_url(self, target_key: str, url: str, wait_selector: str None) - dict: 执行单个URL的扫描 with sync_playwright() as p: # 启动浏览器 browser p.chromium.launch( headlessTrue, args[ --no-sandbox, --disable-blink-featuresAutomationControlled, --disable-dev-shm-usage ] ) context browser.new_context( user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, viewport{width: 1280, height: 800} ) page context.new_page() try: # 访问页面并等待关键元素 page.goto(url, timeout60000, wait_untildomcontentloaded) # 如果指定了等待选择器等待它出现 if wait_selector: page.wait_for_selector(wait_selector, timeout15000) else: # 兜底等待让JavaScript完成渲染 time.sleep(3) # 提取页面HTML html page.content() # 解析数据 parser self.parser_class(html) data parser.parse() # 补全元信息 data[_target_key] target_key data[_source_url] url data[_scanned_at] time.strftime(%Y-%m-%d %H:%M:%S) # 存入快照 checksum self.storage.save_snapshot(target_key, url, data) return {target_key: target_key, data: data, checksum: checksum} except Exception as e: # 记录异常但不要中断整个调度 print(f[ERROR] 扫描失败 {target_key}: {e}) self.storage.log_error(target_key, url, str(e)) return {target_key: target_key, error: str(e)} finally: # 保证浏览器释放 context.close() browser.close()注意Playwright在headless模式下的浏览器指纹特征相对明显容易被平台识别。我加了--disable-blink-featuresAutomationControlled参数同时用了一个真实浏览器的User-Agent。这两个设置能解决部分基本的反爬检测但如果平台使用了更高级的指纹识别就需要引入stealth插件或者使用更复杂的应对策略。采集时务必遵守目标平台的robots协议和法律法规只采集合法公开数据。5.4 实现数据比对与变化检测比对模块是整个雷达系统的核心。我封装了一个通用Comparator类支持数字、字符串、列表三种字段类型的差异检测class FieldComparator: 字段比对器 staticmethod def diff(old_data: dict, new_data: dict) - list: 比对两组数据返回字段差异列表 每个差异项包含字段名、旧值、新值、变化类型 diffs [] all_keys set(old_data.keys()) | set(new_data.keys()) for key in all_keys: old_val old_data.get(key, ) new_val new_data.get(key, ) # 完全相等跳过 if old_val new_val: continue # 判断变化类型 if old_val : change_type 新增 elif new_val : change_type 消失 else: change_type 修改 diffs.append({ field: key, old_value: old_val, new_value: new_val, change_type: change_type }) return diffs这里需要注意的是不同类型字段的“相等”判断逻辑不同。字符串要strip再比数字要round保留两位再比列表要排序后再比。否则会出现明明没变化但checksum不一致的情况白白浪费通知额度。5.5 完整流程串起来main入口把调度器、Runner、Comparator和Notifier在main.py里组装起来整个系统就能跑了import logging from scanner.runner import ScanRunner from scanner.product import ProductParser from detector.comparator import FieldComparator from detector.rules import NotificationRules from notifier.console import ConsoleNotifier from notifier.feishu import FeishuNotifier from storage.database import SQLiteStorage from config.settings import TARGET_GROUPS def main(): # 初始化各模块 storage SQLiteStorage(plfm_radar.db) comparator FieldComparator() rules NotificationRules(threshold0.05, silence_minutes30) notifier FeishuNotifier(webhook_urlhttps://open.feishu.cn/open-apis/bot/v2/hook/xxxx) # 组装扫描Runner runner ScanRunner( parser_classProductParser, storagestorage, comparatorcomparator, rulesrules, notifiernotifier ) # 从配置读取监测目标列表 targets storage.load_targets() for group_name, urls in TARGET_GROUPS.items(): for target_key, url in urls.items(): runner.register_target(target_key, url, groupgroup_name) # 启动调度器 scheduler setup_scheduler(runner) scheduler.start() # 保持主线程运行 logging.info(PLFM_RADAR 已启动正在等待扫描任务...) try: while True: time.sleep(60) except KeyboardInterrupt: scheduler.shutdown() logging.info(系统已停止) if __name__ __main__: main()6. 常见问题与排查技巧实录系统跑起来之后真正的挑战才刚刚开始。我在实际运行的头两周里遇到了不少问题这里把最典型的几个整理成速查表都是真实踩过的坑。6.1 问题速查表症状可能原因排查方法解决方案扫描任务不定期中断浏览器进程崩溃查看日志中是否有Playwright异常增加失败重试机制单次失败不停止调度数据全是空的页面选择器失效手动打开页面检查DOM结构页面改版导致选择器过时需要更新解析规则价格数据出现9999平台反爬机制触发了假数据对比多个时间点的历史快照增加连续确认机制多次一致才通知通知发送失败Webhook失效或网络异常检查飞书机器人配置增加重试和备用通知渠道邮件扫描速度越来越慢历史数据堆积、数据库膨胀检查数据库大小定期归档旧快照只保留最近30天数据频繁被平台限流扫描频率过高查看HTTP返回码降低频率、增加随机延迟、使用代理池6.2 我踩过最深的坑页面结构突然改版某平台在没有任何预告的情况下改版了前端框架所有CSS类名全部换新。我的解析器一夜之间全部失效监测系统变成了瞎子。当时是凌晨两点等我早上看到通知的时候一晚上扫描的全部是空数据。这次教训让我做了三个改进第一解析器加了“自检”功能。每次解析后检查关键字段是否为空如果连续3次都为空自动触发一个“解析器异常”的紧急通知提醒人工介入。第二所有解析器升级为“多路径抽取”。同一个字段同时配置CSS选择器和XPath或者配置正则表达式作为兜底只要有一条路径能抽出值就算成功。第三建立了一个“页面结构指纹”机制。每次扫描记录页面中某些稳定元素的特征比如标题标签的内容、某些数据属性的值。当指纹和之前明显不同时说明页面结构变了自动降级处理并通知管理员。6.3 关于反爬策略的经验之谈这里必须说明采集行为一定要在合规前提下进行。我说的反爬应对指的是应对常见的频率限制和基础防护而不是突破平台的安全机制。实际项目中我的做法是扫描频率保持在温和水平不对目标平台造成压力每次请求之间加入随机延迟2-5秒模拟人类浏览节奏所有数据只用于内部分析不做任何公开传播或商业转售有能力就优先使用平台提供的公开API页面采集只是补充方案一个很现实的情况是如果你把扫描频率控制在一小时一次绝大多数平台的普通页面都不会触发任何风控。真正的问题往往出在“盲目加速”上——采集速度越快被封的风险越大数据的可靠性反而下降。雷达系统要的是持续稳定的情报流不是一次性的大规模数据落地。6.4 监控与自愈机制最后一个重大的设计经验是雷达系统本身也需要监控。如果一个监测任务已经连续失败6小时应该有人知道这件事而不是等到用户发现数据不对才来排查。我加了一个“心跳”机制系统每完成一轮扫描就向一个状态页面写入心跳时间戳。外部监控比如一个简单的定时请求检查心跳是否正常更新如果超过15分钟没有更新就触发告警。同时扫描任务内部的失败重试逻辑也是必须的。我用的是指数退避策略——第一次失败等1分钟重试第二次等2分钟第三次等4分钟最多重试5次。这样做既给了临时网络问题恢复的时间又不会因为持续重试给平台造成额外压力。7. 场景扩展与更多应用可能PLFM_RADAR虽然是围绕“平台监测”这个场景做的但拆开来看它的核心能力其实是“定时扫描公开页面 - 提取结构化字段 - 比对历史快照 - 分级通知”。这个能力可以平移应用到很多场景。7.1 电商价格与库存监测这是最直接的应用。注册一批竞品商品链接设置好价格波动阈值系统就能在你睡觉的时候盯住竞品调价动作。配合之前说的历史快照表还能画出竞品的价格走势曲线对定价决策很有参考价值。库存状态监测也很好用——某个热销商品突然有货了第一时间收到通知抢单的时间窗口能缩短到分钟级。7.2 内容平台榜单追踪我后来把解析器扩展到了内容平台的榜单页面比如热播剧排名、音乐榜单、App Store下载榜等。每天固定时间扫描排行榜生成榜单变化报告“新进榜单”和“排名下滑”都会触发通知。这对做内容运营和热点追踪的人来说等于省去了人工盯屏的时间。7.3 词条监控与舆情预警如果把扫描对象从商品页换成搜索页把提取字段从价格换成“结果数量”雷达系统就变成了一个简单的舆情预警工具。比如监测某个关键词在搜索页的排位变化、相关文章数量的增长速度。虽然不如专业的舆情系统全面但对个人和中小团队来说性价比很高。7.4 资源与服务状态巡检我还用过这套逻辑监测自己部署的API服务的可用性——定时访问健康检查端点提取状态码和响应时间配合通知模块做异常告警。相比商业监控工具自建这套系统的优势在于完全可控、逻辑透明、没有按量计费。8. 最后的经验沉淀这个项目从构思到跑稳定前后花了大概一周时间。我最深的体会是做一个监测系统难点不在于页面解析写得多漂亮而在于“稳定”二字——稳定的调度、稳定的数据、稳定的通知、稳定的自我运行。为此牺牲掉的首版功能集中在“过度采集”上。如果你打算复刻这个项目我给三个最实际的建议第一上线第一周先跑“影子模式”——系统正常扫描、正常存储、正常判定但通知只发到自己私人的测试群。观察几天数据质量确认无误后再接入正式通知渠道。我一开始就接入了正式通知结果误报轰炸了同事两天体验非常尴尬。第二监测目标从3个以内起步跑通全流程后再逐步添加。目标越多解析规则维护成本越高页面改版的影响面也越大。小而精起步比大而全更稳健。第三数据库备份要自动化。SQLite虽然轻量但历史快照文件是核心资产。我用了一个简单的定时任务每天凌晨把数据库文件复制到备份目录保留最近7天的副本。这个操作成本极低但能避免数据丢失时的灭顶之灾——毕竟雷达的历史快照就是它存在的全部意义。最后说一句心里话PLFM_RADAR这套东西技术门槛并不高每一个模块单独拎出来都是别人写过无数遍的常规操作。但当它们组合在一起形成了一个“持续盯守、主动报告”的系统时价值的提升是指数级的。自动化并不是什么神秘高深的事它就藏在把这些常规操作耐心地、严谨地串联起来的过程里。希望这篇复盘对你有所启发。