Update app.py

app.py

@@ -1,73 +1,204 @@
+import time
+import re
 import gradio as gr
+from typing import Dict, Any, List, Tuple
 from transformers import AutoTokenizer, AutoModelForQuestionAnswering, AutoModelForCausalLM, pipeline
 
-# Modelli
+# ================= CONFIGURAZIONE =================
+
 MODEL_DEBERTA = "osiria/deberta-italian-question-answering"
 MODEL_GEPPETTO = "LorenzoDeMattei/GePpeTto"
 
-# Pipeline DeBERTa (estrattivo)
-tok_deb = AutoTokenizer.from_pretrained(MODEL_DEBERTA)
-mdl_deb = AutoModelForQuestionAnswering.from_pretrained(MODEL_DEBERTA)
-qa_deb = pipeline("question-answering", model=mdl_deb, tokenizer=tok_deb, device=-1)
+# Domande standard (chiave, domanda)
+DOMANDE_IT: List[Tuple[str, str]] = [
+    ("tipologia_fattura", "Qual è la tipologia della fattura? / Qual è la natura o il tipo del documento?"),
+    ("numero_fattura", "Qual è il numero della fattura? / Qual è il codice identificativo della fattura?"),
+    ("data_fattura", "Qual è la data della fattura? / Quando è stata emessa la fattura?"),
+    ("data_scadenza", "Qual è la data di scadenza del pagamento? / Entro quale termine deve essere saldata la fattura?"),
+    ("emittente", "Chi è l'emittente? / Qual è la ragione sociale dell'emittente?"),
+    ("indirizzo_emittente", "Qual è l'indirizzo dell'emittente? / Dove si trova la sede dell'emittente?"),
+    ("piva_emittente", "Qual è il codice fiscale o P.IVA dell'emittente? / Qual è la partita IVA dell'azienda?"),
+    ("email_emittente", "Qual è l'email dell'emittente? / Qual è l'indirizzo di posta elettronica dell'emittente?"),
+    ("telefono_emittente", "Qual è il numero di telefono dell'emittente? / Qual è il recapito telefonico dell'emittente?"),
+    ("beneficiario", "Chi è il beneficiario? / Qual è l'intestatario della fattura?"),
+    ("indirizzo_beneficiario", "Qual è l'indirizzo del beneficiario? / Dove risiede il beneficiario?"),
+    ("cf_beneficiario", "Qual è il codice fiscale del beneficiario? / Qual è l'identificativo fiscale del cliente?"),
+    ("email_beneficiario", "Qual è l'email del beneficiario? / Qual è l'indirizzo di posta elettronica del cliente?"),
+    ("descrizione_servizio", "Qual è la descrizione del prodotto/servizio? / Come viene denominato l’articolo o la prestazione?"),
+    ("quantita_servizio", "Qual è la quantità del prodotto/servizio? / Quante unità o ore sono state fatturate?"),
+    ("prezzo_unitario", "Qual è il prezzo unitario del prodotto/servizio? / Qual è il costo per singola unità o ora?"),
+    ("totale_servizio", "Qual è il totale del prodotto/servizio? / Qual è l’importo complessivo relativo all’articolo o servizio?"),
+    ("importo_imponibile", "Qual è l'importo imponibile? / Qual è il subtotale prima dell'IVA?"),
+    ("aliquota_iva", "Qual è l'aliquota IVA applicata? / Qual è la percentuale di IVA?"),
+    ("importo_iva", "Qual è l'importo IVA? / Qual è il valore dell'imposta applicata?"),
+    ("importo_totale", "Qual è l'importo totale del documento? / Qual è la somma complessiva da pagare?"),
+    ("condizioni_pagamento", "Quali sono le condizioni di pagamento? / Quali sono i termini di pagamento previsti?"),
+    ("modalita_pagamento", "Qual è la modalità di pagamento prevista? / Quale metodo di pagamento è indicato?"),
+    ("iban", "Qual è l'IBAN indicato? / Qual è il numero di conto per il bonifico?"),
+    ("bic_swift", "Qual è il BIC/SWIFT indicato? / Qual è il codice bancario internazionale?"),
+    ("intestatario_conto", "A chi è intestato il conto? / Chi è il titolare del conto bancario?"),
+    ("note", "Quali sono le note aggiuntive riportate nella fattura? / Quali osservazioni sono incluse nel documento?"),
+    ("firma_digitale", "Qual è la firma digitale? / Chi ha firmato digitalmente la fattura?"),
+    ("cig", "Qual è il CIG di riferimento? / Qual è il codice identificativo di gara?"),
+    ("stato_liquidazione", "Qual è lo stato di liquidazione? / Qual è la situazione del pagamento?"),
+    ("codice_destinatario", "Qual è il codice destinatario? / Qual è l'identificativo del destinatario?"),
+    ("causale", "Qual è la causale della fattura? / Qual è la motivazione o descrizione del pagamento?")
+]
 
-# Pipeline GePpeTto (generativo)
-tok_gepp = AutoTokenizer.from_pretrained(MODEL_GEPPETTO)
-mdl_gepp = AutoModelForCausalLM.from_pretrained(MODEL_GEPPETTO)
-qa_gepp = pipeline("text-generation", model=mdl_gepp, tokenizer=tok_gepp, device=-1)
+# ================= CACHE MODELLI =================
+LOADED: Dict[str, Any] = {}
 
-def ensemble_invoice_qa(md_text: str, question: str):
-    results = {}
+def get_deberta_pipeline():
+    if "deb" in LOADED: return LOADED["deb"]
+    tok = AutoTokenizer.from_pretrained(MODEL_DEBERTA)
+    mdl = AutoModelForQuestionAnswering.from_pretrained(MODEL_DEBERTA)
+    qa = pipeline("question-answering", model=mdl, tokenizer=tok, handle_impossible_answer=True, top_k=1, device=-1)
+    LOADED["deb"] = qa
+    return qa
 
-    # Estrattivo (DeBERTa)
-    try:
-        res_deb = qa_deb(question=question, context=md_text)
-        results["DeBERTa (estrattivo)"] = {
-            "risposta": res_deb["answer"].strip(),
-            "confidenza": round(res_deb["score"], 3)
+def get_geppetto_pipeline():
+    if "gepp" in LOADED: return LOADED["gepp"]
+    tok = AutoTokenizer.from_pretrained(MODEL_GEPPETTO)
+    mdl = AutoModelForCausalLM.from_pretrained(MODEL_GEPPETTO)
+    gen = pipeline("text-generation", model=mdl, tokenizer=tok, device=-1)
+    LOADED["gepp"] = gen
+    return gen
+
+# ================= UTILITY =================
+
+def preprocess_markdown(text: str) -> str:
+    if not text: return ""
+    text = re.sub(r'\|[\s-]+\|', ' ', text)   # ripulisce separatori tabella
+    text = text.replace('|', ' ')
+    text = text.replace('**', '').replace('##', '')
+    # mapping semantico leggero
+    text = text.replace('P.IVA', 'partita IVA').replace('PIVA', 'partita IVA')
+    text = re.sub(r'\s+', ' ', text).strip()
+    return text
+
+def chunk_text(text: str, max_chars: int = 3000, overlap: int = 200) -> List[str]:
+    if len(text) <= max_chars: return [text]
+    chunks = []
+    i = 0
+    while i < len(text):
+        end = min(i + max_chars, len(text))
+        chunks.append(text[i:end])
+        i = end - overlap
+        if i < 0: i = 0
+    return chunks
+
+# ================= LOGICA PRINCIPALE =================
+
+def analyze_invoice(md_text: str, custom_question_it: str):
+    logs: List[str] = []
+    final_output: Dict[str, Any] = {}
+
+    if len(md_text.strip()) < 10:
+        return {"Error": "Testo troppo breve"}, "⚠️ Inserisci almeno 10 caratteri."
+
+    clean_text = preprocess_markdown(md_text)
+    chunks = chunk_text(clean_text, max_chars=3000, overlap=200)
+    logs.append(f"📄 Testo originale: {len(md_text)} chars | Pulito: {len(clean_text)} chars | Chunks: {len(chunks)}")
+
+    qa_deb = get_deberta_pipeline()
+    gen_gepp = get_geppetto_pipeline()
+
+    # 1) DeBERTa: QA estrattivo su tutte le domande + opzionale
+    t_start_deb = time.time()
+    deb_res: Dict[str, Any] = {}
+    success_count = 0
+
+    def ask_all_chunks(question: str) -> Tuple[str, float]:
+        best_answer, best_score = "", 0.0
+        for c in chunks:
+            try:
+                r = qa_deb(question=question, context=c)
+                ans = r.get("answer", "").strip()
+                score = float(r.get("score", 0.0))
+                if score > best_score and ans:
+                    best_answer, best_score = ans, score
+            except Exception as e:
+                logs.append(f"❌ Errore QA chunk: {str(e)}")
+        return best_answer, best_score
+
+    for key, question_text in DOMANDE_IT:
+        answer, score = ask_all_chunks(question_text)
+        status = "Successo" if score > 0.05 and answer else "Non Trovato"
+        if status == "Successo": success_count += 1
+        deb_res[key] = {
+            "domanda": question_text,
+            "risposta": answer,
+            "confidenza": round(score, 3),
+            "status": status
         }
-    except Exception as e:
-        results["DeBERTa (estrattivo)"] = {"errore": str(e)}
 
-    # Generativo (GePpeTto)
-    try:
-        prompt = f"Domanda: {question}\nContesto: {md_text}\nRisposta:"
-        res_gepp = qa_gepp(prompt, max_new_tokens=64, do_sample=False)
-        results["GePpeTto (generativo)"] = {
-            "risposta": res_gepp[0]["generated_text"].replace(prompt, "").strip()
+    custom_q = custom_question_it.strip()
+    if custom_q:
+        answer, score = ask_all_chunks(custom_q)
+        status = "Successo" if score > 0.05 and answer else "Non Trovato"
+        if status == "Successo": success_count += 1
+        deb_res["domanda_opzionale"] = {
+            "domanda": custom_q,
+            "risposta": answer,
+            "confidenza": round(score, 3),
+            "status": status
         }
+
+    t_elapsed_deb = round(time.time() - t_start_deb, 2)
+    final_output["DeBERTa (estrattivo)"] = deb_res
+    logs.append(f"✅ DeBERTa completato in {t_elapsed_deb}s | Successi: {success_count}/{len(DOMANDE_IT) + (1 if custom_q else 0)}")
+
+    # 2) GePpeTto: generativo su tutte le domande in blocco
+    t_start_gepp = time.time()
+    try:
+        # Costruzione prompt conciso per ridurre rumore
+        prompt_lines = ["Rispondi in elenco puntato alle seguenti domande sulla fattura:"]
+        for _, q in DOMANDE_IT:
+            prompt_lines.append(f"- {q}")
+        if custom_q:
+            prompt_lines.append(f"- {custom_q}")
+        prompt_lines.append("\nContesto:")
+        prompt_lines.append(clean_text[:4000])  # taglio prudenziale su CPU
+        prompt_lines.append("\nRisposte (usa un punto per ogni domanda, senza inventare dati):")
+
+        prompt = "\n".join(prompt_lines)
+        gen = gen_gepp(prompt, max_new_tokens=256, do_sample=False)
+        generative_text = gen[0]["generated_text"].replace(prompt, "").strip()
+        final_output["GePpeTto (generativo)"] = {"risposte": generative_text}
+        t_elapsed_gepp = round(time.time() - t_start_gepp, 2)
+        logs.append(f"✅ GePpeTto completato in {t_elapsed_gepp}s")
     except Exception as e:
-        results["GePpeTto (generativo)"] = {"errore": str(e)}
+        final_output["GePpeTto (generativo)"] = {"errore": str(e)}
+        logs.append(f"❌ Errore GePpeTto: {e}")
 
-    return results
+    return final_output, "\n".join(logs)
 
-# ================== UI Gradio ==================
+# ================== UI GRADIO ==================
 with gr.Blocks(theme=gr.themes.Base()) as demo:
-    gr.Markdown("# 🧾 Invoice QA: Ensemble DeBERTa + GePpeTto")
-    gr.Markdown("Confronto tra risposte estrattive (DeBERTa) e generative (GePpeTto).")
+    gr.Markdown("# 🧾 Invoice QA: Domande standard + opzionale (DeBERTa estrattivo & GePpeTto generativo)")
+    gr.Markdown("Risposte estrattive strutturate per tutte le domande e un blocco generativo riassuntivo, con log e tempi.")
 
     with gr.Row():
         with gr.Column(scale=1):
             md_input = gr.Textbox(
-                label="Testo Fattura (.MD) / Contesto", 
-                lines=15, 
+                label="Testo Fattura (.MD) / Contesto",
+                lines=18,
                 placeholder="Incolla qui il contenuto Markdown/Testo della fattura..."
             )
-            
+
             custom_q_input = gr.Textbox(
-                label="Domanda (in Italiano)",
-                placeholder="Es: Qual è l'IBAN per il pagamento?"
+                label="Domanda opzionale (in Italiano)",
+                placeholder="Es: Qual è il riferimento d'ordine?"
             )
 
-            btn = gr.Button("🔍 Analizza Documento", variant="primary")
-        
+            btn = gr.Button("🔍 Analizza documento", variant="primary")
+
         with gr.Column(scale=1):
-            out_json = gr.JSON(label="Risultati Ensemble (Estrattivo vs Generativo)")
+            out_json = gr.JSON(label="Risultati estrattivi (DeBERTa) e generativi (GePpeTto)")
+            with gr.Accordion("📝 Log di Sistema (Tempi e Debug)", open=False):
+                out_log = gr.Textbox(label="Process Log", lines=12)
 
-    btn.click(
-        fn=ensemble_invoice_qa, 
-        inputs=[md_input, custom_q_input], 
-        outputs=[out_json]
-    )
+    btn.click(fn=analyze_invoice, inputs=[md_input, custom_q_input], outputs=[out_json, out_log])
 
 if __name__ == "__main__":
     demo.launch()