""" graph_mail.py ------------- Tenká vrstva nad Microsoft Graph API pro ČTENÍ schránky a STAHOVÁNÍ příloh. Používá stejnou app registraci (application permissions) jako Knihovny/EmailMessagingGraph.py. Pro čtení cizí schránky a příloh musí mít ta app registrace grant **Mail.Read** (Application). Pokud chybí, Graph vrátí 403 a je potřeba oprávnění doplnit v Azure portálu. """ import base64 import msal import requests from functools import lru_cache from typing import Iterator # ========================= # CONFIG (sdíleno s EmailMessagingGraph.py) # ========================= TENANT_ID = "7d269944-37a4-43a1-8140-c7517dc426e9" CLIENT_ID = "4b222bfd-78c9-4239-a53f-43006b3ed07f" CLIENT_SECRET = "Txg8Q~MjhocuopxsJyJBhPmDfMxZ2r5WpTFj1dfk" AUTHORITY = f"https://login.microsoftonline.com/{TENANT_ID}" SCOPE = ["https://graph.microsoft.com/.default"] GRAPH = "https://graph.microsoft.com/v1.0" @lru_cache(maxsize=1) def _token() -> str: app = msal.ConfidentialClientApplication( CLIENT_ID, authority=AUTHORITY, client_credential=CLIENT_SECRET ) tok = app.acquire_token_for_client(scopes=SCOPE) if "access_token" not in tok: raise RuntimeError(f"Graph auth failed: {tok}") return tok["access_token"] def _headers() -> dict: return {"Authorization": f"Bearer {_token()}"} def _get(url: str, params: dict | None = None) -> dict: r = requests.get(url, headers=_headers(), params=params, timeout=60) r.raise_for_status() return r.json() def _post(url: str, body: dict) -> dict: r = requests.post( url, headers={**_headers(), "Content-Type": "application/json"}, json=body, timeout=60, ) r.raise_for_status() return r.json() if r.content else {} def _patch(url: str, body: dict) -> dict: r = requests.patch( url, headers={**_headers(), "Content-Type": "application/json"}, json=body, timeout=60, ) r.raise_for_status() return r.json() if r.content else {} def inbox_folder_ids(mailbox: str) -> list[tuple[str, str]]: """ Vrátí [(folder_id, display_name), ...] pro Inbox a jeho přímé podsložky. Záměrně vynechává Junk, Deleted, Sent, Drafts. """ inbox = _get(f"{GRAPH}/users/{mailbox}/mailFolders/inbox") folders = [(inbox["id"], inbox.get("displayName", "Inbox"))] data = _get( f"{GRAPH}/users/{mailbox}/mailFolders/{inbox['id']}/childFolders", {"$top": 100, "$select": "id,displayName"}, ) for f in data.get("value", []): folders.append((f["id"], f.get("displayName", ""))) return folders def list_messages(mailbox: str, folder_id: str, since_iso: str) -> Iterator[dict]: """ Vrací zprávy s přílohou ve složce přijaté od `since_iso` (ISO 8601 UTC, Z). Stránkuje přes @odata.nextLink. Tělo se vrací jako text (Prefer header). """ url = f"{GRAPH}/users/{mailbox}/mailFolders/{folder_id}/messages" params = { "$filter": f"hasAttachments eq true and receivedDateTime ge {since_iso}", "$select": "id,subject,from,receivedDateTime,bodyPreview,body,hasAttachments", "$top": 50, } headers = {**_headers(), "Prefer": 'outlook.body-content-type="text"'} while url: r = requests.get(url, headers=headers, params=params, timeout=60) r.raise_for_status() data = r.json() yield from data.get("value", []) url = data.get("@odata.nextLink") params = None # nextLink už obsahuje všechny parametry def list_attachments(mailbox: str, message_id: str) -> list[dict]: """Metadata příloh (id, name, contentType, size, @odata.type).""" data = _get( f"{GRAPH}/users/{mailbox}/messages/{message_id}/attachments", {"$top": 50}, ) return data.get("value", []) def download_attachment(mailbox: str, message_id: str, attachment_id: str) -> bytes: """Stáhne bajty jedné fileAttachment.""" data = _get( f"{GRAPH}/users/{mailbox}/messages/{message_id}/attachments/{attachment_id}" ) content = data.get("contentBytes") if not content: raise RuntimeError(f"Příloha nemá contentBytes (typ {data.get('@odata.type')})") return base64.b64decode(content) # --------------------------------------------------------------------------- # Zápisové operace (vyžadují Mail.ReadWrite Application) # --------------------------------------------------------------------------- def ensure_category(mailbox: str, name: str, color: str = "preset4") -> None: """Zajistí kategorii v master-listu schránky (s barvou). Idempotentní.""" data = _get(f"{GRAPH}/users/{mailbox}/outlook/masterCategories", {"$top": 200}) if any(c.get("displayName") == name for c in data.get("value", [])): return _post( f"{GRAPH}/users/{mailbox}/outlook/masterCategories", {"displayName": name, "color": color}, ) def add_category(mailbox: str, message_id: str, name: str) -> None: """Přidá kategorii ke zprávě (zachová stávající).""" msg = _get( f"{GRAPH}/users/{mailbox}/messages/{message_id}", {"$select": "categories"} ) cats = msg.get("categories") or [] if name not in cats: _patch( f"{GRAPH}/users/{mailbox}/messages/{message_id}", {"categories": cats + [name]}, ) def ensure_folder_path(mailbox: str, parts: list[str]) -> str: """ Zajistí cestu složek pod Inboxem (vytvoří chybějící). `parts` jsou názvy podsložek, např. ["ProcessedByAgent", "Invoices"]. Vrátí id poslední složky. """ parent_id = _get(f"{GRAPH}/users/{mailbox}/mailFolders/inbox")["id"] for name in parts: children = _get( f"{GRAPH}/users/{mailbox}/mailFolders/{parent_id}/childFolders", {"$top": 200, "$select": "id,displayName"}, ) match = next( (f for f in children.get("value", []) if f.get("displayName") == name), None ) if match is None: match = _post( f"{GRAPH}/users/{mailbox}/mailFolders/{parent_id}/childFolders", {"displayName": name}, ) parent_id = match["id"] return parent_id def move_message(mailbox: str, message_id: str, dest_folder_id: str) -> str: """Přesune zprávu do složky. Vrací NOVÉ id (move id mění).""" res = _post( f"{GRAPH}/users/{mailbox}/messages/{message_id}/move", {"destinationId": dest_folder_id}, ) return res.get("id", message_id) def send_mail(sender: str, to, subject: str, html_body: str) -> None: """Odešle HTML e-mail přes Graph (vyžaduje Mail.Send Application).""" to_list = [to] if isinstance(to, str) else list(to) _post( f"{GRAPH}/users/{sender}/sendMail", { "message": { "subject": subject, "body": {"contentType": "HTML", "content": html_body}, "toRecipients": [{"emailAddress": {"address": a}} for a in to_list], }, "saveToSentItems": True, }, )