""" OutlookConnector — Frank Outlook Module (Layer 1) Connessione a Outlook via COM/MAPI usando pywin32. Outlook deve essere aperto e loggato prima dell'uso. Guardrail: nessun delete definitivo — le email vengono sempre spostate in [Frank_Cestino] invece di essere eliminate. """ try: from loguru import logger except ImportError: import logging logger = logging.getLogger(__name__) class OutlookConnector: """Wrapper COM/MAPI per Outlook. Richiede Outlook aperto e loggato.""" def __init__(self): self.app = None self.namespace = None def connect(self) -> bool: """ Connette a Outlook via COM (late binding). Usa Dispatch, NON GetActiveObject — funziona anche se Outlook non era già aperto. Restituisce True se connessione riuscita. """ try: import win32com.client self.app = win32com.client.Dispatch('Outlook.Application') self.namespace = self.app.GetNamespace('MAPI') account = self.namespace.CurrentUser.Name logger.info(f"Connesso a Outlook — utente: {account}") return True except Exception as e: logger.error(f"Connessione Outlook fallita: {e}") return False def list_folders(self, parent=None, recursive: bool = False) -> list[dict]: """ Elenca le cartelle di Outlook. Args: parent: folder MAPI da cui partire (None = root Inbox) recursive: se True, elenca anche le sottocartelle Returns: Lista di dict: [{'name': str, 'path': str, 'count': int}, ...] """ if not self.namespace: raise RuntimeError("Non connesso. Chiama connect() prima.") if parent is None: parent = self.namespace.GetDefaultFolder(6) # 6 = olFolderInbox results = [] try: for folder in parent.Folders: entry = { 'name': folder.Name, 'path': folder.FolderPath, 'count': folder.Items.Count } results.append(entry) if recursive: results.extend(self.list_folders(parent=folder, recursive=True)) except Exception as e: logger.error(f"Errore listando cartelle: {e}") return results def get_emails(self, folder_name: str = 'Inbox', n: int = 20, unread_only: bool = False) -> list[dict]: """ Legge le ultime N email da una cartella. Usa .Restrict() per filtrare in modo efficiente (evita di caricare tutta la mailbox in memoria). Args: folder_name: nome cartella ('Inbox' = Posta in arrivo) n: numero massimo di email da restituire unread_only: se True, restituisce solo le non lette Returns: Lista di dict per ogni email: {'subject', 'sender', 'sender_email', 'received', 'body_preview', 'unread', 'entry_id'} """ if not self.namespace: raise RuntimeError("Non connesso. Chiama connect() prima.") # Recupera la cartella folder = self._get_folder_by_name(folder_name) if folder is None: logger.error(f"Cartella '{folder_name}' non trovata") return [] items = folder.Items items.Sort('[ReceivedTime]', True) # Ordine decrescente (più recenti prima) # Applica filtro non letti se richiesto if unread_only: items = items.Restrict('[UnRead] = True') results = [] count = 0 try: for item in items: if count >= n: break # Gestisce solo messaggi email (non appuntamenti, task, ecc.) if item.Class != 43: # 43 = olMail continue try: body_preview = (item.Body or '')[:200].replace('\r\n', ' ').replace('\n', ' ') received = item.ReceivedTime if hasattr(received, 'strftime'): received_str = received.strftime('%Y-%m-%d %H:%M') else: received_str = str(received) results.append({ 'subject': item.Subject or '(nessun oggetto)', 'sender': item.SenderName or '', 'sender_email': item.SenderEmailAddress or '', 'received': received_str, 'body_preview': body_preview, 'unread': item.UnRead, 'entry_id': item.EntryID }) count += 1 except Exception as e: logger.warning(f"Errore leggendo email: {e}") continue except Exception as e: logger.error(f"Errore iterando email in '{folder_name}': {e}") return results def move_email(self, entry_id: str, target_folder_name: str) -> bool: """ Sposta un'email in una cartella target. GUARDRAIL: non usare per cancellare email — usa sempre [Frank_Cestino] come destinazione "cestino" reversibile. Args: entry_id: EntryID dell'email (da get_emails) target_folder_name: nome della cartella destinazione Returns: True se operazione riuscita """ if not self.namespace: raise RuntimeError("Non connesso. Chiama connect() prima.") try: # Recupera email tramite ID mail_item = self.namespace.GetItemFromID(entry_id) # Recupera o crea la cartella target inbox = self.namespace.GetDefaultFolder(6) # olFolderInbox target_folder = self.get_or_create_folder(inbox, target_folder_name) subject = mail_item.Subject or '(nessun oggetto)' mail_item.Move(target_folder) logger.info(f"Email spostata in '{target_folder_name}': {subject}") return True except Exception as e: logger.error(f"Errore spostando email {entry_id}: {e}") return False def get_or_create_folder(self, parent_folder, name: str): """ Cerca una sottocartella per nome in parent_folder. Se non esiste, la crea. Args: parent_folder: folder MAPI padre name: nome della cartella da cercare/creare Returns: Folder object MAPI """ try: for folder in parent_folder.Folders: if folder.Name == name: return folder # Non trovata — creala new_folder = parent_folder.Folders.Add(name) logger.info(f"Cartella '{name}' creata") return new_folder except Exception as e: logger.error(f"Errore get_or_create_folder '{name}': {e}") raise def send_email(self, to: str, subject: str, body: str) -> bool: """ Invia un'email tramite Outlook. Args: to: destinatario (indirizzo email) subject: oggetto body: corpo del messaggio (testo semplice) Returns: True se invio riuscito """ if not self.app: raise RuntimeError("Non connesso. Chiama connect() prima.") try: mail = self.app.CreateItem(0) # 0 = olMailItem mail.To = to mail.Subject = subject mail.Body = body mail.Send() logger.info(f"Email inviata a {to}: {subject}") return True except Exception as e: logger.error(f"Errore invio email a {to}: {e}") return False def _get_folder_by_name(self, folder_name: str): """ Recupera una cartella per nome semplice. 'Inbox' mappa sulla cartella Posta in arrivo nativa. Per altre cartelle, cerca per nome in primo livello dell'Inbox. """ if folder_name.lower() in ('inbox', 'posta in arrivo'): return self.namespace.GetDefaultFolder(6) # olFolderInbox # Cerca tra le sottocartelle dell'Inbox inbox = self.namespace.GetDefaultFolder(6) for folder in inbox.Folders: if folder.Name == folder_name: return folder # Cerca a livello root (store principale) try: for store in self.namespace.Stores: root = store.GetRootFolder() for folder in root.Folders: if folder.Name == folder_name: return folder except Exception: pass return None