Spaces:
Sleeping
Sleeping
File size: 4,078 Bytes
4c83ab6 0a367ef 4c83ab6 0a367ef 4c83ab6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | 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
|