modules/external_gpu.py

# type: ignore
"""
modules/external_gpu.py
Backend GPU externo (Kaggle T4 4-bit / Lightning / RunPod) — fallback
quando o ZeroGPU esgota a quota.
- A Space chama EXTERNAL_GPU_URL/generate (ver kaggle_backend.py).
- Sem quota ZeroGPU não há CUDA local: 4-bit SÓ faz sentido numa GPU
  externa real (T4 16GB aguenta Lexi 8B em 4-bit ≈ 5.5GB).
"""
import os
from typing import Optional, List, Dict

try:
    from loguru import logger  # type: ignore
except Exception:
    class _Dummy:
        def info(self, *a, **k): pass
        def success(self, *a, **k): pass
        def warning(self, *a, **k): pass
        def error(self, *a, **k): pass
    logger = _Dummy()  # type: ignore


_last_status: str = "nunca tentada"


def get_external_gpu_url() -> str:
    return (os.getenv("EXTERNAL_GPU_URL") or "").strip().rstrip("/")


def is_external_gpu_configured() -> bool:
    return bool(get_external_gpu_url())


def get_last_status() -> str:
    """Último resultado (para mostrar no chat/status)."""
    return _last_status


def _set_status(msg: str) -> None:
    global _last_status
    _last_status = msg


def generate_external_gpu(
    prompt: str,
    system_prompt: Optional[str] = None,
    context_history: Optional[List[Dict]] = None,
    max_tokens: int = 512,
    timeout: int = 120,
) -> Optional[str]:
    """POST {url}/generate → texto ou None (qualquer falha = None, sem exceção)."""
    url = get_external_gpu_url()
    if not url:
        _set_status("off (sem EXTERNAL_GPU_URL nos Secrets)")
        return None
    try:
        import requests  # type: ignore
        # 1) probe rápido: túnel morto (Kaggle parado/URL rodada) falha aqui
        # em vez de esperar o timeout cheio do /generate.
        try:
            h = requests.get(f"{url}/health",
                             headers={"ngrok-skip-browser-warning": "1"},
                             timeout=15)
            if h.status_code != 200:
                _set_status("túnel morto? (reinicia o Kaggle e atualiza EXTERNAL_GPU_URL)")
        except Exception:
            pass
        r = requests.post(
            f"{url}/generate",
            json={
                "prompt": prompt,
                "system_prompt": system_prompt or "",
                "history": context_history or [],
                "max_tokens": max_tokens,
            },
            headers={"ngrok-skip-browser-warning": "1"},
            timeout=timeout,
        )
        if r.status_code == 200:
            text = (r.json().get("text") or "").strip()
            if text:
                logger.success("[EXT-GPU] Resposta via backend externo")
                _set_status("OK (última chamada com sucesso)")
                return text
            _set_status("falhou (200 vazio)")
            logger.warning(f"[EXT-GPU] 200 vazio: {r.text[:200]}")
            return None
        ctype = (r.headers.get("content-type") or "")
        if "html" in ctype or r.text.lstrip().lower().startswith(("<!doctype", "<html")):
            _set_status("túnel morto (reinicia o Kaggle e atualiza EXTERNAL_GPU_URL)")
            logger.warning("[EXT-GPU] túnel ngrok morto (HTML em vez de JSON)")
            return None
        try:
            detail = r.json().get("error", "") or r.json().get("trace", "")
        except Exception:
            detail = r.text
        _set_status(f"falhou (HTTP {r.status_code}: {str(detail)[:120]})")
        logger.warning(f"[EXT-GPU] HTTP {r.status_code}: {str(detail)[:300]}")
        return None
    except Exception as e:
        _set_status(f"falhou (inalcançável: {str(e)[:120]})")
        logger.warning(f"[EXT-GPU] indisponível ({e})")
        return None