91 lines
3 KiB
Python
91 lines
3 KiB
Python
import re
|
|
|
|
_MESSAGE_ID_PATTERN = re.compile(r"<[^>]+>")
|
|
|
|
|
|
def canonical_message_id(value: str | None) -> str | None:
|
|
"""Normalize to angle-bracket form while preserving original case."""
|
|
if value is None:
|
|
return None
|
|
cleaned = value.strip()
|
|
if not cleaned:
|
|
return None
|
|
if not cleaned.startswith("<"):
|
|
cleaned = f"<{cleaned.strip('<>')}>"
|
|
return cleaned
|
|
|
|
|
|
def folded_message_id(value: str | None) -> str | None:
|
|
"""Case-insensitive lookup key for Message-ID matching."""
|
|
canonical = canonical_message_id(value)
|
|
return canonical.casefold() if canonical else None
|
|
|
|
|
|
def normalize_message_id(value: str | None) -> str | None:
|
|
"""Preserve case — Gmail threading requires exact Message-ID tokens."""
|
|
return canonical_message_id(value)
|
|
|
|
|
|
def internet_message_id_of_message(message: dict) -> str | None:
|
|
"""Resolve Message-ID, preferring raw MIME header case when stored value was folded."""
|
|
stored = canonical_message_id(
|
|
message.get("internet_message_id") or message.get("message_id")
|
|
)
|
|
raw_header = (message.get("raw_headers") or {}).get("message_id")
|
|
raw_canonical = canonical_message_id(raw_header)
|
|
if raw_canonical and stored:
|
|
if folded_message_id(raw_canonical) == folded_message_id(stored):
|
|
return raw_canonical
|
|
return stored or raw_canonical
|
|
|
|
|
|
def message_id_fields(value: str | None) -> dict[str, str]:
|
|
canonical = canonical_message_id(value)
|
|
if not canonical:
|
|
return {}
|
|
folded = folded_message_id(canonical)
|
|
return {
|
|
"message_id": canonical,
|
|
"internet_message_id": canonical,
|
|
"message_id_folded": folded or canonical.casefold(),
|
|
}
|
|
|
|
|
|
def parse_message_id_list(header_value: str | None) -> list[str]:
|
|
if not header_value:
|
|
return []
|
|
folded_header = re.sub(r"\s+", " ", header_value.strip())
|
|
seen: set[str] = set()
|
|
ordered: list[str] = []
|
|
for match in _MESSAGE_ID_PATTERN.findall(folded_header):
|
|
canonical = canonical_message_id(match)
|
|
key = folded_message_id(canonical)
|
|
if canonical and key and key not in seen:
|
|
seen.add(key)
|
|
ordered.append(canonical)
|
|
if not ordered:
|
|
single = canonical_message_id(folded_header)
|
|
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:
|
|
canonical = canonical_message_id(value)
|
|
key = folded_message_id(canonical)
|
|
if canonical and key and key not in seen:
|
|
seen.add(key)
|
|
chain.append(canonical)
|
|
parent = canonical_message_id(parent_message_id)
|
|
parent_key = folded_message_id(parent)
|
|
if parent and parent_key and parent_key not in seen:
|
|
chain.append(parent)
|
|
return " ".join(chain)
|