#!/usr/bin/env python3
"""Read-only health/reboot observer with durable heartbeats; never restarts the box."""
import argparse
from datetime import datetime
import hashlib
import json
import os
from pathlib import Path
import re
import signal
import subprocess
import tempfile
import threading

ORDERLY = re.compile(r"systemd-shutdown|Reached target.*(?:System Reboot|System Power Off)|systemd\[1\]: (?:Rebooting|Powering off)|reboot: Restarting system|reboot: Power down", re.I)
FAILURE = re.compile(r"Kernel panic - not syncing|Oops:|BUG: unable to handle|watchdog:.*(?:hard LOCKUP|soft lockup)|Out of memory: Killed process|oom-kill:|thermal.*critical", re.I)
EVIDENCE = re.compile(r"panic|oops|watchdog|oom|out of memory|thermal|overheat|shutdown|reboot|crash|segfault|I/O error|ext4.*error|mmc.*(?:error|timeout)|voltage|hung task", re.I)


def atomic_write(path, text):
    path.parent.mkdir(parents=True, exist_ok=True)
    name = None
    try:
        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent, delete=False) as f:
            name = f.name
            os.chmod(name, 0o600)
            f.write(text); f.flush(); os.fsync(f.fileno())
        os.replace(name, path); name = None
        if os.name == "posix":
            fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
            try: os.fsync(fd)
            finally: os.close(fd)
    finally:
        if name: os.unlink(name)


def command(args, timeout=4, limit=24576):
    try:
        r = subprocess.run(args, capture_output=True, text=True, errors="replace", timeout=timeout,
                           env={**os.environ, "LC_ALL": "C", "SYSTEMD_COLORS": "0"})
        return {"code": r.returncode, "text": r.stdout[-limit:], "error": r.stderr[-1024:]}
    except (OSError, subprocess.TimeoutExpired) as e:
        return {"code": -1, "text": "", "error": str(e)[:1024]}


def read(path, default=""):
    try: return Path(path).read_text().strip()
    except (OSError, ValueError): return default


def classify(evidence):
    text = "\n".join(evidence.get(k, {}).get("text", "") for k in ("tail", "warnings", "pstore_new"))
    if FAILURE.search(text): return "发现内核异常/OOM等线索，不能单独认定为重启原因"
    if ORDERLY.search(text): return "发现正常关机或重启流程记录，发起者需结合原始日志核对"
    return "重启原因未确认：无充分关机/故障证据，不能区分断电、硬复位等情况"


def transition(previous, boot_id, now, evidence=None):
    if previous is not None and (previous.get("version") != 1 or type(previous.get("reboots")) is not int):
        raise ValueError("Unsupported/corrupt saved state; preserve it for diagnosis")
    state = dict(previous or {"version": 1, "reboots": 0, "started_at": now, "history": []})
    old = state.get("boot_id")
    event = "boot_baseline" if not old else "observer_restart_same_boot"
    if old and old != boot_id:
        event = "boot_change"; state["reboots"] += 1
        state["history"] = (state.get("history", []) + [{"previous_boot_id": old, "boot_id": boot_id,
            "previous_last_seen": state.get("last_seen"), "previous_uptime_seconds": state.get("uptime_seconds"),
            "detected_at": now, "assessment": classify(evidence or {})}])[-32:]
    state.update(boot_id=boot_id, observer_started_at=now)
    return state, event


class Observer:
    def __init__(self, logs, state_dir):
        self.logs, self.state_dir = logs, state_dir
        logs.mkdir(parents=True, exist_ok=True); state_dir.mkdir(parents=True, exist_ok=True)
        self.state_file = state_dir / "overnight-state.json"
        self.state = None

    def log(self, event, **values):
        now = datetime.now().astimezone()
        path = self.logs / ("overnight-health-" + now.strftime("%Y%m%d") + ".log")
        if path.exists() and path.stat().st_size > 8*1024*1024:
            for i in (2, 1):
                source = path.with_suffix(path.suffix + "." + str(i))
                if source.exists(): os.replace(source, path.with_suffix(path.suffix + "." + str(i+1)))
            os.replace(path, path.with_suffix(path.suffix + ".1"))
        record = dict(time=now.isoformat(timespec="seconds"), event=event, **values)
        with path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n"); f.flush(); os.fsync(f.fileno())
        # Bound only our own logs; never touch kiosk/stream/application logs.
        days = sorted(self.logs.glob("overnight-health-????????.log"))
        for old in days[:-7]:
            if old == path: continue
            old.unlink(missing_ok=True)
            for i in (1, 2, 3): old.with_suffix(old.suffix + "." + str(i)).unlink(missing_ok=True)

    def prior_evidence(self, boot_id):
        # Boot IDs remain meaningful even when RTC/NTP changes wall-clock time.
        if not re.fullmatch(r"[0-9a-f-]{32,36}", boot_id):
            return {"tail": {"text": "", "error": "invalid saved boot ID"}}
        base = ["journalctl", "--no-pager", "-b", boot_id, "-o", "short-monotonic"]
        return {"tail": command(base+["-n", "100"]), "warnings": command(base+["-p", "warning", "-n", "120"]),
                "last": command(["last", "-x", "-n", "10"])}

    def start(self):
        boot_id = read("/proc/sys/kernel/random/boot_id")
        if not re.fullmatch(r"[0-9a-f-]{36}", boot_id): raise ValueError("Cannot read kernel boot ID")
        previous = json.loads(self.state_file.read_text()) if self.state_file.exists() else None
        try:
            with Path('/var/lib/vdi-observer/pstore.json').open('r', encoding='utf-8') as f:
                pstore = json.loads(f.read(1024*1024))
        except (OSError, ValueError) as e: pstore = {"read_error": str(e)}
        # Compare contents, not capture time; an old ramoops archive is not a new crash.
        texts = {root+"/"+name: entry["text"] for root, files in pstore.get("sources", {}).items()
                 for name, entry in files.items() if isinstance(entry, dict) and "text" in entry}
        digest = hashlib.sha256(json.dumps(texts, sort_keys=True).encode()).hexdigest()
        evidence = self.prior_evidence(previous["boot_id"]) if previous and previous.get("boot_id") != boot_id else None
        if evidence is not None and previous.get("pstore_digest") and previous["pstore_digest"] != digest:
            evidence["pstore_new"] = {"text": "\n".join(texts.values()), "note": "pstore contents changed since previous heartbeat; cause still needs review"}
        self.state, event = transition(previous, boot_id, datetime.now().astimezone().isoformat(timespec="seconds"), evidence)
        self.state["pstore_digest"] = digest
        self.log(event, boot_id=boot_id, reboots=self.state["reboots"], previous_last_seen=(previous or {}).get("last_seen"), evidence=evidence)
        if evidence: self.log("restart_assessment", assessment=self.state["history"][-1]["assessment"])
        self.log("boot_evidence", kernel=os.uname().release,
                 journal=command(["journalctl", "--list-boots", "--no-pager", "-n", "8"]), pstore=pstore,
                 watchdog_bootstatus={str(p): read(p) for p in Path("/sys/class/watchdog").glob("watchdog*/bootstatus")})

    def sample(self):
        uptime = float(read("/proc/uptime").split()[0]); now = datetime.now().astimezone()
        processes = []
        for path in Path("/proc").glob("[0-9]*/cmdline"):
            try:
                args = path.read_bytes().decode(errors="replace").strip("\0").split("\0"); exe = Path(args[0]).name
                if exe not in ("moonlight", "kiosk-shell", "vdi-kiosk-monitor"): continue
                item = {"pid": int(path.parent.name), "name": exe}
                if exe == "moonlight" and "stream" in args:
                    i = args.index("stream"); item["host"] = args[i+1] if i+1 < len(args) else ""
                    for flag in ("--resolution", "--fps"):
                        if flag in args and args.index(flag)+1 < len(args): item[flag[2:]] = args[args.index(flag)+1]
                processes.append(item)
            except (OSError, ValueError, IndexError): continue
        temps = {p.parent.name: int(read(p, "0"))/1000 for p in Path("/sys/class/thermal").glob("thermal_zone*/temp")}
        memory = {s.split(":")[0]: s.split(":")[1].strip() for s in read("/proc/meminfo").splitlines() if s.startswith(("MemAvailable:", "MemTotal:", "SwapFree:"))}
        display = command(["env", "DISPLAY=:0", "XAUTHORITY="+str(Path.home()/".Xauthority"), "xrandr", "--current"], timeout=3)
        display["text"] = "\n".join(s for s in display["text"].splitlines() if s.startswith("Screen ") or " connected" in s or " disconnected" in s)
        network = {p.name: {"operstate": read(p/"operstate"), "rx_errors": read(p/"statistics/rx_errors"), "tx_errors": read(p/"statistics/tx_errors")}
                   for p in Path("/sys/class/net").iterdir() if p.name != "lo"}
        if self.state.get("last_seen") and self.state.get("sample_boot_id") == self.state["boot_id"]:
            wall = now.timestamp()-self.state.get("last_wall_seconds", now.timestamp())
            mono = uptime-self.state.get("uptime_seconds", uptime)
            if abs(wall-mono) > 120: self.log("clock_adjustment", wall_delta=wall, uptime_delta=mono)
        fs = os.statvfs("/")
        values = dict(boot_id=self.state["boot_id"], reboots=self.state["reboots"], uptime_seconds=uptime,
                      load=read("/proc/loadavg").split()[:3], memory=memory, temperatures_c=temps,
                      processes=processes, display=display, network=network, root_available_mb=fs.f_bavail*fs.f_frsize//(1024*1024))
        self.log("sample", **values)
        self.state.update(last_seen=now.isoformat(timespec="seconds"), last_wall_seconds=now.timestamp(), uptime_seconds=uptime,
                          sample_boot_id=self.state["boot_id"], last_sample=values)
        self.save()

    def save(self):
        atomic_write(self.state_file, json.dumps(self.state, ensure_ascii=False, indent=2)+"\n")
        history = self.state.get("history", [])
        lines = ["盒子过夜观察汇总", "记录启用时间："+self.state["started_at"], "最近记录时间："+self.state.get("last_seen", ""),
                 "记录启用后重启次数："+str(self.state["reboots"]), "当前启动编号："+self.state["boot_id"],
                 "当前开机时长（秒）："+str(self.state.get("uptime_seconds", "")),
                 "最近一次重启判断："+(history[-1]["assessment"] if history else "观察期间尚未发现重启"),
                 "详细数据见同目录 overnight-health-YYYYMMDD.log；每分钟更新。",
                 "服务重启不计为盒子重启；无故障证据时不推断具体断电原因。"]
        atomic_write(self.logs/"overnight-summary.txt", "\n".join(lines)+"\n")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--once", action="store_true"); parser.add_argument("--report", action="store_true")
    args = parser.parse_args()
    logs = Path(os.environ.get("VDI_OBSERVER_LOG_DIR", str(Path.home()/"vdi-terminal/logs")))
    state_dir = Path(os.environ.get("VDI_OBSERVER_STATE_DIR", str(Path.home()/"vdi-terminal/state")))
    if args.report: print((logs/"overnight-summary.txt").read_text()); return
    interval = int(os.environ.get("VDI_OBSERVER_INTERVAL", "60"))
    if not 10 <= interval <= 3600: parser.error("Interval must be 10..3600 seconds")
    os.umask(0o077); observer = Observer(logs, state_dir)
    import fcntl
    with (state_dir/"overnight-observer.lock").open("w") as lock:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        stopped = threading.Event(); stop_signal = []
        def on_signal(signum, frame): stop_signal.append(signum); stopped.set()
        signal.signal(signal.SIGTERM, on_signal); signal.signal(signal.SIGINT, on_signal)
        observer.start(); observer.sample()
        if args.once: return
        ticks = 0
        while not stopped.wait(interval):
            observer.sample(); ticks += 1
            if ticks % max(1, 600//interval) == 0:
                r = command(["journalctl", "-k", "-b", "0", "-p", "warning", "-n", "50", "--no-pager", "-o", "short-monotonic"])
                r["text"] = "\n".join(s for s in r["text"].splitlines() if EVIDENCE.search(s))
                observer.log("kernel_warning_clues", **r)
        # SIGTERM may just be a service stop. Never call it an OS reboot.
        observer.log("observer_stop", signal=stop_signal, boot_id=observer.state["boot_id"])
        observer.state["observer_stop_signal"] = stop_signal; observer.save()


if __name__ == "__main__": main()
