267 lines
8.5 KiB
Python
267 lines
8.5 KiB
Python
"""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)
|