Blame for scripts/75_cli.py

#!/usr/bin/env python3
"""KHTST-RUST — CLI MLOps unificada (v0.10.0, crítica Qwen3.8 lote 3 nº 2).

**Por que FAÇADA e não substituição:** a proposta original ("substituir
por um CLI declarativo") violaria duas regras do projeto: (a) os scripts
numerados são a trilha de rastreabilidade das memórias r-XX (cada prova
cita "scripts 39-74"); (b) aprimoramentos não removem funcionalidade
eficaz. O CLI unifica o ENTRY POINT (um só comando para todo o pipeline)
delegando via subprocess aos scripts canônicos — zero duplicação de
lógica, zero remoção — e acrescenta o que não existia: validação de
integridade criptográfica dos checkpoints ANTES do resume (Teo CP-1..3,
r-35 §4) e gestão de estados locais.

Uso:
    python scripts/75_cli.py versao
    python scripts/75_cli.py treinar --validar-checkpoints
    python scripts/75_cli.py verificar-checkpoints
    python scripts/75_cli.py estados [--apagar]
    python scripts/75_cli.py coletar|suite|telemetria|publicar [--confirmar]
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Optional

import typer

app = typer.Typer(help="Pipeline KHTST-RUST v0.10.0 — fachada unificada (r-35 §6)")

RAIZ = Path(__file__).resolve().parent.parent
RESULTADOS = RAIZ / "resultados"
CHECKPOINTS = [
    RESULTADOS / "tokenizador_conceitos.json",
    RESULTADOS / "gerador_conceitos.json",
]

# scripts canônicos por estágio (trilha r-XX preservada)
DELEGACAO = {
    "coletar": "70_coleta_shards_v090.py",
    "treinar": "71_treino_v090.py",
    "suite": "77_suite_v010.py",
    "telemetria": "76_telemetria_v010.py",
    "publicar": "78_publicar_v010.py",
}

# padrões de estados de treino (mesma lista da varredura pós-publicação)
PADROES_ESTADO = [
    "*.pt", "*.npz", "*.ckpt", "*.safetensors", "*.estado.json",
    "**/*.pt", "**/*.npz", "**/*.ckpt", "**/*.safetensors", "**/*.estado.json",
]


def _delega(estagio: str, extra: list[str] | None = None) -> int:
    script = DELEGACAO[estagio]
    cmd = [sys.executable, str(RAIZ / "scripts" / script)] + (extra or [])
    typer.echo(f"→ {estagio}: {' '.join(cmd)}", err=True)
    return subprocess.run(cmd, cwd=RAIZ).returncode


@app.command()
def versao() -> None:
    """Versão do núcleo Rust e informação de build."""
    import khtst_rust as k
    typer.echo(k.versao())
    typer.echo(k.info_build())


@app.command()
def coletar() -> None:
    """Coleta shards reais (Wikipédia PT-BR, streaming controlado)."""
    raise typer.Exit(_delega("coletar"))


def _valida_ou_regenera(ck: Path, k) -> str:
    """Política de resume com prova de integridade (CP-1..CP-3):
    - não existe              → 'ausente'   (treina)
    - manifesto bate (SHA256) → 'válido'    (reusa — rápido)
    - manifesto divergente    → 'corrompido' (APAGA e retreina: nunca
      reusa estado em silêncio)
    - sem manifesto (legado)  → 'legado'    (reusa; o carregador legado
      falha alto em JSON truncado — detecção por parse)
    """
    if not ck.exists():
        return "ausente"
    if k.checkpoint_valido(str(ck)):
        return "válido"
    tem_manifesto = k.checkpoint_manifesto(str(ck)) is not None
    if tem_manifesto:
        ck.unlink(missing_ok=True)
        Path(str(ck) + ".manifest.json").unlink(missing_ok=True)
        return "corrompido (apagado — será retreinado)"
    return "legado (sem manifesto; parse do carregador detecta truncamento)"


@app.command()
def treinar(
    validar_checkpoints: bool = typer.Option(
        True, "--validar-checkpoints/--nao-validar",
        help="Valida SHA-256 dos checkpoints antes do resume (Teo CP-3)."),
) -> None:
    """Pipeline A/B/C (script 71) com pré-validação de integridade."""
    import khtst_rust as k
    if validar_checkpoints:
        for ck in CHECKPOINTS:
            estado = _valida_ou_regenera(ck, k)
            typer.echo(f"checkpoint {ck.name}: {estado}")
    raise typer.Exit(_delega("treinar"))


@app.command()
def verificar_checkpoints() -> None:
    """Relatório de integridade (SHA-256 + tamanho) dos checkpoints."""
    import khtst_rust as k
    achou = False
    for ck in RESULTADOS.rglob("*.json"):
        if ck.name.endswith(".manifest.json"):
            continue
        manifesto = k.checkpoint_manifesto(str(ck))
        if manifesto is None:
            continue
        achou = True
        m = json.loads(manifesto)
        valido = k.checkpoint_valido(str(ck))
        typer.echo(
            f"{'✓' if valido else '✗'} {ck.name}: sha256={m['hash_sha256'][:16]}… "
            f"tamanho={m['tamanho']}B algoritmo={m['algoritmo']} "
            f"versão={m['versao']}")
    if not achou:
        typer.echo("nenhum checkpoint com manifesto encontrado")


@app.command()
def suite() -> None:
    """Suíte consolidada de gates v0.10.0 (script 77)."""
    raise typer.Exit(_delega("suite"))


@app.command()
def telemetria() -> None:
    """Coleta telemetria do projeto v0.10.0 (script 76)."""
    raise typer.Exit(_delega("telemetria"))


@app.command()
def publicar(
    confirmar: bool = typer.Option(False, "--confirmar",
                                   help="Executa o upload único ao HF (script 74)."),
) -> None:
    """Publicação organizada no HF — exige --confirmar."""
    if not confirmar:
        typer.echo("publicação não confirmada: use --confirmar (upload único, "
                   "HF_TOKEN apenas via variável de ambiente)")
        raise typer.Exit(2)
    if not os.environ.get("HF_TOKEN"):
        typer.echo("HF_TOKEN ausente no ambiente — defina antes de publicar")
        raise typer.Exit(2)
    raise typer.Exit(_delega("publicar"))


@app.command()
def estados(
    apagar: bool = typer.Option(False, "--apagar",
                                help="Apaga os estados encontrados (regeneráveis)."),
) -> None:
    """Varre (e opcionalmente apaga) estados de treino do ambiente local."""
    achados: list[Path] = []
    for padrao in PADROES_ESTADO:
        achados.extend(RAIZ.rglob(padrao))
    achados = sorted(set(p for p in achados if "target" not in p.parts
                         and ".git" not in p.parts))
    if not achados:
        typer.echo("0 estados de treino no ambiente local")
        return
    total = 0
    for p in achados:
        tam = p.stat().st_size
        total += tam
        typer.echo(f"{'apagando' if apagar else 'achado'}: {p.relative_to(RAIZ)} "
                   f"({tam / 1e6:.1f} MB)")
        if apagar:
            p.unlink()
    typer.echo(f"total: {len(achados)} arquivos, {total / 1e6:.1f} MB"
               + (" — apagados" if apagar else " (use --apagar para remover)"))


if __name__ == "__main__":
    app()