Spaces:
Sleeping
Sleeping
| import json | |
| import os | |
| import shutil | |
| import uuid | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| def _safe_filename(name: str) -> str: | |
| return "".join(c if (c.isalnum() or c in "._- ") else "_" for c in name).strip() | |
| def build_email_dir(base_path: Path, received_datetime: str, message_id: str) -> Path: | |
| try: | |
| dt = datetime.fromisoformat(received_datetime.replace("Z", "+00:00")) | |
| except ValueError: | |
| dt = datetime.now(timezone.utc) | |
| date_folder = dt.strftime("%Y-%m-%d") | |
| time_prefix = dt.strftime("%Y%m%d_%H%M%S") | |
| folder_name = f"{time_prefix}_{message_id[:8]}" | |
| return base_path / date_folder / folder_name | |
| async def write_email( | |
| email_data: dict, | |
| attachment_bytes: list[tuple[str, str, bytes]], | |
| base_path: Path, | |
| ) -> Path: | |
| received = email_data.get("receivedDateTime", "") | |
| message_id = email_data.get("id", str(uuid.uuid4())) | |
| final_dir = build_email_dir(base_path, received, message_id) | |
| tmp_dir = base_path / ".tmp" / str(uuid.uuid4()) | |
| tmp_dir.mkdir(parents=True, exist_ok=True) | |
| try: | |
| body = email_data.get("body", {}) | |
| if body.get("contentType", "").lower() == "html": | |
| (tmp_dir / "body.html").write_text(body.get("content", ""), encoding="utf-8") | |
| else: | |
| (tmp_dir / "body.txt").write_text(body.get("content", ""), encoding="utf-8") | |
| if email_data.get("uniqueBody"): | |
| unique = email_data["uniqueBody"] | |
| if unique.get("contentType", "").lower() == "html": | |
| (tmp_dir / "body_unique.html").write_text( | |
| unique.get("content", ""), encoding="utf-8" | |
| ) | |
| else: | |
| (tmp_dir / "body_unique.txt").write_text( | |
| unique.get("content", ""), encoding="utf-8" | |
| ) | |
| stored_attachments = [] | |
| if attachment_bytes: | |
| att_dir = tmp_dir / "attachments" | |
| att_dir.mkdir() | |
| for filename, content_type, raw in attachment_bytes: | |
| safe_name = _safe_filename(filename) or "attachment" | |
| dest = att_dir / safe_name | |
| dest.write_bytes(raw) | |
| stored_attachments.append( | |
| { | |
| "filename": safe_name, | |
| "content_type": content_type, | |
| "size_bytes": len(raw), | |
| } | |
| ) | |
| from_addr = email_data.get("from", {}).get("emailAddress", {}) | |
| metadata = { | |
| "message_id": message_id, | |
| "subject": email_data.get("subject", ""), | |
| "from": {"name": from_addr.get("name", ""), "address": from_addr.get("address", "")}, | |
| "to": [ | |
| {"name": r.get("emailAddress", {}).get("name", ""), | |
| "address": r.get("emailAddress", {}).get("address", "")} | |
| for r in email_data.get("toRecipients", []) | |
| ], | |
| "cc": [ | |
| {"name": r.get("emailAddress", {}).get("name", ""), | |
| "address": r.get("emailAddress", {}).get("address", "")} | |
| for r in email_data.get("ccRecipients", []) | |
| ], | |
| "received_datetime": received, | |
| "has_attachments": email_data.get("hasAttachments", False), | |
| "attachment_count": len(stored_attachments), | |
| "attachments": stored_attachments, | |
| "graph_web_link": email_data.get("webLink", ""), | |
| "stored_at": datetime.now(timezone.utc).isoformat(), | |
| } | |
| (tmp_dir / "metadata.json").write_text( | |
| json.dumps(metadata, indent=2), encoding="utf-8" | |
| ) | |
| final_dir.parent.mkdir(parents=True, exist_ok=True) | |
| if final_dir.exists(): | |
| # Duplicate webhook delivery — email already stored; discard temp copy. | |
| shutil.rmtree(tmp_dir, ignore_errors=True) | |
| return final_dir | |
| os.rename(tmp_dir, final_dir) | |
| return final_dir | |
| except Exception: | |
| shutil.rmtree(tmp_dir, ignore_errors=True) | |
| raise | |