淘宝抽赏余量监控简记

一、背景

朋友在淘宝玩一个成就抽赏活动，想实时监控剩余抽赏张数的变化。页面是个内嵌在 WebView 中的小程序，数据通过加密 API（淘宝 MTop 协议）传输，直接抓包不可行。手机端没有 API 调用权限，只能在手机上开着页面看。

我的方案是放弃破解加密，改用视觉+无障碍方案：用 ADB 连手机，通过 Android 自带的 uiautomator dump 命令读取当前界面的 Accessibility Tree，直接从系统层面拿到所有文字信息（包括剩余张数）。每次检查前自动点击 tab 触发页面刷新，确保数据是最新的。

变化时通过企业微信 API 发送通知到手机，也支持 macOS 系统通知。

二、原理

整体流程:

• 手机连接电脑，开启 USB 调试
• Python 脚本通过 ADB 控制手机
• 每次检查: 点 tab → 等 0.5s → uiautomator dump → 解析 XML → 提取剩余张数
• 数字变化时: 企业微信 API → 消息推送到微信

使用 uiautomator dump 而不是 OCR 或截图的原因:

• Uiautomator 从系统层面直接拿到 TextView 的文字内容和坐标，不需要任何 OCR 引擎
• 不依赖 Tesseract、EasyOCR 等第三方库，零安装成本
• 速度快、准确率 100%（系统提供的数据，不是识别结果）
• 数据包含完整的节点树和坐标信息，可以精确定位到数字区域

三、文件结构

taobao-chouwan-monitor/
├── fast_monitor.py        # 主脚本（后台监控 + 企业微信通知）
├── uia_monitor.py         # 详细解析版（带完整奖品数据和中奖记录）
├── visual_monitor.py      # OCR 方案（备选，需要 tesseract/easyocr）
├── monitor.log            # 运行日志
├── monitor_stdout.log     # 后台进程标准输出
└── .uia_state.json        # 状态文件（记录上次剩余数和历史）

核心文件是 fast_monitor.py，其他为备选或辅助。

四、环境准备

4.1 电脑端

需要 macOS 或 Linux 系统，安装 ADB（Android Debug Bridge）:

# macOS
brew install android-platform-tools

# Linux (Ubuntu/Debian)
sudo apt install adb

# 验证安装
adb --version

Python 3.6+ 即可，不需要安装任何第三方库（仅使用标准库）。

4.2 手机端

• 安卓手机（不限品牌）
• 开启开发者选项: 设置 → 关于手机 → 连点 7 次版本号
• 开启 USB 调试: 设置 → 开发者选项 → USB 调试
• 用 USB 线连接电脑
• 手机上弹出"允许 USB 调试"提示 → 点击允许

验证连接:

adb devices
# 应显示: <设备号>       device

4.3 企业微信通知（可选）

如果需要手机收到变化通知，需要注册企业微信自建应用:

1. 登录 企业微信管理后台
2. 「应用管理」→ 「自建」→ 「创建应用」
3. 填写应用名称，选择可见范围（你的企业微信号）
4. 创建后拿到:
   • corpid: 「我的企业」页面底部
   • corpsecret: 自建应用的 Secret
   • agentid: 自建应用的 AgentId（纯数字）
   • 接收人 userid: 通讯录中你的账号
5. 在「微信插件」中开启「允许成员接收应用消息」，这样才能推送到个人微信

如果不配置企业微信，脚本只会在 Mac 上弹系统通知（通过 osascript）。

五、完整代码

核心脚本 fast_monitor.py:

#!/usr/bin/env python3
"""
淘宝成就抽赏 - 后台监控（企业微信通知）
变化时自动发消息到企业微信（可转发个人微信）

用法:
  python3 fast_monitor.py                     # 前台实时输出
  python3 fast_monitor.py --daemon --interval 30  # 后台运行
  pkill -f fast_monitor.py                     # 停止
  
配置: 文件头部 WX_* 变量
"""
import subprocess, sys, re, json, os, time, urllib.request
import xml.etree.ElementTree as ET
from datetime import datetime

# ═══════ 企业微信配置（需手动修改） ═══════
WX_CORPID = "your_corpid"          # 企业微信 corp ID
WX_CORPSECRET = "your_corpsecret"  # 自建应用 Secret
WX_AGENTID = "1000002"             # 自建应用 AgentId
WX_USER = "userid"                 # 通知接收人 userid

# ═══════ 监控配置 ═══════
ADB = "adb"
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
STATE_FILE = os.path.join(SCRIPT_DIR, ".uia_state.json")
LOG_FILE = os.path.join(SCRIPT_DIR, "monitor.log")
TOKEN_FILE = os.path.join(SCRIPT_DIR, ".wx_token.json")

HISTORY = []


# ---------- 企业微信通知 ----------
WX_TOKEN = None
WX_TOKEN_EXPIRES = 0


def wx_get_token():
    """获取企业微信 access token（带文件缓存）"""
    global WX_TOKEN, WX_TOKEN_EXPIRES
    now = time.time()
    if WX_TOKEN and now < WX_TOKEN_EXPIRES - 60:
        return WX_TOKEN

    if os.path.exists(TOKEN_FILE):
        try:
            with open(TOKEN_FILE) as f:
                d = json.load(f)
            if d.get("expires_at", 0) > now + 60:
                WX_TOKEN = d["token"]
                WX_TOKEN_EXPIRES = d["expires_at"]
                return WX_TOKEN
        except:
            pass

    url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={WX_CORPID}&corpsecret={WX_CORPSECRET}"
    try:
        req = urllib.request.Request(url)
        resp = urllib.request.urlopen(req, timeout=10)
        data = json.loads(resp.read())
        if data.get("errcode") == 0:
            WX_TOKEN = data["access_token"]
            WX_TOKEN_EXPIRES = now + data["expires_in"]
            with open(TOKEN_FILE, "w") as f:
                json.dump({"token": WX_TOKEN, "expires_at": WX_TOKEN_EXPIRES}, f)
            return WX_TOKEN
        else:
            log(f"WX token error: {data}")
    except Exception as e:
        log(f"WX token request failed: {e}")
    return None


def wx_send(title, text, emoji="🎯"):
    """发送企业微信文本消息"""
    token = wx_get_token()
    if not token:
        return False

    content = f"{emoji} {title}\n{text}"
    if len(content) > 2000:
        content = content[:1997] + "..."

    msg_data = {
        "touser": WX_USER,
        "msgtype": "text",
        "agentid": int(WX_AGENTID),
        "text": {"content": content},
        "safe": 0,
        "enable_duplicate_check": 1,
        "duplicate_check_interval": 30,
    }

    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
    try:
        req = urllib.request.Request(
            url, data=json.dumps(msg_data).encode("utf-8"),
            headers={"Content-Type": "application/json"}
        )
        resp = urllib.request.urlopen(req, timeout=10)
        result = json.loads(resp.read())
        if result.get("errcode") == 0:
            log(f"WX sent: {title}")
            return True
        else:
            log(f"WX send error: {result}")
    except Exception as e:
        log(f"WX send failed: {e}")
    return False


# ---------- ADB 控制 ----------
def log(msg):
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(f"[{ts}] {msg}\n")


def refresh_and_dump():
    """点击 tab 刷新页面，然后 dump UI 树"""
    # 先切到奖池赏品（强制卸载当前视图）
    subprocess.run([ADB, "shell", "input", "tap", "821", "804"],
                   capture_output=True, timeout=10)
    time.sleep(0.3)
    # 再切回赏品余量（触发数据重新加载）
    subprocess.run([ADB, "shell", "input", "tap", "612", "804"],
                   capture_output=True, timeout=10)
    time.sleep(0.5)

    subprocess.run([ADB, "shell", "uiautomator", "dump", "/sdcard/ui.xml"],
                   capture_output=True, timeout=30)
    xml_text = subprocess.run(
        [ADB, "shell", "cat", "/sdcard/ui.xml"],
        capture_output=True, text=True, timeout=30
    ).stdout
    subprocess.run([ADB, "shell", "rm", "/sdcard/ui.xml"],
                   capture_output=True, timeout=10)

    try:
        root = ET.fromstring(xml_text)
    except ET.ParseError:
        return None, None

    items = []
    for elem in root.iter():
        text = elem.get("text", "")
        clazz = elem.get("class", "")
        if text and "TextView" in clazz:
            b = list(map(int, re.findall(r'\d+', elem.get("bounds", "0 0 0 0"))))
            if len(b) == 4:
                items.append((text, b[0], b[1], b[2], b[3]))

    for i, (text, x1, y1, x2, y2) in enumerate(items):
        if text.strip() == "剩余:":
            r, s = None, None
            if i + 1 < len(items):
                try:
                    r = int(items[i + 1][0].strip())
                except ValueError:
                    pass
            if i + 2 < len(items):
                m = re.match(r"/(\d+)张", items[i + 2][0])
                if m:
                    s = int(m.group(1))
            return r, s
    return None, None


def os_notify(title, text):
    """macOS 通知中心"""
    subprocess.run(["osascript", "-e",
        f'display notification "{text}" with title "{title}" sound name "default"'],
        capture_output=True, timeout=5)


def load_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f)
    return {"last_remaining": None, "total_drops": 0, "total_cycles": 0}


def save_state(state):
    os.makedirs(SCRIPT_DIR, exist_ok=True)
    with open(STATE_FILE, "w") as f:
        json.dump(state, f, ensure_ascii=False, indent=2)


def check_device():
    r = subprocess.run([ADB, "devices"], capture_output=True, text=True, timeout=10)
    for line in r.stdout.split("\n"):
        if line.strip() and "\tdevice" in line:
            return True
    return False


# ---------- Main ----------
def main():
    daemon = False
    interval = 30

    for arg in sys.argv[1:]:
        if arg == "--daemon":
            daemon = True
        elif arg.startswith("--interval="):
            interval = int(arg.split("=")[1])

    if not check_device():
        msg = "手机未连接！请插 USB 并开启调试"
        log(msg)
        os_notify("抽赏监控", msg)
        if daemon:
            wx_send("监控异常", msg, "⚠️")
        print(msg)
        sys.exit(1)

    state = load_state()
    last = state.get("last_remaining")
    total_drops = state.get("total_drops", 0)
    total_cycles = state.get("total_cycles", 0)
    start_time = time.time()

    if daemon:
        print(f"后台监控启动（每 {interval}s，企业微信通知已开启）")
        print(f"   日志: {LOG_FILE}")
        log(f"=== 监控启动 (interval={interval}s) ===")
        wx_send("监控已启动",
                f"抽赏监控已启动\n每 {interval} 秒检查一次\n变化时自动通知")
    else:
        print(f"监控启动（每 {interval}s）Ctrl+C 停止")
        print(f"企业微信通知: 开启")

    try:
        while True:
            cycle_start = time.time()
            total_cycles += 1
            remaining, stock = refresh_and_dump()

            if remaining is not None:
                if last is not None and remaining != last:
                    diff = last - remaining
                    if diff > 0:
                        total_drops += diff
                        pct = remaining / stock * 100 if stock else 0
                        msg = (f"减少 {diff} 张: {last} → {remaining}\n"
                               f"剩余: {remaining}/{stock} ({pct:.1f}%)")
                        log(msg)
                        os_notify("抽赏", f"减少 {diff} 张")
                        wx_send("抽赏消耗", msg)
                    else:
                        msg = f"补货 {-diff} 张: {last} → {remaining}"
                        log(msg)
                        if abs(diff) >= 10:
                            wx_send("抽赏补货", msg)

                last = remaining

                if total_cycles % 10 == 0:
                    state["last_remaining"] = remaining
                    state["total_drops"] = total_drops
                    state["total_cycles"] = total_cycles
                    state["last_stock"] = stock
                    state.setdefault("history", []).append({
                        "t": datetime.now().isoformat(),
                        "r": remaining,
                    })
                    if len(state["history"]) > 200:
                        state["history"] = state["history"][-200:]
                    save_state(state)

                if daemon and total_cycles % 720 == 0:
                    elapsed_h = int((time.time() - start_time) // 3600)
                    wx_send("抽赏日报",
                        f"运行 {elapsed_h}h\n"
                        f"当前: {remaining}/{stock}\n"
                        f"累计消耗: {total_drops} 张")

            if not daemon and remaining is not None:
                pct = remaining / stock * 100 if stock else 0
                line = f"[{datetime.now().strftime('%H:%M:%S')}]  {remaining:>4d}/{stock:<4d}  ({pct:.1f}%)"
                print(line)

            elapsed = time.time() - cycle_start
            time.sleep(max(0.0, interval - elapsed))

    except KeyboardInterrupt:
        mins = int((time.time() - start_time) // 60)
        summary = f"运行 {mins} 分, {total_cycles} 轮, 累计降 {total_drops} 张"
        if daemon:
            wx_send("监控已停止", summary)
        print(f"\n{summary}")
        save_state(state)


if __name__ == "__main__":
    main()

辅助脚本 uia_monitor.py（详细解析版，可查看完整奖品列表和中奖记录）:

#!/usr/bin/env python3
"""
淘宝成就抽赏 - 详细解析版
通过 uiautomator dump 读取完整 Accessibility Tree
可查看奖品余量、中奖记录、抽赏按钮等信息

用法:
  python3 uia_monitor.py              # 单次读取
  python3 uia_monitor.py --watch      # 持续监控
  python3 uia_monitor.py --watch --interval 60  # 每60秒
"""
import subprocess, sys, re, json, os, time, argparse
import xml.etree.ElementTree as ET
from datetime import datetime

ADB = "adb"
STATE_FILE = os.path.join(os.path.dirname(__file__), ".uia_state.json")


def adb_shell(cmd):
    r = subprocess.run([ADB, "shell", cmd],
                       capture_output=True, text=True, timeout=30)
    if r.returncode != 0:
        raise RuntimeError(f"ADB: {r.stderr.strip()}")
    return r.stdout.strip()


def check_device():
    r = subprocess.run([ADB, "devices"],
                       capture_output=True, text=True, timeout=10)
    for line in r.stdout.split("\n"):
        if line.strip() and "\tdevice" in line:
            return True
    print("设备未连接")
    sys.exit(1)


def dump_ui():
    adb_shell("uiautomator dump /sdcard/ui.xml")
    xml_text = subprocess.run(
        [ADB, "shell", "cat", "/sdcard/ui.xml"],
        capture_output=True, text=True, timeout=30
    ).stdout
    adb_shell("rm /sdcard/ui.xml")
    return xml_text


def get_all_texts(xml_text):
    """解析 XML，返回所有 TextView 的 (text, bounds)"""
    items = []
    try:
        root = ET.fromstring(xml_text)
    except ET.ParseError:
        return items
    for elem in root.iter():
        text = elem.get("text", "")
        clazz = elem.get("class", "")
        if not text or "TextView" not in clazz:
            continue
        b = list(map(int, re.findall(r'\d+',
                 elem.get("bounds", "0 0 0 0"))))
        if len(b) == 4:
            items.append((text, b[0], b[1], b[2], b[3]))
    return items


def parse_data(xml_text):
    """解析 UI 数据"""
    items = get_all_texts(xml_text)
    result = {
        "total_remaining": None,
        "total_stock": None,
        "prizes": [],
        "recent_wins": [],
        "visible_counts": [],
        "draw_info": {},
    }

    # 解析剩余总张数
    for i, (text, x1, y1, x2, y2) in enumerate(items):
        if text.strip() == "剩余:":
            if i + 1 < len(items):
                try:
                    result["total_remaining"] = int(items[i + 1][0].strip())
                except ValueError:
                    pass
            if i + 2 < len(items):
                m = re.match(r"/(\d+)张", items[i + 2][0])
                if m:
                    result["total_stock"] = int(m.group(1))
            break

    # 解析余量模式 X/Y
    prize_re = re.compile(r"^(\d+)/(\d+)$")
    for text, x1, y1, x2, y2 in items:
        m = prize_re.match(text)
        if m and x2 > x1 + 2 and x2 > 20:
            result["visible_counts"].append({
                "current": int(m.group(1)),
                "total": int(m.group(2)),
                "y": (y1 + y2) // 2,
            })

    # 解析中奖记录
    skip = {"规则", "剩余:", "赏品余量", "奖池赏品",
            "中奖记录", "只赠送不售卖"}
    wins = []
    i = 0
    while i < len(items):
        text, x1, y1, x2, y2 = items[i]
        if re.match(r"^\d{2}-\d{2}$", text) and x1 == 0:
            date_val = text
            time_val = ""
            winner = ""
            prize = ""
            if i + 1 < len(items) and \
               re.match(r"^\d{2}:\d{2}:\d{2}$", items[i+1][0]):
                time_val = items[i+1][0]
                for j in range(i + 2, min(i + 6, len(items))):
                    t, bx1, by1, bx2, by2 = items[j]
                    if bx1 != 0 or bx2 != 0:
                        break
                    if "获得" in t:
                        winner = t.replace("获得", "").strip()
                    elif winner and not t.startswith("/") and t not in skip:
                        prize = t
                if winner and prize:
                    wins.append({"date": date_val, "time": time_val,
                                 "winner": winner, "prize": prize})
        i += 1
    result["recent_wins"] = wins[:50]

    return result


def format_output(data):
    lines = []
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    lines.append(f"═══ 淘宝成就抽赏 @ {ts} ═══")

    tr = data.get("total_remaining")
    ts_val = data.get("total_stock")
    if tr is not None:
        pct = f" ({tr/ts_val*100:.1f}%)" if ts_val else ""
        lines.append(f"剩余    {tr} / {ts_val}{pct}")

    if data.get("visible_counts"):
        lines.append("")
        lines.append("奖品余量:")
        for vc in data["visible_counts"]:
            lines.append(f"  {vc['current']:>3d}/{vc['total']:<4d}")

    if data.get("recent_wins"):
        lines.append("")
        lines.append("最新中奖:")
        for w in data["recent_wins"][:8]:
            lines.append(f"  {w['date']} {w['time']}  "
                         f"{w['winner']} → {w['prize']}")

    return "\n".join(lines)


def load_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f)
    return {}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--watch", "-w", action="store_true")
    parser.add_argument("--interval", "-i", type=int, default=60)
    args = parser.parse_args()

    check_device()
    state = load_state()

    if not args.watch:
        xml_text = dump_ui()
        data = parse_data(xml_text)
        print(format_output(data))
        return

    print(f"监控模式（每 {args.interval}s）")
    try:
        while True:
            xml_text = dump_ui()
            data = parse_data(xml_text)
            output = format_output(data)
            print("\033[2J\033[H", end="")
            print(output)
            time.sleep(args.interval)
    except KeyboardInterrupt:
        print("\n已停止")


if __name__ == "__main__":
    main()

六、配置与使用

6.1 快速上手（只监控 + 弹系统通知）

# 1. 下载脚本
mkdir taobao-chouwan-monitor && cd taobao-chouwan-monitor
# 将上面的 fast_monitor.py 保存到此目录

# 2. 连接手机并打开淘宝抽赏页面
adb devices  # 确认设备已连接

# 3. 前台运行一次测试
python3 fast_monitor.py

# 4. 后台运行
nohup python3 -u fast_monitor.py --daemon --interval=30 \
  > monitor_stdout.log 2>&1 &

# 5. 停止
pkill -f fast_monitor.py

6.2 配置企业微信通知

编辑 fast_monitor.py 头部配置项:

WX_CORPID = "ww****************"  # 你的企业 ID
WX_CORPSECRET = "****"           # 应用 Secret
WX_AGENTID = "1000002"           # 应用 AgentId
WX_USER = "YourUserID"           # 接收人 userid

6.3 调整监控频率

# 每 10 秒（最高频率，注意耗电）
nohup python3 -u fast_monitor.py --daemon --interval=10 \
  > monitor_stdout.log 2>&1 &

# 每 5 分钟（省电模式）
nohup python3 -u fast_monitor.py --daemon --interval=300 \
  > monitor_stdout.log 2>&1 &

6.4 查看详情（奖品列表 + 中奖记录）

python3 uia_monitor.py

输出示例:

═══ 淘宝成就抽赏 @ 2026-07-20 09:40:04 ═══
剩余    209 / 2000 (10.5%)

奖品余量:
    1/1
   27/300
    0/1
  183/1700

最新中奖:
  07-20 09:19:58  阿**i → 萌粒手办盲盒
  07-20 08:58:41  j**g  → 冷变杯-1个
  07-20 08:52:16  柚**6 → 萌粒手办盲盒

七、运行效果

后台运行时，日志文件 monitor.log 记录每次变化:

[2026-07-20 10:05:09] === 监控启动 (interval=30s) ===
[2026-07-20 10:05:09] WX sent: 监控已启动
[2026-07-20 10:10:15] 减少 1 张: 209 → 208
[2026-07-20 10:10:15] WX sent: 抽赏消耗

企业微信收到的通知示例:

🎯 抽赏消耗
减少 1 张: 209 → 208
剩余: 208/2000 (10.4%)

八、注意事项

• 手机屏幕不要关: uiautomator dump 需要屏幕亮着才能正确获取数据。如果手机息屏，先 adb shell input keyevent 26 亮屏后再 dump
• USB 保持连接: 断开后 ADB 无法访问，脚本会报错并发通知提醒
• 耗电: uiautomator dump 每次约 2-3 秒，建议间隔不低于 10 秒。日常挂机 30-60 秒比较合理
• 不依赖网络: 所有数据从本地 ADB 获取，不走网络接口。只有发通知时才需要网络
• 企业微信 token: 自动缓存到 .wx_token.json，有效期 7200 秒（2 小时），到期自动刷新
• 页面结构变化: 如果淘宝更新了页面布局，剩余: 文字的位置可能变化，需要调整 refresh_and_dump() 中的坐标。坐标通过 uiautomator dump 输出的 bounds 获取

更新: 单个 tab 点击不会触发刷新，因为当前 tab 已激活。改为先切换到「奖池赏品」再切回「赏品余量」。代码中 refresh_and_dump() 函数已更新。

九、其他方案对比

方案	难度	可靠性	说明
uiautomator dump	低	高	本方案，零依赖，直接从系统拿文字，推荐
Playwright 浏览器自动化	中	低	需要 GUI 环境，页面结构复杂，选择器难适配
手机抓包 + API 调用	高	中	MTop 签名加密，破解成本高，且签名会过期
OCR（Tesseract/EasyOCR）	中	中	需要安装 OCR 引擎，中文识别准确率取决于字体和背景

简记。