利用Python爬虫进行房源与票务实时变价提醒

在寻找租赁房源或抢购演唱会、景区门票和特价机票时，标的价格往往随供需关系、房东调价或平台动态定价策略而频繁变动。人工反复刷新网页不仅耗时费力，更容易错失瞬息即逝的低价窗口。利用 Python 爬虫构建一套自动化的房源与票务实时变价监控提醒系统，核心路径并不复杂：通过定时任务发起网络请求获取目标标的的最新价格，结合本地持久化存储进行新旧价格的差值与比对，当价格跌幅满足预设阈值且通过防抖过滤后，触发即时通讯工具的 Webhook 接口推送通知。本文将按照实际落地步骤，系统讲解这一监控体系的设计、实现与容错运维。

一、系统工作原理与架构设计

变价监控系统的本质是“时序数据的差异化告警”。一个高可用、不漏报且不产生无效噪音的监控系统，需要由数据采集、数据清洗与提取、状态持久化、变价决策引擎以及消息推送网关五个关键环节组成。

在设计该系统时，数据源通常分为两类：一类是直接返回结构化数据的 JSON API（常见于移动端 APP 接口或现代前端页面的异步请求），另一类是需要解析 HTML DOM 结构的网页。爬虫抓取到原始响应后，清洗出标的唯一标识（如房源 ID 或票次航班号）与数字类型的当前单价。持久化层负责记录每一个标的的首次基准价与历史变动快照。当新抓取价格出现变动时，决策引擎负责计算差价幅度，过滤微小波动与重复告警，最后交由通知通道向用户直达推送。

二、目标接口分析与数据提取实现

在编写爬虫之前，必须在浏览器开发者工具（F12 -> Network）中确认目标平台的价格加载方式。房源类站点（如贝壳找房、安居客等公开房源页面）通常采用服务端直出渲染，价格直接写在 HTML 标签中；而票务类平台（如部分航空客票、演出票务或景点分销平台）为了高并发性能，多采用异步拉取 JSON 接口的方式呈现余票与实时报价。

无论哪种形态，抓取时都需要维护统一的 HTTP 请求头。重点包含 User-Agent、Accept 以及合法的 Referer。使用 requests.Session() 可以复用底层的 TCP 连接，提升轮询效率并自动管理 Cookies。

import re
import requests
from bs4 import BeautifulSoup

class PropertyTicketScraper:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,application/json,*/*;q=0.8",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
        })

    def fetch_house_price(self, url: str) -> dict:
        """从房源页面解析当前租金或售价"""
        resp = self.session.get(url, timeout=10)
        resp.raise_for_status()
        soup = BeautifulSoup(resp.text, "html.parser")
        
        # 示例：定位房源标题与价格节点（具体选择器根据目标站点DOM结构调整）
        title_node = soup.select_one(".house-title, h1.title, .detail-title")
        price_node = soup.select_one(".price-num, .total-price, .house-price")
        
        if not price_node:
            raise ValueError("未能匹配到房源价格节点，请检查页面结构是否变动")

        title = title_node.get_text(strip=True) if title_node else "未知房源"
        price_text = price_node.get_text(strip=True)
        # 利用正则提取纯数字浮点值
        price_match = re.search(r"(\d+(?:\.\d+)?)", price_text.replace(",", ""))
        if not price_match:
            raise ValueError(f"无法从内容 '{price_text}' 提取有效价格数值")
            
        current_price = float(price_match.group(1))
        return {
            "title": title,
            "price": current_price,
            "raw_text": price_text
        }

    def fetch_ticket_price(self, api_url: str, params: dict) -> dict:
        """从票务接口解析当前最低余票或特价机票"""
        resp = self.session.get(api_url, params=params, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        
        # 示例结构：解析票务返回对象
        ticket_info = data.get("data", {})
        lowest_price = float(ticket_info.get("lowest_price", 0.0))
        available_seats = int(ticket_info.get("remain_count", 0))
        
        return {
            "title": ticket_info.get("flight_or_show_name", "监控票次"),
            "price": lowest_price,
            "available_seats": available_seats
        }

三、变价监控系统的五步实施流程

搭建一个生产级长期运行的变价监控系统，不能仅仅停留在单次脚本测试。必须将其拆解为标准化的实施步骤，保证每个模块输入输出清晰无歧义。

实施这五个步骤时，应避免将抓取逻辑与报警逻辑紧密耦合。如果抓取函数内直接调用推送接口，一旦抓取超时或返回格式稍有变动，极易导致整个告警管道崩溃或重复乱发。将持久化层置于两者之间，可以实现状态隔离与异常兜底。

四、价格状态持久化与变价判定引擎

单纯在 Python 内存全局变量中保存上次抓取价格是非常脆弱的。一旦脚本因网络波动重启或系统重启，所有历史基准价就会丢失，导致重启后的首次抓取出现漏报或误报。因此，推荐使用 Python 内置的 SQLite 数据库保存轻量级时序快照。

import sqlite3
import time

class PriceTrackerDB:
    def __init__(self, db_path: str = "price_monitor.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS item_prices (
                    item_id TEXT PRIMARY KEY,
                    item_type TEXT,
                    title TEXT,
                    initial_price REAL,
                    last_price REAL,
                    lowest_price REAL,
                    last_alert_time INTEGER,
                    updated_at INTEGER
                )
            """)
            conn.commit()

    def get_item(self, item_id: str) -> dict:
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM item_prices WHERE item_id = ?", (item_id,))
            row = cursor.fetchone()
            return dict(row) if row else None

    def record_or_update(self, item_id: str, item_type: str, title: str, current_price: float, last_alert_time: int = 0):
        now = int(time.time())
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            existing = self.get_item(item_id)
            if not existing:
                # 首次抓取，以当前价格建立基准
                cursor.execute("""
                    INSERT INTO item_prices 
                    (item_id, item_type, title, initial_price, last_price, lowest_price, last_alert_time, updated_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                """, (item_id, item_type, title, current_price, current_price, current_price, 0, now))
            else:
                new_lowest = min(existing["lowest_price"], current_price)
                alert_time = last_alert_time if last_alert_time > 0 else existing["last_alert_time"]
                cursor.execute("""
                    UPDATE item_prices 
                    SET title = ?, last_price = ?, lowest_price = ?, last_alert_time = ?, updated_at = ?
                    WHERE item_id = ?
                """, (title, current_price, new_lowest, alert_time, now, item_id))
            conn.commit()

在判定逻辑上，不能单纯以“当前价格 < 上次价格”作为告警依据。很多电商或租房平台存在1元或几分钱的微调，如果频繁弹窗会极大消耗用户注意力。科学的变价判定机制需要满足“阈值门槛”与“防抖静音窗口”双重条件。

变价判定函数示例代码如下：

def should_trigger_alert(existing_record: dict, current_price: float, drop_threshold_abs: float = 100.0, drop_threshold_rate: float = 0.05, cooldown_seconds: int = 1800) -> tuple:
    """
    判定是否应当触发警报
    :return: (is_alert: bool, reason: str)
    """
    if not existing_record:
        return False, "首次收录，已设置初始参考价"

    last_price = existing_record["last_price"]
    now = int(time.time())
    last_alert_time = existing_record.get("last_alert_time", 0)

    # 计算降幅
    price_diff = last_price - current_price
    if price_diff <= 0:
        return False, f"未出现降价（当前: {current_price}, 上次: {last_price}）"

    drop_rate = price_diff / last_price
    # 判断是否满足绝对金额门槛或百分比门槛
    if price_diff < drop_threshold_abs and drop_rate < drop_threshold_rate:
        return False, f"降价幅度较小 (降幅 {price_diff:.1f}元 / {drop_rate*100:.2f}%)，未达到告警门槛"

    # 防抖冷却判定
    if (now - last_alert_time) < cooldown_seconds:
        return False, f"处于冷却静音期内 (上次通知时间: {last_alert_time})，本次暂不推送"

    return True, f"价格显著下调 {price_diff:.1f} 元 (降幅 {drop_rate*100:.2f}%)"

五、即时通知渠道接入与多端推送

在告警触达方面，最轻量且免搭建独立 App 的方式是使用现代协作软件的群机器人 Webhook（如企业微信、钉钉或飞书）。这类工具不仅支持标准 HTTP POST 请求，还能直接渲染 Markdown 格式，帮助读者在手机锁屏或电脑桌面上一目了然地看到标的降价信息与直达链接。

def send_webhook_alert(webhook_url: str, title: str, old_price: float, new_price: float, target_url: str):
    """通过企业微信/钉钉/飞书机器人发送Markdown变价卡片"""
    drop_val = old_price - new_price
    drop_percent = (drop_val / old_price) * 100
    
    # 构造 Markdown 消息体（以企业微信格式为例，其他平台语法基本一致）
    message_content = f"""### 🚨 发现价格变动提醒！
**标的名称**：{title}
**变动状态**：降价 {drop_val:.2f} 元（跌幅 {drop_percent:.1f}%）
**原参考价**：~~￥{old_price:.2f}~~
**当前现价**：**￥{new_price:.2f}**
**发现时间**：{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}

[👉 点击直达原始页面查看详情]({target_url})
"""
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "content": message_content
        }
    }
    
    try:
        response = requests.post(webhook_url, json=payload, timeout=8)
        if response.status_code == 200:
            print(f"[{title}] 变价推送发送成功")
        else:
            print(f"推送接口返回异常状态码: {response.status_code}, 内容: {response.text}")
    except Exception as e:
        print(f"网络异常导致推送失败: {e}")

六、生产级稳定性保障与工程治理总结

将变价爬虫部署到生产服务器长期运行时，单靠 while True: time.sleep(60) 很容易因为网络断开、目标页面偶发502或结构调整而悄然退出。必须从反爬对抗、错误隔离、进程保活三个维度建立健全的治理规范。

针对房源与票务两类不同业务场景，由于业务周期的巨大差异，监控配置策略也存在明确的分野：

综上所述，利用 Python 爬虫构建房源与票务的变价提醒，并非简单的请求循环，而是一套将网络采集、数据持久化、决策过滤与即时通信有机串联的微型系统。通过清晰的模块解耦、可靠的状态记录和科学的降噪规则，便能以极低的技术成本，建立起敏锐、稳定且实用的自动化情报触达管道。