"""Pure parsing helpers for IMAP responses (unit-testable, no I/O). aioimaplib returns responses as lists of ``bytes``/``bytearray`` lines. A FETCH with ``BODY[]`` yields, per message, a header line ending ``... {}`` followed by a ``bytearray`` literal, then a lone ``)``. LIST lines carry SPECIAL-USE flags. """ from __future__ import annotations import re from dataclasses import dataclass from datetime import datetime from imapclient.imap_utf7 import decode as imap_utf7_decode from imapclient.imap_utf7 import encode as imap_utf7_encode # --- SPECIAL-USE mapping (SPEC §4 special_use values) ----------------------- _SPECIAL_USE_FLAGS = { "\\sent": "sent", "\\drafts": "drafts", "\\trash": "trash", "\\junk": "junk", "\\archive": "archive", "\\all": "archive", # Gmail "All Mail" acts as archive } _LIST_RE = re.compile(rb'^\(([^)]*)\)\s+(?:"([^"]*)"|NIL)\s+(.+)$') _UID_RE = re.compile(rb"UID\s+(\d+)") _FLAGS_RE = re.compile(rb"FLAGS\s+\(([^)]*)\)") _GM_LABELS_RE = re.compile(rb"X-GM-LABELS\s+\(([^)]*)\)") _INTERNALDATE_RE = re.compile(rb'INTERNALDATE\s+"([^"]+)"') _LITERAL_RE = re.compile(rb"\{(\d+)\}\s*$") _FETCH_HEADER_RE = re.compile(rb"^\d+\s+FETCH\s+\(", re.IGNORECASE) @dataclass(frozen=True) class ParsedMailbox: path: str special_use: str | None @dataclass(frozen=True) class ParsedFetch: uid: int flags: frozenset[str] internal_date: datetime | None raw: bytes | None # None when the fetch did not request the body gmail_labels: tuple[str, ...] = () # custom Gmail labels (X-GM-LABELS), system ones dropped def _decode_mailbox_name(raw: bytes) -> str: name = raw.strip() if name.startswith(b'"') and name.endswith(b'"'): name = name[1:-1] # IMAP mailbox names are modified UTF-7 (SPEC §4: store UTF-7 decoded). return str(imap_utf7_decode(bytes(name))) def encode_mailbox_name(name: str) -> str: """Modified UTF-7 wire form of a mailbox name (inverse of ``_decode_mailbox_name``). Mailbox names are stored/displayed as Unicode but IMAP commands (SELECT/STATUS/APPEND) require them mUTF-7 encoded. ASCII names are unchanged; non-ASCII names — e.g. Gmail's localized ``[Gmail]/Wysłane`` — get encoded so the server can find them.""" return str(imap_utf7_encode(name).decode("ascii")) def parse_list_line(line: bytes) -> ParsedMailbox | None: """Parse one LIST response line into a :class:`ParsedMailbox` (or None). Non-selectable mailboxes are skipped (return None): ``\\Noselect`` / ``\\NonExistent`` names are pure hierarchy containers that hold no messages and cannot be SELECTed — e.g. Gmail's ``[Gmail]`` root or a parent label like ``Serial numbers`` that only groups child labels.""" m = _LIST_RE.match(line.strip()) if not m: return None flags = m.group(1).lower().split() if b"\\noselect" in flags or b"\\nonexistent" in flags: return None name = _decode_mailbox_name(m.group(3)) special: str | None = None for flag in flags: mapped = _SPECIAL_USE_FLAGS.get(flag.decode("ascii", "ignore")) if mapped: special = mapped break if special is None and name.upper() == "INBOX": special = "inbox" return ParsedMailbox(path=name, special_use=special) def parse_internaldate(raw: bytes | str) -> datetime | None: """Parse an IMAP INTERNALDATE (``d-Mon-Y H:M:S +ZZZZ``) to an aware datetime.""" text = raw.decode("ascii", "ignore") if isinstance(raw, bytes) else raw text = text.strip() try: return datetime.strptime(text, "%d-%b-%Y %H:%M:%S %z") except ValueError: return None def _parse_flag_token(tok: bytes) -> str: return tok.decode("ascii", "ignore") def parse_fetch_header(line: bytes) -> tuple[int | None, frozenset[str], datetime | None]: """Extract (uid, flags, internaldate) from a FETCH header line.""" uid_m = _UID_RE.search(line) uid = int(uid_m.group(1)) if uid_m else None flags_m = _FLAGS_RE.search(line) if flags_m: flags = frozenset(_parse_flag_token(t) for t in flags_m.group(1).split()) else: flags = frozenset() date_m = _INTERNALDATE_RE.search(line) internal = parse_internaldate(date_m.group(1)) if date_m else None return uid, flags, internal def _tokenize_gm_labels(blob: bytes) -> list[bytes]: """Split an ``X-GM-LABELS (...)`` body into raw tokens, keeping quoted strings (which may contain spaces) whole. Escaped chars inside quotes are unescaped.""" tokens: list[bytes] = [] i, n = 0, len(blob) while i < n: ch = blob[i : i + 1] if ch in (b" ", b"\t"): i += 1 continue if ch == b'"': j = i + 1 buf = bytearray() while j < n and blob[j : j + 1] != b'"': if blob[j : j + 1] == b"\\" and j + 1 < n: buf += blob[j + 1 : j + 2] j += 2 continue buf += blob[j : j + 1] j += 1 tokens.append(bytes(buf)) i = j + 1 else: j = i while j < n and blob[j : j + 1] not in (b" ", b"\t"): j += 1 tokens.append(blob[i:j]) i = j return tokens def parse_gm_labels(header: bytes) -> tuple[str, ...]: """Custom Gmail labels from a FETCH header's ``X-GM-LABELS`` (SPEC: Gmail sync). System labels (``\\Inbox``, ``\\Sent``, ``\\Important``, ``\\Starred``, …) are dropped — only user labels become chips. Names are modified-UTF-7 decoded, like mailbox names, so e.g. ``Zg&AAI...`` → the real Unicode label.""" m = _GM_LABELS_RE.search(header) if not m: return () out: list[str] = [] for tok in _tokenize_gm_labels(m.group(1)): if not tok or tok.startswith(b"\\"): # system label continue try: out.append(str(imap_utf7_decode(tok))) except Exception: # noqa: BLE001 - fall back to a best-effort ASCII decode out.append(tok.decode("utf-8", "ignore")) return tuple(out) def parse_fetch_lines(lines: list[bytes | bytearray]) -> list[ParsedFetch]: """Walk a full FETCH response, returning one :class:`ParsedFetch` per message. Handles both metadata-only fetches (FLAGS/INTERNALDATE) and body fetches (a ``{size}`` literal announced on the header line, body on the next line). """ results: list[ParsedFetch] = [] i = 0 n = len(lines) while i < n: line = lines[i] if not isinstance(line, (bytes, bytearray)) or not _FETCH_HEADER_RE.match(bytes(line)): i += 1 continue header = bytes(line) uid, flags, internal = parse_fetch_header(header) raw: bytes | None = None if _LITERAL_RE.search(header) and i + 1 < n: body = lines[i + 1] raw = bytes(body) i += 2 # Skip the trailing ")" continuation line if present. if ( i < n and isinstance(lines[i], (bytes, bytearray)) and bytes(lines[i]).strip() == b")" ): i += 1 else: i += 1 if uid is not None: results.append( ParsedFetch( uid=uid, flags=flags, internal_date=internal, raw=raw, gmail_labels=parse_gm_labels(header), ) ) return results _VANISHED_RE = re.compile(rb"^VANISHED\s+(?:\(EARLIER\)\s+)?(\S+)", re.IGNORECASE) def _expand_uid_set(token: bytes) -> list[int]: """Expand an IMAP uid-set (``1,3:5,9``) to a flat list of ints (``*`` skipped).""" out: list[int] = [] for part in token.split(b","): part = part.strip() lo_b, sep, hi_b = part.partition(b":") if sep: if lo_b.isdigit() and hi_b.isdigit(): lo, hi = int(lo_b), int(hi_b) out.extend(range(min(lo, hi), max(lo, hi) + 1)) elif part.isdigit(): out.append(int(part)) return out def parse_vanished(lines: list[bytes | bytearray]) -> list[int]: """UIDs from QRESYNC ``VANISHED [(EARLIER)] `` responses (RFC 7162). aioimaplib strips the leading ``* `` before buffering the line, so we match on the bare ``VANISHED`` token. Both the SELECT-time ``(EARLIER)`` form and the live-expunge form are accepted. """ out: list[int] = [] for line in lines: if not isinstance(line, (bytes, bytearray)): continue m = _VANISHED_RE.match(bytes(line).strip()) if m: out.extend(_expand_uid_set(m.group(1))) return out def parse_search_uids(lines: list[bytes | bytearray]) -> list[int]: """Parse a UID SEARCH response's first line into a list of ints.""" if not lines: return [] first = lines[0] if not isinstance(first, (bytes, bytearray)): return [] return [int(tok) for tok in bytes(first).split() if tok.isdigit()] def parse_select_untagged(lines: list[bytes | bytearray]) -> dict[str, int]: """Extract UIDVALIDITY / UIDNEXT / HIGHESTMODSEQ / EXISTS from SELECT lines.""" out: dict[str, int] = {} patterns = { "uidvalidity": re.compile(rb"UIDVALIDITY\s+(\d+)"), "uidnext": re.compile(rb"UIDNEXT\s+(\d+)"), "highestmodseq": re.compile(rb"HIGHESTMODSEQ\s+(\d+)"), "exists": re.compile(rb"^(\d+)\s+EXISTS"), } for line in lines: if not isinstance(line, (bytes, bytearray)): continue b = bytes(line) for key, pat in patterns.items(): m = pat.search(b) if m: out[key] = int(m.group(1)) return out