rag-from-scratch/step-7/check_env.py

#!/usr/bin/env python3
"""
Step 7 — Verifica ambiente

Controlla che tutti i prerequisiti per la vettorizzazione siano soddisfatti:
  1. ollama è nel PATH e risponde
  2. Almeno un modello di embedding è scaricato
  3. Almeno un modello LLM è scaricato
  4. chromadb è importabile

Output: report a schermo con ✅ / ❌ per ogni componente.
Nessun file scritto. Exit 0 se tutto OK, 1 altrimenti.

Uso:
    python step-7/check_env.py
"""

import shutil
import subprocess
import sys
from pathlib import Path


# ─── Lista canonica di modelli embedding supportati ───────────────────────────
# Ordine: prima scelta → ultima scelta (come da README step-7)
EMBED_MODELS = [
    "qwen3-embedding",
    "nomic-embed-text-v2-moe",
    "bge-m3",
    "nomic-embed-text",
    "mxbai-embed-large",
    "paraphrase-multilingual",
    "all-minilm",
]


def _is_embed(model_name: str) -> bool:
    """True se il modello è riconosciuto come embedding (lista canonica o keyword)."""
    base = model_name.split(":")[0].lower()
    return any(base == e or base.startswith(e) for e in EMBED_MODELS) or "embed" in base


# ─── Modelli configurati in step-9/config.py ─────────────────────────────────
# Per spostare config.py alla root: cambia solo la riga qui sotto.
sys.path.insert(0, str(Path(__file__).parent.parent / "step-9"))
try:
    from config import EMBED_MODEL as CONFIGURED_EMBED, OLLAMA_MODEL as CONFIGURED_LLM
except Exception:
    CONFIGURED_EMBED = None
    CONFIGURED_LLM = None

REQUIRED_LIBS = ["chromadb"]


# ─── Checks ───────────────────────────────────────────────────────────────────

def check_ollama_in_path() -> bool:
    """Verifica che ollama sia nel PATH."""
    found = shutil.which("ollama") is not None
    if found:
        print("✅ ollama trovato nel PATH")
    else:
        print("❌ ollama non trovato nel PATH")
        print("   → Installa con: curl -fsSL https://ollama.com/install.sh | sh")
    return found


def check_ollama_running() -> list[str] | None:
    """
    Esegue 'ollama list' e ritorna la lista dei modelli disponibili.
    Ritorna None se ollama non risponde.
    """
    try:
        result = subprocess.run(
            ["ollama", "list"],
            capture_output=True, text=True, timeout=10
        )
        if result.returncode != 0:
            print("❌ ollama non risponde (errore all'avvio)")
            print("   → Avvia il servizio con: ollama serve")
            return None
        lines = result.stdout.strip().splitlines()
        models = []
        for line in lines[1:]:  # salta l'header
            parts = line.split()
            if parts:
                models.append(parts[0])
        print("✅ ollama risponde correttamente")
        return models
    except FileNotFoundError:
        print("❌ ollama non trovato (FileNotFoundError)")
        return None
    except subprocess.TimeoutExpired:
        print("❌ ollama non risponde (timeout)")
        print("   → Avvia il servizio con: ollama serve")
        return None


def _match(model_name: str, available: list[str]) -> str | None:
    """
    Ritorna il nome completo del modello trovato in 'available' che corrisponde
    a 'model_name' (confronto per prefisso), oppure None.
    """
    for m in available:
        if m == model_name or m.startswith(model_name + ":") or m.startswith(model_name + "-"):
            return m
    return None


def check_embed_model(available: list[str]) -> bool:
    """Verifica che il modello di embedding configurato sia disponibile."""
    if CONFIGURED_EMBED:
        print(f"   modello configurato (step-9/config.py): {CONFIGURED_EMBED}")
        found = _match(CONFIGURED_EMBED, available)
        if found:
            print(f"✅ embedding disponibile: {found}")
            return True
        print(f"❌ {CONFIGURED_EMBED} non trovato in Ollama")
        print(f"   → ollama pull {CONFIGURED_EMBED}")
        return False
    # fallback: config.py non leggibile
    found = next((m for m in available if _is_embed(m)), None)
    if found:
        print(f"✅ modello embedding trovato: {found}")
        return True
    print("❌ nessun modello di embedding trovato")
    print(f"   → Prima scelta: ollama pull qwen3-embedding:0.6b")
    return False


def check_llm_model(available: list[str]) -> bool:
    """Verifica che il modello LLM configurato sia disponibile."""
    if CONFIGURED_LLM:
        print(f"   modello configurato (step-9/config.py): {CONFIGURED_LLM}")
        found = _match(CONFIGURED_LLM, available)
        if found:
            print(f"✅ LLM disponibile: {found}")
            return True
        print(f"❌ {CONFIGURED_LLM} non trovato in Ollama")
        print(f"   → ollama pull {CONFIGURED_LLM}")
        return False
    # fallback: config.py non leggibile
    llm_candidates = [m for m in available if not _is_embed(m)]
    if llm_candidates:
        print(f"✅ modello LLM trovato: {llm_candidates[0]}")
        return True
    print("❌ nessun modello LLM trovato")
    print(f"   → Consigliato per 8 GB RAM: ollama pull qwen3.5:4b")
    return False


def check_library(lib: str) -> bool:
    """Verifica che una libreria Python sia importabile."""
    try:
        __import__(lib)
        print(f"✅ {lib} importabile")
        return True
    except ImportError:
        print(f"❌ {lib} non importabile")
        print(f"   → Installa con: pip install {lib}")
        return False


# ─── Entry point ──────────────────────────────────────────────────────────────

def main() -> int:
    print("─── Step 7 — Verifica ambiente ───────────────────────────────────────\n")

    results: list[bool] = []

    # 1. ollama nel PATH
    in_path = check_ollama_in_path()
    results.append(in_path)

    # 2. ollama risponde + modelli
    if in_path:
        available = check_ollama_running()
        if available is None:
            results.extend([False, False, False])
        else:
            results.append(True)
            results.append(check_embed_model(available))
            results.append(check_llm_model(available))
    else:
        results.extend([False, False, False])
        print("⚠️  modelli non verificabili (ollama non trovato)")

    # 3. Librerie Python
    print()
    for lib in REQUIRED_LIBS:
        results.append(check_library(lib))

    # ── Riepilogo ─────────────────────────────────────────────────────────────
    print()
    print("──────────────────────────────────────────────────────────────────────")
    all_ok = all(results)
    if all_ok:
        print("✅ Ambiente pronto — procedi con la vettorizzazione:")
        print("   python step-8/ingest.py --stem <nome>")
    else:
        n_fail = sum(1 for r in results if not r)
        print(f"⚠️  {n_fail} problema/i rilevato/i — risolvi prima di procedere con step-8.")

    return 0 if all_ok else 1


if __name__ == "__main__":
    sys.exit(main())