Spaces:
Sleeping
Sleeping
Update app.py
Browse files
app.py
CHANGED
@@ -1,194 +1,386 @@
|
|
1 |
import streamlit as st
|
2 |
import time
|
3 |
-
import
|
4 |
-
import
|
5 |
-
import
|
6 |
-
from PIL import Image
|
7 |
-
from together import Together
|
8 |
import os
|
9 |
-
|
10 |
-
from
|
11 |
-
|
12 |
-
|
13 |
-
|
14 |
-
|
15 |
-
|
16 |
-
|
17 |
-
|
18 |
-
|
19 |
-
|
20 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
21 |
)
|
22 |
|
23 |
-
|
24 |
-
|
25 |
-
|
26 |
-
|
27 |
-
|
28 |
-
|
29 |
-
|
30 |
-
|
31 |
-
|
32 |
-
"
|
33 |
-
|
34 |
-
|
35 |
-
|
36 |
-
|
37 |
-
|
38 |
-
|
39 |
-
|
40 |
-
"
|
41 |
-
|
42 |
-
|
43 |
-
|
44 |
-
|
45 |
-
|
46 |
-
"
|
47 |
-
|
48 |
-
"stile_immagine": (
|
49 |
-
"A vintage logo design prompt inspired by bygone eras. Emphasizes handcrafted details, worn textures, "
|
50 |
-
"and a nostalgic atmosphere, ideal for artisanal products or brands with a long-standing tradition."
|
51 |
-
)
|
52 |
-
},
|
53 |
-
"Geometric": {
|
54 |
-
"nome": "Geometric",
|
55 |
-
"stile_immagine": (
|
56 |
-
"A geometric logo design prompt that leverages simple, precise shapes, clean lines, and symmetry. "
|
57 |
-
"Communicates order, professionalism, and a rational approach to design."
|
58 |
-
)
|
59 |
-
},
|
60 |
-
"Typographic": {
|
61 |
-
"nome": "Typographic",
|
62 |
-
"stile_immagine": (
|
63 |
-
"A typographic logo design prompt focused on the creative use of lettering. "
|
64 |
-
"Bold typography paired with minimal color usage highlights the strength of word-based identities."
|
65 |
-
)
|
66 |
-
},
|
67 |
-
}
|
68 |
-
|
69 |
-
class Logo(BaseModel):
|
70 |
-
nome: str
|
71 |
-
descrizione: str
|
72 |
-
english_description:str
|
73 |
-
|
74 |
-
class Loghi(BaseModel):
|
75 |
-
loghi: list[Logo]
|
76 |
-
|
77 |
-
def generate_ai(num_loghi, tema, creativita):
|
78 |
-
prompt = (
|
79 |
-
f"Genera {num_loghi} prompt per la generazione immagini che trasmetta questo: {tema}"
|
80 |
-
"Sii molto SINTETICO e usa SIMBOLI stilizzati e non troppi oggetti. Restituisci il risultato in formato JSON seguendo lo schema fornito")
|
81 |
-
completion = clientOpenAI.beta.chat.completions.parse(
|
82 |
-
model=MODEL,
|
83 |
-
messages=[
|
84 |
-
{"role": "system", "content": f"Sei un assistente utile per la generazione di IDEE per la generazioni immagini su questo tema: {tema}."},
|
85 |
-
{"role": "user", "content": prompt},
|
86 |
-
],
|
87 |
-
temperature=creativita,
|
88 |
-
response_format=Loghi,
|
89 |
-
)
|
90 |
-
loghi = completion.choices[0].message.parsed
|
91 |
-
print(loghi)
|
92 |
-
return loghi
|
93 |
-
|
94 |
-
# Funzione per generare le immagini, con gestione errori e retry dopo 10 secondi
|
95 |
-
def generate_image(prompt, max_retries=5):
|
96 |
-
client = Together(api_key=api_together)
|
97 |
-
retries = 0
|
98 |
-
while retries < max_retries:
|
99 |
try:
|
100 |
-
|
101 |
-
|
102 |
-
|
103 |
-
width=1024,
|
104 |
-
height=1024,
|
105 |
-
steps=4,
|
106 |
-
n=1,
|
107 |
-
response_format="b64_json"
|
108 |
-
)
|
109 |
-
return response.data # Una lista di oggetti con attributo b64_json
|
110 |
except Exception as e:
|
111 |
-
print(f"Errore
|
112 |
-
|
113 |
-
|
114 |
-
|
115 |
-
|
116 |
-
|
117 |
-
|
118 |
-
|
119 |
-
|
120 |
-
|
121 |
-
|
122 |
-
|
123 |
-
|
124 |
-
|
125 |
-
|
126 |
-
|
127 |
-
|
128 |
-
|
129 |
-
|
130 |
-
|
131 |
-
|
132 |
-
|
133 |
-
|
134 |
-
|
135 |
-
|
136 |
-
|
137 |
-
|
138 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
139 |
else:
|
140 |
-
|
141 |
-
|
142 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
143 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
144 |
def main():
|
145 |
-
st.
|
146 |
-
st.
|
147 |
-
|
148 |
-
|
149 |
-
|
150 |
-
|
151 |
-
|
152 |
-
|
153 |
-
|
154 |
-
|
155 |
-
|
156 |
-
|
157 |
-
|
158 |
-
|
159 |
-
|
160 |
-
|
161 |
-
|
162 |
-
|
163 |
-
|
164 |
-
|
165 |
-
|
166 |
-
|
167 |
-
|
168 |
-
|
169 |
-
|
170 |
-
|
171 |
-
|
172 |
-
|
173 |
-
|
174 |
-
|
175 |
-
|
176 |
-
|
177 |
-
|
178 |
-
|
179 |
-
|
180 |
-
|
181 |
-
|
182 |
-
|
183 |
-
|
184 |
-
|
185 |
-
|
186 |
-
|
187 |
-
|
188 |
-
|
189 |
-
|
190 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
191 |
|
192 |
if __name__ == "__main__":
|
193 |
-
st.set_page_config(page_title="Logo Generator AI", page_icon="🎨", layout="wide")
|
194 |
main()
|
|
|
1 |
import streamlit as st
|
2 |
import time
|
3 |
+
import google.generativeai as genai
|
4 |
+
from pydantic import ValidationError
|
5 |
+
import mimetypes
|
|
|
|
|
6 |
import os
|
7 |
+
import settings_ai
|
8 |
+
from settings_ai import Documento, Articolo
|
9 |
+
import pandas as pd
|
10 |
+
from PyPDF2 import PdfReader, PdfWriter
|
11 |
+
import json
|
12 |
+
from azure.core.credentials import AzureKeyCredential
|
13 |
+
from azure.ai.documentintelligence import DocumentIntelligenceClient
|
14 |
+
from azure.ai.documentintelligence.models import AnalyzeDocumentRequest
|
15 |
+
from streamlit_pdf_viewer import pdf_viewer
|
16 |
+
import io
|
17 |
+
from PyPDF2 import PdfReader, PdfWriter
|
18 |
+
import fitz
|
19 |
+
import re
|
20 |
+
import io
|
21 |
+
from collections import Counter
|
22 |
+
|
23 |
+
GENERATION_CONFIG = settings_ai.GENERATION_CONFIG
|
24 |
+
SYSTEM_INSTRUCTION = settings_ai.SYSTEM_INSTRUCTION
|
25 |
+
USER_MESSAGE = settings_ai.USER_MESSAGE
|
26 |
+
API_KEY_GEMINI = settings_ai.API_KEY_GEMINI
|
27 |
+
|
28 |
+
# Configura il modello Gemini
|
29 |
+
genai.configure(api_key=API_KEY_GEMINI)
|
30 |
+
model = genai.GenerativeModel(
|
31 |
+
model_name="gemini-2.0-flash",
|
32 |
+
generation_config=GENERATION_CONFIG,
|
33 |
+
system_instruction=SYSTEM_INSTRUCTION
|
34 |
)
|
35 |
|
36 |
+
# Upload File a GEMINI
|
37 |
+
def upload_to_gemini(path: str, mime_type: str = None):
|
38 |
+
"""Carica un file su Gemini e ne ritorna l'oggetto file."""
|
39 |
+
file = genai.upload_file(path, mime_type=mime_type)
|
40 |
+
print(f"Uploaded file '{file.display_name}' as: {file.uri}")
|
41 |
+
return file
|
42 |
+
|
43 |
+
# Attesa Upload Files
|
44 |
+
def wait_for_files_active(files):
|
45 |
+
"""Attende che i file siano nello stato ACTIVE su Gemini."""
|
46 |
+
print("Waiting for file processing...")
|
47 |
+
for name in (f.name for f in files):
|
48 |
+
file_status = genai.get_file(name)
|
49 |
+
while file_status.state.name == "PROCESSING":
|
50 |
+
print(".", end="", flush=True)
|
51 |
+
time.sleep(10)
|
52 |
+
file_status = genai.get_file(name)
|
53 |
+
if file_status.state.name != "ACTIVE":
|
54 |
+
raise Exception(f"File {file_status.name} failed to process")
|
55 |
+
print("\n...all files ready")
|
56 |
+
|
57 |
+
# Chiamata API Gemini
|
58 |
+
def send_message_to_gemini(chat_session, message, max_attempts=3):
|
59 |
+
"""Tenta di inviare il messaggio tramite la chat_session, riprovando fino a max_attempts in caso di eccezioni, con un delay di 10 secondi tra i tentativi. """
|
60 |
+
for attempt in range(max_attempts):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
61 |
try:
|
62 |
+
print(f"Generazione AI con PROMPT: {message}")
|
63 |
+
response_local = chat_session.send_message(message)
|
64 |
+
return response_local
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
65 |
except Exception as e:
|
66 |
+
print(f"Errore in send_message (tentativo {attempt+1}/{max_attempts}): {e}")
|
67 |
+
if attempt < max_attempts - 1:
|
68 |
+
print("Riprovo tra 10 secondi...")
|
69 |
+
time.sleep(10)
|
70 |
+
raise RuntimeError(f"Invio messaggio fallito dopo {max_attempts} tentativi.")
|
71 |
+
|
72 |
+
# Unisce i rettangoli evidenziati (se codice e descrizione stanno su linee diverse viene mostrato un solo rettangolo evidenziato)
|
73 |
+
def merge_intervals(intervals):
|
74 |
+
"""Unisce gli intervalli sovrapposti. Gli intervalli sono tuple (y0, y1)."""
|
75 |
+
if not intervals:
|
76 |
+
return []
|
77 |
+
intervals.sort(key=lambda x: x[0])
|
78 |
+
merged = [intervals[0]]
|
79 |
+
for current in intervals[1:]:
|
80 |
+
last = merged[-1]
|
81 |
+
if current[0] <= last[1]:
|
82 |
+
merged[-1] = (last[0], max(last[1], current[1]))
|
83 |
+
else:
|
84 |
+
merged.append(current)
|
85 |
+
return merged
|
86 |
+
|
87 |
+
# Evidenzia le corrispondenze
|
88 |
+
def highlight_text_in_pdf(input_pdf_bytes, text_list):
|
89 |
+
"""Crea rettangoli rossi che evidenziano i testi trovati, unendo gli intervalli sovrapposti in un unico rettangolo. """
|
90 |
+
pdf_document = fitz.open(stream=input_pdf_bytes, filetype="pdf")
|
91 |
+
patterns = []
|
92 |
+
for text in text_list:
|
93 |
+
text_pattern = re.escape(text).replace(r'\ ', r'\s*')
|
94 |
+
patterns.append(text_pattern)
|
95 |
+
pattern = r'\b(' + '|'.join(patterns) + r')\b'
|
96 |
+
regex = re.compile(pattern, re.IGNORECASE)
|
97 |
+
highlight_color = (1, 0, 0) # rosso
|
98 |
+
for page in pdf_document:
|
99 |
+
page_text = page.get_text()
|
100 |
+
matches = list(regex.finditer(page_text))
|
101 |
+
intervals = []
|
102 |
+
for match in matches:
|
103 |
+
match_text = match.group(0)
|
104 |
+
text_instances = page.search_for(match_text)
|
105 |
+
for inst in text_instances:
|
106 |
+
text_height = inst.y1 - inst.y0
|
107 |
+
new_y0 = inst.y0 - 0.1 * text_height
|
108 |
+
new_y1 = inst.y1 + 0.1 * text_height
|
109 |
+
intervals.append((new_y0, new_y1))
|
110 |
+
merged_intervals = merge_intervals(intervals)
|
111 |
+
for y0, y1 in merged_intervals:
|
112 |
+
full_width_rect = fitz.Rect(page.rect.x0, y0, page.rect.x1, y1)
|
113 |
+
rect_annot = page.add_rect_annot(full_width_rect)
|
114 |
+
rect_annot.set_colors(stroke=highlight_color, fill=highlight_color)
|
115 |
+
rect_annot.set_opacity(0.15)
|
116 |
+
rect_annot.update()
|
117 |
+
output_stream = io.BytesIO()
|
118 |
+
pdf_document.save(output_stream)
|
119 |
+
pdf_document.close()
|
120 |
+
output_stream.seek(0)
|
121 |
+
return output_stream
|
122 |
+
|
123 |
+
# Formattazione Euro
|
124 |
+
def format_euro(amount):
|
125 |
+
formatted = f"{amount:,.2f}"
|
126 |
+
formatted = formatted.replace(",", "X").replace(".", ",").replace("X", ".")
|
127 |
+
return f"€ {formatted}"
|
128 |
+
|
129 |
+
# Testo da PDF
|
130 |
+
def pdf_to_text(path_file: str) -> str:
|
131 |
+
""" Estrae e concatena il testo da tutte le pagine del PDF. """
|
132 |
+
reader = PdfReader(path_file)
|
133 |
+
full_text = ""
|
134 |
+
for page in reader.pages:
|
135 |
+
page_text = page.extract_text() or ""
|
136 |
+
full_text += page_text + "\n"
|
137 |
+
return full_text
|
138 |
+
|
139 |
+
# Funzione che verifica se gli articoli sono corretti (facendo un partsing PDF to TEXT)
|
140 |
+
def verify_articles(file_path: str, chunk_document):
|
141 |
+
''' La funzione trasforma il PDF in TESTO e cerca se ogni articolo è presente (al netto degli spazi) '''
|
142 |
+
if not file_path.lower().endswith(".pdf"):
|
143 |
+
for articolo in chunk_document.Articoli:
|
144 |
+
articolo.Verificato = 2
|
145 |
+
return None
|
146 |
+
pdf_text = pdf_to_text(file_path)
|
147 |
+
if '□' in pdf_text:
|
148 |
+
for articolo in chunk_document.Articoli:
|
149 |
+
articolo.Verificato = 2
|
150 |
+
return None
|
151 |
+
for articolo in chunk_document.Articoli:
|
152 |
+
articolo.Verificato = 1 if articolo.CodiceArticolo and (articolo.CodiceArticolo in pdf_text) else 0
|
153 |
+
if articolo.Verificato == 0:
|
154 |
+
articolo.Verificato = 1 if articolo.CodiceArticolo and (articolo.CodiceArticolo.replace(" ", "") in pdf_text.replace(" ", "")) else 0
|
155 |
+
if not any(articolo.Verificato == 0 for articolo in chunk_document.Articoli):
|
156 |
+
return None
|
157 |
+
unverified_articles = [articolo for articolo in chunk_document.Articoli if articolo.Verificato == 0]
|
158 |
+
json_unverified_articles = json.dumps([articolo.model_dump() for articolo in unverified_articles])
|
159 |
+
return json_unverified_articles
|
160 |
+
|
161 |
+
# Funzione ausiliaria che elabora un file (o un chunk) inviandolo tramite send_message_to_gemini
|
162 |
+
def process_document_splitted(file_path: str, chunk_label: str, use_azure: bool = False) -> Documento:
|
163 |
+
""" Elabora il file (o il chunk) inviandolo a Gemini e processando la risposta:
|
164 |
+
- Determina il mime type.
|
165 |
+
- Effettua l'upload tramite upload_to_gemini e attende che il file sia attivo.
|
166 |
+
- Avvia una chat con il file caricato e invia il messaggio utente.
|
167 |
+
- Tenta fino a 3 volte di validare il JSON ottenuto, filtrando gli Articoli.
|
168 |
+
- Se presenti errori, RIPROCESSA il documento 3 volta passando il risultato precedente, In questo modo riesce a gestire gli errori in modo più preciso!
|
169 |
+
Ritorna l'istanza di Documento validata. """
|
170 |
+
mime_type, _ = mimetypes.guess_type(file_path)
|
171 |
+
if mime_type is None:
|
172 |
+
mime_type = "application/octet-stream"
|
173 |
+
if not use_azure:
|
174 |
+
files = [upload_to_gemini(file_path, mime_type=mime_type)]
|
175 |
+
wait_for_files_active(files)
|
176 |
+
chat_history = [{ "role": "user","parts": [files[0]]}]
|
177 |
+
chat_session = model.start_chat(history=chat_history)
|
178 |
+
max_validation_attempts = 3
|
179 |
+
max_number_reprocess = 3
|
180 |
+
chunk_document = None
|
181 |
+
|
182 |
+
for i in range(max_number_reprocess):
|
183 |
+
print(f"Reprocessamento {i+1} di {max_number_reprocess} per il chunk {chunk_label}")
|
184 |
+
response = None
|
185 |
+
for attempt in range(max_validation_attempts):
|
186 |
+
message = USER_MESSAGE
|
187 |
+
if i > 0:
|
188 |
+
message += f". Attenzione, RIPROVA perché i seguenti articoli sono da ESCLUDERE in quanto ERRATI! {json_unverified_articles}"
|
189 |
+
if not use_azure:
|
190 |
+
response = send_message_to_gemini(chat_session, message)
|
191 |
+
else:
|
192 |
+
chunk_document = analyze_invoice_azure(file_path)
|
193 |
+
try:
|
194 |
+
if not use_azure:
|
195 |
+
chunk_document = Documento.model_validate_json(response.text)
|
196 |
+
chunk_document.Articoli = [
|
197 |
+
art for art in chunk_document.Articoli
|
198 |
+
if art.CodiceArticolo.startswith(("AVE", "AV", "3V", "44"))
|
199 |
+
and art.TotaleNonIvato != 0
|
200 |
+
and art.CodiceArticolo not in ("AVE", "AV", "3V", "44")
|
201 |
+
]
|
202 |
+
break
|
203 |
+
except ValidationError as ve:
|
204 |
+
print(f"Errore di validazione {chunk_label} (tentativo {attempt+1}/{max_validation_attempts}): {ve}")
|
205 |
+
if attempt < max_validation_attempts - 1:
|
206 |
+
print("Riprovo tra 5 secondi...")
|
207 |
+
time.sleep(5)
|
208 |
+
else:
|
209 |
+
raise RuntimeError(f"Superato il numero massimo di tentativi di validazione {chunk_label}.")
|
210 |
+
|
211 |
+
json_unverified_articles = verify_articles(file_path, chunk_document)
|
212 |
+
if not json_unverified_articles:
|
213 |
+
return chunk_document
|
214 |
+
return chunk_document
|
215 |
+
|
216 |
+
# Funzione principale che elabora il documento
|
217 |
+
def process_document(path_file: str, number_pages_split: int, use_azure: bool = False) -> Documento:
|
218 |
+
""" Elabora il documento in base al tipo di file:
|
219 |
+
1. Se il file non è un PDF, lo tratta "così com'è" (ad esempio, come immagine) e ne processa il contenuto tramite process_document_splitted.
|
220 |
+
2. Se il file è un PDF e contiene più di 5 pagine, lo divide in chunk da 5 pagine.
|
221 |
+
Per ogni chunk viene effettuato l’upload, viene elaborato il JSON e validato.
|
222 |
+
I chunk successivi vengono aggregati nel Documento finale e, se il PDF ha più
|
223 |
+
di 5 pagine, al termine il campo TotaleMerce viene aggiornato con il valore riportato dall’ultimo chunk. """
|
224 |
+
mime_type, _ = mimetypes.guess_type(path_file)
|
225 |
+
if mime_type is None:
|
226 |
+
mime_type = "application/octet-stream"
|
227 |
+
if use_azure:
|
228 |
+
number_pages_split = 2
|
229 |
+
if not path_file.lower().endswith(".pdf"):
|
230 |
+
print("File non PDF: elaborazione come immagine.")
|
231 |
+
documento_finale = process_document_splitted(path_file, chunk_label="(immagine)", use_azure=use_azure)
|
232 |
+
return documento_finale
|
233 |
+
|
234 |
+
reader = PdfReader(path_file)
|
235 |
+
total_pages = len(reader.pages)
|
236 |
+
documento_finale = None
|
237 |
+
ultimo_totale_merce = None
|
238 |
+
|
239 |
+
for chunk_index in range(0, total_pages, number_pages_split):
|
240 |
+
writer = PdfWriter()
|
241 |
+
for page in reader.pages[chunk_index:chunk_index + number_pages_split]:
|
242 |
+
writer.add_page(page)
|
243 |
+
temp_filename = "temp_chunk_"
|
244 |
+
if use_azure:
|
245 |
+
temp_filename+="azure_"
|
246 |
+
temp_filename += f"{chunk_index // number_pages_split}.pdf"
|
247 |
+
with open(temp_filename, "wb") as temp_file:
|
248 |
+
writer.write(temp_file)
|
249 |
+
chunk_label = f"(chunk {chunk_index // number_pages_split})"
|
250 |
+
chunk_document = process_document_splitted(temp_filename, chunk_label=chunk_label, use_azure=use_azure)
|
251 |
+
if hasattr(chunk_document, "TotaleImponibile"):
|
252 |
+
ultimo_totale_merce = chunk_document.TotaleImponibile
|
253 |
+
if documento_finale is None:
|
254 |
+
documento_finale = chunk_document
|
255 |
else:
|
256 |
+
documento_finale.Articoli.extend(chunk_document.Articoli)
|
257 |
+
os.remove(temp_filename)
|
258 |
+
|
259 |
+
if total_pages > number_pages_split and ultimo_totale_merce is not None:
|
260 |
+
documento_finale.TotaleImponibile = ultimo_totale_merce
|
261 |
+
if documento_finale is None:
|
262 |
+
raise RuntimeError("Nessun documento elaborato.")
|
263 |
+
|
264 |
+
# Controlli aggiuntivi: Se esiste un AVE non possono esistere altri articoli non ave. Se articoli DOPPI segnalo!
|
265 |
+
if any(articolo.CodiceArticolo.startswith("AVE") for articolo in documento_finale.Articoli):
|
266 |
+
documento_finale.Articoli = [articolo for articolo in documento_finale.Articoli if articolo.CodiceArticolo.startswith("AVE")]
|
267 |
+
combinazioni = [(articolo.CodiceArticolo, articolo.TotaleNonIvato) for articolo in documento_finale.Articoli]
|
268 |
+
conta_combinazioni = Counter(combinazioni)
|
269 |
+
for articolo in documento_finale.Articoli:
|
270 |
+
if conta_combinazioni[(articolo.CodiceArticolo, articolo.TotaleNonIvato)] > 1:
|
271 |
+
articolo.Verificato = False
|
272 |
+
return documento_finale
|
273 |
|
274 |
+
# Analizza Fattura con AZURE
|
275 |
+
def analyze_invoice_azure(file_path: str):
|
276 |
+
"""Invia il file (dal percorso specificato) al servizio prebuilt-invoice e restituisce il risultato dell'analisi."""
|
277 |
+
# Apri il file in modalità binaria e leggi il contenuto
|
278 |
+
with open(file_path, "rb") as file:
|
279 |
+
file_data = file.read()
|
280 |
+
client = DocumentIntelligenceClient(endpoint=settings_ai.ENDPOINT_AZURE, credential=AzureKeyCredential(settings_ai.API_AZURE))
|
281 |
+
poller = client.begin_analyze_document("prebuilt-invoice", body=file_data)
|
282 |
+
result = poller.result()
|
283 |
+
return parse_invoice_to_documento_azure(result)
|
284 |
+
|
285 |
+
# Parsing Fattura con AZURE
|
286 |
+
def parse_invoice_to_documento_azure(result) -> Documento:
|
287 |
+
""" Parssa il risultato dell'analisi e mappa i campi rilevanti nel modello Pydantic Documento. """
|
288 |
+
if not result.documents:
|
289 |
+
raise ValueError("Nessun documento analizzato trovato.")
|
290 |
+
invoice = result.documents[0]
|
291 |
+
invoice_id_field = invoice.fields.get("InvoiceId")
|
292 |
+
numero_documento = invoice_id_field.value_string if invoice_id_field and invoice_id_field.value_string else ""
|
293 |
+
invoice_date_field = invoice.fields.get("InvoiceDate")
|
294 |
+
data_str = invoice_date_field.value_date.isoformat() if invoice_date_field and invoice_date_field.value_date else ""
|
295 |
+
subtotal_field = invoice.fields.get("SubTotal")
|
296 |
+
if subtotal_field and subtotal_field.value_currency:
|
297 |
+
totale_imponibile = subtotal_field.value_currency.amount
|
298 |
+
else:
|
299 |
+
invoice_total_field = invoice.fields.get("InvoiceTotal")
|
300 |
+
totale_imponibile = invoice_total_field.value_currency.amount if invoice_total_field and invoice_total_field.value_currency else 0.0
|
301 |
+
articoli = []
|
302 |
+
items_field = invoice.fields.get("Items")
|
303 |
+
if items_field and items_field.value_array:
|
304 |
+
for item in items_field.value_array:
|
305 |
+
product_code_field = item.value_object.get("ProductCode")
|
306 |
+
codice_articolo = product_code_field.value_string if product_code_field and product_code_field.value_string else ""
|
307 |
+
amount_field = item.value_object.get("Amount")
|
308 |
+
totale_non_ivato = amount_field.value_currency.amount if amount_field and amount_field.value_currency else 0.0
|
309 |
+
articolo = Articolo(
|
310 |
+
CodiceArticolo=codice_articolo,
|
311 |
+
TotaleNonIvato=totale_non_ivato,
|
312 |
+
Verificato=None
|
313 |
+
)
|
314 |
+
articoli.append(articolo)
|
315 |
+
|
316 |
+
documento = Documento(
|
317 |
+
TipoDocumento="Fattura",
|
318 |
+
NumeroDocumento=numero_documento,
|
319 |
+
Data=data_str,
|
320 |
+
TotaleImponibile=totale_imponibile,
|
321 |
+
Articoli=articoli
|
322 |
+
)
|
323 |
+
return documento
|
324 |
+
|
325 |
+
# Front-End con Streamlit
|
326 |
def main():
|
327 |
+
st.set_page_config(page_title="Import Fatture AI", page_icon="✨")
|
328 |
+
st.title("Import Fatture AI ✨")
|
329 |
+
st.sidebar.title("Caricamento File")
|
330 |
+
uploaded_files = st.sidebar.file_uploader("Seleziona uno o più PDF", type=["pdf", "jpg", "jpeg", "png"], accept_multiple_files=True)
|
331 |
+
model_ai = st.sidebar.selectbox("Modello", ['Gemini Flash 2.0', 'Azure Intelligence'])
|
332 |
+
use_azure = True if model_ai == 'Azure Intelligence' else False
|
333 |
+
number_pages_split = st.sidebar.slider('Split Pagine', 1, 30, 2)
|
334 |
+
if st.sidebar.button("Importa", type="primary", use_container_width=True):
|
335 |
+
if not uploaded_files:
|
336 |
+
st.warning("Nessun file caricato!")
|
337 |
+
else:
|
338 |
+
for uploaded_file in uploaded_files:
|
339 |
+
st.subheader(f"📄 {uploaded_file.name}")
|
340 |
+
file_path = uploaded_file.name
|
341 |
+
with open(file_path, "wb") as f:
|
342 |
+
f.write(uploaded_file.getbuffer())
|
343 |
+
|
344 |
+
with st.spinner(f"Elaborazione in corso"):
|
345 |
+
try:
|
346 |
+
doc = process_document(uploaded_file.name, number_pages_split, use_azure=use_azure)
|
347 |
+
totale_non_ivato_verificato = sum(articolo.TotaleNonIvato for articolo in doc.Articoli if articolo.Verificato == 1)
|
348 |
+
totale_non_ivato_non_verificato = sum(articolo.TotaleNonIvato for articolo in doc.Articoli if articolo.Verificato != 1)
|
349 |
+
totale_non_ivato = totale_non_ivato_verificato + totale_non_ivato_non_verificato
|
350 |
+
st.write(
|
351 |
+
f"- **Tipo**: {doc.TipoDocumento}\n"
|
352 |
+
f"- **Numero**: {doc.NumeroDocumento}\n"
|
353 |
+
f"- **Data**: {doc.Data}\n"
|
354 |
+
f"- **Articoli Compatibili**: {len(doc.Articoli)}\n"
|
355 |
+
f"- **Totale Documento**: {format_euro(doc.TotaleImponibile)}\n"
|
356 |
+
)
|
357 |
+
if totale_non_ivato_non_verificato > 0:
|
358 |
+
st.error(f"Totale Ave Non Verificato: {format_euro(totale_non_ivato_verificato)}")
|
359 |
+
elif totale_non_ivato != 0:
|
360 |
+
st.success(f"Totale Ave Verificato: {format_euro(totale_non_ivato_verificato)}")
|
361 |
+
df = pd.DataFrame([{k: v for k, v in Articolo.model_dump().items() if k != ""} for Articolo in doc.Articoli])
|
362 |
+
if 'Verificato' in df.columns:
|
363 |
+
df['Verificato'] = df['Verificato'].apply(lambda x: "✅" if x == 1 else "❌" if x == 0 else "❓" if x == 2 else x)
|
364 |
+
if totale_non_ivato > 0:
|
365 |
+
st.dataframe(df, use_container_width=True ,column_config={"TotaleNonIvato": st.column_config.NumberColumn("Totale non Ivato",format="€ %.2f")})
|
366 |
+
st.json(doc.model_dump(), expanded=False)
|
367 |
+
if totale_non_ivato == 0:
|
368 |
+
st.info(f"Non sono presenti articoli 'AVE'")
|
369 |
+
if uploaded_file and file_path.lower().endswith(".pdf"):
|
370 |
+
list_art = list_art = [articolo.CodiceArticolo for articolo in doc.Articoli] + [articolo.DescrizioneArticolo for articolo in doc.Articoli]
|
371 |
+
if list_art:
|
372 |
+
new_pdf = highlight_text_in_pdf(uploaded_file.getvalue(), list_art)
|
373 |
+
pdf_viewer(input=new_pdf.getvalue(), width=1200)
|
374 |
+
else:
|
375 |
+
pdf_viewer(input=uploaded_file.getvalue(), width=1200)
|
376 |
+
else:
|
377 |
+
st.image(file_path)
|
378 |
+
st.divider()
|
379 |
+
except Exception as e:
|
380 |
+
st.error(f"Errore durante l'elaborazione di {uploaded_file.name}: {e}")
|
381 |
+
finally:
|
382 |
+
if os.path.exists(file_path):
|
383 |
+
os.remove(file_path)
|
384 |
|
385 |
if __name__ == "__main__":
|
|
|
386 |
main()
|