Vladimir Buzalka
199 lines
7.0 KiB
Python
199 lines
7.0 KiB
Python
"""
|
|
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,
|
|
},
|
|
)
|