ykxmz/windows/lanremote/netcheck.py

# -*- coding: utf-8 -*-
"""网络自检 —— 确认手表的数据包到底有没有到达这台电脑。

这个工具回答一个二选一的问题：

    手表的连接请求，究竟「到了电脑但被拒绝」，还是「压根没到」？

用法：
    由 ykxmz.exe --netcheck 调用（已打包进 EXE，不需要装 Python）

会监听 90 秒，可以按 Ctrl+C 提前结束。

编码说明：
    脚本启动时会把控制台代码页设成 UTF-8(65001) 再输出，
    所以不论系统默认代码页是什么，中文都不会乱码。
"""
from __future__ import annotations

import socket
import subprocess
import sys
import threading
import time
from datetime import datetime

TCP_PORT = 8765
UDP_PORT = 8766
LISTEN_SECONDS = 90

_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0)

# WSAEADDRINUSE / WSAEACCES
_ADDR_IN_USE = (10048, 10013)


def setup_console() -> None:
    """把控制台和标准输出都设成 UTF-8，避免中文乱码。"""
    try:
        import ctypes
        ctypes.windll.kernel32.SetConsoleOutputCP(65001)
        ctypes.windll.kernel32.SetConsoleCP(65001)
    except Exception:
        pass
    for stream in (sys.stdout, sys.stderr):
        try:
            stream.reconfigure(encoding="utf-8", errors="replace")
        except Exception:
            pass


def line(text: str = "") -> None:
    print(text, flush=True)


def title(text: str) -> None:
    line()
    line("=" * 62)
    line(f"  {text}")
    line("=" * 62)


def stamp() -> str:
    return datetime.now().strftime("%H:%M:%S")


def local_ips() -> list[tuple[str, str]]:
    found: list[tuple[str, str]] = []
    try:
        probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        probe.connect(("8.8.8.8", 80))
        addr = probe.getsockname()[0]
        probe.close()
        if addr and not addr.startswith("127."):
            found.append((addr, "默认路由"))
    except OSError:
        pass
    try:
        for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
            addr = info[4][0]
            if addr and not addr.startswith("127.") and addr not in [a for a, _ in found]:
                found.append((addr, "主机名"))
    except OSError:
        pass
    return found


def run(cmd: list[str], timeout: int = 15) -> str:
    try:
        proc = subprocess.run(cmd, capture_output=True, timeout=timeout,
                              creationflags=_NO_WINDOW,
                              encoding="mbcs", errors="replace")
        return (proc.stdout or "") + (proc.stderr or "")
    except Exception as exc:                              # noqa: BLE001
        return f"({exc})"


def agent_running() -> bool:
    """被控端是不是正在运行（它会占着 8765/8766）。"""
    text = run(["tasklist", "/FI", "IMAGENAME eq ykxmz.exe"])
    return "ykxmz" in text


def stop_agent() -> bool:
    """结束被控端进程，好让自检能绑上端口。"""
    run(["taskkill", "/IM", "ykxmz.exe", "/F"])
    time.sleep(1.5)
    return not agent_running()


def ask_stop_agent() -> bool:
    """询问是否关掉被控端。非交互环境直接返回 False。"""
    if not sys.stdin.isatty():
        return False
    line()
    line("  被控端正在运行，占着 8765/8766，本自检绑不上端口。")
    line("  要抓包必须先让它让出端口。")
    line()
    try:
        answer = input("  >>> 现在自动关掉被控端并继续抓包吗？(Y/N) ").strip().lower()
    except Exception:
        return False
    if not answer.startswith("y"):
        line("  好，那本次不抓包。")
        line("  （这不是故障 —— 被控端在跑就说明它能收到包；")
        line("    想抓包时先在它窗口点「退出」再重跑本脚本。）")
        return False
    if stop_agent():
        line("  已关闭被控端，继续抓包。")
        line("  抓完记得重新启动它。")
        return True
    line("  [警告] 关闭失败，可能是权限不够。请手动在它窗口点「退出」。")
    return False


def show_network() -> str:
    title("1. 本机网络")
    ips = local_ips()
    if ips:
        for addr, tag in ips:
            line(f"  本机 IP：{addr}    [{tag}]")
        line()
        line(f"  >>> 手表上就填这个 IP：{ips[0][0]} <<<")
    else:
        line("  没能读到本机 IP（网络可能没连上）")

    line()
    wlan = run(["netsh", "wlan", "show", "interfaces"])
    ssid = None
    for raw in wlan.splitlines():
        text = raw.strip()
        if text.startswith("SSID") and "BSSID" not in text:
            ssid = text.split(":", 1)[-1].strip()
    if ssid:
        line(f"  当前 WiFi：{ssid}")
        line("  （手表必须连的是同一个 WiFi）")
    else:
        line("  当前不是 WiFi 连接，或有线连接")

    return ips[0][0] if ips else ""


def show_ports() -> None:
    title("2. 端口占用情况")
    tcp = run(["netstat", "-ano", "-p", "TCP"])
    udp = run(["netstat", "-ano", "-p", "UDP"])
    tcp_hit = [l.strip() for l in tcp.splitlines()
               if f":{TCP_PORT}" in l and "LISTENING" in l.upper()]
    udp_hit = [l.strip() for l in udp.splitlines() if f":{UDP_PORT}" in l]

    if tcp_hit:
        for item in tcp_hit:
            line(f"  TCP {TCP_PORT}  已被监听： {item}")
    else:
        line(f"  TCP {TCP_PORT}  没有监听")
    if udp_hit:
        for item in udp_hit:
            line(f"  UDP {UDP_PORT}  已被监听： {item}")
    else:
        line(f"  UDP {UDP_PORT}  没有监听")

    line()
    if agent_running():
        line("  >>> 被控端（ykxmz.exe）正在运行。")
        line("      它会占着 8765/8766，本自检就绑不上这两个端口。")
        line("      想用自检抓包：先在它窗口上点「退出」，再重跑本脚本。")
        line("      其实不用自检也可以 —— 被控端在跑就说明它能收包。")
    else:
        line("  被控端没有在运行。想连手表的话需要先把它启动。")


def show_firewall() -> None:
    title("3. 防火墙放行状态")
    for name in ("LAN Remote TCP", "LAN Remote UDP", "LAN Remote App TCP"):
        out = run(["netsh", "advfirewall", "firewall", "show", "rule",
                   f"name={name}"])
        if not out.strip() or "No rules match" in out or "没有与指定条件" in out:
            line(f"  {name}：未放行")
        else:
            enabled = action = ""
            for raw in out.splitlines():
                text = raw.strip()
                if text.startswith("Enabled") or text.startswith("已启用"):
                    enabled = text.split(":", 1)[-1].strip()
                elif text.startswith("Action") or text.startswith("操作"):
                    action = text.split(":", 1)[-1].strip()
            line(f"  {name}：已放行   Enabled={enabled or '?'}  Action={action or '?'}")


# ----------------------------------------------------------------------
# 监听：这段才是关键
# ----------------------------------------------------------------------

received_tcp: list[str] = []
received_udp: list[str] = []
bind_errors: list[str] = []


def _bind(sock: socket.socket, port: int) -> bool:
    """独占绑定。Windows 上必须用 SO_EXCLUSIVEADDRUSE，
    否则第二个进程也能绑上，抓包结果会失真。"""
    if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
        try:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
        except OSError:
            pass
    else:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    try:
        sock.bind(("0.0.0.0", port))
        return True
    except OSError as exc:
        reason = ""
        if exc.errno in _ADDR_IN_USE or (exc.winerror or 0) in _ADDR_IN_USE:
            reason = "端口被占用"
        bind_errors.append(f"{port}: {reason or exc}")
        return False


def tcp_listener(stop: threading.Event) -> None:
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    if not _bind(server, TCP_PORT):
        line(f"  !! 无法监听 TCP {TCP_PORT}")
        if agent_running():
            line("     原因：被控端正在运行，占着这个端口。")
            line("     -> 先在它窗口上点「退出」，再重跑本自检。")
        return
    server.listen(8)
    server.settimeout(0.5)
    line(f"  正在监听 TCP {TCP_PORT} …")
    while not stop.is_set():
        try:
            conn, addr = server.accept()
        except socket.timeout:
            continue
        except OSError:
            break
        src = f"{addr[0]}:{addr[1]}"
        received_tcp.append(src)
        line(f"  [{stamp()}] >>> 收到 TCP 连接：来自 {src}  <<<")
        try:
            conn.sendall(b"LANREMOTE-SELFTEST\n")
        except OSError:
            pass
        finally:
            try:
                conn.close()
            except OSError:
                pass
    server.close()


def udp_listener(stop: threading.Event) -> None:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    if not _bind(sock, UDP_PORT):
        line(f"  !! 无法监听 UDP {UDP_PORT}")
        if agent_running():
            line("     原因：被控端正在运行，占着这个端口。")
        return
    sock.settimeout(0.5)
    line(f"  正在监听 UDP {UDP_PORT} …")
    while not stop.is_set():
        try:
            data, addr = sock.recvfrom(2048)
        except socket.timeout:
            continue
        except OSError:
            break
        src = f"{addr[0]}:{addr[1]}"
        received_udp.append(src)
        preview = data[:80].decode("utf-8", errors="replace")
        line(f"  [{stamp()}] >>> 收到 UDP 包：来自 {src}  {preview} <<<")
    sock.close()


def run_check(seconds: int = 90) -> int:
    global LISTEN_SECONDS
    LISTEN_SECONDS = int(seconds or 90)
    setup_console()

    title("局域网遥控 · 网络自检")
    line("  这个工具监听端口，看手表的请求有没有到达这台电脑。")

    show_network()
    show_ports()
    show_firewall()

    title(f"4. 开始监听（{LISTEN_SECONDS} 秒）")
    if agent_running():
        ask_stop_agent()
        line()
    line("  现在去手表上：打开 App →「设备诊断」→ 底部填上面的 IP → 点「测一下」")
    line()

    stop = threading.Event()
    for target, args in ((tcp_listener, (stop,)), (udp_listener, (stop,))):
        threading.Thread(target=target, args=args, daemon=True).start()
        time.sleep(0.2)

    deadline = time.time() + LISTEN_SECONDS
    try:
        while time.time() < deadline:
            time.sleep(0.5)
            left = int(deadline - time.time())
            print(f"\r  剩余 {left:3d} 秒 …   已收到 TCP {len(received_tcp)} 个 / "
                  f"UDP {len(received_udp)} 个", end="", flush=True)
    except KeyboardInterrupt:
        pass
    stop.set()
    line()
    line()

    title("5. 结论")
    if received_tcp or received_udp:
        line(f"  TCP 收到 {len(received_tcp)} 个连接，UDP 收到 {len(received_udp)} 个包。")
        line()
        line("  >>> 网络是通的！手表的数据包能到达这台电脑。")
        line("      既然包能到，连不上就只可能是：")
        line("        * 被控端没在运行")
        line("        * 或端口被别的程序占了")
    elif bind_errors and agent_running():
        line("  没能抓包 —— 被控端正占着 8765/8766。")
        line()
        line("  >>> 这其实说明被控端是好的。要抓包请先退出它再重跑。")
    else:
        line("  TCP 和 UDP 都一个包都没收到。")
        line()
        line("  >>> 手表的数据包根本没到这台电脑。依次排查：")
        line("      1) 电脑是不是休眠/关机了")
        line("      2) 手表连的 WiFi 和电脑是不是同一个")
        line("      3) 路由器有没有开「AP 隔离 / 客户端隔离」")
        line("      4) 电脑上是不是有 VPN / 加速器在改路由表")
    line()
    if sys.stdin.isatty():
        line("  按回车退出…")
        try:
            input()
        except Exception:
            pass
    return 0


if __name__ == "__main__":
    # 直接 `python netcheck.py [秒数]` 也能用。
    # 原来这里写的是 main()，但本文件没有 main() ——
    # 入口是 run_check()，直接跑会 NameError。
    sys.exit(run_check(int(sys.argv[1]) if len(sys.argv) > 1 else 90))