#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
check_ports.py —— 门店本机端口自检（99XX 段统一）工具。

用途（用户需求 2026-10-16）：我们的桌面客户端把「自己要用的端口」统一收敛到了 **99XX** 段，
避免和门店机器上其它软件（尤其是华视电子身份证阅读器占用的 9001）抢端口、互相影响。

这个脚本让门店 / 技术员一键扫描本机，看看我们预留的 99XX 端口有没有被其它软件占了、
占了的会不会冲突、该怎么让位或处理。

用法（在门店电脑上，有 Python 即可；没有 Python 可让技术员远程协助）：
    python check_ports.py            # 文本报告（推荐，一目了然）
    python check_ports.py --json     # 机器可读（便于远程排障时收集上报）

纯标准库，不依赖任何第三方包；Windows / macOS / Linux 都能跑
（Windows 下能进一步查出占用端口的进程名，其它系统只能判断是否被占用）。
"""
import json
import socket
import subprocess
import sys

# —— 我们统一使用的端口段（99XX）——
# 角色说明必须与 hotel_client / relay_service / sidecar / fd.html / hotel_client_gui.html 一致。
OUR_PORTS = [
    (9901, "客户端主端口（配置页 + 卡务）", "main"),
    (9902, "公安回读/上报中转（候选①）", "relay"),
    (9903, "公安回读/上报中转（候选②）", "relay"),
    (9904, "公安回读/上报中转（候选③）", "relay"),
    (9905, "公安回读/上报中转（候选④）", "relay"),
    (9906, "公安回读/上报中转（候选⑤）", "relay"),
    (9907, "自带 Chromium 远程调试（单窗口多标签）", "cdp"),
    (9908, "主端口让位候选①（9901 被占时）", "main"),
    (9909, "主端口让位候选②（9901 被占时）", "main"),
    (9917, "CDP 让位候选①（9907 被占时）", "cdp"),
    (9918, "CDP 让位候选②（9907 被占时）", "cdp"),
]

# 这些端口不是我们的，但和「端口冲突」排查强相关：
#   9001       = 华视电子身份证阅读器常驻端口（我们已弃用；门店若看到它被占＝正常，
#                但千万别把它配给我们的服务，否则公安会读不到身份证）
#   8765/8767  = 旧版制卡/读证端口（仅向后兼容探测，新装不要用）
INFO_PORTS = [
    (9001, "华视电子身份证阅读器（我们弃用，被占＝正常）", "forbidden"),
    (8765, "旧版制卡端口（向后兼容，新装不用）", "legacy"),
    (8767, "旧版读证端口（向后兼容，新装不用）", "legacy"),
]


def port_free(port):
    """用 socket.bind 试一下：能 bind 上就是空闲，否则被占用（只绑一下就释放，不真监听）。"""
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        s.bind(("127.0.0.1", port))
        return True
    except OSError:
        return False
    finally:
        try:
            s.close()
        except Exception:  # noqa: BLE001
            pass


def owner_of(port):
    """尽可能查出占用该端口的进程 (pid, name)；查不到返回 (None, None)。"""
    pid = None
    try:
        out = subprocess.run(
            ["netstat", "-ano", "-p", "TCP"],
            capture_output=True, text=True, timeout=20,
        ).stdout
        for line in out.splitlines():
            cols = line.split()
            # 形如：TCP    127.0.0.1:9902    0.0.0.0:0    LISTENING    1234
            if len(cols) >= 5 and cols[0].upper() == "TCP":
                local = cols[1]
                if local.endswith(":%d" % port) or local.endswith("]:%d" % port):
                    state = cols[3] if len(cols) > 4 else ""
                    if state.upper() in ("LISTENING", "ESTABLISHED"):
                        pid = cols[-1]
                        break
    except Exception:  # noqa: BLE001
        pass
    if not pid:
        return None, None
    try:
        pid = int(pid)
    except ValueError:
        return None, None
    name = _proc_name(pid)
    return pid, name


def _proc_name(pid):
    try:
        out = subprocess.run(
            ["tasklist", "/fi", "PID eq %d" % pid, "/fo", "csv", "/nh"],
            capture_output=True, text=True, timeout=20,
        ).stdout.strip()
        # 形如："python.exe","1234","Console","1","12,345 K"
        if "," in out:
            return out.split(",")[0].strip('"')
    except Exception:  # noqa: BLE001
        pass
    return "PID:%d" % pid


def scan():
    rows = []
    for port, role, group in OUR_PORTS:
        free = port_free(port)
        pid, name = (None, None) if free else owner_of(port)
        rows.append({
            "port": port, "role": role, "group": group,
            "ours": True, "free": free, "pid": pid, "name": name,
        })
    for port, role, group in INFO_PORTS:
        free = port_free(port)
        pid, name = (None, None) if free else owner_of(port)
        rows.append({
            "port": port, "role": role, "group": group,
            "ours": False, "free": free, "pid": pid, "name": name,
        })
    return rows


def conflicts_of(rows):
    return [r for r in rows if r["ours"] and not r["free"]]


def render_text(rows):
    line = "=" * 72
    print(line)
    print(" 云AI酒店 PMS · 本机端口自检（99XX 段统一）")
    from datetime import datetime
    print(" 时间：" + datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    print(line)
    print("【我们统一使用的端口段（99XX）】")
    print("  9901           客户端主端口（配置页 + 卡务）")
    print("  9902~9906      公安回读/上报中转（自动挑空闲端口拉起）")
    print("  9907           自带 Chromium 远程调试（单窗口多标签）")
    print("  9908 / 9909    主端口让位候选（9901 被占时）")
    print("  9917 / 9918    CDP 让位候选（9907 被占时）")
    print("  9001  ⛔禁用    华视电子身份证阅读器（我们已弃用，请勿占用我们的端口段）")
    print("  8765 / 8767    仅旧版向后兼容探测（不推荐新装）")
    print(line)
    print("【扫描结果】")
    print("  %-6s %-30s %-7s %s   %s" % ("端口", "角色", "状态", "标记", "占用进程(PID)"))
    print("  " + "-" * 66)
    for r in rows:
        status = "空闲" if r["free"] else "被占用"
        who = "—" if r["free"] else ("%s (PID %s)" % (r["name"], r["pid"]))
        if not r["ours"] and not r["free"]:
            mark = "ℹ️"   # 别人（华视等）的端口，被占是预期
        elif r["ours"] and not r["free"]:
            mark = "⚠️"   # 我们的端口被抢，要留意
        else:
            mark = "✅"
        print("  %-6s %-30s %-7s %s   %s" % (r["port"], r["role"][:30], status, mark, who))
    print(line)
    conflicts = conflicts_of(rows)
    print("【结论】")
    if not conflicts:
        print("  ✅ 我们的 99XX 端口全部空闲，无冲突，可直接使用。")
    else:
        print("  ⚠️ 检测到 %d 处我们预留的端口被其它软件占用：" % len(conflicts))
        for r in conflicts:
            print("     · 端口 %d（%s）被 %s (PID %s) 占用"
                  % (r["port"], r["role"], r["name"], r["pid"]))
        print("  → 我们的服务会**自动让位**到候选端口（如 9902 被占 → 9903），一般不影响使用；")
        print("    若想彻底避免，请在该软件里改端口或关闭它。")
        own = [r for r in conflicts if r["name"] and "HotelPMS" in str(r["name"])]
        if own:
            print("  ⚠️ 有端口被【我们自己的另一个实例】占用，疑似客户端多开，请确认只运行一个客户端。")
    hsh = next((r for r in rows if r["port"] == 9001), None)
    if hsh and not hsh["free"]:
        print("  ℹ️ 端口 9001 被占用（华视电子身份证阅读器，属正常）；我们已弃用 9001，")
        print("    若客户端日志提示抢 9001，请确认没把回读端口配成 9001（应选 9902 起）。")
    print(line)


def main():
    rows = scan()
    if "--json" in sys.argv[1:]:
        from datetime import datetime
        print(json.dumps({
            "generated_at": datetime.now().isoformat(timespec="seconds"),
            "ports": rows,
            "conflicts": [r["port"] for r in conflicts_of(rows)],
        }, ensure_ascii=False, indent=2))
    else:
        render_text(rows)
    # 退出码：有冲突返回 1（便于自动化判断），无冲突返回 0
    return 1 if conflicts_of(rows) else 0


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