app_gradio.py

import hashlib
import json
import re
from pathlib import Path
from typing import Dict, List

import gradio as gr
import numpy as np

try:
    from sentence_transformers import SentenceTransformer
except Exception:  # pragma: no cover
    SentenceTransformer = None


STOPWORDS = {
    "della",
    "delle",
    "dello",
    "degli",
    "dati",
    "sono",
    "come",
    "questa",
    "questo",
    "nella",
    "nelle",
    "anche",
    "molto",
    "dove",
    "quando",
    "with",
    "that",
    "from",
    "have",
    "your",
    "will",
    "about",
    "parlami",
    "dimmi",
    "spiegami",
    "cosa",
    "quale",
}


def normalizza_testo(t: str) -> str:
    t = (t or "").replace("\n", " ").replace("\t", " ").strip()
    return re.sub(r"\s+", " ", t)


def tokenizza(testo: str) -> List[str]:
    candidati = re.findall(r"[A-Za-z0-9_]+", testo.lower())
    return [t for t in candidati if len(t) >= 4 and t not in STOPWORDS]


class LocalHashEmbedder:
    def __init__(self, dim: int = 384):
        self.dim = int(dim)
        self.name = f"local-hash-{self.dim}"

    def _tokens(self, text: str) -> List[str]:
        candidati = re.findall(r"[A-Za-z0-9_]+", text.lower())
        return [t for t in candidati if len(t) >= 3 and t not in STOPWORDS]

    def encode(self, texts):
        if isinstance(texts, str):
            texts = [texts]
        out = np.zeros((len(texts), self.dim), dtype="float32")
        for i, text in enumerate(texts):
            clean = normalizza_testo(text)
            toks = self._tokens(clean) or clean.lower().split()
            for tok in toks:
                h = int(hashlib.sha1(tok.encode("utf-8", errors="ignore")).hexdigest(), 16)
                idx = h % self.dim
                sign = -1.0 if ((h >> 8) & 1) else 1.0
                out[i, idx] += sign
            norm = float(np.linalg.norm(out[i]))
            if norm > 0:
                out[i] /= norm
        return out


def inizializza_embedder():
    model_name = "paraphrase-multilingual-MiniLM-L12-v2"
    local_path = Path("aio_models") / model_name
    if SentenceTransformer is not None and local_path.exists():
        try:
            model = SentenceTransformer(str(local_path), local_files_only=True)
            return model, f"sentence-transformers(local-path): {local_path}"
        except Exception:
            pass
    return LocalHashEmbedder(dim=384), "local-hash-embedder (showcase fallback)"


def vectorizza(model, testi: List[str]) -> np.ndarray:
    v = model.encode(testi)
    arr = np.array(v, dtype="float32")
    norms = np.linalg.norm(arr, axis=1, keepdims=True)
    norms[norms == 0] = 1.0
    return arr / norms


def carica_corpus() -> List[Dict[str, str]]:
    here = Path(__file__).resolve().parent
    path = here / "demo_corpus.json"
    data = json.loads(path.read_text(encoding="utf-8"))
    for r in data:
        r["text"] = normalizza_testo(r.get("text", ""))
    return data


def prepara_engine():
    model, backend = inizializza_embedder()
    records = carica_corpus()
    texts = [r["text"] for r in records]
    mat = vectorizza(model, texts)
    return model, backend, records, mat


def cerca(query: str, model, records, mat: np.ndarray, top_k: int = 5):
    qv = vectorizza(model, [query])[0]
    sims = mat @ qv
    qtok = set(tokenizza(query))
    out = []
    for i, s in enumerate(sims):
        rec = records[i]
        ttok = set(tokenizza(rec["text"]))
        overlap = len(qtok.intersection(ttok)) if qtok else 0
        lex = (overlap / max(1, len(qtok))) if qtok else 0.0
        score = 0.6 * float(s) + 0.4 * float(lex)
        out.append(
            {
                "score": score,
                "sim": float(s),
                "lex": float(lex),
                "domain": rec.get("domain", "generale"),
                "source": rec.get("source", "showcase"),
                "text": rec.get("text", ""),
            }
        )
    out.sort(key=lambda x: x["score"], reverse=True)
    return out[:top_k]


MODEL, BACKEND, RECORDS, MAT = prepara_engine()


def esegui_ricerca(query: str, top_k: int):
    query = normalizza_testo(query)
    if not query:
        return "Inserisci una domanda.", f"Backend: {BACKEND} | Records demo: {len(RECORDS)}"

    risultati = cerca(query, MODEL, RECORDS, MAT, top_k=int(top_k))
    lines = []
    for i, r in enumerate(risultati, start=1):
        lines.append(
            f"### {i}. [{r['domain']}] score={r['score']:.3f} sim={r['sim']:.3f} lex={r['lex']:.3f}\n"
            f"source: `{r['source']}`\n\n{r['text']}\n"
        )
    lines.append(
        "\n---\nLicenza Showcase: uso e studio liberi, uso commerciale solo su autorizzazione scritta. "
        "Contatto: `info@rthitalia.com`"
    )
    return "\n".join(lines), f"Backend: {BACKEND} | Records demo: {len(RECORDS)}"


with gr.Blocks(title="AIO System Core - Public Showcase") as demo:
    gr.Markdown("# AIO System Core - Public Showcase")
    gr.Markdown(
        "Demo pubblica controllata. Core proprietario e corpus completo restano privati."
    )
    stato = gr.Markdown(f"Backend: {BACKEND} | Records demo: {len(RECORDS)}")

    with gr.Row():
        query = gr.Textbox(
            label="Inserisci una domanda",
            value="Parlami del Mediterraneo e della geopolitica energetica",
            lines=3,
        )
    top_k = gr.Slider(label="Top K", minimum=3, maximum=10, value=5, step=1)
    run = gr.Button("Esegui ricerca", variant="primary")
    output = gr.Markdown()

    run.click(esegui_ricerca, inputs=[query, top_k], outputs=[output, stato])


if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)