#!/usr/bin/env python3
"""
Host protocol probes · safe measure only · feeds SPV HUDs
No listeners opened · outbound/local checks · rides BGF spirit

  wb probe           # all
  wb probe ssh dns icmp
  pack embeds snapshot as window.__BGRTX_PROBES__
"""
from __future__ import annotations

import json
import os
import shutil
import socket
import subprocess
import time
from pathlib import Path
from typing import Any, Dict, List, Optional

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "settings" / "probes.json"


def _run(cmd: List[str], timeout: float = 3.0) -> tuple[int, str, str]:
    try:
        p = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout,
            env={**os.environ, "LC_ALL": "C"},
        )
        return p.returncode, (p.stdout or "").strip(), (p.stderr or "").strip()
    except FileNotFoundError:
        return 127, "", "missing"
    except subprocess.TimeoutExpired:
        return 124, "", "timeout"


def _tcp_open(host: str, port: int, timeout: float = 0.8) -> Dict[str, Any]:
    t0 = time.monotonic()
    try:
        with socket.create_connection((host, port), timeout=timeout):
            ms = int((time.monotonic() - t0) * 1000)
            return {"ok": True, "ms": ms, "host": host, "port": port}
    except OSError as e:
        return {"ok": False, "error": str(e)[:80], "host": host, "port": port}


def probe_icmp(host: str = "1.1.1.1") -> Dict[str, Any]:
    if not shutil.which("ping"):
        return {"ok": False, "error": "no ping", "proto": "icmp"}
    code, out, err = _run(["ping", "-c", "1", "-W", "2", host], timeout=4)
    # parse time=xx
    ms = None
    for token in (out + " " + err).replace("=", " ").split():
        if token.replace(".", "").isdigit() and "time" in (out + err):
            pass
    import re

    m = re.search(r"time[=<]([\d.]+)", out + err)
    if m:
        ms = float(m.group(1))
    return {
        "ok": code == 0,
        "proto": "icmp",
        "host": host,
        "ms": ms,
        "field": int(ms) | 1 if ms is not None else 1,
    }


def probe_dns(name: str = "biggrinrtx.com") -> Dict[str, Any]:
    t0 = time.monotonic()
    try:
        infos = socket.getaddrinfo(name, None)
        ms = int((time.monotonic() - t0) * 1000)
        addrs = sorted({i[4][0] for i in infos})[:6]
        return {
            "ok": bool(addrs),
            "proto": "dns",
            "name": name,
            "addrs": addrs,
            "ms": ms,
            "field": ms | 1,
        }
    except socket.gaierror as e:
        return {"ok": False, "proto": "dns", "name": name, "error": str(e)[:80]}


def probe_tcp(host: str, port: int, proto: str = "tcp") -> Dict[str, Any]:
    r = _tcp_open(host, port)
    r["proto"] = proto
    if r.get("ms") is not None:
        r["field"] = int(r["ms"]) | 1
    return r


def probe_ssh(host: str = "127.0.0.1", port: int = 22) -> Dict[str, Any]:
    r = probe_tcp(host, port, "ssh")
    # banner peek optional
    if r.get("ok"):
        try:
            s = socket.create_connection((host, port), timeout=0.6)
            s.settimeout(0.4)
            try:
                ban = s.recv(64).decode("utf-8", "replace").strip()
                r["banner"] = ban[:60]
            except OSError:
                pass
            s.close()
        except OSError:
            pass
    return r


def probe_https(host: str = "biggrinrtx.com", port: int = 443) -> Dict[str, Any]:
    return probe_tcp(host, port, "https")


def probe_http(host: str = "example.com", port: int = 80) -> Dict[str, Any]:
    return probe_tcp(host, port, "http")


def probe_ntp() -> Dict[str, Any]:
    # just clock local
    return {
        "ok": True,
        "proto": "ntp",
        "local_epoch": int(time.time()),
        "iso": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "field": (int(time.time()) & 0xFFFF) | 1,
        "note": "local clock sample · not NTP query",
    }


def probe_serial() -> Dict[str, Any]:
    ports = sorted(Path("/dev").glob("ttyUSB*")) + sorted(Path("/dev").glob("ttyACM*"))
    return {
        "ok": True,
        "proto": "serial",
        "devices": [str(p) for p in ports[:12]],
        "n": len(ports),
        "field": (len(ports) or 1) | 1,
    }


def probe_git() -> Dict[str, Any]:
    root = ROOT
    # walk up for .git
    p = root
    gitdir = None
    for _ in range(6):
        if (p / ".git").exists():
            gitdir = p
            break
        if p.parent == p:
            break
        p = p.parent
    if not gitdir or not shutil.which("git"):
        return {"ok": False, "proto": "git", "error": "no git repo/tool"}
    code, out, _ = _run(["git", "-C", str(gitdir), "rev-parse", "--abbrev-ref", "HEAD"])
    code2, sha, _ = _run(["git", "-C", str(gitdir), "rev-parse", "--short", "HEAD"])
    return {
        "ok": code == 0,
        "proto": "git",
        "branch": out if code == 0 else None,
        "sha": sha if code2 == 0 else None,
        "root": str(gitdir),
        "field": 1,
    }


def probe_rtmp() -> Dict[str, Any]:
    # stream status from stream_io if present
    try:
        import stream_io as S

        st = S.status()
        return {
            "ok": True,
            "proto": "rtmp",
            "running": st.get("running"),
            "key_set": st.get("key_set"),
            "video": st.get("video_device"),
            "audio": st.get("audio_device"),
            "field": 1 if st.get("running") else 3,
        }
    except Exception as e:
        return {"ok": False, "proto": "rtmp", "error": str(e)[:60]}


PROBERS = {
    "icmp": lambda: probe_icmp(),
    "dns": lambda: probe_dns(),
    "ssh": lambda: probe_ssh(),
    "https": lambda: probe_https(),
    "http": lambda: probe_http(),
    "ntp": lambda: probe_ntp(),
    "serial": lambda: probe_serial(),
    "git": lambda: probe_git(),
    "rtmp": lambda: probe_rtmp(),
    "tcp": lambda: probe_tcp("1.1.1.1", 443, "tcp"),
    "sftp": lambda: probe_ssh(),  # same port measure
    "scp": lambda: probe_ssh(),
    "telnet": lambda: probe_tcp("127.0.0.1", 23, "telnet"),
    "smtp": lambda: probe_tcp("127.0.0.1", 25, "smtp"),
    "imap": lambda: probe_tcp("127.0.0.1", 993, "imap"),
    "mqtt": lambda: probe_tcp("127.0.0.1", 1883, "mqtt"),
    "vnc": lambda: probe_tcp("127.0.0.1", 5900, "vnc"),
    "socks": lambda: probe_tcp("127.0.0.1", 1080, "socks"),
    "ws": lambda: probe_https(),
    "udp": lambda: {"ok": True, "proto": "udp", "note": "connectionless · no probe bind", "field": 1},
    "ftp": lambda: probe_tcp("127.0.0.1", 21, "ftp"),
    "rsync": lambda: probe_tcp("127.0.0.1", 873, "rsync"),
}


def run_all(which: Optional[List[str]] = None) -> Dict[str, Any]:
    keys = which or list(PROBERS.keys())
    results: Dict[str, Any] = {}
    t0 = time.time()
    for k in keys:
        fn = PROBERS.get(k)
        if not fn:
            results[k] = {"ok": False, "error": "unknown proto"}
            continue
        try:
            results[k] = fn()
        except Exception as e:
            results[k] = {"ok": False, "proto": k, "error": str(e)[:80]}
    blob = {
        "updated": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "elapsed_ms": int((time.time() - t0) * 1000),
        "probes": results,
        "law": "measure only · no permanent listeners · SPV free",
    }
    try:
        OUT.parent.mkdir(parents=True, exist_ok=True)
        OUT.write_text(json.dumps(blob, indent=2) + "\n")
    except OSError:
        pass
    return blob


def export_for_pack() -> Dict[str, Any]:
    return run_all()


if __name__ == "__main__":
    import sys

    args = sys.argv[1:]
    print(json.dumps(run_all(args or None), indent=2))