从零搭建自动化监控系统(附完整代码)
目录导读
- 为什么需要检测网页文本变化? – 应用场景与核心价值
- 技术选型:哪种脚本最适合你? – 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酱通道。
版权声明:NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
足球直播
足球直播
足球直播
篮球直播
篮球直播
篮球直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
亚冠直播
亚冠直播
亚冠直播
欧冠直播
欧冠直播
欧冠直播
英超直播
英超直播
英超直播
西甲直播
西甲直播
西甲直播
中超直播
中超直播
中超直播
足球直播
足球直播
足球直播
篮球直播
篮球直播
篮球直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
亚冠直播
亚冠直播
亚冠直播
欧冠直播
欧冠直播
欧冠直播
英超直播
英超直播
英超直播
西甲直播
西甲直播
西甲直播
中超直播
中超直播
中超直播
足球直播
足球直播
足球直播
篮球直播
篮球直播
篮球直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
亚冠直播
亚冠直播
亚冠直播
欧冠直播
欧冠直播
欧冠直播
英超直播
英超直播
英超直播
西甲直播
西甲直播
西甲直播
中超直播
中超直播
中超直播
足球直播
足球直播
足球直播
篮球直播
篮球直播
篮球直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
亚冠直播
亚冠直播
亚冠直播
欧冠直播
欧冠直播
欧冠直播
英超直播
英超直播
英超直播
西甲直播
西甲直播
西甲直播
中超直播
中超直播
中超直播
曼联直播
曼联直播
曼联直播
日职联直播
日职联直播
日职联直播
澳超直播
澳超直播
澳超直播
阿森纳直播
阿森纳直播
阿森纳直播
足球直播
足球直播
足球直播
NBA直播
NBA直播
NBA直播
曼联直播
曼联直播
曼联直播
日职联直播
日职联直播
日职联直播
澳超直播
澳超直播
澳超直播
阿森纳直播
阿森纳直播
阿森纳直播
足球直播
足球直播
足球直播
NBA直播
NBA直播
NBA直播
曼联直播
曼联直播
曼联直播
日职联直播
日职联直播
日职联直播
澳超直播
澳超直播
澳超直播
阿森纳直播
阿森纳直播
阿森纳直播
足球直播
足球直播
足球直播
NBA直播
NBA直播
NBA直播
曼联直播
曼联直播
曼联直播
日职联直播
日职联直播
日职联直播
澳超直播
澳超直播
澳超直播
阿森纳直播
阿森纳直播
阿森纳直播
足球直播
足球直播
足球直播
NBA直播
NBA直播
NBA直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
欧冠直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
篮球直播
篮球直播
篮球直播
篮球直播
篮球直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
中超直播
中超直播
中超直播
中超直播
中超直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
篮球直播
篮球直播
篮球直播
篮球直播
篮球直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
中超直播
中超直播
中超直播
中超直播
中超直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播
中超直播