从零搭建自动化监控系统(附完整代码)
目录导读
- 为什么需要检测网页文本变化? – 应用场景与核心价值
- 技术选型:哪种脚本最适合你? – Python vs JavaScript vs 浏览器扩展
- 核心实现原理 – DOM监听、定时轮询与哈希比对
- 手把手实战:Python脚本检测网页文本变化 – 完整代码+逐行解析
- 进阶方案:无头浏览器+WebSocket实时推送 – 应对动态加载页面
- 常见问题FAQ – 内存泄漏、反爬、误报处理
为什么需要检测网页文本变化?
1 典型应用场景
- 竞品监控:盯住对手官网的价格、促销文案、新闻动态
- 政策追踪:政府部门网站的政策文件更新(如FDA、工信部)
- 票务/抢购:检测售票网站的余票状态文字变化
- 学术预警:期刊网站“录用状态”从“Under Review”变为“Accepted”
2 手动监控的痛点
- 每小时刷新10个网页?一天就要240次操作
- 动态加载内容(Ajax请求)在“查看源代码”中不可见
- 记忆上次文本状态?人脑会遗忘
技术选型:哪种脚本最适合你?
| 方案 | 适用场景 | 学习成本 | 抗动态加载 | 资源占用 |
|---|---|---|---|---|
| Python+requests+BeautifulSoup | 纯静态HTML页面 | 低 | 低 | |
| Python+Selenium/Playwright | 现代动态页面(Vue/React) | 中 | 高 | |
| JavaScript守候脚本 | 浏览器扩展/油猴脚本 | 中 | 低 | |
| 云函数+无头浏览器 | 高频率生产环境监控 | 高 | 弹性 |
对大多数用户,推荐Python+Selenium,兼顾易用性与动态页面兼容性。

核心实现原理
1 三种检测策略对比
策略A: 定时轮询(最常用)
┌─────────┐ 每隔N秒 ┌─────────┐
│ 抓取文本 │ ──────────→ │ 比对哈希 │
└─────────┘ └─────────┘
缺点:存在延迟、浪费资源
策略B: DOM监听(MutationObserver,前端专用)
┌─────────────────────────────┐
│ 当目标节点DOM变化时立即触发 │
│ 适合浏览器扩展/油猴脚本 │
└─────────────────────────────┘
缺点:不能跨标签页/跨网站
策略C: 可视化比对(截屏+OCR)
┌──────────┐ ┌───────────┐
│ 截图区域 │ → │ OCR识别文字 │
└──────────┘ └───────────┘
适用:验证码/图片文本变化(罕见)
2 哈希比对 vs 全文比对
- 全文比对:存储上次文本,直接对比差异(适合短文,但占内存)
- 哈希比对:用MD5/SHA256对文本生成指纹,仅存指纹(推荐)
import hashlib
def get_text_hash(text):
return hashlib.md5(text.encode('utf-8')).hexdigest()
手把手实战:Python脚本检测网页文本变化
1 环境准备
pip install selenium beautifulsoup4 python-dotenv # 下载ChromeDriver并放入PATH:https://chromedriver.chromium.org/
2 完整脚本(带注释)
import time
import hashlib
import smtplib
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from email.message import EmailMessage
import json
# ========== 配置区域 ==========
URL = "https://example.com/pricing"
CSS_SELECTOR = ".price-amount" # 目标元素CSS选择器
CHECK_INTERVAL = 60 # 秒
ALERT_EMAIL = "you@example.com"
SMTP_SERVER = "smtp.example.com"
# =============================
def get_price_text(driver):
"""获取目标元素的文本内容"""
try:
element = driver.find_element(By.CSS_SELECTOR, CSS_SELECTOR)
return element.text.strip()
except Exception as e:
print(f"获取元素失败: {e}")
return None
def send_alert(old_text, new_text, url):
"""发送邮件通知(可替换为钉钉/微信)"""
msg = EmailMessage()
msg.set_content(f"网页文本发生变化!\n页面:{url}\n旧内容:{old_text}\n新内容:{new_text}")
msg["Subject"] = "网页变更提醒"
msg["From"] = "monitor@example.com"
msg["To"] = ALERT_EMAIL
with smtplib.SMTP(SMTP_SERVER) as server:
server.send_message(msg)
def main():
# 配置无头浏览器(不显示界面)
chrome_options = Options()
chrome_options.add_argument("--headless") # 无头模式
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
driver = webdriver.Chrome(options=chrome_options)
driver.get(URL)
time.sleep(3) # 等待JS渲染
# 初始化:获取第一次文本作为基准
previous_hash = None
previous_text = None
print(f"开始监控 {URL},每{CHECK_INTERVAL}秒检查一次...")
try:
while True:
current_text = get_price_text(driver)
if current_text is None:
time.sleep(CHECK_INTERVAL)
continue
# 计算当前文本哈希
current_hash = hashlib.md5(current_text.encode('utf-8')).hexdigest()
if previous_hash is None:
# 首次运行,记录基准
previous_hash = current_hash
previous_text = current_text
print(f"初始化文本: {current_text}")
elif current_hash != previous_hash:
# 发生变化!
print(f"⚠️ 文本变化!\n旧: {previous_text}\n新: {current_text}")
send_alert(previous_text, current_text, URL)
# 更新基准
previous_hash = current_hash
previous_text = current_text
else:
print(f"✓ 未变化: {current_text[:50]}...")
time.sleep(CHECK_INTERVAL)
driver.refresh() # 刷新页面获取最新内容
time.sleep(3)
except KeyboardInterrupt:
print("监控停止")
finally:
driver.quit()
if __name__ == "__main__":
main()
3 脚本运行效果
开始监控 https://example.com/pricing,每60秒检查一次...
初始化文本: $99/month
✓ 未变化: $99/month...
✓ 未变化: $99/month...
⚠️ 文本变化!
旧: $99/month
新: $129/month
[邮件已发送]
4 关键优化点
- 降低误报:过滤空格、换行符差异(使用
re.sub(r'\s+', '', text)) - 智能刷新:动态页面每5次检测刷新1次浏览器,避免内存暴涨
- 多元素监控:用
CSS_SELECTOR = "div.price, span.discount"同时监控多个元素
进阶方案:无头浏览器+WebSocket实时推送
1 痛点解决
- 问题:轮询刷新造成页面闪烁,且消耗GPU资源
- 方案:使用
pyppeteer+websockets实现浏览器内监听
2 核心代码片段
# 使用pyppeteer(异步无头浏览器)
import asyncio
from pyppeteer import launch
async def monitor_real_time():
browser = await launch(headless=True)
page = await browser.newPage()
# 注入MutationObserver脚本
await page.evaluateOnNewDocument("""
window.mutationHash = '';
const observer = new MutationObserver(() => {
const target = document.querySelector('.price');
if (target) {
window.mutationHash = target.textContent.trim();
}
});
observer.observe(document.body, {childList: true, subtree: true, characterData: true});
""")
await page.goto("https://example.com")
previous = ""
while True:
# 直接获取页面变量,无需刷新
current = await page.evaluate('window.mutationHash')
if current and current != previous:
print(f"实时变化: {current}")
previous = current
await asyncio.sleep(2)
此方案不刷新页面,通过监听DOM变动实时获取最新文本,性能极佳。
常见问题FAQ
Q1: 脚本总是检测到“变化”,但实际文本并没变?
原因:页面包含动态内容(如当前时间戳、随机广告ID)。
解决:在get_text_hash前用正则过滤\d{10,}和data-id=".*?"等模式。
Q2: 网站有反爬机制怎么办?
对策:
- 轮换User-Agent(
fake_useragent库) - 设置随机延迟(
time.sleep(random.uniform(3,8))) - 使用代理IP池(付费服务如Oxylabs)
Q3: 长时间运行后内存暴涨?
原因:Selenium浏览器未正确关闭/泄漏。
修复:每100次检测重启浏览器:
if check_count % 100 == 0:
driver.quit()
driver = webdriver.Chrome(options=chrome_options)
driver.get(URL)
Q4: 如何监控多个页面?
方案A:多线程
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=5) as executor:
futures = [executor.submit(monitor_page, url) for url in urls]
方案B:Redis队列
每个监控任务推送到Redis,由消费者进程处理(适合大规模监控)。
Q5: 可以监控需要登录的页面吗?
可以,启动浏览器时加载已有的Cookie:
driver.get("https://example.com/login")
# 手动登录一次,然后保存Cookie
cookies = driver.get_cookies()
json.dump(cookies, open("cookies.json","w"))
# 后续运行加载
driver.get("https://example.com")
for cookie in json.load(open("cookies.json")):
driver.add_cookie(cookie)
driver.get("https://example.com/protected-page")
真正好用的文本监控脚本需要这三层
- 精准抓取:CSS选择器定位+Ajax等待(
WebDriverWait) - 智能噪声过滤:时间戳、动态ID、空格规范化
- 可靠通知:邮件/钉钉/企业微信webhook各备一份
这份脚本适合服务器部署(如阿里云轻量服务器,成本5元/月),24小时监控竞争对手官网,如需推送至手机,建议配置Pushbullet或Server酱通道。