agentic.email/backend/app/services/message_ids.py
2026-10-06 01:51:23 +05:30

51 lines
1.5 KiB
Python

import re
_MESSAGE_ID_PATTERN = re.compile(r"<[^>]+>")
def normalize_message_id(value: str | None) -> str | None:
if value is None:
return None
cleaned = value.strip()
if not cleaned:
return None
if not cleaned.startswith("<"):
cleaned = f"<{cleaned.strip('<>')}>"
return cleaned.lower()
def parse_message_id_list(header_value: str | None) -> list[str]:
if not header_value:
return []
folded = re.sub(r"\s+", " ", header_value.strip())
seen: set[str] = set()
ordered: list[str] = []
for match in _MESSAGE_ID_PATTERN.findall(folded):
normalized = normalize_message_id(match)
if normalized and normalized not in seen:
seen.add(normalized)
ordered.append(normalized)
if not ordered:
single = normalize_message_id(folded)
if single:
ordered.append(single)
return ordered
def first_message_id(header_value: str | None) -> str | None:
values = parse_message_id_list(header_value)
return values[0] if values else None
def build_references_header(references: list[str], parent_message_id: str) -> str:
chain: list[str] = []
seen: set[str] = set()
for value in references:
normalized = normalize_message_id(value)
if normalized and normalized not in seen:
seen.add(normalized)
chain.append(normalized)
parent = normalize_message_id(parent_message_id)
if parent and parent not in seen:
chain.append(parent)
return " ".join(chain)