import msal import requests import warnings import base64 import os from datetime import datetime, timedelta from dotenv import load_dotenv warnings.filterwarnings("ignore") # ============================================================================ # CONFIGURACIÓN # ============================================================================ # Cargar variables de entorno desde .env (ruta absoluta: no depende del # directorio de trabajo desde el que se lance el proceso) load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), ".env")) # Configuración de la API CLIENT_ID = os.getenv("CLIENT_ID", "") CLIENT_SECRET = os.getenv("CLIENT_SECRET", "") TENANT_ID = os.getenv("TENANT_ID", "") DOWNLOAD_FOLDER = os.getenv("DOWNLOAD_FOLDER","") VERIFY_SSL = os.getenv("VERIFY_SSL", "False").lower() == "true" # ============================================================================ def get_access_token() -> str: """ Obtiene un token de acceso de Microsoft Graph usando las credenciales configuradas en config.json (Client ID, Client Secret y Tenant ID). """ authority = f"https://login.microsoftonline.com/{TENANT_ID}" app = msal.ConfidentialClientApplication( CLIENT_ID, authority=authority, client_credential=CLIENT_SECRET, verify=VERIFY_SSL ) scopes = ["https://graph.microsoft.com/.default"] result = app.acquire_token_for_client(scopes=scopes) if result: if "access_token" in result: return result["access_token"] else: error = result.get("error") error_description = result.get("error_description") correlation_id = result.get("correlation_id") raise Exception(f"Error al obtener token: {error} - {error_description} (ID: {correlation_id})") # ============================================================================ def get_emails(user_email: str, token: str, fecha_inicio: str, fecha_fin: str) -> list[dict]: """ Obtiene los correos de una cuenta específica en un rango de fechas. Retorna una lista de JSON con: id, fecha (ajustada -5h), sender, subject y has_attach. """ headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } dt_inicio_utc = datetime.strptime(fecha_inicio, "%Y-%m-%d") + timedelta(hours=5) dt_fin_utc = datetime.strptime(fecha_fin, "%Y-%m-%d") + timedelta(days=1, hours=5) filter_query = ( f"receivedDateTime ge {dt_inicio_utc.strftime('%Y-%m-%dT%H:%M:%SZ')} and " f"receivedDateTime lt {dt_fin_utc.strftime('%Y-%m-%dT%H:%M:%SZ')}" ) params = { "$filter": filter_query, "$select": "id,receivedDateTime,hasAttachments,subject,from", "$top": 50 } url = f"https://graph.microsoft.com/v1.0/users/{user_email}/mailFolders/Inbox/messages" all_mails = [] while url: response = requests.get( url, headers=headers, params=params, verify=VERIFY_SSL ) if response.status_code != 200: raise Exception(f"Error Graph API ({user_email}): {response.status_code} - {response.text}") data = response.json() messages = data.get("value", []) for msg in messages: utc_str = msg.get("receivedDateTime") dt_utc = datetime.fromisoformat(utc_str.replace("Z", "+00:00")) dt_local = dt_utc - timedelta(hours=5) from_obj = msg.get("from", {}) sender = from_obj.get("emailAddress", {}).get("address", "Desconocido") all_mails.append({ "id": msg.get("id"), "fecha": dt_local.strftime("%Y-%m-%d %H:%M:%S"), "sender": sender, "subject": msg.get("subject", ""), "has_attach": msg.get("hasAttachments") }) url = data.get("@odata.nextLink") params = None return all_mails # ============================================================================ def download_attachments(user_email: str, message_id: str, token: str, download_path: str) -> list[str]: """ Descarga los adjuntos de un mensaje específico y los guarda en la ruta indicada. Retorna una lista con los nombres de los archivos descargados. """ if not os.path.exists(download_path): os.makedirs(download_path, exist_ok=True) headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } url = f"https://graph.microsoft.com/v1.0/users/{user_email}/messages/{message_id}/attachments" response = requests.get(url, headers=headers, verify=VERIFY_SSL) if response.status_code != 200: raise Exception(f"Error al obtener adjuntos ({message_id}): {response.status_code} - {response.text}") attachments = response.json().get("value", []) downloaded_files = [] for attachment in attachments: if attachment.get("@odata.type") == "#microsoft.graph.fileAttachment": file_name = attachment.get("name") content_bytes = attachment.get("contentBytes") if content_bytes: file_path = os.path.join(download_path, file_name) with open(file_path, "wb") as f: f.write(base64.b64decode(content_bytes)) downloaded_files.append(file_name) return downloaded_files # ============================================================================ def send_email(access_token: str, from_email: str, to_recipients: list[str], subject: str, html_body: str, attachment_paths: list[str] | None = None, cc_recipients: list[str] | None = None) -> bool: """ Envía un correo electrónico con cuerpo HTML y múltiples adjuntos PDF. :param access_token: Token de acceso de Microsoft Graph API. :param from_email: Correo electrónico del remitente (cuenta de Office 365). :param to_recipients: Lista de correos electrónicos de los destinatarios. :param subject: Asunto del correo. :param html_body: Contenido del correo en formato HTML. :param attachment_paths: Lista de rutas a los archivos PDF que se adjuntarán. :param cc_recipients: Lista de correos electrónicos en copia (CC). :return: True si se envió correctamente, False en caso contrario. """ url = f"https://graph.microsoft.com/v1.0/users/{from_email}/sendMail" headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json' } verify_ssl = False #True if os.getenv('OUTLOOK_VERIFY_SSL').lower() == 'true' else False # Construir lista de destinatarios para Graph API recipients_payload = [] for email in to_recipients: recipients_payload.append({ "emailAddress": { "address": email } }) # Construir lista de destinatarios en copia (CC) para Graph API cc_payload = [] for email in (cc_recipients or []): cc_payload.append({ "emailAddress": { "address": email } }) # Construir lista de adjuntos attachments_payload = [] if attachment_paths: for path in attachment_paths: if os.path.exists(path): with open(path, "rb") as f: content_bytes = f.read() # Codificar el contenido en Base64 encoded_content = base64.b64encode(content_bytes).decode('utf-8') filename = os.path.basename(path) ext = os.path.splitext(filename)[1].lower() content_type = { ".pdf": "application/pdf", ".xml": "application/xml", ".json": "application/json", }.get(ext, "application/octet-stream") attachments_payload.append({ "@odata.type": "#microsoft.graph.fileAttachment", "name": filename, "contentType": content_type, "contentBytes": encoded_content }) else: print(f"Advertencia: El archivo no existe y no se adjuntará: {path}") # Estructura del mensaje para la API email_data = { "message": { "subject": subject, "body": { "contentType": "HTML", "content": html_body }, "toRecipients": recipients_payload, "ccRecipients": cc_payload, "attachments": attachments_payload }, "saveToSentItems": "true" } try: response = requests.post(url, headers=headers, json=email_data, verify=verify_ssl) if response.status_code == 202: print(f" ├── SUCCESS: Correo enviado exitosamente a: {', '.join(to_recipients)}") return True else: print(f" ├── ERROR: Al enviar correo: {response.status_code} - {response.text}") return False except Exception as e: print(f" ├── ERROR: Excepcion al enviar correo: {e}") return False # ============================================================================