Compare commits

..

13 Commits

Author SHA1 Message Date
Alessandro Battilani
a35e0b118b disattivazione messaggi bat e timeout prima di uscire 2026-01-21 16:44:40 +01:00
eec665d231 Update ArchiviazioneMail.py 2026-01-21 10:44:19 +00:00
Alessandro Battilani
f14d032348 Merge branch 'main' of https://gitbat.duckdns.org/alex/ArchiveMail 2026-01-21 11:30:23 +01:00
Alessandro Battilani
49386bceca tolti warning ruff 2026-01-21 11:23:57 +01:00
Alessandro Battilani
c11f7eaa7b default n mesi 2026-01-20 09:21:49 +01:00
Alessandro Battilani
1081ba9e1b input mesi 2026-01-15 15:08:41 +01:00
Alessandro Battilani
7f3e047aff tolto locale che non serve più 2026-01-15 10:00:11 +01:00
Alessandro Battilani
7282352401 pre eliminazione gestione allegati 2026-01-15 09:53:03 +01:00
Alessandro Battilani
1868859800 cui native 2026-01-11 14:44:07 +01:00
Alessandro Battilani
0080fa8586 con gui 2026-01-11 14:36:28 +01:00
Alessandro Battilani
16c271349f gui v1 2026-01-10 16:20:08 +01:00
Alessandro Battilani
9cf6040a59 senza gui 2026-01-10 15:36:02 +01:00
Alessandro Battilani
2d9426e87f prima release 2026-01-09 19:01:12 +01:00
6 changed files with 287 additions and 618 deletions

3
ArchiviaEmail.bat Normal file
View File

@@ -0,0 +1,3 @@
@echo off
uv run .\ArchiviazioneMail.py
timeout /t 5

View File

@@ -1,272 +0,0 @@
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
import win32com.client
# --- 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: str) -> str:
"""Calcola l'MD5 per identificare file identici ed evitare duplicati."""
hasher = hashlib.md5()
with open(file_path, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b""):
hasher.update(chunk)
return hasher.hexdigest()
class ArchiverGUI:
"""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"
}
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')
# Avvia la logica in un thread dedicato per non bloccare l'interfaccia
await run.io_bound(self.run_logic, archive_name, onedrive_path, months_limit)
self.running = False
if self.progress >= 1.0:
self.current_mail = "Archiviazione completata con successo!"
ui.notify('Processo terminato!', type='positive')
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(OutlookConstants.FOLDER_INBOX)
try:
archive_root = namespace.Folders.Item(archive_name)
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
# Calcolo data limite per il filtro
cutoff_date = datetime.now() - timedelta(days=int(months_limit) * 30)
filter_str = f"[ReceivedTime] < '{cutoff_date.strftime('%d/%m/%Y %H:%M')}'"
items = inbox.Items.Restrict(filter_str)
items.Sort("[ReceivedTime]", True)
total = items.Count
if total == 0:
self.current_mail = "Nessuna mail trovata con i criteri selezionati."
self.progress = 1.0
logger.info("Nessuna email da archiviare")
return
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:
item = items.Item(i)
rt = item.ReceivedTime
received_time = datetime(rt.year, rt.month, rt.day, rt.hour, rt.minute)
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
self.process_attachments(item, onedrive_path, received_time, processed_files)
item.Save()
# 2. PREPARAZIONE CARTELLA DESTINAZIONE
year_str = str(received_time.year)
month_name = self.MONTH_NAMES[received_time.month]
month_folder_name = f"{received_time.month:02d}-{month_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:
logger.error(f"Errore elaborazione email {i}: {e}")
continue
except Exception as 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: str):
"""Recupera una cartella se esiste, altrimenti la crea."""
try:
return parent.Folders.Item(name)
except Exception:
logger.debug(f"Cartella '{name}' non trovata, creazione...")
return parent.Folders.Add(name)
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 != OutlookConstants.MAIL_CLASS:
return
date_prefix = received_time.strftime("%Y-%m-%d")
# Usiamo un ciclo reverse sugli allegati per la rimozione sicura
count = mail_item.Attachments.Count
for j in range(count, 0, -1):
try:
att = mail_item.Attachments.Item(j)
# Filtro: Solo file reali, escludendo immagini nelle firme (Inline)
if att.Type != OutlookConstants.ATTACHMENT_TYPE_FILE:
continue
is_inline = False
try:
# 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 Exception:
pass
if is_inline:
continue
# Salvataggio fisico del file
temp_path = os.path.join(os.environ['TEMP'], att.FileName)
att.SaveAsFile(temp_path)
f_hash = get_file_hash(temp_path)
# Nome file univoco con Data e Hash corto
unique_name = f"{date_prefix}_{f_hash[:6]}_{att.FileName}"
dest_path = os.path.join(onedrive_path, unique_name)
if f_hash not in processed_files:
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 = (
f"<div style='background:#f3f3f3; padding:10px; border:1px dotted #666; margin:10px 0; font-family:Arial; font-size:12px;'>"
f"<b>📎 Allegato spostato su OneDrive ({date_prefix}):</b><br>"
f"<a href='file:///{dest_path}'>{att.FileName}</a></div>"
)
mail_item.HTMLBody = link_html + mail_item.HTMLBody
# RIMOZIONE FISICA DALLA MAIL
mail_item.Attachments.Remove(j)
except Exception as e:
logger.error(f"Errore elaborazione allegato {j}: {e}")
# --- INTERFACCIA GRAFICA (NICEGUI) ---
archiver = ArchiverGUI()
@ui.page('/')
def main_page():
ui.query('body').style('background-color: #f0f2f5; font-family: "Segoe UI", Tahoma, Geneva, Verdana, sans-serif;')
with ui.header().classes('items-center justify-between bg-blue-9 shadow-4 q-pa-md'):
ui.label('Outlook Smart Archiver').classes('text-h5 text-white font-bold')
ui.icon('inventory_2', size='lg', color='white')
with ui.column().classes('w-full max-w-3xl mx-auto q-pa-lg gap-6'):
# CARD CONFIGURAZIONE
with ui.card().classes('w-full q-pa-md shadow-2 border-l-4 border-blue-9'):
ui.label('Configurazione').classes('text-h6 text-blue-9 mb-2')
arc_name = ui.input('Nome dell\'Archivio Online',
value='Archivio Online - Nome.Cognome@intesasanpaolo.com').classes('w-full')
od_path = ui.input('Percorso locale OneDrive per Allegati',
value=r'C:\Users\--UTENZA--\OneDrive - Intesa SanPaolo\Allegati_Outlook').classes('w-full')
with ui.row().classes('w-full items-center mt-4 bg-grey-2 q-pa-sm rounded'):
ui.label('Mesi da mantenere nella Inbox:')
slider = ui.slider(min=0, max=24, value=6).classes('col px-4')
ui.badge().bind_text_from(slider, 'value').classes('text-lg bg-blue-9')
# CARD PROGRESSO
with ui.card().classes('w-full q-pa-md shadow-2'):
ui.label('Avanzamento').classes('text-h6 mb-2')
with ui.column().classes('w-full items-center'):
ui.label().bind_text_from(archiver, 'current_mail').classes('text-grey-8 text-italic mb-2 text-center')
# Contenitore della barra
with ui.linear_progress().bind_value_from(archiver, 'progress').props('stripe size=40px color=blue-9').classes('rounded-borders relative shadow-1') as p:
# Etichetta percentuale centrata e troncata
ui.label().bind_text_from(archiver, 'progress', backward=lambda x: f'{x * 100:.2f}%') \
.classes('absolute-center text-blue text-weight-bold') \
.style('text-shadow: 1px 1px 3px rgba(0,0,0,0.4); font-size: 1.2rem;')
with ui.row().classes('w-full justify-center mt-8 gap-4'):
start_btn = ui.button('AVVIA ARCHIVIAZIONE', icon='rocket_launch', color='green-8',
on_click=lambda: archiver.start_archiving(arc_name.value, od_path.value, slider.value)) \
.classes('q-px-lg')
start_btn.bind_enabled_from(archiver, 'running', backward=lambda x: not x)
ui.button('STOP', icon='block', color='red-8',
on_click=lambda: setattr(archiver, 'running', False)) \
.classes('q-px-lg').bind_enabled_from(archiver, 'running')
ui.markdown('--- \n *Ricorda: Chiudi Outlook prima di avviare il processo per evitare blocchi.*').classes('text-center text-grey-6 text-xs')
# Avvio dell'applicazione
if __name__ == '__main__':
ui.run(native=True, window_size=(800, 600), title='Outlook Archiver Pro')

View File

@@ -1,184 +1,122 @@
import win32com.client import win32com.client
import os
import hashlib
from datetime import datetime, timedelta from datetime import datetime, timedelta
from tqdm import tqdm from tqdm import tqdm
import time import time
import locale import locale
# Imposta la localizzazione in italiano
try:
locale.setlocale(locale.LC_TIME, "it_IT.UTF-8") # Per Windows/Linux moderni
except: # noqa: E722
try:
locale.setlocale(locale.LC_TIME, "ita_ita") # Specifica per Windows
except: # noqa: E722
print("Impossibile impostare il locale italiano, uso i nomi manuali.")
# --- CONFIGURAZIONE --- # --- CONFIGURAZIONE ---
ARCHIVE_NAME = "Archivio online - alessandro.battilani@intesasanpaolo.com" ARCHIVE_NAME = "Archivio online - alessandro.battilani@intesasanpaolo.com"
ONEDRIVE_PATH = r"C:\Users\U086304\OneDrive - Intesa SanPaolo\Allegati_Outlook" # MONTHS_LIMIT = 3
if not os.path.exists(ONEDRIVE_PATH):
os.makedirs(ONEDRIVE_PATH)
print(f"Cartella creata: {ONEDRIVE_PATH}")
MONTHS_LIMIT = 5
# ---------------------- # ----------------------
def get_file_hash(file_path): # Imposta la localizzazione in italiano
hasher = hashlib.md5() try:
with open(file_path, 'rb') as f: locale.setlocale(locale.LC_TIME, "it_IT.UTF-8")
for chunk in iter(lambda: f.read(4096), b""): except: # noqa: E722
hasher.update(chunk) try:
return hasher.hexdigest() locale.setlocale(locale.LC_TIME, "ita_ita")
except: # noqa: E722
print("Locale italiano non impostato, uso nomi manuali.")
def main(): def main():
if not os.path.exists(ONEDRIVE_PATH):
os.makedirs(ONEDRIVE_PATH)
print("Connessione a Outlook in corso...") print("Connessione a Outlook in corso...")
try: try:
outlook = win32com.client.Dispatch("Outlook.Application") outlook = win32com.client.Dispatch("Outlook.Application")
namespace = outlook.GetNamespace("MAPI") namespace = outlook.GetNamespace("MAPI")
inbox = namespace.GetDefaultFolder(6)
archive_root = namespace.Folders.Item(ARCHIVE_NAME) archive_root = namespace.Folders.Item(ARCHIVE_NAME)
# 6 = Inbox (ReceivedTime), 5 = Sent Items (SentOn)
folders_to_process = [
(namespace.GetDefaultFolder(6), "[ReceivedTime]", "ReceivedTime"),
(namespace.GetDefaultFolder(5), "[SentOn]", "SentOn")
]
except Exception as e: except Exception as e:
print(f"Errore di connessione: {e}") print(f"Errore connessione: {e}")
return return
stringa_mounths_limit = input("\n Quanti mesi vuoi tenere in linea? (default 3)") or '3'
# Convertiamo la stringa in un numero intero
MONTHS_LIMIT = int(stringa_mounths_limit)
cutoff_date = datetime.now() - timedelta(days=MONTHS_LIMIT * 30) cutoff_date = datetime.now() - timedelta(days=MONTHS_LIMIT * 30)
filter_date_str = cutoff_date.strftime("%d/%m/%Y %H:%M") filter_date_str = cutoff_date.strftime("%d/%m/%Y %H:%M")
filter_str = f"[ReceivedTime] < '{filter_date_str}'"
items = inbox.Items.Restrict(filter_str)
items.Sort("[ReceivedTime]", True)
total_items = items.Count
if total_items == 0:
print("Nessuna email da archiviare.")
return
print(f"Trovate {total_items} email vecchie. Inizio archiviazione...")
processed_files = {}
archived_count = 0
mesi_it = {1: "Gennaio", 2: "Febbraio", 3: "Marzo", 4: "Aprile", 5: "Maggio", 6: "Giugno", 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"} 7: "Luglio", 8: "Agosto", 9: "Settembre", 10: "Ottobre", 11: "Novembre", 12: "Dicembre"}
with tqdm(total=total_items, desc="Archiviazione", unit="mail", colour='green') as pbar: for source_folder, filter_attr, time_attr in folders_to_process:
for i in range(total_items, 0, -1): print(f"\n--- Analisi cartella: {source_folder.Name} ---")
try:
item = items.Item(i) filter_str = f"{filter_attr} < '{filter_date_str}'"
items = source_folder.Items.Restrict(filter_str)
items.Sort(filter_attr, True)
total_items = items.Count
if total_items == 0:
print(f"Nessuna mail più vecchia di {MONTHS_LIMIT} mesi in {source_folder.Name}.")
continue
archived_count = 0
# tqdm posizionato esternamente per monitorare il progresso reale
with tqdm(total=total_items, desc=f"Archiviazione {source_folder.Name}", unit="mail", colour='green') as pbar:
for i in range(total_items, 0, -1):
try:
item = items.Item(i)
if not hasattr(item, time_attr):
pbar.update(1)
continue
# Recupero data dinamico
rt = getattr(item, time_attr)
received_time = datetime(rt.year, rt.month, rt.day, rt.hour, rt.minute)
anno_str = str(received_time.year)
nome_mese = mesi_it[received_time.month]
pbar.set_description(f"Archiviazione {source_folder.Name} - Mese: {nome_mese} {anno_str}")
# Gestione struttura cartelle (Source -> Anno -> Mese)
try:
arch_type = archive_root.Folders.Item(source_folder.Name)
except: # noqa: E722
arch_type = archive_root.Folders.Add(source_folder.Name)
try:
y_f = arch_type.Folders.Item(anno_str)
except: # noqa: E722
y_f = arch_type.Folders.Add(anno_str)
month_folder_name = f"{received_time.month:02d}-{nome_mese}"
try:
target_folder = y_f.Folders.Item(month_folder_name)
except: # noqa: E722
target_folder = y_f.Folders.Add(month_folder_name)
# --- TENTA LO SPOSTAMENTO CON RETRY (Reintegrato) ---
archived_item = None
for tentativo in range(3):
try:
archived_item = item.Move(target_folder)
if archived_item:
archived_item.Save()
time.sleep(0.4) # Pausa vitale
break
except: # noqa: E722
time.sleep(1) # Attesa per server busy
if archived_item:
archived_count += 1
pbar.set_postfix(successo=archived_count)
else:
pbar.set_postfix(error="Move fallito")
except Exception as e:
pbar.set_postfix(err=str(e)[:15])
# Update della barra sempre alla fine del ciclo i
pbar.update(1) pbar.update(1)
if not hasattr(item, 'ReceivedTime'): print("\nOperazione conclusa con successo.")
continue
rt = item.ReceivedTime
received_time = datetime(rt.year, rt.month, rt.day, rt.hour, rt.minute)
anno_str = str(received_time.year)
nome_mese = mesi_it[received_time.month]
pbar.set_description(f"Mese: {nome_mese} {anno_str}")
# Cartelle
month_folder_name = f"{received_time.month:02d}-{nome_mese}"
try:
y_f = archive_root.Folders.Item(anno_str)
except: # noqa: E722
y_f = archive_root.Folders.Add(anno_str)
try:
target_folder = y_f.Folders.Item(month_folder_name)
except: # noqa: E722
target_folder = y_f.Folders.Add(month_folder_name)
# --- TENTA LO SPOSTAMENTO CON RETRY ---
archived_item = None
for tentativo in range(3):
try:
archived_item = item.Move(target_folder)
if archived_item:
archived_item.Save()
time.sleep(0.5) # Pausa vitale per il server
break
except: # noqa: E722
time.sleep(1) # Aspetta se il server è occupato
if not archived_item:
pbar.set_postfix(error="Move fallito")
continue
# --- GESTIONE ALLEGATI (SOLO FILE VERI) ---
if archived_item.Class == 43 and archived_item.Attachments.Count > 0:
has_changed = False
attachments = [archived_item.Attachments.Item(j) for j in range(1, archived_item.Attachments.Count + 1)]
for att in attachments:
try:
# 1. Filtro base: solo allegati di tipo "Valore" (file)
if att.Type != 1:
continue
# 2. FILTRO AVANZATO: Salta le immagini nelle firme (Inline Images)
# Controlliamo se l'allegato ha un "Content-ID" (tipico delle immagini incorporate)
try:
prop_accessor = att.PropertyAccessor
cid = prop_accessor.GetProperty("http://schemas.microsoft.com/mapi/proptag/0x3712001E")
if cid: # Se ha un CID, è un'immagine nel testo/firma
continue
except:
# Se non riesce a leggere la proprietà, procediamo con cautela
# (spesso i file veri non hanno questa proprietà)
pass
# 3. FILTRO ESTENSIONI: Opzionale, se vuoi escludere png/jpg a prescindere
# ext = os.path.splitext(att.FileName)[1].lower()
# if ext in ['.png', '.jpg', '.jpeg', '.gif']: continue
# --- Procedura di salvataggio con DATA nel nome ---
temp_path = os.path.join(os.environ['TEMP'], att.FileName)
att.SaveAsFile(temp_path)
f_hash = get_file_hash(temp_path)
# Creiamo il prefisso con la data (es. 2025-02-09_)
date_prefix = received_time.strftime("%Y-%m-%d")
if f_hash not in processed_files:
# Il nome file sarà: DATA_HASH_NOMEORIGINALE.ext
# Usiamo l'hash corto (primi 6 caratteri) per non avere nomi infiniti
unique_name = f"{date_prefix}_{f_hash[:6]}_{att.FileName}"
dest_path = os.path.join(ONEDRIVE_PATH, unique_name)
if not os.path.exists(dest_path):
os.replace(temp_path, dest_path)
processed_files[f_hash] = dest_path
else:
dest_path = processed_files[f_hash]
os.remove(temp_path)
# Il link nella mail punterà al nuovo nome con la data
link_html = (
f"<div style='border:1px solid #ccc; padding:8px; margin:10px 0; "
f"background-color:#f3f3f3; font-family:Arial; font-size:12px;'>"
f"<b>📎 Allegato spostato su OneDrive ({date_prefix}):</b><br>"
f"<a href='file:///{dest_path}'>{att.FileName}</a></div>"
)
archived_item.HTMLBody = link_html + archived_item.HTMLBody
att.Delete()
has_changed = True
except Exception as e:
continue
if has_changed:
archived_item.Save()
archived_count += 1
pbar.set_postfix(archiviate=archived_count)
except Exception as e:
pbar.set_postfix(last_err=str(e)[:15])
continue
print(f"\nCompletato. Archiviate: {archived_count}")
if __name__ == "__main__": if __name__ == "__main__":
main() main()

4
uv.lock generated
View File

@@ -120,7 +120,7 @@ source = { virtual = "." }
dependencies = [ dependencies = [
{ name = "nicegui" }, { name = "nicegui" },
{ name = "pywebview" }, { name = "pywebview" },
{ name = "pywin32", marker = "sys_platform == 'win32'" }, { name = "pywin32" },
{ name = "tqdm" }, { name = "tqdm" },
] ]
@@ -128,7 +128,7 @@ dependencies = [
requires-dist = [ requires-dist = [
{ name = "nicegui", specifier = ">=3.5.0" }, { name = "nicegui", specifier = ">=3.5.0" },
{ name = "pywebview", specifier = ">=6.1" }, { name = "pywebview", specifier = ">=6.1" },
{ name = "pywin32", marker = "sys_platform == 'win32'", specifier = ">=311" }, { name = "pywin32", specifier = ">=311" },
{ name = "tqdm", specifier = ">=4.67.1" }, { name = "tqdm", specifier = ">=4.67.1" },
] ]