wienen
eb53bc123a
Prompt since previous commit: execute step 1 Centralized route targets, source-folder policy, and modified UTF-7 mailbox quoting. Verification: py_compile, route/folder consistency, and git diff --check passed.
97 lines
2.7 KiB
Python
97 lines
2.7 KiB
Python
"""Shared IMAP mailbox-name helpers."""
|
|
|
|
import base64
|
|
import re
|
|
|
|
|
|
def to_mutf7(s: str) -> str:
|
|
"""Encode a mailbox name as IMAP modified UTF-7."""
|
|
res = []
|
|
buf = []
|
|
for c in s:
|
|
if 0x20 <= ord(c) <= 0x7E and c != "&":
|
|
if buf:
|
|
b64 = (
|
|
base64.b64encode("".join(buf).encode("utf-16-be"))
|
|
.decode("ascii")
|
|
.rstrip("=")
|
|
.replace("/", ",")
|
|
)
|
|
res.append(f"&{b64}-")
|
|
buf.clear()
|
|
res.append(c)
|
|
elif c == "&":
|
|
if buf:
|
|
b64 = (
|
|
base64.b64encode("".join(buf).encode("utf-16-be"))
|
|
.decode("ascii")
|
|
.rstrip("=")
|
|
.replace("/", ",")
|
|
)
|
|
res.append(f"&{b64}-")
|
|
buf.clear()
|
|
res.append("&-")
|
|
else:
|
|
buf.append(c)
|
|
if buf:
|
|
b64 = (
|
|
base64.b64encode("".join(buf).encode("utf-16-be"))
|
|
.decode("ascii")
|
|
.rstrip("=")
|
|
.replace("/", ",")
|
|
)
|
|
res.append(f"&{b64}-")
|
|
return "".join(res)
|
|
|
|
|
|
def from_mutf7(s: str) -> str:
|
|
"""Decode an IMAP modified UTF-7 mailbox name."""
|
|
res = []
|
|
i = 0
|
|
while i < len(s):
|
|
if s[i] == "&":
|
|
j = s.index("-", i + 1)
|
|
if j == i + 1:
|
|
res.append("&")
|
|
else:
|
|
encoded = s[i + 1 : j].replace(",", "/")
|
|
pad = (4 - len(encoded) % 4) % 4
|
|
res.append(
|
|
base64.b64decode(encoded + "=" * pad).decode("utf-16-be")
|
|
)
|
|
i = j + 1
|
|
else:
|
|
res.append(s[i])
|
|
i += 1
|
|
return "".join(res)
|
|
|
|
|
|
def quote_mailbox(folder: str) -> str:
|
|
"""Return an IMAP-safe quoted mailbox name."""
|
|
escaped = to_mutf7(folder).replace("\\", "\\\\").replace('"', r"\"")
|
|
return f'"{escaped}"'
|
|
|
|
|
|
def parse_folder_name(raw: bytes) -> str | None:
|
|
"""Extract and decode the mailbox name from an IMAP LIST response."""
|
|
decoded = raw.decode("ascii", errors="replace")
|
|
match = re.match(r'\(.*?\)\s+".*?"\s+(.*)', decoded)
|
|
if not match:
|
|
return None
|
|
name = match.group(1).strip().strip('"')
|
|
if not name:
|
|
return None
|
|
try:
|
|
return from_mutf7(name)
|
|
except Exception:
|
|
return name
|
|
|
|
|
|
def list_folders(mail) -> list[str]:
|
|
"""Return decoded folder names from an IMAP connection."""
|
|
status, raw_folders = mail.list()
|
|
if status != "OK" or not raw_folders:
|
|
return []
|
|
folders = [parse_folder_name(item) for item in raw_folders if item]
|
|
return sorted(folder for folder in folders if folder)
|