#!/usr/bin/env python3
"""电影院空间音频 —— 看门狗。

30 秒扫一轮, 逐项核对 config/组件配置单.json 里的期望状态, **发现问题直接修**
(不是只报警)。同类问题有冷却期, 连续修不好就转为只报警, 不再反复折腾。

用法:
    python3 音频守护.py            # 30 秒循环(前台, 可实时看)
    python3 音频守护.py --once     # 只跑一轮(给 cron 用; 正常时几乎无输出)
    python3 音频守护.py --check    # 只检查不修(干跑)
    python3 音频守护.py --interval 10

★ 杀进程一律用 /proc/<pid>/cmdline 精确判断, 绝不用 pgrep -f(会匹配到自己那条命令行
  把自己杀掉, 2026-09-13 实测 exit -15)。
"""
from __future__ import annotations

import argparse
import json
import os
import struct
import subprocess
import sys
import time
from typing import Any

ROOT = os.path.dirname(os.path.abspath(__file__))
CFG_PATH = os.path.join(ROOT, "config", "组件配置单.json")
CLI = os.path.join(ROOT, "空间音频")
PANEL_SH = os.path.join(ROOT, "web", "启动.sh")
STATE = os.path.expanduser("~/.local/state/cinema-spatial")
LOG_PATH = os.path.join(STATE, "guardian.log")
SHM = "/dev/shm/collaplex-loudness"


# ---------------------------------------------------------------- 基础设施

def sh(*args: str, timeout: int = 25) -> str:
    """跑命令取 stdout; 失败返回空串(看门狗不许因为一条命令失败就崩)。"""
    try:
        r = subprocess.run(args, capture_output=True, text=True, timeout=timeout)
        return r.stdout or ""
    except (subprocess.TimeoutExpired, OSError):
        return ""


def log(msg: str, quiet: bool = False) -> None:
    """写日志(带时间戳) + 打到屏幕。"""
    line = "%s  %s" % (time.strftime("%m-%d %H:%M:%S"), msg)
    if not quiet:
        print(line, flush=True)
    try:
        os.makedirs(STATE, exist_ok=True)
        with open(LOG_PATH, "a", encoding="utf-8") as f:
            f.write(line + "\n")
    except OSError:
        pass


def load_cfg() -> dict[str, Any]:
    try:
        with open(CFG_PATH, encoding="utf-8") as f:
            data: dict[str, Any] = json.load(f)
        return data
    except (OSError, json.JSONDecodeError) as e:
        log("! 读不到配置单 %s (%s), 用内置默认值" % (CFG_PATH, e))
        return {}


# ---------------------------------------------------------------- 采集

_DUMP_CACHE: dict[str, Any] = {"t": 0.0, "data": []}


def dump_nodes(max_age: float = 2.0) -> list[dict[str, Any]]:
    """pw-dump 结果缓存(默认 2 秒) —— 一轮检查要查十几次节点, 不能每次都跑 pw-dump。"""
    now = time.time()
    cached = _DUMP_CACHE["data"]
    if now - float(_DUMP_CACHE["t"]) < max_age and cached:
        return list(cached)
    try:
        data: list[dict[str, Any]] = json.loads(sh("pw-dump", timeout=25) or "[]")
    except json.JSONDecodeError:
        data = []
    _DUMP_CACHE["t"] = now
    _DUMP_CACHE["data"] = data
    return list(data)


def node_ids(name: str) -> list[int]:
    """某个 node.name 的全部 id(正常应恰好 1 个)。"""
    out: list[int] = []
    for o in dump_nodes():
        props = (o.get("info") or {}).get("props") or {}
        if props.get("node.name") == name:
            out.append(int(o["id"]))
    return out


def vol_of(nid: int) -> tuple[float, bool]:
    """(音量, 是否静音); 读不到返回 (-1.0, False)。"""
    txt = sh("wpctl", "get-volume", str(nid), timeout=8)
    if not txt:
        return (-1.0, False)
    vol = -1.0
    for tok in txt.replace(":", " ").split():
        try:
            vol = float(tok)
            break
        except ValueError:
            continue
    return (vol, "MUTED" in txt.upper())


def default_sink() -> str:
    txt = sh("wpctl", "inspect", "@DEFAULT_AUDIO_SINK@", timeout=8)
    for line in txt.split("\n"):
        if "node.name" in line:
            return line.split('"')[1] if '"' in line else ""
    return ""


def dsp_procs() -> list[int]:
    """DSP 进程 pid 列表 —— 走 /proc 精确匹配, 不用 pgrep -f(会自杀)。"""
    out: list[int] = []
    try:
        entries = os.listdir("/proc")
    except OSError:
        return out
    for entry in entries:
        if not entry.isdigit():
            continue
        try:
            with open("/proc/%s/cmdline" % entry, "rb") as f:
                cmd = f.read().replace(b"\0", b" ").decode("utf-8", "replace")
        except OSError:
            continue
        if "loudness_norm.py" in cmd:
            out.append(int(entry))
    return sorted(out)


def slots() -> list[tuple[int, float, float]]:
    """每个槽位 (块数n, 距今秒, 增益dB); 读不到返回空。

    ★ 时间戳必须用 time.monotonic() 相减 —— DSP 写进共享区 +16 的是 monotonic
    (它内部判活也是这么比的)。用 time.time() 会得到十几亿秒的荒谬差值, 守护就会
    误判"槽位僵死"然后疯狂重建。2026-09-13 干跑时抓到。
    """
    out: list[tuple[int, float, float]] = []
    try:
        with open(SHM, "rb") as f:
            raw = f.read()
    except OSError:
        return out
    for i in range(2):
        b = i * 32
        try:
            gain = struct.unpack_from("<f", raw, b + 12)[0]
            ts = struct.unpack_from("<d", raw, b + 16)[0]
            n = struct.unpack_from("<I", raw, b + 24)[0]
        except struct.error:
            return out
        out.append((n, time.monotonic() - ts, gain))
    return out


def out_links() -> list[str]:
    """cinema_spatial_up_out:output_FL/FR 连到了谁。"""
    out: list[str] = []
    for block in sh("pw-link", "-l", timeout=10).split("\n\n"):
        if "cinema_spatial_up_out:output_FL" in block or "cinema_spatial_up_out:output_FR" in block:
            for line in block.split("\n"):
                if "|->" in line:
                    out.append(line.strip().lstrip("|-> ").strip())
    return out


def panel_ok() -> bool:
    txt = sh("curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
             "--noproxy", "*", "--max-time", "4", "http://127.0.0.1:8788/", timeout=10)
    return txt.strip() == "200"


def phys_name() -> str:
    """物理输出设备全名(按配置单里的前缀匹配, 因为 profile 变了名字会变)。"""
    for o in dump_nodes():
        props = (o.get("info") or {}).get("props") or {}
        name = str(props.get("node.name") or "")
        if props.get("media.class") == "Audio/Sink" and "EDIFIER" in name:
            return name
    return ""


# ---------------------------------------------------------------- 修复动作

def fix_kill_dsp() -> str:
    """SIGKILL 掉全部 DSP 进程(重建会起干净的一对)。"""
    killed = 0
    for pid in dsp_procs():
        try:
            os.kill(pid, 9)
            killed += 1
        except OSError:
            pass
    time.sleep(1.0)
    return "杀掉 %d 个 DSP 进程" % killed


def fix_rebuild() -> str:
    """重建链路(CLI 里已含 kill_old_dsp, 重启 PipeWire 前先收旧 DSP)。"""
    if not os.path.exists(CLI):
        return "! 找不到 %s" % CLI
    out = subprocess.run(["bash", CLI, "重建"], capture_output=True, text=True,
                         timeout=200, cwd=ROOT).stdout or ""
    tail = [ln.strip() for ln in out.split("\n") if ln.strip()][-2:]
    return "重建链路 (%s)" % (" / ".join(tail) if tail else "完成")


def fix_set_default(name: str) -> str:
    ids = node_ids(name)
    if not ids:
        return "! 找不到节点 %s, 改为重建" % name
    sh("wpctl", "set-default", str(ids[0]), timeout=10)
    return "把默认输出切回 %s (id %d)" % (name, ids[0])


def fix_unit_gain(name: str, nid: int) -> str:
    sh("wpctl", "set-volume", str(nid), "1.000", timeout=10)
    return "把 %s 音量纠回 1.000" % name


def fix_panel() -> str:
    if not os.path.exists(PANEL_SH):
        return "! 找不到 %s" % PANEL_SH
    subprocess.Popen(["bash", PANEL_SH], cwd=ROOT, stdout=subprocess.DEVNULL,
                     stderr=subprocess.DEVNULL, start_new_session=True)
    return "拉起面板 (8788)"


def fix_link() -> str:
    """尝试运行期重连(比重建温和)。用"连完再查"判断成败 —— pw-link 成功也不打输出。"""
    dev = phys_name()
    if not dev:
        return "! 找不到物理设备"
    for ch in ("FL", "FR"):
        sh("pw-link", "cinema_spatial_up_out:output_%s" % ch, "%s:playback_%s" % (dev, ch), timeout=10)
    time.sleep(0.8)
    links = out_links()
    if ("%s:playback_FL" % dev in " ".join(links)) and ("%s:playback_FR" % dev in " ".join(links)):
        return "重连成功 → %s" % dev.split(".")[-1]
    return "! 重连失败"


# ---------------------------------------------------------------- 检查项

class Problem:
    """一个问题: 类型(用于冷却) + 描述 + 修复动作。"""

    def __init__(self, kind: str, level: str, desc: str, fix: Any) -> None:
        self.kind = kind          # 冷却键
        self.level = level        # 🔴 / 🟡 / 🟢
        self.desc = desc
        self.fix = fix            # 可调用对象, 返回修复说明


def check_all(cfg: dict[str, Any]) -> list[str]:
    """返回问题列表(每项是可读的中文描述)。"""
    probs: list[str] = []
    want_procs = int((cfg.get("DSP") or {}).get("进程数", 2))
    fresh_s = float((cfg.get("DSP") or {}).get("槽位新鲜秒", 5))
    want_sink = str(cfg.get("默认输出", {}).get("必须是", "cinema_spatial_up_sink"))
    min_prio = int(((cfg.get("节点") or {}).get("虚拟声卡最低优先级", 2000)))
    unit_nodes: list[str] = list((cfg.get("节点") or {}).get("必须单位增益", []))
    all_nodes: list[str] = list((cfg.get("节点") or {}).get("全部", []))
    v_sinks: list[str] = list((cfg.get("节点") or {}).get("虚拟声卡", []))

    # 1) DSP 进程数
    procs = dsp_procs()
    if len(procs) != want_procs:
        probs.append("🔴 DSP 进程数 = %d (应为 %d)%s"
                     % (len(procs), want_procs, " —— 孤儿抢共享槽位会整链静音" if len(procs) > want_procs else ""))

    # 2) 槽位新鲜度
    for i, (n, age, _g) in enumerate(slots()):
        if age > fresh_s:
            probs.append("🔴 DSP 槽%d 已 %.0f 秒没更新(僵死) —— pipe 同步管道会卡死" % (i, age))

    # 3) 节点份数
    for name in all_nodes:
        cnt = len(node_ids(name))
        if cnt != 1:
            probs.append("🟡 节点 %s 出现 %d 份(应为 1 份) —— 双份会各建一套链" % (name, cnt))

    # 4) 中间级必须单位增益
    for name in unit_nodes:
        for nid in node_ids(name):
            v, _m = vol_of(nid)
            if v >= 0.0 and abs(v - 1.0) > 0.005:
                probs.append("🟡 %s 音量 = %.3f (应为 1.000) —— 中间级被写低会凭空掉十几 dB" % (name, v))

    # 5) 默认输出
    got = default_sink()
    if got and got != want_sink:
        probs.append("🔴 默认输出是 %s (应为 %s) —— 声音跑到别的设备去了" % (got, want_sink))

    # 6) 虚拟声卡优先级
    for name in v_sinks:
        for o in dump_nodes():
            props = (o.get("info") or {}).get("props") or {}
            if props.get("node.name") != name:
                continue
            prio = props.get("priority.session")
            try:
                pv = int(prio) if prio is not None else -1
            except (TypeError, ValueError):
                pv = -1
            if pv < min_prio:
                probs.append("🟡 %s 优先级 = %s (< %d) —— 默认设备会被 HDMI 抢走且点不回来"
                             % (name, prio, min_prio))

    # 7) 链输出连线
    links = out_links()
    dev = phys_name()
    if dev:
        need = ["%s:playback_FL" % dev, "%s:playback_FR" % dev]
        miss = [d for d in need if not any(d in l for l in links)]
        if miss:
            probs.append("🔴 链输出没连到物理设备(缺 %s) —— 看着全对却静音" % ", ".join(miss))
    elif links == []:
        probs.append("🔴 链输出没有任何连线, 且找不到物理设备")

    # 8) 面板
    if not panel_ok():
        probs.append("🟢 面板 8788 无响应")

    # 9) 物理输出被静音 / 近乎归零
    # ★★ 2026-09-14 修正: 上一版按"物理输出应等于虚拟声卡"去纠, 结果把老板调好的 0.35
    #    顶到 1.0 → 进削波保护、动态被压扁 —— 面板"波形顶满"而听感反而更小(老 bug 复现)。
    #    实际约定: **物理输出音量是个档位**(存档里多少就是多少, 0.35 就很响), 响度由 DSP
    #    归一化保证。所以这里只兜"静音 / 近乎归零"这种明显故障, **绝不改档位数值**。
    pn = phys_name()
    if pn:
        for pid in node_ids(pn):
            pv, pm = vol_of(pid)
            if pm:
                probs.append("🔴 物理输出 %s 被静音 —— 链上全对却没声音" % pn)
            elif pv >= 0.0 and pv < 0.02:
                probs.append("🔴 物理输出音量 %.3f (近乎归零) —— 只剩这一级还有问题" % pv)
            break

    return probs


# ---------------------------------------------------------------- 主流程

def run_round(cfg: dict[str, Any], dry: bool, cd: dict[str, float], cooldown_s: float,
              fails: dict[str, int]) -> int:
    """跑一轮检查+修复, 返回问题数。"""
    probs = check_all(cfg)
    if not probs:
        return 0
    for desc in probs:
        key, action = _pick_fix(desc)
        last = cd.get(key, 0.0)
        if time.time() - last < cooldown_s:
            log("%s (冷却中, %.0fs 后再处理)" % (desc, cooldown_s - (time.time() - last)))
            continue
        if fails.get(key, 0) >= 3:
            log("%s  ! 已连续 3 次修不好, 暂停自动修复(只报警)" % desc)
            continue
        if dry:
            log("%s  [--check 干跑, 不动手] %s" % (desc, action))
            continue
        log("%s  → 修复: %s" % (desc, action))
        cd[key] = time.time()
        try:
            result = _do_fix(key, cfg)
        except Exception as e:                    # 看门狗不许因为一次修复失败就崩
            result = "修复异常: %s" % e
        log("    %s" % result)
        if "!" in result or "失败" in result or "异常" in result:
            fails[key] = fails.get(key, 0) + 1
        else:
            fails[key] = 0
    return len(probs)


def _pick_fix(desc: str) -> tuple[str, str]:
    """从问题描述挑出 (冷却键, 将要做什么)。"""
    if "DSP 进程数" in desc or "僵死" in desc:
        return ("dsp", "杀掉全部 DSP 后重建链路")
    if "没连到物理设备" in desc or "没有任何连线" in desc:
        return ("link", "运行期 pw-link 重连, 失败则重建")
    if "默认输出是" in desc:
        return ("default", "把默认输出切回虚拟声卡")
    if "优先级" in desc:
        return ("prio", "重建配置(生成器带 SINK_PRIO)")
    if "份(应为 1 份)" in desc:
        return ("dup", "重建链路(生成器会清理同名 conf)")
    if "应为 1.000" in desc:
        return ("unitgain", "把中间级音量纠回 1.000")
    if "物理输出" in desc:
        return ("physvol", "把物理输出对齐到虚拟声卡(音量真源)")
    if "面板" in desc:
        return ("panel", "拉起面板")
    return ("misc", "重建链路")


def _do_fix(key: str, cfg: dict[str, Any]) -> str:
    """执行修复。"""
    if key == "dsp":
        killed = fix_kill_dsp()
        return "%s; %s" % (killed, fix_rebuild())
    if key == "link":
        r = fix_link()
        if "!" in r or "失败" in r:
            return "%s; 回落 -> %s" % (r, fix_rebuild())
        return r
    if key == "default":
        want = str(cfg.get("默认输出", {}).get("必须是", "cinema_spatial_up_sink"))
        r = fix_set_default(want)
        if r.startswith("!"):
            return "%s; 回落 -> %s" % (r, fix_rebuild())
        return r
    if key in ("prio", "dup", "misc"):
        return fix_rebuild()
    if key == "unitgain":
        done: list[str] = []
        for name in (cfg.get("节点") or {}).get("必须单位增益", []):
            for nid in node_ids(name):
                v, _m = vol_of(nid)
                if v >= 0.0 and abs(v - 1.0) > 0.005:
                    done.append(fix_unit_gain(name, nid))
        return "; ".join(done) if done else "无需纠正"
    if key == "panel":
        return fix_panel()
    if key == "physvol":
        return fix_phys_volume(cfg)
    return fix_rebuild()


def fix_phys_volume(cfg: dict[str, Any]) -> str:
    """兜"物理输出被静音 / 近乎归零"。

    ★★ 2026-09-14 修正: 本函数**不再把物理输出"对齐"到虚拟声卡** —— 那是错的假设。
    物理输出音量是一个**档位**(WirePlumber 存档里多少就是多少, 实测 0.35 就很响),
    响度由 DSP 归一化负责。把它顶到 1.0 会进削波保护、动态被压扁: 面板波形顶满
    而听感反而更小(老板当场抓出来的老 bug 复现)。所以这里只解除静音, 且只在
    "近乎归零"时把档位救到一个安全值, 正常档位一律不动。
    """
    pn = phys_name()
    if not pn:
        return "! 找不到物理输出"
    pids = node_ids(pn)
    if not pids:
        return "! 物理输出节点不存在"
    pid = pids[0]
    before, bm = vol_of(pid)
    if bm:
        sh("wpctl", "set-mute", str(pid), "0")
        after, am = vol_of(pid)
        if am:
            return "! 静音解不掉(仍静音)"
        return "物理输出原本静音 -> 已解除(音量 %.3f 不动)" % after
    if 0.0 <= before < 0.02:
        sh("wpctl", "set-volume", str(pid), "0.35")
        after, _am = vol_of(pid)
        return "物理输出 %.3f 近乎归零 -> 救回安全档位 %.3f" % (before, after)
    return "物理输出 %.3f 正常, 无需处理(档位不归看门狗管)" % before


def main() -> None:
    ap = argparse.ArgumentParser(description="电影院空间音频看门狗")
    ap.add_argument("--once", action="store_true", help="只跑一轮")
    ap.add_argument("--check", action="store_true", help="只检查不修")
    ap.add_argument("--interval", type=float, default=0.0, help="轮询间隔秒(默认取配置单)")
    args = ap.parse_args()

    cfg = load_cfg()
    guard = cfg.get("看门狗") or {}
    interval = args.interval or float(guard.get("间隔秒", 30))
    cooldown_s = float(guard.get("冷却秒", 90))

    cd: dict[str, float] = {}
    fails: dict[str, int] = {}

    log("看门狗启动: 间隔 %.0fs, 冷却 %.0fs, %s"
        % (interval, cooldown_s, "干跑模式" if args.check else "自动修复"))
    if args.once or args.check:
        n = run_round(cfg, args.check, cd, cooldown_s, fails)
        log("本轮检查完成: %s" % ("一切正常" if n == 0 else "发现 %d 个问题" % n))
        return

    silent_ok = 0
    while True:
        try:
            n = run_round(cfg, False, cd, cooldown_s, fails)
            if n == 0:
                silent_ok += 1
                if silent_ok % 20 == 1:            # 每 10 分钟留一条心跳, 证明还活着
                    log("心跳: 链路正常 (DSP %d 进程, 默认 %s)"
                        % (len(dsp_procs()), default_sink() or "?"))
        except KeyboardInterrupt:
            log("看门狗退出(手动中断)")
            return
        except Exception as e:                     # 兜底: 不许崩
            log("! 本轮异常: %s" % e)
        time.sleep(interval)


if __name__ == "__main__":
    sys.exit(main())