ykxmz/_tests/diag_all_commands.py

# -*- coding: utf-8 -*-
"""逐条实测：启动打包后的 EXE，把**手表会发的每一个命令**都真发一遍。

用来定位用户报的"功能页连不上电脑"、"语音页全不能用"、"很多功能不能用"。
每条都记耗时和结果，慢的、失败的、超时的一目了然。
"""
from __future__ import annotations

import json
import shutil
import socket
import subprocess
import sys
import time
from pathlib import Path

ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote")
EXE = ROOT / "dist_exe" / "ykxmz" / "ykxmz.exe"
PORT = 8911

# 手表上真正会发的命令（从 MainActivity 里抓的）
COMMANDS = [
    ("ping", {}),
    ("feature_list", {}),
    ("voice_list", {}),
    ("audio_list", {}),
    ("perf", {}),
    ("disk", {}),
    ("battery", {}),
    ("proc_list", {}),
    ("clip_history", {}),
    ("info", {}),
    ("clip_get", {}),
    ("say", {"text": "测试"}),
    ("notify", {"title": "标题", "text": "内容"}),
    ("say_clip", {}),
    ("feature_run", {"name": "world_clock", "params": {}}),
    ("feature_run", {"name": "uptime", "params": {}}),
    ("feature_run", {"name": "note_add", "params": {"text": "x"}}),
    ("feature_run", {"name": "note_list", "params": {}}),
    ("feature_run", {"name": "restrict_status", "params": {}}),
    ("feature_run", {"name": "site_list", "params": {}}),
    ("media", {"action": "volup"}),
]


class Talk:
    def __init__(self, port, timeout=30.0):
        self.sock = socket.create_connection(("127.0.0.1", port), timeout=timeout)
        self.sock.settimeout(timeout)
        self.buf = b""

    def call(self, obj, timeout=30.0):
        started = time.time()
        self.sock.sendall((json.dumps(obj, ensure_ascii=False) + "\n").encode("utf-8"))
        deadline = started + timeout
        while True:
            while b"\n" not in self.buf:
                if time.time() > deadline:
                    raise TimeoutError("超时")
                chunk = self.sock.recv(262144)
                if not chunk:
                    raise ConnectionError("断开")
                self.buf += chunk
            line, _, self.buf = self.buf.partition(b"\n")
            message = json.loads(line.decode("utf-8"))
            if "event" in message:
                continue
            return message, time.time() - started

    def close(self):
        try:
            self.sock.close()
        except OSError:
            pass


def main() -> int:
    if not EXE.is_file():
        print(f"  找不到 {EXE}")
        return 1

    d = EXE.parent
    real = d / "data"
    bak = None
    if real.exists():
        bak = d / "data_bk_diag"
        if bak.exists():
            shutil.rmtree(bak, ignore_errors=True)
        shutil.move(str(real), str(bak))

    proc = None
    try:
        proc = subprocess.Popen(
            [str(EXE), "--port", str(PORT), "--add-device", "10"],
            cwd=str(d), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

        code = ""
        settings = d / "data" / "settings.json"
        deadline = time.time() + 30
        while time.time() < deadline:
            time.sleep(0.4)
            try:
                code = json.loads(settings.read_text(
                    encoding="utf-8")).get("pairing_code", "")
            except Exception:                             # noqa: BLE001
                code = ""
            if code:
                try:
                    p = socket.create_connection(("127.0.0.1", PORT), timeout=1)
                    p.close()
                    break
                except OSError:
                    continue
        print(f"  已启动，配对码 {code}\n")

        talk = Talk(PORT)
        reply, _ = talk.call({"cmd": "pair", "deviceId": "diag",
                              "deviceName": "诊断", "code": code})
        print(f"  配对: {reply.get('ok')}\n")

        print(f"  {'命令':<42} {'耗时':>7}  结果")
        print("  " + "-" * 76)
        slow, failed = [], []
        for cmd, extra in COMMANDS:
            payload = {"cmd": cmd}
            payload.update(extra)
            label = cmd + (f" {extra.get('name') or extra.get('action') or ''}"
                           if extra else "")
            try:
                reply, cost = talk.call(payload, timeout=30)
                ok = reply.get("ok")
                note = str(reply.get("note") or reply.get("error") or "")[:38]
                flag = "OK " if ok else "FAIL"
                if not ok:
                    failed.append((label, note))
                if cost > 3:
                    slow.append((label, cost))
                print(f"  {label:<42} {cost:>6.1f}s  {flag}  {note}")
            except Exception as exc:                      # noqa: BLE001
                failed.append((label, f"{type(exc).__name__}: {exc}"))
                print(f"  {label:<42} {'--':>7}  CRASH  {type(exc).__name__}: {exc}")
        talk.close()

        print()
        print("=" * 78)
        if slow:
            print(f"  慢命令（>3 秒）{len(slow)} 条：")
            for label, cost in slow:
                print(f"    {label}  {cost:.1f}s")
        if failed:
            print(f"  失败 {len(failed)} 条：")
            for label, note in failed:
                print(f"    {label}  ->  {note}")
        if not slow and not failed:
            print("  全部正常，没有慢命令")

    finally:
        if proc:
            try:
                proc.terminate()
                proc.wait(timeout=8)
            except Exception:                             # noqa: BLE001
                try:
                    proc.kill()
                except Exception:                         # noqa: BLE001
                    pass
        time.sleep(1)
        if real.exists():
            shutil.rmtree(real, ignore_errors=True)
        if bak and bak.exists():
            shutil.move(str(bak), str(real))
    return 0


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