Spaces:
Sleeping
Sleeping
File size: 1,921 Bytes
c75bbd7 | 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 | """AgentAZAll message format — compose & parse plain-text messages."""
from pathlib import Path
from typing import Optional, Tuple
from .helpers import generate_id, now_str
def format_message(from_a, to_a, subject, body, msg_id=None, attachments=None) -> Tuple[str, str]:
"""Build a plain-text message string. Returns (content, msg_id)."""
if not msg_id:
msg_id = generate_id(from_a, to_a, subject)
lines = [
f"From: {from_a}",
f"To: {to_a}",
f"Subject: {subject}",
f"Date: {now_str()}",
f"Message-ID: {msg_id}",
"Status: new",
]
if attachments:
lines.append(f"Attachments: {', '.join(Path(a).name for a in attachments)}")
lines += ["", "---", body]
return "\n".join(lines), msg_id
def parse_message(path) -> Tuple[Optional[dict], Optional[str]]:
"""Parse a message file into (headers_dict, body_text)."""
p = Path(path)
if not p.exists():
return None, None
text = p.read_text(encoding="utf-8", errors="replace")
headers: dict = {}
body_lines: list = []
in_body = False
for line in text.split("\n"):
if not in_body:
if line.strip() == "---":
in_body = True
continue
if ":" in line:
k, _, v = line.partition(":")
headers[k.strip()] = v.strip()
else:
body_lines.append(line)
return headers, "\n".join(body_lines)
def parse_headers_only(path) -> dict:
"""Parse only message headers (faster — stops at '---')."""
headers = {}
with open(path, "r", encoding="utf-8", errors="replace") as f:
for line in f:
line = line.rstrip("\n")
if line.strip() == "---":
break
if ":" in line:
k, _, v = line.partition(":")
headers[k.strip()] = v.strip()
return headers
|