AURA-chatbot / textutils.py
Flavio Casadei Della Chiesa
versione aggiornata con ollama
ad14e53
import pandas as pd
import re
import PyPDF2
import unicodedata
class DocumentProcessor:
"""
Classe per elaborare il testo dei documenti:
- Rimuove righe inutili basandosi su ESCLUDI_RE.
- Unisce righe spezzate in paragrafi coerenti, mantenendo gli elenchi puntati al paragrafo precedente.
- Suddivide il testo in paragrafi ben formattati.
"""
# 🔹 Lista di espressioni regolari per rimuovere righe indesiderate
ESCLUDI_RE = [
r'Pagina\s+\d+\s+di\s+\d+', # "Pagina x di y"
r'^Foglio\s+\d+', # "Foglio 3"
r'^\s*$', # Righe vuote
r'^Codice\s+Documento:\s+\w+', # "Codice Documento: ABC123"
r'^Firma\s+Digitale', # "Firma Digitale"
]
@staticmethod
def normalizza_testo_avanzato(testo):
if pd.isna(testo):
return ""
testo = str(testo).strip()
testo = testo.replace("’", "'").replace("‘", "'") # Sostituire tipi diversi di apostrofi
testo = unicodedata.normalize("NFKD", testo) # Normalizza Unicode (es: accenti, simboli)
return " ".join(testo.split()) # Rimuove spazi multipli
@staticmethod
def spezza_in_frammenti(testo: str, numero_frammenti : int = 1) ->list :
if numero_frammenti <= 0:
raise ValueError("Il numero di frammenti deve essere un intero positivo.")
lunghezza_testo = len(testo)
if numero_frammenti > lunghezza_testo:
return [] # Restituisce una lista vuota se non è possibile dividere
lunghezza_frammento = lunghezza_testo // numero_frammenti # Divisione intera
resto = lunghezza_testo % numero_frammenti # Calcola il resto
frammenti = []
inizio = 0
for i in range(numero_frammenti):
fine = inizio + lunghezza_frammento + (1 if i < resto else 0) # Gestisce il resto
frammenti.append(testo[inizio:fine])
inizio = fine
return frammenti
def estrai_da_pdf(self, pdf_file_path) :
with open(pdf_file_path, "rb") as f:
reader = PyPDF2.PdfReader(f)
full_text = ""
for page in reader.pages:
page_text = page.extract_text() or ""
full_text += page_text
return full_text
def chunk_text_by_paragraph(self,text: str):
"""
Suddivide il testo in paragrafi basandosi su newline.
Mantiene uniti gli elenchi puntati e numerati con il paragrafo precedente.
"""
paragraphs = text.split("\n")
docs = []
for i, para in enumerate(paragraphs):
para = para.strip()
if para:
docs.append({"id": str(i), "text": para})
return docs
def scomponi_in_frammenti(self, testo:str, numero_frammenti: int = 1):
raise NotImplementedError("Questo metodo deve essere implementato nelle sottoclassi.")
def unify_lines(self, text):
"""
Metodo da implementare nelle sottoclassi per suddividere il testo in paragrafi.
"""
return "\n".join(self.dividi_in_paragrafi(text))
def dividi_in_paragrafi(self,pdf_text: str) :
"""
Unisce righe spezzate in paragrafi coerenti, evitando di separare gli elenchi puntati dal paragrafo precedente.
Filtra le righe che corrispondono a qualsiasi espressione regolare contenuta in ESCLUDI_RE.
"""
lines = pdf_text.splitlines()
paragraphs = []
current_line = ""
end_punctuations = (".", "?", "!", ":", ";")
inside_list = False # 🟢 Indica se stiamo dentro un elenco puntato
for line in lines:
line = line.strip()
# 🛑 Rimuove le righe che corrispondono a una delle regex in ESCLUDI_RE
if any(re.search(pattern, line, re.IGNORECASE) for pattern in DocumentProcessor.ESCLUDI_RE):
continue # ❌ Salta la riga
# 🟢 Riconosce un elemento di un elenco puntato (es. "- testo", "• testo", "1. testo")
is_list_item = re.match(r"^(\d+\.\s+|[-•*]\s+).+", line)
# 🟢 Se la riga è vuota e abbiamo testo nel buffer, chiudiamo il paragrafo
if not line:
if current_line:
paragraphs.append(current_line.strip())
current_line = ""
inside_list = False # 🛑 Reset della modalità elenco
continue
# 🟢 Se è un elemento di elenco, lo aggiungiamo direttamente al paragrafo precedente
if is_list_item:
inside_list = True # 🟢 Indichiamo che siamo dentro un elenco
current_line += " " + line # 🔄 Aggiunge la riga all'elenco senza creare un nuovo paragrafo
elif inside_list:
# 🛑 Se eravamo dentro un elenco e ora la riga NON fa parte dell'elenco
current_line += " " + line # ✅ Manteniamo tutto unito
inside_list = False # 🔄 Reset della modalità elenco
elif line.endswith(end_punctuations):
current_line += " " + line
paragraphs.append(current_line.strip())
current_line = ""
else:
current_line += " " + line # ✅ Unisce righe normali
if current_line.strip():
paragraphs.append(current_line.strip())
#i = 0
#for par in paragraphs:
# print(f"Paragrafo {i} - {par}")
# i = i+1
return paragraphs
class ParagraphDocumentProcessor(DocumentProcessor):
def scomponi_in_frammenti(self, testo:str, numero_frammenti: int = 1):
return self.dividi_in_paragrafi(testo)
class WholeTextDocumentProcessor(DocumentProcessor) :
def scomponi_in_frammenti(self, testo:str, numero_frammenti: int = 1):
return [testo]
class SmallFragmentDocumentProcessor(DocumentProcessor):
def scomponi_in_frammenti(self, testo:str, numero_frammenti: int = 1):
return self.dividi_testo_in_frammenti(testo)
def dividi_testo_in_frammenti(self,testo, lunghezza_massima=1000):
frammenti = []
inizio = 0
while inizio < len(testo):
fine = inizio + lunghezza_massima
# Se siamo alla fine del testo, aggiungiamo e usciamo
if fine >= len(testo):
frammenti.append(testo[inizio:].strip())
break
# Cerca l'ultimo spazio prima del limite per evitare di tagliare la parola
fine_corretto = testo.rfind(" ", inizio, fine)
if fine_corretto == -1 or fine_corretto <= inizio:
# Se non troviamo spazi, tagliamo brutalmente
fine_corretto = fine
frammento = testo[inizio:fine_corretto].strip()
frammenti.append(frammento)
inizio = fine_corretto
return frammenti