ykxmz/_build/patch_diag_counters.py

# -*- coding: utf-8 -*-
"""服务端：加 TCP 连接总次数计数（自检要用），diagnostics 直接读 status。"""
from __future__ import annotations

import sys
from pathlib import Path

ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\windows\lanremote")
SERVER = ROOT / "server.py"
DIAG = ROOT / "diagnostics.py"

# ---- server.py: 初始化计数 ----
INIT_OLD = '''        self._accept_thread: threading.Thread | None = None'''
INIT_NEW = '''        self._accept_thread: threading.Thread | None = None
        # 累计收到过多少次 TCP 连接 —— 自检用它判断"手表的包有没有到"
        self._total_connections = 0'''

# ---- server.py: accept 时累加 ----
ACCEPT_OLD = '''                conn, addr = self._sock.accept()'''
ACCEPT_NEW = '''                conn, addr = self._sock.accept()
                self._total_connections += 1'''

# ---- server.py: status 里带上 ----
STATUS_OLD = '''            "discovery": self._discovery.status() if self._discovery else {"running": False},
            "local_ip": local_ip(),'''
STATUS_NEW = '''            "discovery": self._discovery.status() if self._discovery else {"running": False},
            "total_connections": self._total_connections,
            "local_ip": local_ip(),'''

# ---- diagnostics.py: 改成读 status ----
DIAG_OLD = '''    # 服务端计数器
    try:
        status = server.status()
        data["connected"] = status.get("connected", [])
        data["pairing_code"] = status.get("pairing_code", "")
        data["screen_sharing"] = status.get("screen_sharing", False)
        stats = status.get("stats") or {}
        data["tcp_connections"] = int(stats.get("connections", 0) or 0)
    except Exception:                                     # noqa: BLE001
        pass

    # 自动发现的计数器
    try:
        discovery = getattr(server, "_discovery", None)
        if discovery is not None:
            info = discovery.status()
            data["probes"] = int(info.get("answered", 0) or 0)
            data["selfcheck_requests"] = int(info.get("selfcheck_requests", 0) or 0)
            data["discovery_running"] = bool(info.get("running"))
            data["discovery_error"] = info.get("error", "")
        else:
            data["discovery_running"] = False
            data["discovery_error"] = "未启动"
    except Exception:                                     # noqa: BLE001
        data["discovery_running"] = False
'''

DIAG_NEW = '''    # 服务端状态（含自动发现计数、累计 TCP 连接数）
    try:
        status = server.status()
        data["connected"] = status.get("connected", [])
        data["pairing_code"] = status.get("pairing_code", "")
        data["screen_sharing"] = status.get("screen_sharing", False)
        data["tcp_connections"] = int(status.get("total_connections", 0) or 0)

        info = status.get("discovery") or {}
        data["probes"] = int(info.get("answered", 0) or 0)
        data["selfcheck_requests"] = int(info.get("selfcheck_requests", 0) or 0)
        data["discovery_running"] = bool(info.get("running"))
        data["discovery_error"] = info.get("error", "")
    except Exception as exc:                              # noqa: BLE001
        data["discovery_running"] = False
        data["discovery_error"] = str(exc)
'''


def main() -> int:
    ok = True
    text = SERVER.read_text(encoding="utf-8")
    for old, new, tag in ((INIT_OLD, INIT_NEW, "计数初始化"),
                          (ACCEPT_OLD, ACCEPT_NEW, "accept 累加"),
                          (STATUS_OLD, STATUS_NEW, "status 带出")):
        if old not in text:
            print(f"  [失败] server.py {tag} 锚点未匹配")
            ok = False
        else:
            text = text.replace(old, new, 1)
            print(f"  server.py / {tag}: OK")
    if ok:
        SERVER.write_text(text, encoding="utf-8")

    dtext = DIAG.read_text(encoding="utf-8")
    if DIAG_OLD not in dtext:
        print("  [失败] diagnostics.py 锚点未匹配")
        ok = False
    else:
        dtext = dtext.replace(DIAG_OLD, DIAG_NEW, 1)
        DIAG.write_text(dtext, encoding="utf-8")
        print("  diagnostics.py: OK")

    print("\n语法检查：")
    import ast
    for name, path in (("server.py", SERVER), ("diagnostics.py", DIAG)):
        try:
            ast.parse(path.read_text(encoding="utf-8"))
            print(f"  {name}: OK")
        except SyntaxError as exc:
            print(f"  {name}: 语法错误 {exc}")
            ok = False
    return 0 if ok else 1


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