#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Node collector for the LAN dashboard.
Pure stdlib, no root required. Writes one JSON object to stdout as UTF-8 bytes.

Deployed to: server1 (/opt/dashboard/collect.py) and server2 (/home/xiejun/dashboard/collect.py)
"""
import glob
import ipaddress
import json
import os
import re
import signal
import socket
import struct
import subprocess
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timedelta

SAMPLE_DELAY = 0.4

_MISSING = object()          # real sentinel: distinguishes "no default" from "default None"

KEY_SERVICES = [
    "ssh", "cron", "docker", "containerd", "smbd", "nmbd", "winbind", "tailscaled",
    "dashboard", "bedrock", "rclone", "natfrp", "filebrowser", "casaos",
    "mcsm-daemon", "mcsm-web", "nginx", "aqf-research", "minigames", "avahi-daemon",
]

WEB_CHECKS = {
    "xiejun-web": [
        ("Nginx Proxy Manager", "http://127.0.0.1:81/"),
        ("网站 (HTTP 80)", "http://127.0.0.1/"),
        ("FileBrowser", "http://127.0.0.1:8080/"),
        ("Homer 导航页", "http://127.0.0.1:8012/"),
        ("Reader 阅读器", "http://127.0.0.1:3000/"),
        ("Memos 笔记", "http://127.0.0.1:5230/"),
        ("PhotoPrism 相册", "http://127.0.0.1:2342/"),
        ("AQF Research", "http://127.0.0.1:5555/"),
        ("Minigames", "http://127.0.0.1:8025/"),
        ("本仪表盘", "http://127.0.0.1:9090/api/health"),
    ],
    "xiejun-server": [
        # 探测列表必须和机器上真实在跑的服务一致，否则会出现两种误导：
        # 已卸载的应用永远报红，或者端口被别的服务占用却显示“正常”。
        ("CasaOS 网关 (80)", "http://127.0.0.1:80/"),
        ("FileBrowser (8080)", "http://127.0.0.1:8080/"),
    ],
}


# --------------------------------------------------------------------------
# process helpers
# --------------------------------------------------------------------------
def sh(cmd, timeout=8):
    """Run a shell command. Returns (returncode, stdout, stderr). Never raises.

    Uses a new session so a timeout can kill the entire process group instead of
    just the intermediate shell (grandchildren such as du/docker would survive).
    """
    try:
        proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE, start_new_session=True)
    except Exception as exc:
        return 1, "", "%s: %s" % (type(exc).__name__, exc)
    try:
        out, err = proc.communicate(timeout=timeout)
        return proc.returncode, out.decode("utf-8", "replace").strip(), err.decode("utf-8", "replace").strip()
    except subprocess.TimeoutExpired:
        try:
            os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
        except Exception:
            try:
                proc.kill()
            except Exception:
                pass
        try:
            out, err = proc.communicate(timeout=5)
        except Exception:
            out, err = b"", b""
        return 124, out.decode("utf-8", "replace").strip(), "timeout after %ss" % timeout


def run(cmd, timeout=8):
    """stdout only; for callers that do not care about the exit status."""
    return sh(cmd, timeout)[1]


def read(path, default=""):
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            return fh.read().strip()
    except Exception:
        return default


def safe(fn, default=_MISSING, *args):
    """Call fn, returning `default` on failure. With no default, returns an error dict."""
    try:
        return fn(*args)
    except Exception as exc:
        if default is _MISSING:
            return {"error": "%s: %s" % (type(exc).__name__, exc)}
        return default


# --------------------------------------------------------------------------
# cache (private dir, never a predictable world-writable /tmp path)
# --------------------------------------------------------------------------
def _cache_dir():
    for cand in (os.path.join(os.path.expanduser("~"), ".cache", "dsh-dashboard"),
                 "/run/dsh-dashboard"):
        try:
            os.makedirs(cand, mode=0o700, exist_ok=True)
            return cand
        except Exception:
            continue
    return "/tmp"


CACHE_DIR = _cache_dir()


def _cache_path(name):
    return os.path.join(CACHE_DIR, "%s.json" % name)


def _load_cache(path):
    try:
        with open(path, "r", encoding="utf-8") as fh:
            return json.load(fh)
    except Exception:
        return None


def _save_cache(path, obj):
    try:
        with open(path, "w", encoding="utf-8") as fh:
            json.dump(obj, fh)
    except Exception:
        pass


def cached(name, ttl, producer, default=None):
    """Memoise an expensive section, keyed by name."""
    path = _cache_path(name)
    blob = _load_cache(path)
    now = time.time()
    if isinstance(blob, dict) and now - blob.get("ts", 0) < ttl and "v" in blob:
        return blob["v"]
    try:
        value = producer()
    except Exception:
        value = default
    _save_cache(path, {"ts": now, "v": value})
    return value


def human_duration(seconds):
    seconds = int(seconds)
    days, rem = divmod(seconds, 86400)
    hours, rem = divmod(rem, 3600)
    minutes = rem // 60
    if days:
        return "%dd %dh %dm" % (days, hours, minutes)
    if hours:
        return "%dh %dm" % (hours, minutes)
    return "%dm" % minutes


def _is_loopback(addr):
    """True for 127.0.0.0/8, ::1, localhost and IPv4-mapped forms such as ::ffff:127.0.0.1."""
    if not addr:
        return False
    text = addr.strip()
    if text in ("localhost", "*"):
        return False
    if text.startswith("::ffff:"):
        text = text[len("::ffff:"):]
    try:
        return ipaddress.ip_address(text).is_loopback
    except ValueError:
        return text.startswith("127.")


# --------------------------------------------------------------------------
# identity / cpu / memory / uptime / disks
# --------------------------------------------------------------------------
def host_info():
    os_name = ""
    for line in read("/etc/os-release").splitlines():
        if line.startswith("PRETTY_NAME="):
            os_name = line.split("=", 1)[1].strip().strip('"')
    return {
        "hostname": socket.gethostname(),
        "os": os_name or "Linux",
        "kernel": os.uname().release,
        "arch": os.uname().machine,
    }


def _cpu_times():
    parts = read("/proc/stat").splitlines()[0].split()
    vals = [int(x) for x in parts[1:]]
    idle = vals[3] + (vals[4] if len(vals) > 4 else 0)
    return sum(vals), idle


def cpu_info():
    model = ""
    for line in read("/proc/cpuinfo").splitlines():
        if line.lower().startswith("model name"):
            model = line.split(":", 1)[1].strip()
            break
    t1, i1 = _cpu_times()
    time.sleep(SAMPLE_DELAY)
    t2, i2 = _cpu_times()
    dt, di = t2 - t1, i2 - i1
    usage = round((1 - di / dt) * 100, 1) if dt > 0 else 0.0
    load = read("/proc/loadavg").split()[:3]
    try:
        loads = [float(x) for x in load]
    except ValueError:
        loads = []
    return {"model": model, "cores": os.cpu_count(), "usage_pct": usage, "load": loads if len(loads) == 3 else []}


CPU_CHIPS = ("coretemp", "k10temp", "zenpower", "cpu_thermal", "soc_thermal", "cpu-thermal")


def temps():
    chips = []
    for path in sorted(glob.glob("/sys/class/hwmon/hwmon*")):
        chip_name = read(os.path.join(path, "name")) or os.path.basename(path)
        sensors = []
        for inp in sorted(glob.glob(os.path.join(path, "temp*_input"))):
            base = inp[:-6]
            raw = read(inp)
            if not raw or not raw.lstrip("-").isdigit():
                continue
            crit = read(base + "_crit")
            sensors.append({
                "label": read(base + "_label") or os.path.basename(base),
                "c": round(int(raw) / 1000.0, 1),
                "crit_c": round(int(crit) / 1000.0, 1) if crit.lstrip("-").isdigit() else None,
            })
        if sensors:
            chips.append({"chip": chip_name, "sensors": sensors})

    cpu_c = nvme_c = None
    for chip in chips:
        if cpu_c is None and chip["chip"] in CPU_CHIPS:
            cpu_c = max(s["c"] for s in chip["sensors"])
        if nvme_c is None and chip["chip"] == "nvme":
            nvme_c = max(s["c"] for s in chip["sensors"])
    hottest = max((s["c"] for c in chips for s in c["sensors"]), default=None)
    return {"cpu_c": cpu_c, "nvme_c": nvme_c, "hottest_c": hottest, "chips": chips}


def mem_info():
    info = {}
    for line in read("/proc/meminfo").splitlines():
        key, _, rest = line.partition(":")
        fields = rest.strip().split()
        if fields and fields[0].isdigit():
            info[key] = int(fields[0])

    def mb(k):
        return round(info.get(k, 0) / 1024.0, 1)

    total = mb("MemTotal")
    available = mb("MemAvailable")
    used = round(total - available, 1)
    swap_total = mb("SwapTotal")
    swap_free = mb("SwapFree")
    return {
        "total_mb": total, "used_mb": used, "available_mb": available,
        "usage_pct": round(used / total * 100, 1) if total else 0.0,
        "swap_total_mb": swap_total,
        "swap_used_mb": round(swap_total - swap_free, 1),
        "swap_usage_pct": round((swap_total - swap_free) / swap_total * 100, 1) if swap_total else 0.0,
        "cached_mb": mb("Cached"),
    }


def uptime_info():
    raw = read("/proc/uptime").split()
    seconds = float(raw[0]) if raw else 0.0
    boot = None
    btime = re.search(r"btime (\d+)", read("/proc/stat"))
    if btime:
        boot = datetime.fromtimestamp(int(btime.group(1))).isoformat(timespec="seconds")
    return {"uptime_sec": int(seconds), "uptime_human": human_duration(seconds), "boot_time": boot}


SKIP_FS = {"tmpfs", "devtmpfs", "overlay", "squashfs", "proc", "sysfs", "cgroup",
           "cgroup2", "devpts", "mqueue", "debugfs", "tracefs", "securityfs",
           "pstore", "bpf", "autofs", "hugetlbfs", "configfs", "fusectl", "ramfs",
           "efivarfs", "binfmt_misc", "nsfs"}


def disks():
    out, seen = [], set()
    for line in read("/proc/mounts").splitlines():
        parts = line.split()
        if len(parts) < 3:
            continue
        dev, mount, fstype = parts[0], parts[1].replace("\\040", " "), parts[2]
        if fstype in SKIP_FS or mount in seen:
            continue
        if not (dev.startswith("/dev/") or fstype == "fuseblk"):
            continue
        seen.add(mount)
        try:
            st = os.statvfs(mount)
        except OSError:
            continue
        total = st.f_blocks * st.f_frsize
        if total <= 0:
            continue
        out.append({
            "device": dev, "mount": mount, "fstype": fstype,
            "total_gb": round(total / 1e9, 1),
            "used_gb": round((total - st.f_bfree * st.f_frsize) / 1e9, 1),
            "free_gb": round(st.f_bavail * st.f_frsize / 1e9, 1),
            "usage_pct": round((total - st.f_bfree * st.f_frsize) / total * 100, 1),
        })
    out.sort(key=lambda d: d["mount"])
    return out


def failed_mounts():
    """fstab entries that are configured but not mounted right now."""
    mounted = {line.split()[1] for line in read("/proc/mounts").splitlines() if len(line.split()) > 2}
    missing = []
    for line in read("/etc/fstab").splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        f = line.split()
        if len(f) < 3 or f[1] in ("none", "swap"):
            continue
        if f[2] in SKIP_FS:
            continue
        if f[1] not in mounted:
            missing.append({"mount": f[1], "device": f[0], "fstype": f[2]})
    return missing


# --------------------------------------------------------------------------
# docker
# --------------------------------------------------------------------------
def _docker_stats():
    rc, out, _ = sh("docker stats --no-stream --format '{{.Name}}\\t{{.CPUPerc}}\\t{{.MemUsage}}\\t{{.MemPerc}}'", timeout=30)
    stats = {}
    if rc != 0:
        return stats
    for line in out.splitlines():
        parts = line.split("\t")
        if len(parts) >= 4:
            stats[parts[0]] = {"cpu": parts[1], "mem": parts[2], "mem_pct": parts[3]}
    return stats


def containers():
    empty = {"available": False, "running": [], "all": [], "total": 0, "stopped": 0,
             "unhealthy": [], "restarting": [], "error": None}
    if not run("command -v docker"):
        empty["error"] = "未安装 docker"
        return empty

    rc, out, err = sh("docker ps -a --format '{{.Names}}\\t{{.Image}}\\t{{.State}}\\t{{.Status}}'", timeout=15)
    if rc != 0:
        empty["error"] = "docker ps 失败 (rc=%d): %s" % (rc, (err or "daemon 无响应")[:160])
        return empty

    items = []
    for line in out.splitlines():
        parts = line.split("\t")
        if len(parts) >= 4:
            items.append({"name": parts[0], "image": parts[1], "state": parts[2], "status": parts[3]})

    stats_rc, _, _ = 0, "", ""
    stats = cached("docker_stats", 25, _docker_stats)
    for item in items:
        item.update(stats.get(item["name"], {}))

    running = [c for c in items if c["state"] == "running"]
    return {
        "available": True,
        "running": running,
        "all": items,
        "total": len(items),
        "stopped": len(items) - len(running),
        "unhealthy": [c["name"] for c in items if "unhealthy" in c["status"].lower()],
        "restarting": [c["name"] for c in items if c["state"] == "restarting"],
        "stats_ok": bool(stats),
        "error": None,
    }


# --------------------------------------------------------------------------
# ports
# --------------------------------------------------------------------------
def ports():
    """Returns (items, error). Deduplicated per (proto, port) across bind addresses."""
    rc, out, err = sh("ss -tulnpH", timeout=12)
    if rc != 0:
        return [], "ss 失败 (rc=%d): %s" % (rc, (err or "")[:160])

    grouped = {}
    for line in out.splitlines():
        parts = line.split()
        if len(parts) < 5:
            continue
        proto, state, local = parts[0], parts[1], parts[4]
        m = re.search(r'users:\(\("([^"]+)"', line)
        proc = m.group(1) if m else ""
        addr, _, port = local.rpartition(":")
        addr = addr.strip("[]")
        if not port.isdigit():
            continue
        item = grouped.setdefault((proto, int(port)), {
            "proto": proto, "port": int(port), "state": state,
            "addr": addr, "addrs": [], "procs": [],
        })
        if addr not in item["addrs"]:
            item["addrs"].append(addr)
        if proc and proc not in item["procs"]:
            item["procs"].append(proc)

    items = []
    for item in grouped.values():
        addrs = item["addrs"]
        wildcard = [a for a in addrs if a in ("0.0.0.0", "::", "*")]
        item["addr"] = wildcard[0] if wildcard else addrs[0]
        item["public"] = not all(_is_loopback(a) for a in addrs)
        item["proc"] = ", ".join(item["procs"])
        item["addr_count"] = len(addrs)
        items.append(item)
    items.sort(key=lambda p: (not p["public"], p["port"], p["proto"]))
    return items, None


# --------------------------------------------------------------------------
# services / ops
# --------------------------------------------------------------------------
def services():
    rc, out, _ = sh("systemctl list-units --type=service --all --no-legend --no-pager --plain", timeout=15)
    if rc != 0:
        return {"items": [], "failed": [], "system_state": "unknown", "total_loaded": 0,
                "error": "systemctl 不可用"}
    items = []
    for line in out.splitlines():
        f = line.split(None, 4)
        if len(f) < 4 or not f[0].endswith(".service"):
            continue
        name = f[0][:-len(".service")]
        key = name in KEY_SERVICES
        if key or f[2] == "failed":
            items.append({"name": name, "unit": f[0], "active": f[2], "sub": f[3],
                          "key": key, "description": f[4] if len(f) > 4 else ""})
    items.sort(key=lambda s: (s["active"] != "failed", not s["key"], s["name"]))
    return {
        "items": items[:40],
        "failed": [i["unit"] for i in items if i["active"] == "failed"],
        "system_state": run("systemctl is-system-running", timeout=10).strip() or "unknown",
        "total_loaded": len(out.splitlines()),
        "error": None,
    }


def reboot_required():
    for path in ("/run/reboot-required", "/var/run/reboot-required"):
        if os.path.exists(path):
            pkgs = read("/run/reboot-required.pkgs") or read("/var/run/reboot-required.pkgs")
            return {"required": True, "packages": [p for p in pkgs.splitlines() if p.strip()][:12]}
    return {"required": False, "packages": []}


def top_procs():
    """Top processes by CPU / RSS, excluding this collector and its own helpers."""
    self_pid = os.getpid()
    skip = {"ps", "sh", "head", "timeout", "bash"}

    def parse(sort_key):
        _, out, _ = sh("ps -eo pid,user,pcpu,pmem,rss,comm --sort=%s --no-headers 2>/dev/null | head -14" % sort_key, timeout=10)
        rows = []
        for line in out.splitlines():
            f = line.split(None, 5)
            if len(f) < 6:
                continue
            try:
                pid = int(f[0])
                row = {"pid": pid, "user": f[1], "cpu": float(f[2]), "mem": float(f[3]),
                       "rss_mb": round(int(f[4]) / 1024.0, 1), "cmd": f[5]}
            except ValueError:
                continue
            if pid == self_pid or f[5] in skip:
                continue
            rows.append(row)
        return rows[:5]

    return {"by_cpu": parse("-pcpu"), "by_mem": parse("-rss")}


def sessions():
    _, out, _ = sh("who -u", timeout=8)
    items = []
    for line in out.splitlines():
        f = line.split()
        if len(f) < 4:
            continue
        items.append({
            "user": f[0], "tty": f[1], "login": "%s %s" % (f[2], f[3]),
            "idle": f[4] if len(f) > 4 else "",
            "from": f[6].strip("()") if len(f) > 6 else "",
        })
    return items


def recent_errors():
    def producer():
        _, out, _ = sh("journalctl -p err..alert -n 12 --no-pager -o short-iso 2>/dev/null", timeout=30)
        lines = [l.strip() for l in out.splitlines() if l.strip() and not l.startswith("--")]
        # One pass over the JSON stream gives both the count and the sources, so
        # the alert can say WHO is flooding the log instead of just how many.
        sources = {}
        total = None
        hour = None
        cutoff_us = (time.time() - 3600) * 1000000
        _, js, _ = sh("journalctl -p err..alert --since '-24h' --no-pager -q -o json 2>/dev/null", timeout=40)
        if js:
            total = 0
            hour = 0
            for line in js.splitlines():
                line = line.strip()
                if not line:
                    continue
                total += 1
                try:
                    rec = json.loads(line)
                except ValueError:
                    sources["(无法解析)"] = sources.get("(无法解析)", 0) + 1
                    continue
                try:
                    if int(rec.get("__REALTIME_TIMESTAMP") or 0) >= cutoff_us:
                        hour += 1
                except (TypeError, ValueError):
                    pass
                unit = rec.get("_SYSTEMD_UNIT") or rec.get("SYSLOG_IDENTIFIER") or "kernel"
                sources[unit] = sources.get(unit, 0) + 1
        if total is None:
            _, count, _ = sh("journalctl -p err..alert --since '-24h' --no-pager -q 2>/dev/null | wc -l", timeout=30)
            try:
                total = int(count.strip())
            except ValueError:
                total = None
        top = sorted(sources.items(), key=lambda kv: -kv[1])[:4]
        return {"recent": lines[-12:], "count_24h": total, "count_1h": hour,
                "sources": [{"unit": u, "count": n} for u, n in top]}
    return cached("errors", 45, producer,
                  default={"recent": [], "count_24h": None, "count_1h": None, "sources": []})


def _default_gateway():
    for line in read("/proc/net/route").splitlines()[1:]:
        f = line.split()
        if len(f) >= 3 and f[1] == "00000000":
            try:
                return socket.inet_ntoa(struct.pack("<L", int(f[2], 16)))
            except Exception:
                return None
    return None


def _ping(host, count=2, timeout=1):
    if not host or not re.match(r"^[0-9a-zA-Z.\-]+$", host):
        return {"host": host, "ok": False, "avg_ms": None, "loss_pct": 100.0}
    _, out, _ = sh("ping -c %d -W %d -q %s 2>&1" % (count, timeout, host), timeout=count * timeout + 4)
    m = re.search(r"(\d+(?:\.\d+)?)%\s*packet loss", out)
    loss = float(m.group(1)) if m else 100.0
    m = re.search(r"=\s*[\d.]+/([\d.]+)/", out)
    return {"host": host, "ok": m is not None, "avg_ms": float(m.group(1)) if m else None, "loss_pct": loss}


def _tailscale():
    ips = [l.strip() for l in run("tailscale ip -4 2>/dev/null", timeout=8).splitlines() if l.strip()]
    if not ips:
        return {"up": False, "ip": None, "peers_online": 0, "peers_total": 0}
    status = [l for l in run("tailscale status 2>/dev/null", timeout=10).splitlines()[1:] if l.strip()]
    return {"up": True, "ip": ips[0],
            "peers_online": sum(1 for l in status if "offline" not in l.lower()),
            "peers_total": len(status)}


def connectivity():
    def producer():
        gw = _default_gateway()
        return {
            "gateway": _ping(gw) if gw else None,
            "internet": _ping("1.1.1.1"),
            "dns": _ping("223.5.5.5"),
            "tailscale": _tailscale(),
        }
    return cached("conn", 30, producer, default={})


IFACE_SKIP_PREFIX = ("veth", "br-", "virbr", "docker")


def net_interfaces():
    """Interface counters plus throughput derived from the previous sample."""
    now = time.time()
    current = {}
    try:
        names = sorted(os.listdir("/sys/class/net"))
    except OSError:
        names = []
    for name in names:
        if name == "lo" or name.startswith(IFACE_SKIP_PREFIX):
            continue
        rx = read("/sys/class/net/%s/statistics/rx_bytes" % name)
        tx = read("/sys/class/net/%s/statistics/tx_bytes" % name)
        if not (rx.isdigit() and tx.isdigit()):
            continue
        speed = read("/sys/class/net/%s/speed" % name)
        current[name] = {
            "rx": int(rx), "tx": int(tx),
            "state": read("/sys/class/net/%s/operstate" % name) or "unknown",
            "speed_mbps": int(speed) if speed.lstrip("-").isdigit() and int(speed) > 0 else None,
        }

    prev = _load_cache(_cache_path("netprev")) or {}
    out = []
    for name, cur in current.items():
        item = {"name": name, "state": cur["state"], "speed_mbps": cur["speed_mbps"],
                "rx_gb": round(cur["rx"] / 1e9, 2), "tx_gb": round(cur["tx"] / 1e9, 2),
                "rx_bps": None, "tx_bps": None}
        old = prev.get("ifaces", {}).get(name)
        if old:
            dt = now - prev.get("ts", 0)
            if 1 < dt < 900:
                item["rx_bps"] = max(0.0, (cur["rx"] - old.get("rx", cur["rx"])) / dt)
                item["tx_bps"] = max(0.0, (cur["tx"] - old.get("tx", cur["tx"])) / dt)
        out.append(item)
    _save_cache(_cache_path("netprev"), {"ts": now, "ifaces": current})
    out.sort(key=lambda i: (i["state"] != "up", i["name"]))
    return out


def web_checks():
    def producer():
        results = []
        for label, url in WEB_CHECKS.get(socket.gethostname(), []):
            started = time.time()
            code, err = None, None
            try:
                req = urllib.request.Request(url, method="GET", headers={"User-Agent": "dash-probe"})
                with urllib.request.urlopen(req, timeout=4) as resp:
                    code = resp.status
            except urllib.error.HTTPError as exc:
                code = exc.code          # 401/403/404 still proves the service answers
            except Exception as exc:
                err = type(exc).__name__
            results.append({"label": label, "url": url, "code": code, "ok": code is not None,
                            "ms": int((time.time() - started) * 1000), "error": err})
        return results
    return cached("web", 20, producer, default=[])


# --------------------------------------------------------------------------
# minecraft (bedrock)
# --------------------------------------------------------------------------
MC_UNIT = "bedrock.service"
MC_DIR = "/home/xiejun/minecraft"
MC_STATE = os.path.join(MC_DIR, ".wrapper_state.json")
MC_CONSOLE = os.path.join(MC_DIR, "console.log")
MC_BACKUP_ROOT = "/home/xiejun/minecraft-backups"


def _wrapper_state():
    try:
        with open(MC_STATE, "r", encoding="utf-8") as fh:
            return json.load(fh)
    except Exception:
        return {}


def _console_tail(count=10):
    """Last lines of the wrapper's console mirror, read without loading the file."""
    try:
        with open(MC_CONSOLE, "rb") as fh:
            fh.seek(0, os.SEEK_END)
            size = fh.tell()
            fh.seek(max(0, size - 65536))
            text = fh.read().decode("utf-8", "replace")
        lines = [l.rstrip() for l in text.splitlines() if l.strip()]
        return lines[-count:]
    except Exception:
        return []


BACKUP_NAME_RE = re.compile(r"^bds_\d{8}_\d{6}(_[A-Za-z0-9_\-]{1,32})?$")


def _backups():
    """Backup list from manifests only — never walks 10k files on every poll.

    Only directories whose name matches the backup scheme are listed, so legacy
    leftovers in the same folder do not show up as 'broken backups'.
    """
    items = []
    try:
        names = sorted(os.listdir(MC_BACKUP_ROOT), reverse=True)
    except OSError:
        return items
    for name in names:
        path = os.path.join(MC_BACKUP_ROOT, name)
        if not os.path.isdir(path) or not BACKUP_NAME_RE.match(name):
            continue
        manifest = {}
        try:
            with open(os.path.join(path, "manifest.json"), encoding="utf-8") as fh:
                manifest = json.load(fh)
        except Exception:
            pass
        try:
            mtime = os.stat(path).st_mtime
        except OSError:
            continue
        items.append({
            "name": name,
            "created_at": manifest.get("created_at") or datetime.fromtimestamp(mtime).isoformat(timespec="seconds"),
            "label": manifest.get("label", ""),
            "level_name": manifest.get("level_name", ""),
            "total_files": manifest.get("total_files"),
            "total_mb": round((manifest.get("total_bytes") or 0) / 1048576.0, 1),
            "level_dat_sha256": manifest.get("level_dat_sha256"),
            "broken": not bool(manifest),
        })
    return items[:25]


def _unit_uptime_seconds(unit):
    """Unit uptime from the monotonic clock — avoids locale-dependent date parsing."""
    raw = run("systemctl show %s -p ActiveEnterTimestampMonotonic --value" % unit, timeout=8).strip()
    if not raw.isdigit() or int(raw) == 0:
        return None
    mono = read("/proc/uptime").split()
    if not mono:
        return None
    return max(0, int(float(mono[0]) - int(raw) / 1_000_000.0))


def _console_lines(max_bytes=262144):
    """Tail of the console mirror, big enough to span the current server run."""
    try:
        with open(MC_CONSOLE, "rb") as fh:
            fh.seek(0, os.SEEK_END)
            size = fh.tell()
            fh.seek(max(0, size - max_bytes))
            return [l for l in fh.read().decode("utf-8", "replace").splitlines() if l.strip()]
    except Exception:
        return []


def _fifo_has_reader():
    """True when the wrapper holds input.fifo open, i.e. commands will be accepted."""
    wpid = _wrapper_state().get("wrapper_pid")
    if not wpid:
        return False
    try:
        for fd in os.listdir("/proc/%d/fd" % int(wpid)):
            try:
                if os.readlink("/proc/%d/fd/%s" % (int(wpid), fd)).endswith("input.fifo"):
                    return True
            except OSError:
                continue
    except (OSError, ValueError):
        return False
    return False


def _proc_alive(pid):
    """False for missing, zombie or unreadable processes."""
    try:
        with open("/proc/%d/stat" % int(pid), "r") as fh:
            data = fh.read()
        # state is the field right after the (parenthesised) comm
        state = data.rsplit(")", 1)[1].split()[0]
        return state != "Z"
    except Exception:
        return False


def _ps_one(pid):
    _, out, _ = sh("ps -o pid=,pcpu=,pmem=,rss= -p %d --no-headers 2>/dev/null" % int(pid), timeout=6)
    f = out.split()
    if len(f) < 4:
        return {}
    try:
        return {"pid": int(f[0]), "cpu_pct": float(f[1]), "mem_pct": float(f[2]),
                "rss_mb": round(int(f[3]) / 1024.0, 1)}
    except ValueError:
        return {}


def _proc_comm(pid):
    try:
        with open("/proc/%d/comm" % pid) as fh:
            return fh.read().strip()
    except (OSError, ValueError):
        return ""


def _proc_exe(pid):
    """Path behind /proc/<pid>/exe - stable, unlike the process title.

    BDS rewrites its own process name once it finishes starting (it shows up as
    "Main"), so matching on the comm name silently stops working minutes after a
    restart. The executable path never changes.
    """
    try:
        return os.readlink("/proc/%d/exe" % int(pid))
    except (OSError, ValueError):
        return ""


def _is_game_proc(pid):
    exe = _proc_exe(pid)
    if exe.endswith("/bedrock_server"):
        return True
    # Last resort: some builds keep the original title.
    return "bedrock" in _proc_comm(pid)


def _children_of(pid):
    rc, out, _ = sh("pgrep -P %s 2>/dev/null" % pid, timeout=5)
    return [int(x) for x in out.split() if x.strip().isdigit()]


def _find_game_process(ws):
    """Locate the running game process.

    Trusting `ps -C bedrock_server | head -1` is wrong: ps sorts by PID, so any
    leftover zombie of an earlier run sorts first and reports 0% CPU / 0 MB.

    With a plugin loader (Endstone) the wrapper's direct child is the launcher,
    so ws["pid"] is NOT the game. The wrapper records the resolved game PID as
    "game_pid"; if that is missing we walk one level down.
    """
    seen = set()
    for key in ("game_pid", "pid"):
        pid = ws.get(key)
        if not pid or pid in seen or not _proc_alive(pid):
            continue
        seen.add(pid)
        if _is_game_proc(pid):
            info = _ps_one(pid)
            if info:
                info["comm"] = _proc_comm(pid)
                return info
        for child in _children_of(pid):
            if child in seen or not _proc_alive(child):
                continue
            seen.add(child)
            if _is_game_proc(child):
                cinfo = _ps_one(child)
                if cinfo:
                    cinfo["comm"] = _proc_comm(child)
                    return cinfo
    # Last resort: scan every process and keep the one whose exe is the game.
    rc, out, _ = sh("ps -eo pid=,pcpu=,pmem=,rss=,stat= --no-headers 2>/dev/null", timeout=10)
    for line in out.splitlines():
        f = line.split()
        if len(f) < 5 or f[4].startswith("Z"):
            continue
        try:
            pid = int(f[0])
        except ValueError:
            continue
        if not _is_game_proc(pid):
            continue
        try:
            return {"pid": pid, "cpu_pct": float(f[1]), "mem_pct": float(f[2]),
                    "rss_mb": round(int(f[3]) / 1024.0, 1), "comm": _proc_comm(pid)}
        except ValueError:
            continue
    return {}


def _zombies():
    """Leftover zombie game processes (visible so they can be noticed and reaped)."""
    _, out, _ = sh("ps -eo pid=,stat=,comm= --no-headers 2>/dev/null | awk '$2 ~ /^Z/ && $3 == \"bedrock_server\" {print $1}'", timeout=8)
    return [int(x) for x in out.split() if x.isdigit()]


def minecraft(port_items):
    units = run("systemctl list-unit-files %s --no-legend --no-pager" % MC_UNIT)
    if MC_UNIT not in units:
        return None

    unit_active = run("systemctl is-active %s" % MC_UNIT) or "unknown"
    ws = _wrapper_state()
    game_state = ws.get("state") or ("unknown" if unit_active != "active" else "running")
    wrapped = bool(ws)

    props = {}
    for line in run("systemctl show %s -p MainPID -p NRestarts -p MemoryCurrent" % MC_UNIT).splitlines():
        key, _, value = line.partition("=")
        props[key] = value

    # Uptime comes from the wrapper's recorded child start time: the wrapper can
    # restart the game server without the systemd unit ever changing state.
    uptime_sec = None
    started_at = ws.get("started_at")
    if game_state == "running":
        if started_at:
            try:
                uptime_sec = max(0, int((datetime.now() - datetime.fromisoformat(started_at)).total_seconds()))
            except ValueError:
                uptime_sec = None
        if uptime_sec is None:
            uptime_sec = _unit_uptime_seconds(MC_UNIT)

    lines = _console_lines()
    log_source = "console"
    if not lines:
        # Wrapper not installed yet: fall back to the journal so the panel still
        # works.  Scoped to the current run of the unit, not a fixed line window.
        log_source = "journal"
        up = _unit_uptime_seconds(MC_UNIT)
        if up is not None:
            since = (datetime.now() - timedelta(seconds=up + 300)).strftime("%Y-%m-%d %H:%M:%S")
            _, logs, _ = sh("journalctl -u %s --since '%s' --no-pager -o cat" % (MC_UNIT, since), timeout=30)
        else:
            _, logs, _ = sh("journalctl -u %s -n 400 --no-pager -o cat" % MC_UNIT, timeout=20)
        lines = [l for l in logs.splitlines() if l.strip()]

    last_start = None
    for idx, line in enumerate(lines):
        if "Server started." in line:
            last_start = idx

    version = None
    for line in (lines if last_start is None else lines[:last_start + 1]):
        m = re.search(r"Version:\s*([\d.]+)", line)
        if m:
            version = m.group(1)

    connected = {}
    if game_state == "running":
        for line in (lines if last_start is None else lines[last_start:]):
            m = re.search(r"Player connected:\s*([^,]+),\s*xuid:\s*(\d+)", line)
            if m:
                connected[m.group(1).strip()] = m.group(2)
                continue
            m = re.search(r"Player disconnected:\s*([^,]+),", line)
            if m:
                connected.pop(m.group(1).strip(), None)

    last_event = None
    for line in reversed(lines):
        if re.search(r"(Player (connected|disconnected|spawned)|Server started|Version:)", line):
            last_event = line.strip()
            break

    world_bytes = None
    _, du, _ = sh("du -sb '%s/worlds' 2>/dev/null" % MC_DIR, timeout=30)
    if du.split() and du.split()[0].isdigit():
        world_bytes = int(du.split()[0])

    props_file = {}
    for line in read(os.path.join(MC_DIR, "server.properties")).splitlines():
        if "=" in line and not line.strip().startswith("#"):
            key, _, value = line.partition("=")
            props_file[key.strip()] = value.strip()

    proc = _find_game_process(ws) if game_state == "running" else {}
    zombies = _zombies()

    allow_count = 0
    try:
        with open(os.path.join(MC_DIR, "allowlist.json"), encoding="utf-8") as fh:
            allowlist = json.load(fh)
        if isinstance(allowlist, list):
            allow_count = len(allowlist)
    except Exception:
        pass

    # Judge listening state against the *configured* ports and both transports:
    # raknet listens on UDP, nethernet on TCP + UDP 7551, so checking UDP alone
    # would wrongly report "not listening".
    try:
        mc_port = int(props_file.get("server-port", "19132"))
    except ValueError:
        mc_port = 19132
    try:
        mc_port6 = int(props_file.get("server-portv6", str(mc_port + 1)))
    except ValueError:
        mc_port6 = mc_port + 1
    listen_protos = sorted({p["proto"] for p in port_items if p["port"] in (mc_port, mc_port6)})
    port_listening = bool(listen_protos)

    console_lines = [l for l in lines[-60:] if "telemetry" not in l.lower()][-10:]
    mem_current = props.get("MemoryCurrent", "")

    return {
        "unit": MC_UNIT,
        "active": unit_active,                 # systemd unit state
        "game_state": game_state,              # running | stopped | exited | starting
        "wrapped": wrapped,                    # False => old unit, console not wired
        "wrapper_pid": ws.get("wrapper_pid"),
        # The wrapper's direct child (a launcher, when a plugin loader is used).
        "launcher_pid": ws.get("pid"),
        # The actual game process; equals launcher_pid on a plain BDS setup.
        "game_pid": ws.get("game_pid") or ws.get("pid"),
        "version": version,
        "started_at": started_at,
        "uptime_sec": uptime_sec,
        "uptime_human": human_duration(uptime_sec) if uptime_sec is not None else None,
        "players": sorted(connected.keys()),
        "player_count": len(connected),
        "max_players": props_file.get("max-players", "?"),
        "gamemode": props_file.get("gamemode", "?"),
        "difficulty": props_file.get("difficulty", "?"),
        "online_mode": props_file.get("online-mode", "?"),
        "level_name": props_file.get("level-name", "?"),
        "port": mc_port,
        "port6": mc_port6,
        "listen_protos": listen_protos,
        "transport": props_file.get("transport", ""),
        "port_listening": port_listening,
        "world_bytes": world_bytes,
        "world_gb": round(world_bytes / 1e9, 2) if world_bytes else None,
        "allowlist_count": allow_count,
        "memory_current_mb": round(int(mem_current) / 1048576.0, 1) if mem_current.isdigit() else None,
        "restarts": props.get("NRestarts"),
        "process": proc,
        "zombies": zombies,
        "last_event": last_event,
        "console": console_lines,
        "console_fifo": _fifo_has_reader(),
        "log_source": log_source,
        "backups": _backups(),
        "backup_root": MC_BACKUP_ROOT,
    }


# --------------------------------------------------------------------------
def main():
    port_items, port_error = safe(ports, ([], "ports() 异常"))
    data = {
        "collected_at": datetime.now().isoformat(timespec="seconds"),
        "host": safe(host_info, {}),
        "cpu": safe(cpu_info, {}),
        "temps": safe(temps, {}),
        "mem": safe(mem_info, {}),
        "uptime": safe(uptime_info, {}),
        "disks": safe(disks, []),
        "failed_mounts": safe(failed_mounts, []),
        "containers": safe(containers, {"available": False, "error": "containers() 异常"}),
        "ports": port_items,
        "ports_error": port_error,
        "services": safe(services, {}),
        "reboot": safe(reboot_required, {"required": False, "packages": []}),
        "procs": safe(top_procs, {}),
        "sessions": safe(sessions, []),
        "errors": safe(recent_errors, {"recent": [], "count_24h": None, "count_1h": None, "sources": []}),
        "connectivity": safe(connectivity, {}),
        "interfaces": safe(net_interfaces, []),
        "web": safe(web_checks, []),
    }
    data["minecraft"] = safe(minecraft, None, port_items)

    # Emit bytes explicitly: never let the locale decide the encoding of the payload.
    payload = json.dumps(data, ensure_ascii=False).encode("utf-8")
    sys.stdout.buffer.write(payload + b"\n")


if __name__ == "__main__":
    main()
