Idra-Informatica/PecHub
1
0
Fork 0
mirror of https://github.com/idrainformatica/PecFlow.git synced 2026-06-16 12:45:42 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

vbox funzionanti

Browse Source
This commit is contained in:
Matteo Giustini
2026-03-19 11:41:10 +01:00
parent 538d6a6bec
commit b7f7c1f7c0
32 changed files with 6043 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files
worker/app/imap/sync.py
+212 -23
View File
@@ -1,8 +1,8 @@
"""
Logica di sincronizzazione messaggi IMAP – Fase 3 aggiornata.
Logica di sincronizzazione messaggi IMAP – Fase 3 aggiornata + Sent folder.
Responsabilità:
1. Fetch della lista UID > last_sync_uid
1. Fetch della lista UID > last_sync_uid (INBOX e Sent)
2. Download envelope + raw EML per ogni UID
3. Parsing completo EML tramite app.parsers (Fase 3):
- Classificazione tipo PEC (X-Ricevuta / X-TipoRicevuta)
@@ -13,8 +13,11 @@ Responsabilità:
6. Upload allegati su MinIO + inserimento in tabella attachments
7. State machine messaggi outbound (sent→accepted→delivered/anomaly)
tramite X-Riferimento-Message-ID
8. Aggiornamento last_sync_uid e last_sync_at sulla mailbox
8. Aggiornamento last_sync_uid / sent_last_sync_uid sulla mailbox
9. Pubblicazione evento Redis per notifica WebSocket
Nota: ogni cartella IMAP ha un namespace UID separato, quindi la chiave
di idempotenza è (mailbox_id, imap_uid, imap_folder).
"""
import email
@@ -40,6 +43,17 @@ from app.storage.minio_client import upload_attachment, upload_eml
logger = logging.getLogger(__name__)
settings = get_settings()
# Nomi comuni della cartella Sent nei provider PEC italiani (in ordine di priorità)
SENT_FOLDER_CANDIDATES = [
"Sent",
"Sent Items",
"Sent Messages",
"Inviati",
"INBOX.Sent",
"INBOX.Inviati",
"INBOX.Sent Items",
]
# ─── Helper legacy (mantenuti per backward compatibility con i test) ──────────
@@ -122,7 +136,30 @@ def _parse_eml(raw_bytes: bytes) -> dict:
}
# ─── Core sync function ───────────────────────────────────────────────────────
# ─── Sent folder discovery ────────────────────────────────────────────────────
async def _select_sent_folder(
imap_client: aioimaplib.IMAP4 | aioimaplib.IMAP4_SSL,
email_address: str,
) -> str | None:
"""
Prova i nomi comuni della cartella Sent finché uno funziona (SELECT OK).
Se trovata, il client è già selezionato su quella cartella.
Restituisce il nome della cartella trovata, o None.
"""
for candidate in SENT_FOLDER_CANDIDATES:
try:
status, _ = await imap_client.select(candidate)
if status == "OK":
logger.debug(f"[{email_address}] Cartella Sent trovata: {candidate!r}")
return candidate
except Exception:
continue
logger.info(f"[{email_address}] Nessuna cartella Sent trovata (tentativi: {SENT_FOLDER_CANDIDATES})")
return None
# ─── Core sync functions ──────────────────────────────────────────────────────
async def sync_new_messages(
imap_client: aioimaplib.IMAP4 | aioimaplib.IMAP4_SSL,
@@ -131,7 +168,9 @@ async def sync_new_messages(
redis_client: aioredis.Redis,
) -> int:
"""
Sincronizza i messaggi nuovi (UID > last_sync_uid) per la mailbox data.
Sincronizza i messaggi nuovi (UID > last_sync_uid) dalla cartella INBOX.
ATTENZIONE: il client IMAP deve essere già selezionato su INBOX.
Returns:
Numero di nuovi messaggi sincronizzati.
@@ -143,12 +182,12 @@ async def sync_new_messages(
try:
status, search_data = await imap_client.search("UID", search_range)
except Exception as e:
logger.warning(f"[{mailbox.email_address}] SEARCH fallito: {e}")
logger.warning(f"[{mailbox.email_address}] SEARCH INBOX fallito: {e}")
return 0
if status != "OK":
logger.warning(
f"[{mailbox.email_address}] SEARCH status={status} data={search_data}"
f"[{mailbox.email_address}] SEARCH INBOX status={status} data={search_data}"
)
return 0
@@ -162,7 +201,7 @@ async def sync_new_messages(
seq_numbers = seq_numbers[: settings.imap_max_fetch_per_cycle]
logger.info(
f"[{mailbox.email_address}] Trovati {len(seq_numbers)} messaggi nuovi da sincronizzare"
f"[{mailbox.email_address}] Trovati {len(seq_numbers)} messaggi nuovi in INBOX"
)
synced_count = 0
@@ -177,13 +216,16 @@ async def sync_new_messages(
mailbox=mailbox,
db=db,
redis_client=redis_client,
imap_folder="INBOX",
direction="inbound",
state="received",
)
if synced and uid and uid > max_uid_synced:
synced_count += 1
max_uid_synced = uid
except Exception as e:
logger.error(
f"[{mailbox.email_address}] Errore fetch seq {seq}: {e}",
f"[{mailbox.email_address}] Errore fetch INBOX seq {seq}: {e}",
exc_info=True,
)
@@ -197,6 +239,114 @@ async def sync_new_messages(
return synced_count
async def sync_sent_messages(
imap_client: aioimaplib.IMAP4 | aioimaplib.IMAP4_SSL,
mailbox: Mailbox,
db: AsyncSession,
redis_client: aioredis.Redis,
) -> int:
"""
Sincronizza i messaggi nuovi dalla cartella Sent (posta inviata) del server IMAP.
Salva i messaggi come direction='outbound', state='sent'.
Dopo la sync ri-seleziona INBOX per ripristinare il client allo stato normale.
Returns:
Numero di nuovi messaggi inviati sincronizzati.
"""
# Trova e seleziona la cartella Sent
sent_folder = await _select_sent_folder(imap_client, mailbox.email_address)
if not sent_folder:
# Assicurati di tornare in INBOX
try:
await imap_client.select("INBOX")
except Exception:
pass
return 0
# Siamo ora selezionati nella cartella Sent
last_uid = mailbox.sent_last_sync_uid or 0
search_range = f"{last_uid + 1}:*"
try:
status, search_data = await imap_client.search("UID", search_range)
except Exception as e:
logger.warning(f"[{mailbox.email_address}] SEARCH Sent fallito: {e}")
try:
await imap_client.select("INBOX")
except Exception:
pass
return 0
if status != "OK":
try:
await imap_client.select("INBOX")
except Exception:
pass
return 0
raw_seqs = b" ".join(
d if isinstance(d, bytes) else d.encode() for d in search_data
).decode("ascii", errors="ignore").split()
seq_numbers = [s for s in raw_seqs if s.isdigit()]
if not seq_numbers:
logger.debug(f"[{mailbox.email_address}] Nessun messaggio nuovo in {sent_folder!r}")
try:
await imap_client.select("INBOX")
except Exception:
pass
return 0
seq_numbers = seq_numbers[: settings.imap_max_fetch_per_cycle]
logger.info(
f"[{mailbox.email_address}] Trovati {len(seq_numbers)} messaggi nuovi in {sent_folder!r}"
)
synced_count = 0
max_uid_synced = last_uid
for seq in seq_numbers:
try:
uid, synced = await _fetch_and_save_message_by_seq(
imap_client=imap_client,
seq=seq,
last_uid=last_uid,
mailbox=mailbox,
db=db,
redis_client=redis_client,
imap_folder=sent_folder,
direction="outbound",
state="sent",
)
if synced and uid and uid > max_uid_synced:
synced_count += 1
max_uid_synced = uid
except Exception as e:
logger.error(
f"[{mailbox.email_address}] Errore fetch {sent_folder!r} seq {seq}: {e}",
exc_info=True,
)
# Aggiorna sent_last_sync_uid
if max_uid_synced > last_uid:
mailbox.sent_last_sync_uid = max_uid_synced
await db.flush()
await db.commit()
logger.info(
f"[{mailbox.email_address}] Sync Sent completata: {synced_count} messaggi nuovi"
)
# Ri-seleziona INBOX per tornare allo stato normale
try:
await imap_client.select("INBOX")
except Exception as e:
logger.warning(f"[{mailbox.email_address}] Re-SELECT INBOX dopo Sent sync: {e}")
return synced_count
async def _fetch_and_save_message_by_seq(
imap_client: aioimaplib.IMAP4 | aioimaplib.IMAP4_SSL,
seq: str,
@@ -204,10 +354,18 @@ async def _fetch_and_save_message_by_seq(
mailbox: Mailbox,
db: AsyncSession,
redis_client: aioredis.Redis,
imap_folder: str = "INBOX",
direction: str = "inbound",
state: str = "received",
) -> tuple[int | None, bool]:
"""
Fetcha un singolo messaggio per NUMERO DI SEQUENZA (non UID).
Args:
imap_folder: cartella IMAP corrente (per idempotenza e salvataggio)
direction: 'inbound' per INBOX, 'outbound' per Sent
state: 'received' per INBOX, 'sent' per Sent
Returns:
(uid, saved): UID del messaggio e True se salvato, False altrimenti.
"""
@@ -267,6 +425,9 @@ async def _fetch_and_save_message_by_seq(
mailbox=mailbox,
db=db,
redis_client=redis_client,
imap_folder=imap_folder,
direction=direction,
state=state,
)
@@ -279,11 +440,13 @@ async def _fetch_and_save_message(
) -> bool:
"""
Fetcha un singolo messaggio per UID (usato dal job sync_mailbox one-shot).
Sincronizza solo dalla cartella INBOX.
"""
existing = await db.execute(
select(Message.id).where(
Message.mailbox_id == mailbox.id,
Message.imap_uid == uid,
Message.imap_folder == "INBOX",
)
)
if existing.scalar_one_or_none():
@@ -319,6 +482,9 @@ async def _fetch_and_save_message(
mailbox=mailbox,
db=db,
redis_client=redis_client,
imap_folder="INBOX",
direction="inbound",
state="received",
)
@@ -331,26 +497,37 @@ async def _save_message(
mailbox: Mailbox,
db: AsyncSession,
redis_client: aioredis.Redis,
imap_folder: str = "INBOX",
direction: str = "inbound",
state: str = "received",
) -> bool:
"""
Salva un messaggio EML in DB e su MinIO.
Args:
imap_folder: cartella IMAP di provenienza ('INBOX', 'Sent', ecc.)
direction: 'inbound' per posta in arrivo, 'outbound' per posta inviata
state: stato iniziale del messaggio ('received' per inbound, 'sent' per outbound)
Fase 3 – aggiornato per:
- Idempotenza basata su (mailbox_id, imap_uid, imap_folder) – le UID sono
per-cartella in IMAP, quindi lo stesso UID può esistere in INBOX e Sent
- Parser completo (body, allegati, EML-in-EML)
- Classificazione precisa tipo PEC (tutti i provider)
- Salvataggio allegati su MinIO + tabella attachments
- State machine outbound: aggiorna stato messaggio originale alla ricezione ricevuta
- State machine outbound: solo per messaggi inbound (ricevute PEC)
- Collegamento parent_message_id via X-Riferimento-Message-ID
"""
# ── Idempotenza ───────────────────────────────────────────────────────────
# ── Idempotenza: chiave composta (mailbox_id + imap_uid + imap_folder) ────
existing = await db.execute(
select(Message.id).where(
Message.mailbox_id == mailbox.id,
Message.imap_uid == uid,
Message.imap_folder == imap_folder,
)
)
if existing.scalar_one_or_none():
logger.debug(f"[{mailbox.email_address}] UID {uid} già in DB, skip")
logger.debug(f"[{mailbox.email_address}] UID {uid} in {imap_folder!r} già in DB, skip")
return False
# ── Parsing completo EML ──────────────────────────────────────────────────
@@ -361,9 +538,10 @@ async def _save_message(
received_at = datetime.now(UTC)
# ── State machine: trova e aggiorna messaggio outbound ────────────────────
# Solo per messaggi inbound che sono ricevute PEC (non per posta inviata)
parent_message_id: uuid.UUID | None = None
if pec_class.is_receipt and pec_class.riferimento_message_id:
if direction == "inbound" and pec_class.is_receipt and pec_class.riferimento_message_id:
parent_message_id = await _apply_outbound_state_machine(
riferimento_message_id=pec_class.riferimento_message_id,
pec_type=pec_class.pec_type,
@@ -383,30 +561,36 @@ async def _save_message(
except Exception as e:
logger.error(f"[{mailbox.email_address}] Upload EML MinIO UID {uid}: {e}")
# ── Determina received_at / sent_at per messaggi outbound ─────────────────
# Per posta inviata: il campo "date" dell'EML è la data di invio
msg_received_at = received_at if direction == "inbound" else None
msg_sent_at = parsed.date if direction == "outbound" else parsed.date
# ── Salva messaggio in DB ─────────────────────────────────────────────────
message = Message(
id=uuid.uuid4(),
tenant_id=mailbox.tenant_id,
mailbox_id=mailbox.id,
imap_uid=uid,
imap_folder="INBOX",
direction="inbound",
state="received",
imap_folder=imap_folder,
direction=direction,
state=state,
pec_type=pec_class.pec_type,
subject=parsed.subject,
from_address=parsed.from_address,
to_addresses=parsed.to_addresses if parsed.to_addresses else None,
cc_addresses=parsed.cc_addresses if parsed.cc_addresses else None,
message_id_header=parsed.message_id,
sent_at=parsed.date,
received_at=received_at,
sent_at=msg_sent_at,
received_at=msg_received_at,
size_bytes=size_bytes,
body_text=parsed.body_text,
body_html=parsed.body_html,
has_attachments=parsed.has_attachments,
parent_message_id=parent_message_id,
raw_eml_path=eml_path,
is_read=False,
# Messaggi outbound (Sent) sono già stati letti dal mittente
is_read=(direction == "outbound"),
)
db.add(message)
await db.flush() # ottieni message.id prima di salvare gli allegati
@@ -429,6 +613,7 @@ async def _save_message(
"subject": message.subject or "",
"from_address": message.from_address or "",
"pec_type": message.pec_type,
"direction": direction,
"is_receipt": pec_class.is_receipt,
"received_at": received_at.isoformat(),
}
@@ -437,10 +622,9 @@ async def _save_message(
logger.warning(f"[{mailbox.email_address}] Redis publish UID {uid}: {e}")
logger.info(
f"[{mailbox.email_address}] Nuovo messaggio: UID={uid} "
f"pec_type={pec_class.pec_type!r} "
f"subject={message.subject!r} "
f"allegati={len(parsed.attachments)}"
f"[{mailbox.email_address}] Nuovo messaggio: UID={uid} folder={imap_folder!r} "
f"direction={direction!r} pec_type={pec_class.pec_type!r} "
f"subject={message.subject!r} allegati={len(parsed.attachments)}"
)
return True
@@ -506,6 +690,11 @@ async def _save_attachments(
db: sessione DB
"""
for att in attachments:
# Salta i file di sistema PEC (daticert.xml, postacert.eml, smime.p7s, ecc.)
# L'EML grezzo è già conservato su MinIO tramite upload_eml
if att.is_pec_system:
continue
storage_path: str | None = None
try: