编写一个监测指定浏览器指定网页上的一个元素数字变化并实时通过指定音频设备（防止当前音频设备被独占）报...

python
from pathlib import Path
import json, textwrap, zipfile, os

base = Path("/mnt/data/course_seat_monitor")
base.mkdir(exist_ok=True)

monitor_code = r'''#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Course Seat Monitor
监测已打开浏览器指定页面上的某个元素数字变化，并通过指定音频输出设备报警。

用途示例：
1. 列出音频设备：
   python monitor_course_seats.py --list-audio

2. 测试指定音频设备报警：
   python monitor_course_seats.py --test-audio --device "USB"

3. 按配置开始监测：
   python monitor_course_seats.py --config config.json

注意：
- 只做“读页面 + 本地报警”，不自动抢课、不自动提交表单。
- Chrome/Edge 必须用 --remote-debugging-port 启动，程序才能连接到已经登录的页面。
"""

from __future__ import annotations

import argparse
import json
import math
import re
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional

import numpy as np
import sounddevice as sd
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError


@dataclass
class AlertConfig:
    audio_device: str | int | None = None
    frequency_hz: int = 880
    beep_seconds: float = 0.7
    repeat: int = 20
    gap_seconds: float = 0.2
    samplerate: int = 44100


@dataclass
class MonitorConfig:
    cdp_url: str
    target_url_keyword: str
    selector: str
    selector_type: str = "css"  # css / xpath
    extract_regex: str = r"\d+"
    number_index: int = 0
    poll_interval_sec: float = 3.0
    trigger_when: str = "decrease"  # decrease / increase / changed / below_or_equal / above_or_equal
    threshold: Optional[float] = None
    baseline: Optional[float] = None
    reload_page_each_poll: bool = False
    wait_timeout_ms: int = 3000
    alarm_once: bool = False
    alert: AlertConfig = AlertConfig()
    log_file: str = "monitor.log"


def load_config(path: str | Path) -> MonitorConfig:
    data = json.loads(Path(path).read_text(encoding="utf-8"))
    alert_data = data.get("alert", {}) or {}
    alert = AlertConfig(
        audio_device=alert_data.get("audio_device", None),
        frequency_hz=int(alert_data.get("frequency_hz", 880)),
        beep_seconds=float(alert_data.get("beep_seconds", 0.7)),
        repeat=int(alert_data.get("repeat", 20)),
        gap_seconds=float(alert_data.get("gap_seconds", 0.2)),
        samplerate=int(alert_data.get("samplerate", 44100)),
    )
    return MonitorConfig(
        cdp_url=data.get("cdp_url", "http://127.0.0.1:9222"),
        target_url_keyword=data["target_url_keyword"],
        selector=data["selector"],
        selector_type=data.get("selector_type", "css").lower(),
        extract_regex=data.get("extract_regex", r"\d+"),
        number_index=int(data.get("number_index", 0)),
        poll_interval_sec=float(data.get("poll_interval_sec", 3.0)),
        trigger_when=data.get("trigger_when", "decrease"),
        threshold=data.get("threshold", None),
        baseline=data.get("baseline", None),
        reload_page_each_poll=bool(data.get("reload_page_each_poll", False)),
        wait_timeout_ms=int(data.get("wait_timeout_ms", 3000)),
        alarm_once=bool(data.get("alarm_once", False)),
        alert=alert,
        log_file=data.get("log_file", "monitor.log"),
    )


def log_line(msg: str, log_file: str | None = None) -> None:
    line = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}"
    print(line, flush=True)
    if log_file:
        Path(log_file).write_text(
            (Path(log_file).read_text(encoding="utf-8") if Path(log_file).exists() else "") + line + "\n",
            encoding="utf-8",
        )


def list_audio_devices() -> None:
    print(sd.query_devices())


def find_output_device(device: str | int | None) -> Optional[int]:
    """
    返回 sounddevice 可用的输出设备 index。
    device 可以是：
    - None 或空字符串：使用系统默认输出设备
    - 数字：设备序号
    - 字符串：按设备名称不区分大小写模糊匹配
    """
    if device is None or device == "":
        return None

    devices = sd.query_devices()

    if isinstance(device, int):
        info = devices[device]
        if info["max_output_channels"] < 1:
            raise ValueError(f"设备 {device} 不是输出设备：{info['name']}")
        return device

    device_str = str(device).strip()
    if device_str.isdigit():
        idx = int(device_str)
        info = devices[idx]
        if info["max_output_channels"] < 1:
            raise ValueError(f"设备 {idx} 不是输出设备：{info['name']}")
        return idx

    needle = device_str.lower()
    matches: list[tuple[int, str]] = []
    for i, info in enumerate(devices):
        if info["max_output_channels"] >= 1 and needle in info["name"].lower():
            matches.append((i, info["name"]))

    if not matches:
        raise ValueError(f"找不到包含名称 {device_str!r} 的输出设备。请先运行 --list-audio。")

    if len(matches) > 1:
        print("匹配到多个输出设备，将使用第一个；如需精确指定，请用设备序号：")
        for i, name in matches:
            print(f"  {i}: {name}")

    return matches[0][0]


def play_alarm(alert: AlertConfig) -> None:
    device_index = find_output_device(alert.audio_device)
    samplerate = alert.samplerate

    # 生成带淡入淡出的正弦波，减少爆音
    t = np.linspace(0, alert.beep_seconds, int(samplerate * alert.beep_seconds), endpoint=False)
    wave = 0.45 * np.sin(2 * math.pi * alert.frequency_hz * t)
    fade_len = min(int(0.02 * samplerate), len(wave) // 4)
    if fade_len > 0:
        fade = np.linspace(0, 1, fade_len)
        wave[:fade_len] *= fade
        wave[-fade_len:] *= fade[::-1]

    for _ in range(max(1, alert.repeat)):
        sd.play(wave.astype(np.float32), samplerate=samplerate, device=device_index, blocking=True)
        if alert.gap_seconds > 0:
            time.sleep(alert.gap_seconds)


def parse_number(text: str, pattern: str, number_index: int = 0) -> float:
    matches = re.findall(pattern, text)
    if not matches:
        raise ValueError(f"文本中没有匹配到数字。文本={text!r}，regex={pattern!r}")

    if number_index >= len(matches):
        raise ValueError(f"匹配到 {len(matches)} 个数字，但 number_index={number_index} 越界。匹配结果={matches}")

    raw = matches[number_index]
    if isinstance(raw, tuple):
        raw = next((x for x in raw if x), "")

    # 支持 1,234 / 12.5 / 中文文本中的数字
    cleaned = str(raw).replace(",", "").strip()
    return float(cleaned)


def should_trigger(current: float, previous: float, cfg: MonitorConfig) -> bool:
    mode = cfg.trigger_when.lower()
    threshold = cfg.threshold

    if mode == "decrease":
        return current < previous
    if mode == "increase":
        return current > previous
    if mode == "changed":
        return current != previous
    if mode == "below_or_equal":
        if threshold is None:
            raise ValueError("trigger_when=below_or_equal 时必须设置 threshold")
        return current <= float(threshold)
    if mode == "above_or_equal":
        if threshold is None:
            raise ValueError("trigger_when=above_or_equal 时必须设置 threshold")
        return current >= float(threshold)

    raise ValueError(f"不支持的 trigger_when: {cfg.trigger_when}")


def find_target_page(browser, target_url_keyword: str):
    all_pages = []
    for context in browser.contexts:
        all_pages.extend(context.pages)

    matched = [p for p in all_pages if target_url_keyword in p.url]
    if not matched:
        print("当前可见页面：")
        for p in all_pages:
            print("  -", p.url)
        raise RuntimeError(f"没有找到 URL 包含 {target_url_keyword!r} 的页面。请确认该页面已在调试浏览器中打开。")

    if len(matched) > 1:
        print("匹配到多个页面，将使用第一个：")
        for p in matched:
            print("  -", p.url)

    return matched[0]


def read_value_from_page(page, cfg: MonitorConfig) -> tuple[float, str]:
    if cfg.reload_page_each_poll:
        page.reload(wait_until="domcontentloaded", timeout=cfg.wait_timeout_ms)

    if cfg.selector_type == "xpath":
        locator = page.locator(f"xpath={cfg.selector}")
    elif cfg.selector_type == "css":
        locator = page.locator(cfg.selector)
    else:
        raise ValueError("selector_type 只能是 css 或 xpath")

    text = locator.first.text_content(timeout=cfg.wait_timeout_ms)
    if text is None:
        text = ""
    value = parse_number(text, cfg.extract_regex, cfg.number_index)
    return value, text.strip()


def monitor(config_path: str | Path) -> None:
    cfg = load_config(config_path)

    log_line("启动监测程序。", cfg.log_file)
    log_line(f"CDP: {cfg.cdp_url}", cfg.log_file)
    log_line(f"页面 URL 关键词: {cfg.target_url_keyword}", cfg.log_file)
    log_line(f"选择器: {cfg.selector_type}={cfg.selector}", cfg.log_file)

    with sync_playwright() as p:
        browser = p.chromium.connect_over_cdp(cfg.cdp_url)
        page = find_target_page(browser, cfg.target_url_keyword)
        log_line(f"已连接页面: {page.url}", cfg.log_file)

        value, text = read_value_from_page(page, cfg)
        previous = float(cfg.baseline) if cfg.baseline is not None else value
        log_line(f"初始读取: value={value}, text={text!r}, baseline={previous}", cfg.log_file)

        while True:
            try:
                current, raw_text = read_value_from_page(page, cfg)
                log_line(f"当前读取: value={current}, previous={previous}, text={raw_text!r}", cfg.log_file)

                if should_trigger(current, previous, cfg):
                    log_line(f"触发报警：{cfg.trigger_when}，previous={previous}, current={current}", cfg.log_file)
                    try:
                        play_alarm(cfg.alert)
                    except Exception as audio_error:
                        # 音频设备被独占/不可用时，至少在命令行持续响铃和报错
                        log_line(f"音频报警失败：{audio_error!r}", cfg.log_file)
                        for _ in range(10):
                            print("\a", end="", flush=True)
                            time.sleep(0.4)

                    if cfg.alarm_once:
                        log_line("alarm_once=true，程序退出。", cfg.log_file)
                        return

                previous = current
                time.sleep(cfg.poll_interval_sec)

            except KeyboardInterrupt:
                log_line("用户中断，程序退出。", cfg.log_file)
                return

            except PlaywrightTimeoutError as e:
                log_line(f"页面元素读取超时：{e!r}", cfg.log_file)
                time.sleep(cfg.poll_interval_sec)

            except Exception as e:
                log_line(f"读取失败：{e!r}", cfg.log_file)
                time.sleep(cfg.poll_interval_sec)


def main() -> None:
    parser = argparse.ArgumentParser(description="监测指定浏览器指定网页元素数字变化并报警")
    parser.add_argument("--config", default="config.json", help="配置文件路径，默认 config.json")
    parser.add_argument("--list-audio", action="store_true", help="列出可用音频设备")
    parser.add_argument("--test-audio", action="store_true", help="测试报警声音")
    parser.add_argument("--device", default=None, help="测试报警用的音频设备序号或名称片段")
    args = parser.parse_args()

    if args.list_audio:
        list_audio_devices()
        return

    if args.test_audio:
        play_alarm(AlertConfig(audio_device=args.device, repeat=3, gap_seconds=0.25))
        return

    monitor(args.config)


if __name__ == "__main__":
    main()
'''

config_example = {
    "cdp_url": "http://127.0.0.1:9222",
    "target_url_keyword": "test_page.html",
    "selector_type": "css",
    "selector": "#selected-count",
    "extract_regex": "\\d+",
    "number_index": 0,
    "poll_interval_sec": 2,
    "trigger_when": "decrease",
    "threshold": None,
    "baseline": None,
    "reload_page_each_poll": False,
    "wait_timeout_ms": 3000,
    "alarm_once": False,
    "log_file": "monitor.log",
    "alert": {
        "audio_device": "",
        "frequency_hz": 880,
        "beep_seconds": 0.7,
        "repeat": 10,
        "gap_seconds": 0.2,
        "samplerate": 44100
    }
}

test_page = r'''<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8" />
  <title>test_page.html - 选课人数监测测试页</title>
  <style>
    body { font-family: system-ui, "Microsoft YaHei", sans-serif; margin: 40px; line-height: 1.7; }
    .card { border: 1px solid #ddd; border-radius: 12px; padding: 24px; max-width: 560px; }
    #selected-count { font-size: 48px; font-weight: 700; color: #b00020; }
    button { font-size: 18px; margin-right: 12px; padding: 8px 14px; }
    code { background: #f4f4f4; padding: 2px 5px; border-radius: 4px; }
  </style>
</head>
<body>
  <div class="card">
    <h1>选课人数测试页</h1>
    <p>已选人数：<span id="selected-count">30</span> / 30</p>
    <button onclick="change(-1)">模拟有人退课：人数下降</button>
    <button onclick="change(1)">模拟有人选课：人数上升</button>
    <button onclick="setCount(30)">重置为 30</button>
    <p>程序默认监测 CSS 选择器：<code>#selected-count</code>。点击“人数下降”后应触发报警。</p>
  </div>
  <script>
    let n = 30;
    function change(delta) {
      n += delta;
      if (n < 0) n = 0;
      setCount(n);
    }
    function setCount(v) {
      n = v;
      document.querySelector("#selected-count").innerText = String(n);
    }
  </script>
</body>
</html>
'''

requirements = """playwright>=1.45
sounddevice>=0.5
numpy>=1.24
"""

readme = r'''# Course Seat Monitor 使用说明

这个程序用于监测“已打开的 Chrome/Edge 页面”中指定元素的数字变化，并通过指定音频设备报警。
它只读取页面文本并报警，不会自动点击、抢课或提交表单。

## 1. 安装

```bash
python -m pip install -r requirements.txt

2. 启动可连接的 Chrome/Edge

Windows Chrome 示例：

bat
start "" "C:\Program Files\Google\Chrome\Application\chrome.exe" --remote-debugging-port=9222 --user-data-dir="%TEMP%\course-monitor-chrome"

Windows Edge 示例：

bat
start "" "C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe" --remote-debugging-port=9222 --user-data-dir="%TEMP%\course-monitor-edge"

打开后，在这个浏览器窗口里登录选课网站，并进入需要监测的课程页面。

3. 音频设备测试

列出设备：

bash
python monitor_course_seats.py --list-audio

测试指定设备，device 可以填序号或名称片段：

bash
python monitor_course_seats.py --test-audio --device 3
python monitor_course_seats.py --test-audio --device "USB"

如果你担心当前默认音频设备被播放器、游戏、会议软件独占，请选择另一个输出设备，例如 USB 小音箱、显示器 HDMI 音频、蓝牙耳机等。

4. 本地测试页测试

1. 用上面带 remote-debugging-port 的 Chrome/Edge 打开 test_page.html。
2. 保持 config.json 默认配置。
3. 运行：

bash
python monitor_course_seats.py --config config.json

4. 在测试页点击“模拟有人退课：人数下降”，应触发报警。

5. 配置真实选课网站

把 config.json 改成类似这样：

json
{
  "cdp_url": "http://127.0.0.1:9222",
  "target_url_keyword": "courseSelection",
  "selector_type": "css",
  "selector": "#selected-count",
  "extract_regex": "\\d+",
  "number_index": 0,
  "poll_interval_sec": 3,
  "trigger_when": "decrease",
  "threshold": null,
  "baseline": null,
  "reload_page_each_poll": false,
  "wait_timeout_ms": 3000,
  "alarm_once": false,
  "log_file": "monitor.log",
  "alert": {
    "audio_device": "USB",
    "frequency_hz": 880,
    "beep_seconds": 0.7,
    "repeat": 20,
    "gap_seconds": 0.2,
    "samplerate": 44100
  }
}

字段说明：

• target_url_keyword：浏览器标签页 URL 中的一段关键词，用来定位要监测的页面。
• selector_type：css 或 xpath。
• selector：人数所在元素的 CSS/XPath。
• extract_regex：从元素文本中提取数字，默认提取整数。
• number_index：如果文本里有多个数字，选择第几个，0 表示第一个。
  • 例如文本是 已选 29 / 30，默认会取 29。
• trigger_when：
  • decrease：当前数字小于上次数字时报。
  • increase：当前数字大于上次数字时报。
  • changed：数字变化就报。
  • below_or_equal：当前数字 <= threshold 时报。
  • above_or_equal：当前数字 >= threshold 时报。
• baseline：默认 null，表示第一次读取值作为基准。也可以手动填一个数字。
• reload_page_each_poll：是否每次轮询都刷新页面。一般先设 false；如果页面人数不会自动刷新，再考虑 true。
• poll_interval_sec：轮询间隔。建议不要太低，避免给选课系统造成压力或触发风控。
• alert.audio_device：音频输出设备序号或名称片段；空字符串表示默认输出设备。

6. 怎么找 selector

在浏览器里按 F12，点击左上角“选择元素”，点选页面上的“已选人数”数字区域。 优先选择稳定的 id 或 data-* 属性，例如：

css
#selected-count
[data-testid="selected-count"]
.course-capacity .selected

如果没有稳定 CSS，可以右键元素复制 XPath，并把 selector_type 改成 xpath。 '''

run_bat = r'''@echo off chcp 65001 >nul python monitor_course_seats.py --config config.json pause '''

(base / "monitor_course_seats.py").write_text(monitor_code, encoding="utf-8") (base / "config.json").write_text(json.dumps(config_example, ensure_ascii=False, indent=2), encoding="utf-8") (base / "requirements.txt").write_text(requirements, encoding="utf-8") (base / "test_page.html").write_text(test_page, encoding="utf-8") (base / "README.md").write_text(readme, encoding="utf-8") (base / "run_windows.bat").write_text(run_bat, encoding="utf-8")

zip_path = Path("/mnt/data/course_seat_monitor.zip") if zip_path.exists(): zip_path.unlink() with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: for p in base.iterdir(): zf.write(p, arcname=p.name)

zip_path.as_posix()

'/mnt/data/course_seat_monitor.zip'

python
from pathlib import Path
import zipfile, re

base = Path("/mnt/data/course_seat_monitor")
path = base / "monitor_course_seats.py"
text = path.read_text(encoding="utf-8")
text = text.replace("from dataclasses import dataclass", "from dataclasses import dataclass, field")
text = text.replace("    alert: AlertConfig = AlertConfig()\n", "    alert: AlertConfig = field(default_factory=AlertConfig)\n")
path.write_text(text, encoding="utf-8")

zip_path = Path("/mnt/data/course_seat_monitor.zip")
if zip_path.exists():
    zip_path.unlink()
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
    for p in base.iterdir():
        zf.write(p, arcname=p.name)

print(zip_path)