国内电商比价与价保：差价其实是可以要回来的（2026 实操）

三、降价了怎么立刻知道：推到微信

比价这件事真正的瓶颈不是「查到价格」，而是「记得去查」。所以关键环节是把提醒推到你每天一定会看的地方——对国内用户来说就是微信。可选通道如下，都是真实可用的方案：

自建方案为什么值得单独提一下：上面这些第三方推送服务，本质上都依赖服务方持续运营和平台政策。服务调整、接口变更甚至停止维护时，你所有接入过的脚本都要跟着改。message-pusher 把企业微信、飞书、钉钉、Bark、邮件、Telegram、Discord、QQ 等十几种通道统一到一个自建服务里，并且兼容 Server酱 的 API 格式——意味着你以后想换通道，业务代码基本不用动。它用 Go 写成单可执行文件，Docker 部署一条命令，默认管理员密码是公开的，记得改掉。

如果只是想尽快跑起来，直接从 Server酱 开始：扫码拿到 SendKey，调一个 HTTP 接口就能往自己微信发消息，五分钟能通。

四、脚本：便宜的提醒 + 只推「真的降了」

下面这个脚本做三件事：把你要盯的商品存进配置、每次运行时取一次当前价、在「创历史新低」或「低于你设的目标价」时才推送提醒。关键在于「只在有意义的时候打扰你」——每天都推「价格没变」，三次之后你就会屏蔽它。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""降价提醒（国内推送渠道版）

盯住若干个商品页，价格创新低或低于目标价时，通过国内可用的渠道推消息：
企业微信群机器人 / 钉钉群机器人 / 飞书群机器人 / Server酱 / PushPlus /
WxPusher / Bark / 自定义 webhook。

只用标准库，无需 pip install。

用法:
    python price_alert_cn.py --init                # 生成配置模板
    python price_alert_cn.py                       # 跑一轮检查
    python price_alert_cn.py --test-push           # 只测试推送渠道
    python price_alert_cn.py --add URL 目标价       # 追加一个监控项
"""

import argparse
import json
import os
import re
import sqlite3
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime

DB = "price_cn.db"
CFG = "price_cn.json"
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/126.0 Safari/537.36")

# ---------------------------------------------------------------- 推送渠道

def _post(url, payload, timeout=15):
    data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(url, data=data,
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.read().decode("utf-8", "ignore")


def _get(url, timeout=15):
    req = urllib.request.Request(url, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.read().decode("utf-8", "ignore")


def send(title, content, ch):
    """按配置里的渠道推送。返回 (是否成功, 渠道名, 返回摘要)"""
    kind = ch.get("type")
    try:
        if kind == "wecom":
            # 企业微信群机器人：markdown 有长度限制，超长自动降级为 text
            body = {"msgtype": "markdown", "markdown": {"content": content}}
            if len(content) > 3800:
                body = {"msgtype": "text", "text": {"content": content[:2000]}}
            r = _post(ch["webhook"], body)
            return ("ok" in r, "企业微信群机器人", r[:80])

        if kind == "dingtalk":
            r = _post(ch["webhook"], {
                "msgtype": "markdown",
                "markdown": {"title": title, "text": content}})
            return ("ok" in r or "errmsg" in r, "钉钉群机器人", r[:80])

        if kind == "feishu":
            r = _post(ch["webhook"], {
                "msg_type": "text",
                "content": {"text": title + "n" + content}})
            return ("code" in r, "飞书群机器人", r[:80])

        if kind == "serverchan":
            q = urllib.parse.urlencode({"title": title, "desp": content})
            r = _get("https://sctapi.ftqq.com/%s.send?%s" % (ch["key"], q))
            return ("code" in r or "data" in r, "Server酱", r[:80])

        if kind == "pushplus":
            r = _post("https://www.pushplus.plus/send", {
                "token": ch["token"], "title": title,
                "content": content, "template": "markdown"})
            return ("code" in r, "PushPlus", r[:80])

        if kind == "wxpusher":
            r = _post("https://wxpusher.zjiecode.com/api/send/message", {
                "appToken": ch["token"],
                "content": title + "n" + content,
                "contentType": 1,
                "uids": ch["uids"]})
            return ("code" in r, "WxPusher", r[:80])

        if kind == "bark":
            q = urllib.parse.urlencode({"title": title, "body": content})
            r = _get("%s/%s/?%s" % (ch["server"].rstrip("/"), ch["key"], q))
            return ("code" in r, "Bark", r[:80])

        if kind == "webhook":
            r = _post(ch["url"], {"title": title, "content": content})
            return (True, "自定义 webhook", r[:80])

        return (False, kind or "(未配置)", "未知渠道类型")
    except urllib.error.URLError as e:
        return (False, kind, "网络错误: %s" % e)
    except Exception as e:
        return (False, kind, "错误: %s" % e)


def push_all(title, content, cfg):
    for ch in cfg.get("push", []):
        ok, name, info = send(title, content, ch)
        print("  推送 %-14s %s  %s" % (name, "成功" if ok else "失败", info))
        if not ok:
            print("    -> 检查配置：%s" % {k: (v[:6] + "..." if isinstance(v, str)
                                            and len(v) > 10 else v)
                                        for k, v in ch.items()})


# ---------------------------------------------------------------- 抓取与解析

def fetch(url):
    req = urllib.request.Request(url, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=30) as r:
        raw = r.read()
    for enc in ("utf-8", "gbk", "gb18030"):
        try:
            return raw.decode(enc)
        except UnicodeDecodeError:
            continue
    return raw.decode("utf-8", "ignore")


def extract_price(html, item):
    """先按选择器/正则精确取，取不到再全页兜底"""
    pat = item.get("pattern")
    if pat:
        m = re.search(pat, html)
        if m:
            return to_float(m.group(1))
    for p in (r'"price"s*:s*"?([d.]+)"?',
              r'class="[^"]*price[^"]*"[^>]*>s*[¥￥]?s*([d,]+.?d*)',
              r'[¥￥]s*([d,]+.?d*)'):
        m = re.search(p, html)
        if m:
            return to_float(m.group(1))
    return None


def to_float(s):
    try:
        return float(str(s).replace(",", "").strip())
    except (ValueError, AttributeError):
        return None


# ---------------------------------------------------------------- 存储与判定

def init_db():
    con = sqlite3.connect(DB)
    con.execute("""CREATE TABLE IF NOT EXISTS history(
        url TEXT, price REAL, ts TEXT)""")
    con.execute("""CREATE TABLE IF NOT EXISTS notified(
        url TEXT, price REAL, ts TEXT)""")
    con.commit()
    return con


def last_notified(con, url):
    r = con.execute("SELECT price FROM notified WHERE url=? ORDER BY ts DESC LIMIT 1",
                    (url,)).fetchone()
    return r[0] if r else None


def record(con, url, price):
    con.execute("INSERT INTO history VALUES(?,?,?)",
                (url, price, datetime.now().strftime("%Y-%m-%d %H:%M:%S")))
    con.commit()


def mark_notified(con, url, price):
    con.execute("INSERT INTO notified VALUES(?,?,?)",
                (url, price, datetime.now().strftime("%Y-%m-%d %H:%M:%S")))
    con.commit()


def run(cfg, once=True):
    con = init_db()
    hits = []
    for item in cfg.get("watch", []):
        url = item["url"]
        name = item.get("name") or url[:40]
        try:
            html = fetch(url)
        except Exception as e:
            print("  [抓取失败] %s  %s" % (name, e))
            continue

        price = extract_price(html, item)
        if price is None:
            print("  [未取到价格] %s" % name)
            continue

        record(con, url, price)
        hist = con.execute("SELECT MIN(price) FROM history WHERE url=?",
                           (url,)).fetchone()[0]
        target = item.get("target")
        prev = last_notified(con, url)

        why = None
        if target is not None and price <= target:
            why = "低于目标价 %.2f" % target
        elif hist is not None and price < hist:
            why = "刷新历史最低 %.2f -> %.2f" % (hist, price)
        elif prev is not None and prev - price >= (item.get("drop_step") or 0):
            why = "较上次提醒降 %.2f" % (prev - price)

        print("  %-28s 当前 %10.2f  历史最低 %10.2f  %s"
              % (name[:28], price, hist, why or ""))

        if why and (prev is None or price < prev):
            hits.append((name, url, price, why))
            mark_notified(con, url, price)

    if hits:
        lines = ["## 降价提醒", ""]
        for name, url, price, why in hits:
            lines.append("- **%s**" % name)
            lines.append("  - 当前价：**%.2f**  （%s）" % (price, why))
            lines.append("  - %s" % url)
        content = "n".join(lines)
        print("n触发 %d 条提醒，开始推送：" % len(hits))
        push_all("有 %d 个商品降价了" % len(hits), content, cfg)
    else:
        print("n本轮没有触发提醒")
    con.close()
    return hits


# ---------------------------------------------------------------- CLI

TEMPLATE = {
    "push": [
        {"type": "wecom", "webhook": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=改成你的"},
        {"type": "serverchan", "key": "改成你的 SendKey"}
    ],
    "watch": [
        {"name": "示例商品", "url": "https://example.com/item/1",
         "pattern": r'"price"s*:s*"?([d.]+)"?', "target": 999, "drop_step": 30}
    ]
}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--init", action="store_true")
    ap.add_argument("--test-push", action="store_true")
    ap.add_argument("--add", nargs=2, metavar=("URL", "TARGET"))
    a = ap.parse_args()

    if a.init or not os.path.exists(CFG):
        with open(CFG, "w", encoding="utf-8") as f:
            json.dump(TEMPLATE, f, ensure_ascii=False, indent=2)
        print("已生成 %s，填好推送渠道和监控项后重跑" % CFG)
        return

    with open(CFG, encoding="utf-8") as f:
        cfg = json.load(f)

    if a.add:
        cfg.setdefault("watch", []).append(
            {"name": a.add[0][:40], "url": a.add[0],
             "target": float(a.add[1]), "drop_step": 30})
        with open(CFG, "w", encoding="utf-8") as f:
            json.dump(cfg, f, ensure_ascii=False, indent=2)
        print("已添加：%s 目标价 %s" % (a.add[0], a.add[1]))
        return

    if a.test_push:
        push_all("推送测试", "如果你收到这条，说明渠道配置正确。", cfg)
        return

    print("开始检查 %d 个商品" % len(cfg.get("watch", [])))
    run(cfg)


if __name__ == "__main__":
    main()

它只用 Python 标准库，不需要 pip install 任何东西。配置里的推送通道留空就会只在终端打印，填上就发到微信。

工作流程：

1. --init 生成一份带示例的配置文件
2. 把你要盯的商品填进去：名称、页面地址、价格提取用的正则、目标价
3. 先手动跑一次，确认价格抓对了（这是最容易出错的一步，不同页面的结构不一样）
4. 挂到定时任务里，一天一到两次

关于频率，这里再强调一次：一天一到两次足够。比价不是抢购，价格不会在五分钟里变三次。把频率压到最低，既是对平台的基本尊重，也让自己的脚本不容易被风控误伤。

价格提取的正则怎么写？在商品页面按 F12 或查看源代码，找到显示价格的那段 HTML，把数字部分换成捕获组即可。比如页面里是 <span class="price">￥1,299.00</span>，正则就写 class="price">￥([d,]+.d{2})。页面改版后正则必然失效，这时候你会看到「取不到价格」的提示，而不是被推送一个假价格——脚本对这种情况做了保护。

五、把三件事串成一条流程

1. 买之前：先查这个类目支不支持价保、窗口多长，再确认当前价格在历史区间里处于什么位置
2. 买之后：把商品页加进脚本的监听列表，设一个你真正愿意再买一次的目标价
3. 收到提醒：先判断是真降价还是换规格的「低价」，再看是否落在价保窗口内
4. 在窗口内：立刻申请价保，附上商品页截图，保留客服对话记录
5. 超出窗口：如果降幅大、且离下单时间很近，可以联系客服说明情况——能不能通融看平台，但值得问一句
6. 归档：所有订单、价格截图、客服记录按月归档，需要维权时能立刻翻出来（见下面那篇票据归档）

按项目采购，而不是持续订阅

同样的思路也适用于素材：如果一年里真正需要商用图的只是几个项目，按月订阅并不划算——额度月度清零，用得少就等于替平台养用户。按项目按张获取，才是跟着实际需求走的买法。 代下载仅提供文件获取服务，商用授权需自行购买。

搜索并获取需要的素材文件 →

常见问题

价保和「七天无理由」是一回事吗？

不是。两者解决的场景完全不同：价保是「东西没变、价格降了」，退的是差价；七天无理由退货是「你不想要了」，退的是商品。价保有独立的规则、窗口和申请入口，跟退换货流程不共用，所以不要指望退货流程能顺带退差价。

为什么不直接教我用现成的比价插件？

可以用，而且对多数人来说它是最省事的选择——装一个浏览器插件，它会提示历史价格。但你要知道它的三点局限：数据是抓取的，不保证完整和及时；接口可能随时失效；你的浏览行为会经过第三方服务器。所以本文的思路是「插件做粗略参考，自己的记录做决策依据」。

用脚本盯价格会不会有法律风险？

取决于怎么做。人工打开页面、低频查看、只为自己用、不绕过任何技术保护措施、不做数据转售——这些在正常使用范围内。而高频并发抓取、绕过验证码或风控、把数据打包对外提供，就完全是另一回事，通常违反平台条款，还可能触及数据与竞争相关法律。本文脚本默认一天跑一到两次，且不做并发，请保持这个使用方式。

提醒太多了怎么处理？

把目标价设得务实一点。一个简单的标准：定在「如果降到这个价，我今天就会下单」的那个数字上，而不是「我想要的最低价」。另一个办法是给降幅设阈值（比如比上次记录的低 30 元以上才提醒），避免为了几块钱来回打扰自己。

延伸阅读

• AI 帮你赚钱很难，但帮你省钱很简单：四条能立刻落地的路径
• 电费、话费、宽带：找出账单里能砍的部分
• 话费宽带降档被拒怎么办：三步维权路径与证据清单
• 发票、小票、合同本地 OCR 归档：离线识别 + 自动归类