GovOPlaN/govoplan-mail
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases Wiki Activity

intermittent commit

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-14 13:22:11 +02:00
parent 4865de5007
commit b0337b2d26
28 changed files with 2264 additions and 490 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,6 +1,7 @@
from __future__ import annotations
import imaplib
import logging
import re
import socket
import ssl
@@ -21,6 +22,8 @@ from govoplan_mail.backend.dev.mock_mailbox import (
record_imap_append,
)
logger = logging.getLogger(__name__)
class ImapConfigurationError(ValueError):
"""Raised when IMAP settings are incomplete or inconsistent."""
@@ -70,6 +73,10 @@ class ImapMailboxAttachmentInfo:
size_bytes: int
def _log_imap_cleanup_failure(action: str, exc: BaseException) -> None:
logger.debug("IMAP cleanup failed while %s: %s", action, exc, exc_info=True)
@dataclass(frozen=True, slots=True)
class ImapMailboxMessageSummary:
uid: str
@@ -119,6 +126,12 @@ class ImapMailboxMessageListResult:
cursor_reset: bool = False
@dataclass(frozen=True, slots=True)
class ImapMailboxBootstrapResult:
folders: ImapFolderListResult
messages: ImapMailboxMessageListResult
@dataclass(frozen=True, slots=True)
class ImapMailboxMessageResult:
host: str
@@ -170,8 +183,8 @@ def _open_imap(config: ImapConfig) -> imaplib.IMAP4:
except Exception:
try:
client.logout() # type: ignore[possibly-undefined]
except Exception:
pass
except Exception as cleanup_exc:
_log_imap_cleanup_failure("opening connection", cleanup_exc)
raise
@@ -306,58 +319,200 @@ def test_imap_login(*, imap_config: ImapConfig) -> ImapLoginTestResult:
finally:
try:
client.logout()
except Exception:
pass
except Exception as cleanup_exc:
_log_imap_cleanup_failure("testing login", cleanup_exc)
def list_imap_folders(*, imap_config: ImapConfig) -> ImapFolderListResult:
def _mock_imap_folders(*, imap_config: ImapConfig) -> ImapFolderListResult:
host, port = _require_imap_config(imap_config)
records = list_records(limit=500)
folders = []
for item in MOCK_IMAP_FOLDERS:
name = str(item["name"])
count = sum(1 for record in records if _mock_folder_matches(record, name))
folders.append(ImapMailboxInfo(name=name, flags=list(item.get("flags") or []), message_count=count, unseen_count=None))
return ImapFolderListResult(
host=host,
port=port,
security=imap_config.security.value,
folders=folders,
detected_sent_folder="Sent",
)
def _list_imap_folders_on_client(
client: imaplib.IMAP4,
*,
host: str,
port: int,
security: str,
include_status: bool,
) -> ImapFolderListResult:
typ, data = client.list()
if typ != "OK":
raise ImapAppendError(f"IMAP folder listing failed: {data!r}", temporary=True)
parsed: list[tuple[str, set[str]]] = []
folders: list[ImapMailboxInfo] = []
for item in data or []:
extracted = _extract_mailbox_name(item)
if not extracted:
continue
name, flags = extracted
parsed.append((name, flags))
message_count, unseen_count = (
(None, None)
if not include_status or _has_folder_flag(flags, "noselect")
else _imap_folder_status(client, name)
)
folders.append(ImapMailboxInfo(name=name, flags=sorted(flags), message_count=message_count, unseen_count=unseen_count))
return ImapFolderListResult(
host=host,
port=port,
security=security,
folders=folders,
detected_sent_folder=_detect_sent_folder(parsed),
)
def list_imap_folders(*, imap_config: ImapConfig, include_status: bool = True) -> ImapFolderListResult:
"""Return folders visible through IMAP LIST and the best sent-folder guess."""
host, port = _require_imap_config(imap_config)
if is_mock_imap_host(imap_config.host):
records = list_records(limit=500)
folders = []
for item in MOCK_IMAP_FOLDERS:
name = str(item["name"])
count = sum(1 for record in records if _mock_folder_matches(record, name))
folders.append(ImapMailboxInfo(name=name, flags=list(item.get("flags") or []), message_count=count, unseen_count=None))
return ImapFolderListResult(
host=host,
port=port,
security=imap_config.security.value,
folders=folders,
detected_sent_folder="Sent",
)
return _mock_imap_folders(imap_config=imap_config)
client = _open_imap(imap_config)
try:
typ, data = client.list()
if typ != "OK":
raise ImapAppendError(f"IMAP folder listing failed: {data!r}", temporary=True)
parsed: list[tuple[str, set[str]]] = []
folders: list[ImapMailboxInfo] = []
for item in data or []:
extracted = _extract_mailbox_name(item)
if not extracted:
continue
name, flags = extracted
parsed.append((name, flags))
message_count, unseen_count = (None, None) if _has_folder_flag(flags, "noselect") else _imap_folder_status(client, name)
folders.append(ImapMailboxInfo(name=name, flags=sorted(flags), message_count=message_count, unseen_count=unseen_count))
return ImapFolderListResult(
return _list_imap_folders_on_client(
client,
host=host,
port=port,
security=imap_config.security.value,
folders=folders,
detected_sent_folder=_detect_sent_folder(parsed),
include_status=include_status,
)
finally:
try:
client.logout()
except Exception:
pass
except Exception as cleanup_exc:
_log_imap_cleanup_failure("listing folders", cleanup_exc)
def _preferred_mailbox_folder(folders: list[ImapMailboxInfo], requested: str | None, detected_sent_folder: str | None) -> str:
folder_names = {folder.name for folder in folders}
requested = (requested or "").strip()
if requested and requested in folder_names:
return requested
if "INBOX" in folder_names:
return "INBOX"
if detected_sent_folder and detected_sent_folder in folder_names:
return detected_sent_folder
return folders[0].name if folders else (requested or "INBOX")
def load_imap_mailbox_bootstrap(
*,
imap_config: ImapConfig,
folder: str = "INBOX",
limit: int = 50,
offset: int = 0,
include_folder_status: bool = False,
) -> ImapMailboxBootstrapResult:
"""Load folders and one message page through a single IMAP connection."""
host, port = _require_imap_config(imap_config)
if is_mock_imap_host(imap_config.host):
folders = _mock_imap_folders(imap_config=imap_config)
selected_folder = _preferred_mailbox_folder(folders.folders, folder, folders.detected_sent_folder)
messages = list_imap_messages(imap_config=imap_config, folder=selected_folder, limit=limit, offset=offset)
return ImapMailboxBootstrapResult(folders=folders, messages=messages)
client = _open_imap(imap_config)
try:
folders = _list_imap_folders_on_client(
client,
host=host,
port=port,
security=imap_config.security.value,
include_status=include_folder_status,
)
selected_folder = _preferred_mailbox_folder(folders.folders, folder, folders.detected_sent_folder)
selected_folder, limit, offset = _normalize_mailbox_page(folder=selected_folder, limit=limit, offset=offset)
messages = _list_imap_messages_on_client(
client,
host=host,
port=port,
security=imap_config.security.value,
folder=selected_folder,
limit=limit,
offset=offset,
after_uid=None,
expected_uidvalidity=None,
)
return ImapMailboxBootstrapResult(folders=folders, messages=messages)
finally:
try:
client.logout()
except Exception as cleanup_exc:
_log_imap_cleanup_failure("bootstrapping mailbox", cleanup_exc)
def _list_imap_messages_on_client(
client: imaplib.IMAP4,
*,
host: str,
port: int,
security: str,
folder: str,
limit: int,
offset: int,
after_uid: str | None,
expected_uidvalidity: str | None,
) -> ImapMailboxMessageListResult:
total_count, uidvalidity = _select_readonly(client, folder)
if after_uid is None and expected_uidvalidity is None:
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=security,
folder=folder,
messages=messages,
total_count=total_count,
offset=offset,
limit=limit,
uidvalidity=None,
cursor_reset=False,
)
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=security,
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,
)
def _has_folder_flag(flags: set[str], flag: str) -> bool:
@@ -691,9 +846,7 @@ def list_imap_messages(
"""List mailbox messages without mutating read/unread state."""
host, port = _require_imap_config(imap_config)
folder = (folder or "INBOX").strip() or "INBOX"
limit = max(1, min(limit, 100))
offset = max(0, offset)
folder, limit, offset = _normalize_mailbox_page(folder=folder, limit=limit, offset=offset)
if is_mock_imap_host(imap_config.host):
records = [record for record in list_records(limit=500) if _mock_folder_matches(record, folder)]
cursor_reset = False
@@ -736,6 +889,21 @@ def list_imap_messages(
client = _open_imap(imap_config)
try:
total_count, uidvalidity = _select_readonly(client, folder)
if after_uid is None and expected_uidvalidity is None:
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,
uidvalidity=None,
cursor_reset=False,
)
uids = _search_message_uids(client)
cursor_reset = False
effective_after_uid = after_uid
@@ -766,8 +934,12 @@ def list_imap_messages(
finally:
try:
client.logout()
except Exception:
pass
except Exception as cleanup_exc:
_log_imap_cleanup_failure("listing messages", cleanup_exc)
def _normalize_mailbox_page(*, folder: str | None, limit: int, offset: int) -> tuple[str, int, int]:
return (folder or "INBOX").strip() or "INBOX", max(1, min(limit, 100)), max(0, offset)
def _paged_descending_sequences(total_count: int, *, offset: int, limit: int) -> list[str]:
@@ -809,8 +981,8 @@ def get_imap_message(*, imap_config: ImapConfig, folder: str, uid: str) -> ImapM
finally:
try:
client.logout()
except Exception:
pass
except Exception as cleanup_exc:
_log_imap_cleanup_failure("reading message", cleanup_exc)
def append_message_to_sent(
@@ -867,5 +1039,5 @@ def append_message_to_sent(
if client is not None:
try:
client.logout()
except Exception:
pass
except Exception as cleanup_exc:
_log_imap_cleanup_failure("appending sent message", cleanup_exc)