import time import re import gradio as gr from typing import Dict, Any, List, Tuple from transformers import AutoTokenizer, AutoModelForQuestionAnswering, AutoModelForCausalLM, pipeline # ================= CONFIGURAZIONE ================= MODEL_DEBERTA = "osiria/deberta-italian-question-answering" MODEL_GEPPETTO = "LorenzoDeMattei/GePpeTto" # 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?") ] # ================= PIPELINES ================= 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) 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) # ================= UTILITY ================= def preprocess_markdown(text: str) -> str: if not text: return "" text = re.sub(r'\|[\s-]+\|', ' ', text) text = text.replace('|', ' ') text = text.replace('**', '').replace('##', '') text = re.sub(r'\s+', ' ', text).strip() return text # ================= FUNZIONE 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) # 1) DeBERTa: QA estrattivo su tutte le domande + opzionale t_start_deb = time.time() deb_res: Dict[str, Any] = {} success_count = 0 for key, question_text in DOMANDE_IT: try: res = qa_deb(question=question_text, context=clean_text) answer = res["answer"].strip() score = round(res.get("score", 0.0), 3) 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": score, "status": status } except Exception as e: deb_res[key] = {"status": f"Errore inferenza: {str(e)}"} custom_q = custom_question_it.strip() if custom_q: try: res = qa_deb(question=custom_q, context=clean_text) answer = res["answer"].strip() score = round(res.get("score", 0.0), 3) 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": score, "status": status } except Exception as e: deb_res["domanda_opzionale"] = {"status": f"Errore inferenza: {str(e)}"} 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: SOLO domanda opzionale if custom_q: try: t_start_gepp = time.time() short_context = clean_text[:800] # taglio prudenziale prompt = f"Domanda: {custom_q}\nContesto: {short_context}\nRisposta:" res_gepp = qa_gepp(prompt, max_new_tokens=64, do_sample=False) generative_text = res_gepp[0]["generated_text"].replace(prompt, "").strip() final_output["GePpeTto (generativo)"] = {"risposta_opzionale": 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: final_output["GePpeTto (generativo)"] = {"errore": str(e)} logs.append(f"❌ Errore GePpeTto: {e}") else: final_output["GePpeTto (generativo)"] = {"info": "Nessuna domanda opzionale fornita"} return final_output, "\n".join(logs) # ================== UI GRADIO ================== with gr.Blocks(theme=gr.themes.Base()) as demo: gr.Markdown("# 🧾 Invoice QA: Domande standard + opzionale") gr.Markdown("Risposte estrattive (DeBERTa) su tutte le domande e generative (GePpeTto) solo sulla domanda opzionale.") with gr.Row(): with gr.Column(scale=1): md_input = gr.Textbox( label="Testo Fattura (.MD) / Contesto", lines=18, placeholder="Incolla qui il contenuto Markdown/Testo della fattura..." ) custom_q_input = gr.Textbox( label="Domanda opzionale (in Italiano)", placeholder="Es: Qual è il riferimento d'ordine?" ) btn = gr.Button("🔍 Analizza documento", variant="primary") with gr.Column(scale=1): out_json = gr.JSON(label="Risultati (DeBERTa estrattivo + GePpeTto opzionale)") with gr.Accordion("📝 Log di Sistema (Tempi e Debug)", open=False): out_log = gr.Textbox(label="Process Log", lines=12) btn.click(fn=analyze_invoice, inputs=[md_input, custom_q_input], outputs=[out_json, out_log]) if __name__ == "__main__": demo.launch()