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)