#!/usr/bin/env python3
"""X|COOL telemetry receiver: tasks (tarefa.*) and minigames (jogo.*).

One code for both places:
  - imported by servidor.py (the local mocks), mounted at /telemetria and /_llm;
  - run standalone on the VPS behind nginx (DEPLOY-TELEMETRIA.md):
      python3 telemetria_receptor.py --dados /var/lib/xcool-telemetria --porta 8790 --prefixo /midia/telemetria

Routes, relative to a prefix P:
  POST P  (or P/)                 batch {eventos:[...]} (or a bare list) -> validated, appended to jsonl
  GET  P/telemetria.js            the page component
  GET  P/painel/  P/painel/<f>    the reading panel
  GET  P/dados/lista              {tarefas:[...], jogos:[...]}
  GET  P/dados/eventos?tarefa=<t> | ?jogo=<j>   the raw jsonl
Files: <dados>/<tarefa>.jsonl and <dados>/jogos/<jogo>.jsonl.
Standard library only. Nothing personal is stored: no IP, no name (CDC-001); the access log has no address.
A minigame measures no skill and grants no tokens (ADR-XC-028 item 6, ADR-XC-072 item 1): any evidence,
weight, token or skill key inside a game event is refused.
"""
import argparse
import json
import os
import re
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs

HERE = os.path.dirname(os.path.abspath(__file__))
PANEL_DIR = os.path.join(HERE, "painel")
COMPONENT = os.path.join(HERE, "telemetria.js")

MAX_BODY = 512 * 1024          # one batch
MAX_EVENTS = 1000              # per batch
MAX_CARGA = 32 * 1024          # serialized payload of one event
MAX_STR = 600                  # any string inside a payload
MAX_FILE = 200 * 1024 * 1024   # one jsonl; above it the batch is refused (disk guard for a public endpoint)
KEY_RE = re.compile(r"^[a-z0-9_-]{1,64}$")
ID_RE = re.compile(r"^[A-Za-z0-9_-]{4,64}$")
PANEL_FILE_RE = re.compile(r"^[a-z0-9_-]+\.(html|js)$")

# Same names as the game's closed catalog (game/theme/plugins/telemetria-captura.js) plus tarefa.acao,
# tarefa.fala, tarefa.foco and the jogo.* family. Declared keys may be missing; undeclared keys are refused.
CATALOG = {
    "tarefa.inicio": {"nivel", "parametros", "habilidades", "pagina"},
    "tarefa.acao": {"acao", "detalhe", "correta"},
    "tarefa.fala": {"ponto", "fonte", "texto", "latencia_ms"},
    "tarefa.pista": {"nivel", "custo", "fonte"},
    "tarefa.erro": {"erro", "detalhe"},
    "tarefa.foco": {"visivel", "medidas"},          # both modes
    "tarefa.conclusao": {"outcome", "weight", "tokens", "evidencia", "medidas"},
    "tarefa.abandono": {"motivo", "medidas"},
    "economia.concessao": {"quantidade"},
    "jogo.inicio": {"fase", "estilo", "parametros", "pagina"},
    "jogo.acao": {"fase", "de_ms", "ate_ms", "total", "contagens", "pico_por_segundo"},
    "jogo.fase": {"de", "para", "pontos"},
    "jogo.fim": {"motivo", "pontos", "fase", "colocacao", "participantes", "medidas"},
    "jogo.abandono": {"motivo", "fase", "pontos", "medidas"},
}
GAME_TYPES = {t for t in CATALOG if t.startswith("jogo.")} | {"tarefa.foco"}
TASK_TYPES = {t for t in CATALOG if not t.startswith("jogo.")}
OUTCOMES = {"pass", "partial", "fail"}
GAME_ENDS = {"ganhou", "perdeu", "saiu", "terminou"}
GAME_STYLES = {"desafio", "corrida", "livre"}
# Proficiency and economy keys a game event may never carry, at any depth.
GAME_FORBIDDEN = {"evidencia", "skill_evidence", "skill_code", "weight", "tokens", "quantidade", "outcome"}

_lock = threading.Lock()


def _strings_ok(value, depth=0):
    if depth > 6:
        return False
    if isinstance(value, str):
        return len(value) <= MAX_STR
    if isinstance(value, dict):
        return all(isinstance(k, str) and len(k) <= 64 and _strings_ok(v, depth + 1) for k, v in value.items())
    if isinstance(value, list):
        return all(_strings_ok(v, depth + 1) for v in value)
    return value is None or isinstance(value, (bool, int, float))


def _keys_deep(value, depth=0):
    if depth > 6:
        return set()
    if isinstance(value, dict):
        out = set(value)
        for v in value.values():
            out |= _keys_deep(v, depth + 1)
        return out
    if isinstance(value, list):
        out = set()
        for v in value:
            out |= _keys_deep(v, depth + 1)
        return out
    return set()


def validate_event(ev):
    """Returns (clean_event, None) or (None, error)."""
    if not isinstance(ev, dict):
        return None, "event is not an object"
    tipo = ev.get("tipo")
    if tipo not in CATALOG:
        return None, "tipo not in catalog"
    has_task, has_game = "tarefa" in ev, "jogo" in ev
    if has_task == has_game:
        return None, "send exactly one of tarefa or jogo"
    mode = "jogo" if has_game else "tarefa"
    if tipo not in (GAME_TYPES if mode == "jogo" else TASK_TYPES):
        return None, f"{tipo} is not a {mode} event"
    key = ev[mode]
    if not isinstance(key, str) or not KEY_RE.match(key):
        return None, f"{mode} must match ^[a-z0-9_-]{{1,64}}$"
    for k in ("sessao", "encontro"):
        if not isinstance(ev.get(k), str) or not ID_RE.match(ev[k]):
            return None, f"{k} must be an anonymous id"
    ts = ev.get("ts")
    if isinstance(ts, bool) or not isinstance(ts, (int, float)) or not 1.5e12 < ts < 4e12:
        return None, "ts must be ms since epoch"
    t_ms = ev.get("t_ms", 0)
    if isinstance(t_ms, bool) or not isinstance(t_ms, (int, float)) or t_ms < 0:
        return None, "t_ms must be a non-negative number"
    carga = ev.get("carga", {})
    if not isinstance(carga, dict):
        return None, "carga must be an object"
    if mode == "jogo":
        bad = _keys_deep(carga) & GAME_FORBIDDEN
        if bad:
            return None, ("minigame measures no skill and grants no tokens (ADR-XC-028 item 6, ADR-XC-072 item 1): "
                          + ",".join(sorted(bad)))
    extra = set(carga) - CATALOG[tipo]
    if extra:
        return None, "carga key not declared: " + ",".join(sorted(k[:32] for k in extra))
    if len(json.dumps(carga, ensure_ascii=False)) > MAX_CARGA or not _strings_ok(carga):
        return None, "carga too large"
    if tipo == "tarefa.conclusao" and carga.get("outcome") not in OUTCOMES:
        return None, "outcome must be pass|partial|fail"
    if tipo == "jogo.fim":
        if carga.get("motivo") not in GAME_ENDS:
            return None, "motivo must be ganhou|perdeu|saiu|terminou"
        pos, total = carga.get("colocacao"), carga.get("participantes")
        if pos is not None and (isinstance(pos, bool) or not isinstance(pos, int) or pos < 1):
            return None, "colocacao must be an integer >= 1"
        if total is not None and (isinstance(total, bool) or not isinstance(total, int) or total < 1
                                  or (pos is not None and total < pos)):
            return None, "participantes must be an integer >= colocacao"
    if tipo == "jogo.inicio" and carga.get("estilo") is not None and carga["estilo"] not in GAME_STYLES:
        return None, "estilo must be desafio|corrida|livre"
    if tipo == "tarefa.fala" and carga.get("fonte") not in ("llm", "roteiro"):
        return None, "fonte must be llm|roteiro"
    clean = {"v": 1, "tipo": tipo, "ts": int(ts), "t_ms": int(t_ms), "sessao": ev["sessao"],
             "encontro": ev["encontro"], mode: key, "carga": carga,
             "recebido": time.strftime("%Y-%m-%dT%H:%M:%S%z")}
    return clean, None


def _path(data_dir, mode, key):
    return os.path.join(data_dir, "jogos", key + ".jsonl") if mode == "jogo" else os.path.join(data_dir, key + ".jsonl")


def ingest(body, data_dir):
    """Returns (status, payload)."""
    events = body.get("eventos") if isinstance(body, dict) else body
    if not isinstance(events, list) or not events:
        return 400, {"ok": False, "erro": "send {eventos:[...]} with at least one event"}
    if len(events) > MAX_EVENTS:
        return 400, {"ok": False, "erro": f"at most {MAX_EVENTS} events per batch"}
    groups, refused = {}, []
    for i, ev in enumerate(events):
        clean, err = validate_event(ev)
        if clean is None:
            refused.append({"i": i, "erro": err})
        else:
            mode = "jogo" if "jogo" in clean else "tarefa"
            groups.setdefault((mode, clean[mode]), []).append(clean)
    saved = 0
    with _lock:
        for (mode, key), rows in groups.items():
            path = _path(data_dir, mode, key)
            os.makedirs(os.path.dirname(path), exist_ok=True)
            if os.path.exists(path) and os.path.getsize(path) > MAX_FILE:
                refused.extend({"i": -1, "erro": f"{mode} {key}: file full"} for _ in rows)
                continue
            with open(path, "a", encoding="utf-8") as f:
                for r in rows:
                    f.write(json.dumps(r, ensure_ascii=False) + "\n")
            saved += len(rows)
    return (200 if saved else 400), {"ok": saved > 0, "gravados": saved, "recusados": refused}


def listing(data_dir):
    def scan(d):
        out = []
        if os.path.isdir(d):
            for n in sorted(os.listdir(d)):
                if n.endswith(".jsonl") and KEY_RE.match(n[:-6]):
                    st = os.stat(os.path.join(d, n))
                    out.append({"nome": n[:-6], "bytes": st.st_size, "alterado": int(st.st_mtime * 1000)})
        return out
    tarefas = [dict(x, tarefa=x["nome"]) for x in scan(data_dir)]
    jogos = [dict(x, jogo=x["nome"]) for x in scan(os.path.join(data_dir, "jogos"))]
    return {"ok": True, "tarefas": tarefas, "jogos": jogos}


# ── HTTP plumbing shared by servidor.py and the standalone receiver ──────────
def _send(h, status, data, ctype, extra=None):
    h.send_response(status)
    h.send_header("Content-Type", ctype)
    h.send_header("Content-Length", str(len(data)))
    h.send_header("Cache-Control", "no-store")
    for k, v in (extra or {}).items():
        h.send_header(k, v)
    h.end_headers()
    h.wfile.write(data)


def _json(h, status, payload):
    _send(h, status, json.dumps(payload, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8")


def _file(h, path, ctype):
    with open(path, "rb") as f:
        _send(h, 200, f.read(), ctype)


def serve(h, method, prefix, data_dir):
    """Handles the request on handler `h` if its path is under `prefix`. Returns True when handled."""
    route, _, query = h.path.partition("?")
    if route != prefix and not route.startswith(prefix + "/"):
        return False
    rest = route[len(prefix):]
    if method == "POST":
        if rest not in ("", "/"):
            _json(h, 404, {"ok": False, "erro": "not found"})
            return True
        length = int(h.headers.get("Content-Length") or 0)
        if length <= 0 or length > MAX_BODY:
            if length > MAX_BODY:
                h.close_connection = True  # do not read a body we refuse
            _json(h, 413 if length > MAX_BODY else 400, {"ok": False, "erro": "empty or too large body"})
            return True
        try:
            body = json.loads(h.rfile.read(length).decode("utf-8"))
        except ValueError:
            _json(h, 400, {"ok": False, "erro": "body must be JSON"})
            return True
        _json(h, *ingest(body, data_dir))
        return True
    if rest == "/telemetria.js":
        _file(h, COMPONENT, "text/javascript; charset=utf-8")
        return True
    if rest == "/painel":
        _send(h, 301, b"", "text/plain", {"Location": prefix + "/painel/"})
        return True
    if rest.startswith("/painel/"):
        name = rest[len("/painel/"):] or "index.html"
        if not PANEL_FILE_RE.match(name) or not os.path.isfile(os.path.join(PANEL_DIR, name)):
            _json(h, 404, {"ok": False, "erro": "not found"})
            return True
        ctype = "text/html; charset=utf-8" if name.endswith(".html") else "text/javascript; charset=utf-8"
        _file(h, os.path.join(PANEL_DIR, name), ctype)
        return True
    if rest == "/dados/lista":
        _json(h, 200, listing(data_dir))
        return True
    if rest == "/dados/eventos":
        q = parse_qs(query)
        mode = "jogo" if "jogo" in q else "tarefa"
        key = (q.get(mode) or [""])[0]
        path = _path(data_dir, mode, key)
        if not KEY_RE.match(key) or not os.path.isfile(path):
            _json(h, 404, {"ok": False, "erro": f"unknown {mode}"})
            return True
        _file(h, path, "application/x-ndjson; charset=utf-8")
        return True
    if rest in ("", "/"):
        _json(h, 405, {"ok": False, "erro": "POST events here; panel at " + prefix + "/painel/"})
        return True
    return False


def main():
    ap = argparse.ArgumentParser(description="X|COOL telemetry receiver (tasks and minigames)")
    ap.add_argument("--dados", required=True, help="directory for the jsonl files")
    ap.add_argument("--porta", type=int, required=True)
    ap.add_argument("--prefixo", default="/midia/telemetria", help="URL prefix nginx passes through unchanged")
    args = ap.parse_args()
    data_dir = os.path.abspath(args.dados)
    prefix = "/" + args.prefixo.strip("/")
    os.makedirs(data_dir, exist_ok=True)

    class Handler(BaseHTTPRequestHandler):
        server_version = "xcool-telemetria"

        def log_message(self, fmt, *a):  # no client address in the log (CDC-001)
            print(time.strftime("%Y-%m-%dT%H:%M:%S ") + (fmt % a), flush=True)

        def do_POST(self):
            if not serve(self, "POST", prefix, data_dir):
                _json(self, 404, {"ok": False, "erro": "not found"})

        def do_GET(self):
            if not serve(self, "GET", prefix, data_dir):
                _json(self, 404, {"ok": False, "erro": "not found"})

    server = ThreadingHTTPServer(("127.0.0.1", args.porta), Handler)
    print(f"telemetria: {data_dir} on http://127.0.0.1:{args.porta}{prefix} (panel {prefix}/painel/)", flush=True)
    server.serve_forever()


if __name__ == "__main__":
    main()