rag-from-scratch/ollama/check_env.py

#!/usr/bin/env python3
"""
Verifica ambiente Ollama

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 ollama/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 ollama/README.md)
EMBED_MODELS = [
    "qwen3-embedding",
    "nomic-embed-text-v2-moe",
    "bge-m3",
    "nomic-embed-text",
    "mxbai-embed-large",
    "paraphrase-multilingual",
    "all-minilm",
]
EMBED_MODEL_PREFIXES = tuple(EMBED_MODELS)

OLLAMA_SERVE_HINT = "   → Avvia il servizio con: ollama serve"
RECOMMENDED_EMBED_MODEL = "qwen3-embedding:0.6b"
RECOMMENDED_LLM_MODEL = "qwen3.5:4b"


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 base.startswith(EMBED_MODEL_PREFIXES) or "embed" in base


def _parse_ollama_models(raw_output: str) -> list[str]:
    """Estrae i nomi modello dall'output di `ollama list`."""
    models: list[str] = []
    for idx, line in enumerate(raw_output.splitlines()):
        line = line.strip()
        if not line:
            continue
        # Prima riga: header tabellare ("NAME ...")
        if idx == 0 and line.lower().startswith("name"):
            continue
        model_name = line.split(maxsplit=1)[0]
        models.append(model_name)
    return models


sys.path.insert(0, str(Path(__file__).parent.parent))
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 _print_model_list(title: str, models: list[str]) -> None:
    """Stampa in modo uniforme una lista di modelli."""
    if not models:
        print(f"   {title}: nessuno")
        return
    print(f"   {title} ({len(models)}):")
    for model in models:
        print(f"   - {model}")

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(OLLAMA_SERVE_HINT)
            return None
        models = _parse_ollama_models(result.stdout)
        print("✅ ollama risponde correttamente")
        return models
    except FileNotFoundError:
        print("❌ ollama non trovato (FileNotFoundError)")
        return None
    except subprocess.TimeoutExpired:
        print("❌ ollama non risponde (timeout)")
        print(OLLAMA_SERVE_HINT)
        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_configured_model(
    configured_name: str | None,
    available: list[str],
    label: str,
) -> bool | None:
    """
    Se esiste un modello configurato, lo verifica e ritorna True/False.
    Se non è configurato, ritorna None (il chiamante userà il fallback).
    """
    if not configured_name:
        return None

    print(f"   modello configurato (config.py): {configured_name}")
    found = _match(configured_name, available)
    if found:
        print(f"✅ {label} disponibile: {found}")
        return True

    print(f"❌ {configured_name} non trovato in Ollama")
    print(f"   → ollama pull {configured_name}")
    return False


def check_embed_model(available: list[str]) -> bool:
    """Verifica che il modello di embedding configurato sia disponibile."""
    configured_check = _check_configured_model(CONFIGURED_EMBED, available, "embedding")
    if configured_check is not None:
        return configured_check

    # 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 {RECOMMENDED_EMBED_MODEL}")
    return False


def check_llm_model(available: list[str]) -> bool:
    """Verifica che il modello LLM configurato sia disponibile."""
    configured_check = _check_configured_model(CONFIGURED_LLM, available, "LLM")
    if configured_check is not None:
        return configured_check

    # fallback: config.py non leggibile
    first_llm = next((m for m in available if not _is_embed(m)), None)
    if first_llm:
        print(f"✅ modello LLM trovato: {first_llm}")
        return True
    print("❌ nessun modello LLM trovato")
    print(f"   → Consigliato per 8 GB RAM: ollama pull {RECOMMENDED_LLM_MODEL}")
    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("─── Verifica ambiente Ollama ─────────────────────────────────────────\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)
            _print_model_list(
                "modelli embedding rilevati",
                [m for m in available if _is_embed(m)],
            )
            _print_model_list(
                "modelli LLM rilevati",
                [m for m in available if not _is_embed(m)],
            )
            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")
    else:
        n_fail = sum(1 for r in results if not r)
        print(f"⚠️  {n_fail} problema/i rilevato/i — risolvi prima di procedere.")

    return 0 if all_ok else 1


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