alex/ArchiveMail
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ref code

Browse Source
This commit is contained in:
alex
2026-01-11 14:57:32 +01:00
parent 509e557453
commit c175ca6fe2
1 changed files with 96 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
ArchiviaMailGui.py
View File

@@ -1,13 +1,30 @@
import win32com.client
import os
import hashlib
import logging
from datetime import datetime, timedelta
from typing import Dict
from enum import IntEnum
import time
from nicegui import ui, run
# --- CONFIGURAZIONE LOGGING ---
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# --- COSTANTI OUTLOOK ---
class OutlookConstants(IntEnum):
"""Costanti Outlook per migliorare leggibilità"""
FOLDER_INBOX = 6
MAIL_CLASS = 43
ATTACHMENT_TYPE_FILE = 1
# --- LOGICA DI BACKEND ---
def get_file_hash(file_path):
def get_file_hash(file_path: str) -> str:
"""Calcola l'MD5 per identificare file identici ed evitare duplicati."""
hasher = hashlib.md5()
with open(file_path, 'rb') as f:
@@ -16,16 +33,21 @@ def get_file_hash(file_path):
return hasher.hexdigest()
class ArchiverGUI:
def __init__(self):
self.running = False
self.progress = 0.0
self.current_mail = "In attesa di avvio..."
self.mesi_it = {
1: "Gennaio", 2: "Febbraio", 3: "Marzo", 4: "Aprile", 5: "Maggio", 6: "Giugno",
7: "Luglio", 8: "Agosto", 9: "Settembre", 10: "Ottobre", 11: "Novembre", 12: "Dicembre"
"""Gestisce lo stato e la logica di archiviazione delle email Outlook."""
MONTH_NAMES = {
1: "Gennaio", 2: "Febbraio", 3: "Marzo", 4: "Aprile",
5: "Maggio", 6: "Giugno", 7: "Luglio", 8: "Agosto",
9: "Settembre", 10: "Ottobre", 11: "Novembre", 12: "Dicembre"
}
async def start_archiving(self, archive_name, onedrive_path, months_limit):
def __init__(self):
self.running: bool = False
self.progress: float = 0.0
self.current_mail: str = "In attesa di avvio..."
async def start_archiving(self, archive_name: str, onedrive_path: str, months_limit: int) -> None:
"""Avvia il processo di archiviazione in background."""
self.running = True
self.progress = 0.0
ui.notify('Connessione a Outlook in corso...', color='info')
@@ -38,19 +60,23 @@ class ArchiverGUI:
self.current_mail = "Archiviazione completata con successo!"
ui.notify('Processo terminato!', type='positive')
def run_logic(self, archive_name, onedrive_path, months_limit):
def run_logic(self, archive_name: str, onedrive_path: str, months_limit: int) -> None:
"""Logica principale di archiviazione delle email."""
try:
if not os.path.exists(onedrive_path):
os.makedirs(onedrive_path)
logger.info(f"Cartella OneDrive creata: {onedrive_path}")
outlook = win32com.client.Dispatch("Outlook.Application")
namespace = outlook.GetNamespace("MAPI")
inbox = namespace.GetDefaultFolder(6) # 6 = OlFolderInbox
inbox = namespace.GetDefaultFolder(OutlookConstants.FOLDER_INBOX)
try:
archive_root = namespace.Folders.Item(archive_name)
except Exception:
self.current_mail = f"ERRORE: Archivio '{archive_name}' non trovato."
except Exception as e:
error_msg = f"ERRORE: Archivio '{archive_name}' non trovato."
logger.error(error_msg, exc_info=e)
self.current_mail = error_msg
self.running = False
return
@@ -65,14 +91,17 @@ class ArchiverGUI:
if total == 0:
self.current_mail = "Nessuna mail trovata con i criteri selezionati."
self.progress = 1.0
logger.info("Nessuna email da archiviare")
return
processed_files = {}
logger.info(f"Inizio archiviazione di {total} email")
processed_files: Dict[str, str] = {}
# Ciclo dal fondo verso l'inizio per non sballare gli indici di Outlook
for i in range(total, 0, -1):
if not self.running:
self.current_mail = "Processo interrotto dall'utente."
logger.info("Archiviazione interrotta dall'utente")
break
try:
@@ -80,44 +109,52 @@ class ArchiverGUI:
rt = item.ReceivedTime
received_time = datetime(rt.year, rt.month, rt.day, rt.hour, rt.minute)
self.current_mail = f"[{total-i+1}/{total}] {item.Subject[:40]}..."
self.progress = float((total - i + 1) / total)
current_idx = total - i + 1
subject_preview = item.Subject[:40] if item.Subject else "(nessun oggetto)"
self.current_mail = f"[{current_idx}/{total}] {subject_preview}..."
self.progress = float(current_idx / total)
# 1. ELIMINAZIONE ALLEGATI E AGGIUNTA LINK (Mentre è ancora in Inbox)
# Questo garantisce i permessi di scrittura necessari per cancellare i file
# 1. ELIMINAZIONE ALLEGATI E AGGIUNTA LINK
self.process_attachments(item, onedrive_path, received_time, processed_files)
item.Save()
# 2. PREPARAZIONE CARTELLA DESTINAZIONE
anno_str = str(received_time.year)
nome_mese = self.mesi_it[received_time.month]
month_folder_name = f"{received_time.month:02d}-{nome_mese}"
year_str = str(received_time.year)
month_name = self.MONTH_NAMES[received_time.month]
month_folder_name = f"{received_time.month:02d}-{month_name}"
y_f = self.get_or_create_folder(archive_root, anno_str)
target_folder = self.get_or_create_folder(y_f, month_folder_name)
year_folder = self.get_or_create_folder(archive_root, year_str)
target_folder = self.get_or_create_folder(year_folder, month_folder_name)
# 3. SPOSTAMENTO DEFINITIVO
item.Move(target_folder)
# Piccola pausa per dare respiro al server Exchange
time.sleep(0.1)
except Exception as e:
print(f"Errore su singola mail: {e}")
logger.error(f"Errore elaborazione email {i}: {e}")
continue
except Exception as e:
self.current_mail = f"Errore critico: {str(e)}"
error_msg = f"Errore critico: {str(e)}"
logger.critical(error_msg, exc_info=e)
self.current_mail = error_msg
self.running = False
def get_or_create_folder(self, parent, name):
def get_or_create_folder(self, parent, name: str):
"""Recupera una cartella se esiste, altrimenti la crea."""
try:
return parent.Folders.Item(name)
except:
except Exception:
logger.debug(f"Cartella '{name}' non trovata, creazione...")
return parent.Folders.Add(name)
def process_attachments(self, mail_item, onedrive_path, received_time, processed_files):
def process_attachments(self, mail_item, onedrive_path: str, received_time: datetime,
processed_files: Dict[str, str]) -> None:
"""Rimuove gli allegati reali, li salva su OneDrive e inserisce il link nella mail."""
if mail_item.Class != 43: return # 43 = OlMail
if mail_item.Class != OutlookConstants.MAIL_CLASS:
return
date_prefix = received_time.strftime("%Y-%m-%d")
@@ -127,8 +164,8 @@ class ArchiverGUI:
try:
att = mail_item.Attachments.Item(j)
# Filtro: Solo file reali (Type 1), escludendo immagini nelle firme (Inline)
if att.Type != 1:
# Filtro: Solo file reali, escludendo immagini nelle firme (Inline)
if att.Type != OutlookConstants.ATTACHMENT_TYPE_FILE:
continue
is_inline = False
@@ -136,11 +173,12 @@ class ArchiverGUI:
# Controlla il tag MAPI per il Content-ID delle immagini incorporate
if att.PropertyAccessor.GetProperty("http://schemas.microsoft.com/mapi/proptag/0x3712001E"):
is_inline = True
except: pass
except Exception:
pass
if is_inline:
continue
print("da qui gestisco gli allegati")
# Salvataggio fisico del file
temp_path = os.path.join(os.environ['TEMP'], att.FileName)
att.SaveAsFile(temp_path)
@@ -154,9 +192,11 @@ class ArchiverGUI:
if not os.path.exists(dest_path):
os.replace(temp_path, dest_path)
processed_files[f_hash] = dest_path
logger.debug(f"File salvato: {unique_name}")
else:
if os.path.exists(temp_path):
os.remove(temp_path)
logger.debug(f"File duplicato eliminato: {att.FileName}")
# Inserimento link HTML nel corpo della mail
link_html = (
@@ -170,8 +210,7 @@ class ArchiverGUI:
mail_item.Attachments.Remove(j)
except Exception as e:
print(f"Errore su allegato {j}: {e}")
continue
logger.error(f"Errore elaborazione allegato {j}: {e}")
# --- INTERFACCIA GRAFICA (NICEGUI) ---