"""Adapter that delegates to mock_mail or imap_client based on USE_MOCK_MAIL config. Import this module instead of mock_mail/imap_client directly. """ from __future__ import annotations from flask import current_app, g from .mock_mail import ( Message, Thread, thread_key, # non-IMAP state (always from mock/db): all_folders, folder_title, folder_exists as _mock_folder_exists, list_signatures, default_signature, list_shared_for, list_groups, get_group, delete_group, list_rules, delete_rule, get_autoreply, # mock CRUD kept for Phase 3 (rules/groups/signatures still in-memory): add_signature, update_signature, delete_signature, folder_counts as _mock_counts, list_threads as _mock_threads, get_message as _mock_get_message, toggle_star as _mock_star, toggle_important as _mock_important, move_message as _mock_move, delete_message as _mock_delete, mark_all_read as _mock_mark_all_read, bulk_mark_read as _mock_bulk_read, bulk_delete as _mock_bulk_delete, bulk_move as _mock_bulk_move, CURRENT_USER_EMAIL as _MOCK_USER, ) def _use_mock() -> bool: return current_app.config.get("USE_MOCK_MAIL", True) def _user() -> str: """Current user email — from session (Phase 7) or mock default.""" return getattr(g, "user_email", _MOCK_USER) def _password() -> str | None: return getattr(g, "user_password", None) # ── Read operations ─────────────────────────────────────────────────────── def get_list_threads(folder_key: str, filter_: str | None) -> list[Thread]: if _use_mock(): return _mock_threads(folder_key, filter_) from . import imap_client return imap_client.list_threads(folder_key, filter_, _user(), _password()) def get_message(uid: int, folder_key: str = "inbox") -> Message | None: if _use_mock(): return _mock_get_message(uid) from . import imap_client return imap_client.get_message(uid, folder_key, _user(), _password()) def get_folder_counts() -> dict[str, int]: if _use_mock(): return _mock_counts() from . import imap_client return imap_client.folder_counts(_user(), _password()) def check_folder_exists(folder_key: str) -> bool: if _use_mock(): return _mock_folder_exists(folder_key) from . import imap_client return imap_client.folder_exists(folder_key, _user(), _password()) # ── Write operations ────────────────────────────────────────────────────── def do_toggle_star(uid: int, folder_key: str = "inbox") -> bool: if _use_mock(): return _mock_star(uid) from . import imap_client return imap_client.toggle_star(uid, folder_key, _user(), _password()) def do_toggle_important(uid: int, folder_key: str = "inbox") -> bool: if _use_mock(): return _mock_important(uid) from . import imap_client return imap_client.toggle_important(uid, folder_key, _user(), _password()) def do_move(uid: int, from_folder: str, to_folder: str) -> None: if _use_mock(): _mock_move(uid, to_folder) else: from . import imap_client imap_client.move_message(uid, from_folder, to_folder, _user(), _password()) def do_delete(uid: int, folder_key: str) -> None: if _use_mock(): _mock_delete(uid) else: from . import imap_client imap_client.delete_message(uid, folder_key, _user(), _password()) def do_mark_all_read(folder_key: str) -> None: if _use_mock(): _mock_mark_all_read(folder_key) else: from . import imap_client imap_client.mark_all_read(folder_key, _user(), _password()) def do_bulk_mark_read(uids: list[int], folder_key: str) -> None: if _use_mock(): _mock_bulk_read(uids) else: from . import imap_client imap_client.bulk_mark_read(uids, folder_key, _user(), _password()) def do_bulk_delete(uids: list[int], folder_key: str) -> None: if _use_mock(): _mock_bulk_delete(uids) else: from . import imap_client imap_client.bulk_delete(uids, folder_key, _user(), _password()) def get_contacts() -> list[dict]: """Address book: all DMS mailboxes + aliases + emails from inbox/sent + group members.""" seen: dict[str, str] = {} def add(email: str, name: str = ""): if not email or "@" not in email: return e = email.lower().strip() if e not in seen or (name and not seen[e]): seen[e] = name or "" # Server-side mailboxes / aliases (always included). try: from . import dms_config for mb in dms_config.list_mailboxes(): add(mb.email) for al in dms_config.list_aliases(): add(al["from"]) add(al["to"]) except Exception: pass # Mailing groups members from store. try: from . import store owner = _user() for g in store.list_groups(owner): for m in g.members: add(m) for sm in store.list_shared_for(owner): add(sm.email) for m in sm.members: add(m) except Exception: pass # Real IMAP contacts (From/To headers). if not _use_mock(): try: from . import imap_client for c in imap_client.list_contacts(_user(), _password()): add(c["email"], c.get("name", "")) except Exception: pass return [{"email": e, "name": n} for e, n in sorted(seen.items())] def do_save_draft( from_email: str, to_emails: list[str], cc_emails: list[str], bcc_emails: list[str], subject: str, body_html: str, body_text: str = "", important: bool = False, draft_uid: int | None = None, ) -> int | None: """Save (or replace) a draft. Returns new IMAP UID or None in mock mode.""" if _use_mock(): return None from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.utils import formatdate, make_msgid from . import imap_client msg = MIMEMultipart("alternative") msg["From"] = from_email msg["To"] = ", ".join(to_emails) if cc_emails: msg["Cc"] = ", ".join(cc_emails) msg["Subject"] = subject msg["Date"] = formatdate() msg["Message-ID"] = make_msgid() if important: msg["X-Priority"] = "1" if body_text: msg.attach(MIMEText(body_text, "plain", "utf-8")) msg.attach(MIMEText(body_html, "html", "utf-8")) _pass = _password() or current_app.config.get("IMAP_PASSWORD", "admin123") return imap_client.append_draft(msg.as_bytes(), draft_uid, _user(), _pass) def do_bulk_move(uids: list[int], from_folder: str, to_folder: str) -> None: if _use_mock(): _mock_bulk_move(uids, to_folder) else: from . import imap_client imap_client.bulk_move(uids, from_folder, to_folder, _user(), _password()) def do_send( from_email: str, to_emails: list[str], cc_emails: list[str], bcc_emails: list[str], subject: str, body_html: str, body_text: str = "", important: bool = False, attachments: list | None = None, ) -> None: if _use_mock(): # In mock mode, just add to sent folder from datetime import datetime from .mock_mail import _messages, _uid_seq from . import mock_mail as mm uid = next(mm._uid_seq) mm._messages[uid] = Message( uid=uid, folder="sent", from_name=from_email, from_email=from_email, to=", ".join(to_emails), subject=subject, preview=body_text[:100] or "Отправлено", body_html=body_html, date=datetime.now(), ) else: from . import smtp_sender, imap_client _pass = _password() or current_app.config.get("IMAP_PASSWORD", "admin123") raw = smtp_sender.send_message( from_email=from_email, password=_pass, to_emails=to_emails, cc_emails=cc_emails, bcc_emails=bcc_emails, subject=subject, body_html=body_html, body_text=body_text, important=important, attachments=attachments, ) try: imap_client.append_to_folder("sent", raw, _user(), _pass, flags=[r"\Seen"]) except Exception: current_app.logger.warning("APPEND to Sent failed", exc_info=True)