add-ideas/govoplan-mail
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

chore: sync GovOPlaN module split state

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-10 12:51:22 +02:00
parent b4b0c76455
commit d5d5b89603
21 changed files with 1955 additions and 551 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
src/govoplan_mail/backend/sending/imap.py
View File

@@ -115,6 +115,8 @@ class ImapMailboxMessageListResult:
total_count: int
offset: int
limit: int
uidvalidity: str | None = None
cursor_reset: bool = False
@dataclass(frozen=True, slots=True)
@@ -527,15 +529,23 @@ def _parse_fetch_parts_with_sequence(data: list[Any] | tuple[Any, ...] | None) -
return parts
def _select_readonly(client: imaplib.IMAP4, folder: str) -> int:
def _select_readonly(client: imaplib.IMAP4, folder: str) -> tuple[int, str | None]:
typ, data = client.select(folder, readonly=True)
if typ != "OK":
raise ImapAppendError(f"IMAP folder {folder!r} could not be opened read-only: {data!r}", temporary=False)
selected_count = _decode_item(data[0] if data else None).strip()
try:
return max(0, int(selected_count))
total_count = max(0, int(selected_count))
except ValueError:
return 0
total_count = 0
uidvalidity = None
try:
_, uidvalidity_data = client.response("UIDVALIDITY")
if uidvalidity_data:
uidvalidity = _decode_item(uidvalidity_data[0]).strip() or None
except Exception:
uidvalidity = None
return total_count, uidvalidity
def _fetch_message_by_uid(client: imaplib.IMAP4, uid: str) -> tuple[str, list[str], int | None, bytes]:
@@ -590,6 +600,63 @@ def _fetch_message_summaries_by_sequence(client: imaplib.IMAP4, sequences: list[
return summaries
def _search_message_uids(client: imaplib.IMAP4) -> list[str]:
typ, data = client.uid("search", None, "ALL")
if typ != "OK":
raise ImapAppendError(f"IMAP UID search failed: {data!r}", temporary=True)
uids: set[int] = set()
for item in data or []:
for token in _decode_item(item).split():
if token.isdigit():
uids.add(int(token))
return [str(uid) for uid in sorted(uids, reverse=True)]
def _paged_descending_uids(
uids: list[str],
*,
offset: int,
limit: int,
after_uid: str | None = None,
) -> tuple[list[str], int, bool]:
if not uids:
return [], 0, False
cursor_reset = False
if after_uid:
try:
start = uids.index(str(after_uid)) + 1
except ValueError:
start = 0
cursor_reset = True
else:
start = min(max(0, offset), len(uids))
return uids[start:start + limit], start, cursor_reset
def _fetch_message_summaries_by_uid(client: imaplib.IMAP4, uids: list[str], folder: str) -> list[ImapMailboxMessageSummary]:
if not uids:
return []
typ, data = client.uid("fetch", _sequence_set(uids), "(UID FLAGS RFC822.SIZE BODY.PEEK[HEADER.FIELDS (SUBJECT FROM TO CC DATE MESSAGE-ID)])")
if typ != "OK":
raise ImapAppendError(f"IMAP message summary fetch failed: {data!r}", temporary=True)
by_uid: dict[str, ImapMailboxMessageSummary] = {}
for _sequence, fetched_uid, flags, size_bytes, headers in _parse_fetch_parts_with_sequence(data):
if not fetched_uid:
continue
by_uid[fetched_uid] = _message_summary_from_headers(uid=fetched_uid, folder=folder, headers=headers, flags=flags, size_bytes=size_bytes)
summaries: list[ImapMailboxMessageSummary] = []
for uid in uids:
summary = by_uid.get(str(uid))
if summary is not None:
summaries.append(summary)
continue
fetched_uid, flags, size_bytes, raw = _fetch_message_by_uid(client, uid)
summaries.append(_message_summary_from_raw(uid=fetched_uid, folder=folder, raw=raw, flags=flags, size_bytes=size_bytes))
return summaries
def _mock_record_folder(record: dict[str, Any]) -> str:
folder = str(record.get("folder") or "").strip()
if folder:
@@ -612,7 +679,15 @@ def _mock_raw_bytes(record: dict[str, Any]) -> bytes:
return "\n".join(lines).encode("utf-8", errors="replace")
def list_imap_messages(*, imap_config: ImapConfig, folder: str = "INBOX", limit: int = 50, offset: int = 0) -> ImapMailboxMessageListResult:
def list_imap_messages(
*,
imap_config: ImapConfig,
folder: str = "INBOX",
limit: int = 50,
offset: int = 0,
after_uid: str | None = None,
expected_uidvalidity: str | None = None,
) -> ImapMailboxMessageListResult:
"""List mailbox messages without mutating read/unread state."""
host, port = _require_imap_config(imap_config)
@@ -621,7 +696,20 @@ def list_imap_messages(*, imap_config: ImapConfig, folder: str = "INBOX", limit:
offset = max(0, offset)
if is_mock_imap_host(imap_config.host):
records = [record for record in list_records(limit=500) if _mock_folder_matches(record, folder)]
page_records = records[offset:offset + limit]
cursor_reset = False
if expected_uidvalidity and expected_uidvalidity != "mock-v1":
effective_offset = 0
cursor_reset = True
elif after_uid:
record_ids = [str(record.get("id") or "") for record in records]
try:
effective_offset = record_ids.index(str(after_uid)) + 1
except ValueError:
effective_offset = 0
cursor_reset = True
else:
effective_offset = min(offset, len(records))
page_records = records[effective_offset:effective_offset + limit]
messages = [
_message_summary_from_raw(
uid=str(record.get("id") or ""),
@@ -632,14 +720,49 @@ def list_imap_messages(*, imap_config: ImapConfig, folder: str = "INBOX", limit:
)
for record in page_records
]
return ImapMailboxMessageListResult(host=host, port=port, security=imap_config.security.value, folder=folder, messages=messages, total_count=len(records), offset=offset, limit=limit)
return ImapMailboxMessageListResult(
host=host,
port=port,
security=imap_config.security.value,
folder=folder,
messages=messages,
total_count=len(records),
offset=effective_offset,
limit=limit,
uidvalidity="mock-v1",
cursor_reset=cursor_reset,
)
client = _open_imap(imap_config)
try:
total_count = _select_readonly(client, folder)
page_sequences = _paged_descending_sequences(total_count, offset=offset, limit=limit)
messages = _fetch_message_summaries_by_sequence(client, page_sequences, folder)
return ImapMailboxMessageListResult(host=host, port=port, security=imap_config.security.value, folder=folder, messages=messages, total_count=total_count, offset=offset, limit=limit)
total_count, uidvalidity = _select_readonly(client, folder)
uids = _search_message_uids(client)
cursor_reset = False
effective_after_uid = after_uid
effective_offset = offset
if expected_uidvalidity and expected_uidvalidity != uidvalidity:
effective_after_uid = None
effective_offset = 0
cursor_reset = True
page_uids, effective_offset, anchor_missing = _paged_descending_uids(
uids,
offset=effective_offset,
limit=limit,
after_uid=effective_after_uid,
)
messages = _fetch_message_summaries_by_uid(client, page_uids, folder)
return ImapMailboxMessageListResult(
host=host,
port=port,
security=imap_config.security.value,
folder=folder,
messages=messages,
total_count=len(uids) if uids else total_count,
offset=effective_offset,
limit=limit,
uidvalidity=uidvalidity,
cursor_reset=cursor_reset or anchor_missing,
)
finally:
try:
client.logout()