Agent Zero
fb92acb52b
- If mail has deleted_at and is still on IMAP server: - Check sync queue for pending delete operation - If pending delete exists: our IMAP delete failed, keep deleted, retry via queue - If no pending delete: another program restored the mail, restore in DB - This enables true bidirectional sync like Thunderbird/Outlook
2950 lines
106 KiB
Python
2950 lines
106 KiB
Python
"""Service layer for the Mail plugin: encryption, IMAP sync, SMTP send, rules, vacation, PGP."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import json
|
||
import logging
|
||
import mimetypes
|
||
import os
|
||
import re
|
||
import uuid
|
||
from datetime import UTC, datetime, timedelta
|
||
from email import message_from_bytes
|
||
from email.message import EmailMessage
|
||
from email.utils import formataddr, formatdate, make_msgid
|
||
|
||
import aiofiles
|
||
import aioimaplib
|
||
import aiosmtplib
|
||
import nh3
|
||
import pgpy
|
||
from cryptography.fernet import Fernet
|
||
from cryptography.hazmat.primitives import hashes
|
||
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
||
from sqlalchemy import and_, func, or_, select, text
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.config import settings
|
||
from app.core.notifications import create_notification
|
||
from app.plugins.builtins.mail.models import (
|
||
Mail,
|
||
MailAccount,
|
||
MailAttachment,
|
||
MailFolder,
|
||
MailLabel,
|
||
MailLabelAssignment,
|
||
MailRule,
|
||
MailSignature,
|
||
MailTemplate,
|
||
VacationSentLog,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ─── Attachment Storage Helpers ───
|
||
|
||
MAX_ATTACHMENT_SIZE = 25 * 1024 * 1024 # 25 MB
|
||
|
||
|
||
def _decode_mime_filename(filename: str) -> str:
|
||
"""Decode MIME-encoded filename, handling =?charset?Q?...?= and =?charset?B?...?= patterns."""
|
||
if not filename:
|
||
return "attachment"
|
||
# If no MIME encoding pattern, return as-is
|
||
if "=?" not in filename:
|
||
return filename
|
||
try:
|
||
from email.header import decode_header, make_header
|
||
return str(make_header(decode_header(filename)))
|
||
except Exception:
|
||
# Fallback: manually decode Q-encoding if decode_header fails
|
||
# This handles cases where the email parser partially processed the filename
|
||
try:
|
||
import re
|
||
def decode_q(match):
|
||
charset, encoding, encoded = match.group(1), match.group(2).upper(), match.group(3)
|
||
if encoding == 'B':
|
||
import base64
|
||
decoded = base64.b64decode(encoded).decode(charset or 'utf-8', errors='replace')
|
||
else: # Q encoding
|
||
decoded = encoded.replace('_', ' ')
|
||
decoded = re.sub(r'=([0-9A-Fa-f]{2})', lambda m: chr(int(m.group(1), 16)), decoded)
|
||
decoded = decoded.encode('latin-1').decode(charset or 'utf-8', errors='replace')
|
||
return decoded
|
||
return re.sub(r'=\?([^?]+)\?([BbQq])\?([^?]*)\?=', decode_q, filename)
|
||
except Exception:
|
||
return filename
|
||
|
||
|
||
def _sanitize_filename(filename: str) -> str:
|
||
"""Sanitize a filename to prevent path traversal attacks."""
|
||
# Remove any path components — keep only the basename
|
||
filename = os.path.basename(filename or "attachment")
|
||
# Replace potentially dangerous characters
|
||
filename = re.sub(r"[^a-zA-Z0-9._-]", "_", filename)
|
||
# Ensure non-empty
|
||
if not filename:
|
||
filename = "attachment"
|
||
# Limit length
|
||
if len(filename) > 200:
|
||
name, ext = os.path.splitext(filename)
|
||
filename = name[:200 - len(ext)] + ext
|
||
return filename
|
||
|
||
|
||
def _attachment_storage_path(mail_id: uuid.UUID, filename: str) -> str:
|
||
"""Build the on-disk storage path for a mail attachment."""
|
||
safe_name = _sanitize_filename(filename)
|
||
return os.path.join(
|
||
settings.storage_path,
|
||
"mail_attachments",
|
||
str(mail_id),
|
||
safe_name,
|
||
)
|
||
|
||
|
||
async def _save_attachment_to_storage(
|
||
mail_id: uuid.UUID, filename: str, content: bytes
|
||
) -> str:
|
||
"""Save attachment content to disk and return the storage path."""
|
||
storage_path = _attachment_storage_path(mail_id, filename)
|
||
os.makedirs(os.path.dirname(storage_path), exist_ok=True)
|
||
async with aiofiles.open(storage_path, "wb") as f:
|
||
await f.write(content)
|
||
return storage_path
|
||
|
||
|
||
def attachment_to_response(att: MailAttachment) -> dict:
|
||
"""Convert a MailAttachment ORM object to a response dict."""
|
||
return {
|
||
"id": str(att.id),
|
||
"mail_id": str(att.mail_id),
|
||
"filename": att.filename,
|
||
"mime_type": att.mime_type,
|
||
"size_bytes": att.size_bytes,
|
||
"size": att.size_bytes, # alias for frontend compatibility
|
||
"content_id": att.content_id,
|
||
"is_inline": bool(att.content_id),
|
||
"dms_file_id": str(att.dms_file_id) if att.dms_file_id else None,
|
||
}
|
||
|
||
|
||
# ─── AES-256 Encryption (Fernet) ───
|
||
|
||
MAIL_ENCRYPTION_KEY = os.environ.get("MAIL_ENCRYPTION_KEY", "leocrm-mail-encryption-key-2024")
|
||
|
||
|
||
def _derive_key(password: str, salt: bytes = b"leocrm-mail-salt") -> bytes:
|
||
"""Derive a 32-byte Fernet key from a password using PBKDF2."""
|
||
kdf = PBKDF2HMAC(
|
||
algorithm=hashes.SHA256(),
|
||
length=32,
|
||
salt=salt,
|
||
iterations=480000,
|
||
)
|
||
return base64.urlsafe_b64encode(kdf.derive(password.encode()))
|
||
|
||
|
||
_fernet = Fernet(_derive_key(MAIL_ENCRYPTION_KEY))
|
||
|
||
|
||
def encrypt_password(plaintext: str) -> str:
|
||
"""Encrypt a password using AES-256 (Fernet). Returns base64 ciphertext."""
|
||
return _fernet.encrypt(plaintext.encode()).decode()
|
||
|
||
|
||
def decrypt_password(ciphertext: str) -> str:
|
||
"""Decrypt a password encrypted with encrypt_password."""
|
||
return _fernet.decrypt(ciphertext.encode()).decode()
|
||
|
||
|
||
# ─── HTML Sanitization (F-MAIL: no script tags) ───
|
||
|
||
|
||
def sanitize_html(raw_html: str) -> str:
|
||
"""Sanitize HTML using nh3 — removes script tags and dangerous attributes."""
|
||
if not raw_html:
|
||
return ""
|
||
return nh3.clean(
|
||
raw_html,
|
||
tags={
|
||
"a",
|
||
"b",
|
||
"br",
|
||
"div",
|
||
"em",
|
||
"h1",
|
||
"h2",
|
||
"h3",
|
||
"h4",
|
||
"h5",
|
||
"h6",
|
||
"hr",
|
||
"i",
|
||
"img",
|
||
"li",
|
||
"ol",
|
||
"p",
|
||
"span",
|
||
"strong",
|
||
"table",
|
||
"tbody",
|
||
"td",
|
||
"th",
|
||
"thead",
|
||
"tr",
|
||
"u",
|
||
"ul",
|
||
"blockquote",
|
||
"code",
|
||
"pre",
|
||
"font",
|
||
"center",
|
||
},
|
||
attributes={
|
||
"a": {"href", "title", "target"},
|
||
"img": {"src", "alt", "width", "height"},
|
||
"span": {"style"},
|
||
"div": {"style"},
|
||
"font": {"color", "size", "face"},
|
||
"p": {"style"},
|
||
"td": {"style"},
|
||
"th": {"style"},
|
||
},
|
||
)
|
||
|
||
|
||
# ─── IMAP Quota Parser ───
|
||
|
||
|
||
def _parse_imap_quota_response(response) -> int | None:
|
||
"""Parse an IMAP GETQUOTAROOT response and return usage percentage.
|
||
|
||
Looks for lines like:
|
||
* QUOTA "INBOX" (STORAGE 12345 67890)
|
||
where 12345 is used and 67890 is limit.
|
||
Returns the usage percentage as an int, or None if parsing fails.
|
||
"""
|
||
try:
|
||
lines = response.lines if hasattr(response, "lines") else response
|
||
for line in lines:
|
||
if isinstance(line, (bytes, bytearray)):
|
||
line = line.decode("utf-8", errors="replace")
|
||
if not isinstance(line, str):
|
||
continue
|
||
if "QUOTA" not in line.upper():
|
||
continue
|
||
# Extract the parenthesized storage values
|
||
# Pattern: (STORAGE <used> <limit>)
|
||
match = re.search(r"\(STORAGE\s+(\d+)\s+(\d+)\)", line, re.IGNORECASE)
|
||
if match:
|
||
used = int(match.group(1))
|
||
limit = int(match.group(2))
|
||
if limit > 0:
|
||
return int((used / limit) * 100)
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
# ─── Mail Account Service ───
|
||
|
||
|
||
async def create_mail_account(
|
||
db: AsyncSession, *, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict
|
||
) -> MailAccount:
|
||
"""Create a new mail account with encrypted password."""
|
||
account = MailAccount(
|
||
tenant_id=tenant_id,
|
||
user_id=user_id,
|
||
email_address=data["email_address"],
|
||
display_name=data.get("display_name", ""),
|
||
imap_host=data["imap_host"],
|
||
imap_port=data.get("imap_port", 993),
|
||
imap_ssl=data.get("imap_ssl", True),
|
||
smtp_host=data["smtp_host"],
|
||
smtp_port=data.get("smtp_port", 587),
|
||
smtp_tls=data.get("smtp_tls", True),
|
||
username=data.get("username") or data["email_address"],
|
||
encrypted_password=encrypt_password(data["password"]),
|
||
is_shared=data.get("is_shared", False),
|
||
is_active=True,
|
||
sent_folder_imap_name=data.get("sent_folder_imap_name"),
|
||
drafts_folder_imap_name=data.get("drafts_folder_imap_name"),
|
||
spam_folder_imap_name=data.get("spam_folder_imap_name"),
|
||
trash_folder_imap_name=data.get("trash_folder_imap_name"),
|
||
)
|
||
db.add(account)
|
||
await db.flush()
|
||
|
||
# Create INBOX first so subfolders can reference it as parent
|
||
inbox_folder = MailFolder(
|
||
tenant_id=tenant_id,
|
||
account_id=account.id,
|
||
name="Posteingang",
|
||
imap_name="INBOX",
|
||
is_standard=True,
|
||
)
|
||
db.add(inbox_folder)
|
||
await db.flush()
|
||
|
||
# Create standard subfolders under INBOX (IMAP server uses '.' delimiter)
|
||
for fname, imap_name in [
|
||
("Gesendet", "INBOX.Sent"),
|
||
("Entwürfe", "INBOX.Drafts"),
|
||
("Papierkorb", "INBOX.Trash"),
|
||
("Spam", "INBOX.spam"),
|
||
]:
|
||
folder = MailFolder(
|
||
tenant_id=tenant_id,
|
||
account_id=account.id,
|
||
name=fname,
|
||
imap_name=imap_name,
|
||
parent_id=inbox_folder.id,
|
||
is_standard=True,
|
||
)
|
||
db.add(folder)
|
||
await db.flush()
|
||
return account
|
||
|
||
|
||
async def update_mail_account(db: AsyncSession, account: MailAccount, data: dict) -> MailAccount:
|
||
"""Update a mail account, encrypting password if changed."""
|
||
field_map = {
|
||
"email": "email_address",
|
||
"email_address": "email_address",
|
||
"display_name": "display_name",
|
||
"imap_host": "imap_host",
|
||
"imap_port": "imap_port",
|
||
"imap_ssl": "imap_ssl",
|
||
"smtp_host": "smtp_host",
|
||
"smtp_port": "smtp_port",
|
||
"smtp_tls": "smtp_tls",
|
||
"username": "username",
|
||
"is_shared": "is_shared",
|
||
"is_active": "is_active",
|
||
"sent_folder_imap_name": "sent_folder_imap_name",
|
||
"drafts_folder_imap_name": "drafts_folder_imap_name",
|
||
"spam_folder_imap_name": "spam_folder_imap_name",
|
||
"trash_folder_imap_name": "trash_folder_imap_name",
|
||
}
|
||
for api_field, model_field in field_map.items():
|
||
if api_field in data and data[api_field] is not None:
|
||
setattr(account, model_field, data[api_field])
|
||
if "password" in data and data["password"] is not None:
|
||
account.encrypted_password = encrypt_password(data["password"])
|
||
await db.flush()
|
||
await db.refresh(account)
|
||
return account
|
||
|
||
|
||
async def get_account_password(account: MailAccount) -> str:
|
||
"""Decrypt and return the account password (internal use only)."""
|
||
return decrypt_password(account.encrypted_password)
|
||
|
||
|
||
def account_to_response(account: MailAccount) -> dict:
|
||
"""Convert MailAccount to response dict, NEVER including password."""
|
||
return {
|
||
"id": str(account.id),
|
||
"email": account.email_address,
|
||
"email_address": account.email_address,
|
||
"display_name": account.display_name,
|
||
"imap_host": account.imap_host,
|
||
"imap_port": account.imap_port,
|
||
"imap_ssl": account.imap_ssl,
|
||
"smtp_host": account.smtp_host,
|
||
"smtp_port": account.smtp_port,
|
||
"smtp_tls": account.smtp_tls,
|
||
"username": account.username,
|
||
"is_shared": account.is_shared,
|
||
"is_active": account.is_active,
|
||
"sent_folder_imap_name": account.sent_folder_imap_name,
|
||
"drafts_folder_imap_name": account.drafts_folder_imap_name,
|
||
"spam_folder_imap_name": account.spam_folder_imap_name,
|
||
"trash_folder_imap_name": account.trash_folder_imap_name,
|
||
"created_at": account.created_at,
|
||
"updated_at": account.updated_at,
|
||
}
|
||
|
||
|
||
# ─── IMAP Sync Service (F-MAIL-01) ───
|
||
|
||
|
||
# German display names for standard IMAP folders.
|
||
# Keys are full IMAP paths (with dot delimiter) and also bare leaf names.
|
||
IMAP_FOLDER_NAME_MAP = {
|
||
"INBOX": "Posteingang",
|
||
"INBOX.Sent": "Gesendet",
|
||
"INBOX.Drafts": "Entwürfe",
|
||
"INBOX.Trash": "Papierkorb",
|
||
"INBOX.Archive": "Archiv",
|
||
"INBOX.spam": "Spam",
|
||
"INBOX.Spam": "Spam",
|
||
# Bare names (fallback for servers that don't nest under INBOX)
|
||
"Sent": "Gesendet",
|
||
"Sent Items": "Gesendet",
|
||
"Sent Mail": "Gesendet",
|
||
"Drafts": "Entwürfe",
|
||
"Draft": "Entwürfe",
|
||
"Spam": "Spam",
|
||
"Junk": "Spam",
|
||
"Junk Email": "Spam",
|
||
"Junk E-mail": "Spam",
|
||
"Trash": "Papierkorb",
|
||
"Deleted": "Papierkorb",
|
||
"Deleted Items": "Papierkorb",
|
||
"Archive": "Archiv",
|
||
}
|
||
|
||
# Standard IMAP folders that are always considered "standard"
|
||
STANDARD_IMAP_FOLDERS = {
|
||
"INBOX",
|
||
"INBOX.Sent", "INBOX.Drafts", "INBOX.Trash", "INBOX.Archive",
|
||
"INBOX.spam", "INBOX.Spam",
|
||
"Sent", "Sent Items", "Sent Mail",
|
||
"Drafts", "Draft",
|
||
"Spam", "Junk", "Junk Email", "Junk E-mail",
|
||
"Trash", "Deleted", "Deleted Items",
|
||
"Archive",
|
||
}
|
||
|
||
MAX_EMAILS_PER_FOLDER = 2000
|
||
|
||
|
||
def _get_german_folder_name(imap_name: str) -> str:
|
||
"""Return the German display name for a standard IMAP folder, or the original name."""
|
||
if imap_name in IMAP_FOLDER_NAME_MAP:
|
||
return IMAP_FOLDER_NAME_MAP[imap_name]
|
||
# Try the leaf component (after last dot) for unknown nested folders
|
||
leaf = imap_name.rsplit(".", 1)[-1] if "." in imap_name else imap_name
|
||
if leaf in IMAP_FOLDER_NAME_MAP:
|
||
return IMAP_FOLDER_NAME_MAP[leaf]
|
||
return imap_name
|
||
|
||
|
||
def _parse_imap_list_response(response) -> tuple[list[tuple[str, str]], str]:
|
||
"""Parse IMAP LIST response into list of (flags, folder_name) tuples.
|
||
|
||
Handles both "/" and "." delimiters. The IMAP LIST response format is:
|
||
* LIST (\\HasChildren) "." "INBOX"
|
||
* LIST (\\HasNoChildren) "." "INBOX.Sent"
|
||
|
||
Returns (folders, delimiter) where delimiter is the hierarchy separator
|
||
detected from the LIST response (defaults to '.' if not found).
|
||
"""
|
||
folders: list[tuple[str, str]] = []
|
||
delimiter = '.'
|
||
lines = response.lines if hasattr(response, 'lines') else response
|
||
for line in lines:
|
||
if isinstance(line, (bytes, bytearray)):
|
||
text = line.decode('utf-8', errors='replace')
|
||
elif isinstance(line, str):
|
||
text = line
|
||
else:
|
||
continue
|
||
if 'LIST' not in text:
|
||
continue
|
||
# Extract quoted segments — the delimiter is the first quoted string,
|
||
# the folder name is the second.
|
||
parts = text.split('"')
|
||
if len(parts) >= 4:
|
||
delimiter = parts[1]
|
||
folder_name = parts[3]
|
||
flags = parts[0] if parts[0] else ''
|
||
folders.append((flags, folder_name))
|
||
elif len(parts) >= 2:
|
||
folder_name = parts[-2] if len(parts) >= 2 else ''
|
||
if folder_name:
|
||
folders.append(('', folder_name))
|
||
return folders, delimiter
|
||
|
||
|
||
def _build_folder_hierarchy(
|
||
imap_folders: list[tuple[str, str]],
|
||
account_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
existing_folders: dict[str, MailFolder],
|
||
delimiter: str = '.',
|
||
folder_mapping: dict[str, str | None] | None = None,
|
||
) -> list[MailFolder]:
|
||
"""Create or update MailFolder records from IMAP LIST response.
|
||
|
||
Handles delimiter-separated hierarchies (e.g. INBOX.Sent → parent=INBOX).
|
||
Updates existing folders in-place (name, is_standard, parent_id) so
|
||
that stale DB records with wrong imap_name values get corrected.
|
||
"""
|
||
result: list[MailFolder] = []
|
||
|
||
# Build reverse mapping: imap_name -> standard type (sent/drafts/spam/trash)
|
||
mapping_by_imap: dict[str, str] = {}
|
||
if folder_mapping:
|
||
for std_type, imap_name_val in folder_mapping.items():
|
||
if imap_name_val:
|
||
mapping_by_imap[imap_name_val] = std_type
|
||
|
||
# First pass: create or update folder records
|
||
for flags, imap_name in imap_folders:
|
||
if not imap_name:
|
||
continue
|
||
|
||
display_name = _get_german_folder_name(imap_name)
|
||
is_standard = imap_name in STANDARD_IMAP_FOLDERS
|
||
|
||
# If account has explicit folder mapping, mark mapped folders as standard
|
||
if imap_name in mapping_by_imap:
|
||
is_standard = True
|
||
std_type = mapping_by_imap[imap_name]
|
||
if std_type == "sent":
|
||
display_name = "Gesendet"
|
||
elif std_type == "drafts":
|
||
display_name = "Entwürfe"
|
||
elif std_type == "spam":
|
||
display_name = "Spam"
|
||
elif std_type == "trash":
|
||
display_name = "Papierkorb"
|
||
|
||
if imap_name in existing_folders:
|
||
folder = existing_folders[imap_name]
|
||
folder.name = display_name
|
||
folder.is_standard = is_standard
|
||
result.append(folder)
|
||
else:
|
||
folder = MailFolder(
|
||
tenant_id=tenant_id,
|
||
account_id=account_id,
|
||
name=display_name,
|
||
imap_name=imap_name,
|
||
is_standard=is_standard,
|
||
)
|
||
result.append(folder)
|
||
|
||
# Second pass: set parent_id based on IMAP hierarchy
|
||
folder_by_imap_name = {f.imap_name: f for f in result}
|
||
for folder in result:
|
||
if delimiter in folder.imap_name:
|
||
parts = folder.imap_name.split(delimiter)
|
||
parent_imap = delimiter.join(parts[:-1])
|
||
if parent_imap in folder_by_imap_name:
|
||
parent = folder_by_imap_name[parent_imap]
|
||
folder.parent_id = parent.id # may be None for new folders; fixed after flush
|
||
else:
|
||
folder.parent_id = None
|
||
else:
|
||
folder.parent_id = None
|
||
|
||
return result
|
||
|
||
|
||
async def imap_sync_folder(
|
||
db: AsyncSession,
|
||
folder_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
) -> dict:
|
||
"""Incremental sync: only fetch NEW mails since the highest known UID.
|
||
|
||
Like real mail clients: checks for new UIDs only, doesn't re-fetch everything.
|
||
Also checks for deleted mails on the server (vanished UIDs).
|
||
"""
|
||
import logging
|
||
logger = logging.getLogger(__name__)
|
||
|
||
folder = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(MailFolder.id == folder_id, MailFolder.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not folder:
|
||
return {"synced": 0, "error": "Folder not found"}
|
||
|
||
account = (
|
||
await db.execute(
|
||
select(MailAccount).where(
|
||
and_(MailAccount.id == folder.account_id, MailAccount.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not account:
|
||
return {"synced": 0, "error": "Account not found"}
|
||
|
||
if not account.is_active:
|
||
return {"synced": 0, "error": "Account is not active"}
|
||
|
||
# Get the highest known UID in this folder from DB
|
||
max_uid_result = (
|
||
await db.execute(
|
||
select(func.max(Mail.imap_uid)).where(
|
||
and_(
|
||
Mail.folder_id == folder_id,
|
||
Mail.tenant_id == tenant_id,
|
||
Mail.imap_uid.is_not(None),
|
||
)
|
||
)
|
||
)
|
||
).scalar()
|
||
|
||
password = await get_account_password(account)
|
||
client = None
|
||
try:
|
||
client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
|
||
await client.wait_hello_from_server()
|
||
await client.login(account.username, password)
|
||
|
||
select_resp = await client.select(folder.imap_name)
|
||
if select_resp.result != 'OK':
|
||
return {"synced": 0, "error": f"Cannot select folder {folder.imap_name}"}
|
||
|
||
# Incremental: only search for UIDs greater than our highest known UID
|
||
if max_uid_result:
|
||
try:
|
||
max_uid_int = int(max_uid_result)
|
||
search_resp = await client.uid_search(f'UID {max_uid_int + 1}:*')
|
||
except (ValueError, TypeError):
|
||
search_resp = await client.uid_search('ALL')
|
||
else:
|
||
# First sync — get last 50
|
||
search_resp = await client.uid_search('ALL')
|
||
|
||
uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b''
|
||
if isinstance(uids_raw, (bytes, bytearray)):
|
||
uids = uids_raw.split()
|
||
else:
|
||
uids = []
|
||
|
||
# First sync: limit to last 50
|
||
if not max_uid_result and len(uids) > MAX_EMAILS_PER_FOLDER:
|
||
uids = uids[-MAX_EMAILS_PER_FOLDER:]
|
||
|
||
synced_count = 0
|
||
for uid in uids:
|
||
uid_str = uid.decode() if isinstance(uid, bytes) else str(uid)
|
||
|
||
# Skip if we already have this UID (can happen with UID * search)
|
||
existing_mail = (
|
||
await db.execute(
|
||
select(Mail).where(
|
||
and_(
|
||
Mail.account_id == account.id,
|
||
Mail.folder_id == folder.id,
|
||
Mail.imap_uid == uid_str,
|
||
Mail.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
# Also check by message_id — but ONLY within the same folder
|
||
# to prevent cross-folder moves. A mail in Trash and INBOX with
|
||
# the same message_id are separate copies (like real mail clients).
|
||
if not existing_mail:
|
||
# Fetch headers first to get Message-ID
|
||
fetch_hdr = await client.uid('fetch', uid_str, '(BODY.PEEK[HEADER.FIELDS (MESSAGE-ID)])')
|
||
hdr_raw = None
|
||
for line in (fetch_hdr.lines if hasattr(fetch_hdr, 'lines') else fetch_hdr):
|
||
if isinstance(line, bytearray):
|
||
hdr_raw = bytes(line)
|
||
break
|
||
if hdr_raw:
|
||
hdr_msg = message_from_bytes(hdr_raw)
|
||
peek_msg_id = hdr_msg.get("Message-ID", "")
|
||
if peek_msg_id:
|
||
existing_mail = (
|
||
await db.execute(
|
||
select(Mail).where(
|
||
and_(
|
||
Mail.account_id == account.id,
|
||
Mail.message_id == peek_msg_id,
|
||
Mail.folder_id == folder.id,
|
||
Mail.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
if existing_mail:
|
||
# Bidirectional sync logic:
|
||
# - Mail has deleted_at AND is still on IMAP server
|
||
# - If pending delete in sync queue → our IMAP delete failed, DON'T restore
|
||
# - If no pending delete → another program restored it, DO restore
|
||
# - Mail has no deleted_at → normal update
|
||
if existing_mail.deleted_at is not None:
|
||
# Check if there's a pending delete in the sync queue
|
||
from app.plugins.builtins.mail.models import MailSyncQueue
|
||
pending_delete = (
|
||
await db.execute(
|
||
select(MailSyncQueue).where(
|
||
and_(
|
||
MailSyncQueue.mail_id == existing_mail.id,
|
||
MailSyncQueue.operation == "delete",
|
||
MailSyncQueue.status == "pending",
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if pending_delete:
|
||
# Our IMAP delete failed and is retrying — don't restore
|
||
logger.info("imap_sync_folder: skipping mail %s (deleted, pending IMAP delete in queue)", existing_mail.id)
|
||
continue
|
||
else:
|
||
# No pending delete → another program restored the mail → restore in DB
|
||
existing_mail.deleted_at = None
|
||
logger.info("imap_sync_folder: restored mail %s (no pending delete, another program restored it)", existing_mail.id)
|
||
# Update folder_id and imap_uid if mail moved to a different folder
|
||
if existing_mail.folder_id != folder.id:
|
||
existing_mail.folder_id = folder.id
|
||
existing_mail.imap_uid = uid_str
|
||
elif not existing_mail.imap_uid:
|
||
existing_mail.imap_uid = uid_str
|
||
continue
|
||
|
||
# Fetch and parse the email
|
||
fetch_resp = await client.uid('fetch', uid_str, '(RFC822)')
|
||
raw_email = None
|
||
for line in (fetch_resp.lines if hasattr(fetch_resp, 'lines') else fetch_resp):
|
||
if isinstance(line, bytearray):
|
||
raw_email = bytes(line)
|
||
break
|
||
if raw_email is None:
|
||
continue
|
||
if isinstance(raw_email, str):
|
||
raw_email = raw_email.encode()
|
||
|
||
msg = message_from_bytes(raw_email)
|
||
body_text = ""
|
||
body_html = ""
|
||
attachments = []
|
||
|
||
if msg.is_multipart():
|
||
for part in msg.walk():
|
||
ct = part.get_content_type()
|
||
if ct == "text/plain":
|
||
payload = part.get_payload(decode=True)
|
||
if payload:
|
||
body_text = payload.decode("utf-8", errors="replace")
|
||
elif ct == "text/html":
|
||
payload = part.get_payload(decode=True)
|
||
if payload:
|
||
body_html = payload.decode("utf-8", errors="replace")
|
||
elif part.get_filename():
|
||
payload_bytes = part.get_payload(decode=True) or b""
|
||
attachments.append({
|
||
"filename": part.get_filename(),
|
||
"mime_type": ct,
|
||
"size": len(payload_bytes),
|
||
"content": payload_bytes,
|
||
})
|
||
else:
|
||
ct = msg.get_content_type()
|
||
payload = msg.get_payload(decode=True)
|
||
if payload:
|
||
decoded = payload.decode("utf-8", errors="replace")
|
||
if ct == "text/html":
|
||
body_html = decoded
|
||
else:
|
||
body_text = decoded
|
||
|
||
from email.header import decode_header, make_header
|
||
|
||
def _decode_mime_header(value: str) -> str:
|
||
if not value:
|
||
return ""
|
||
try:
|
||
return str(make_header(decode_header(value)))
|
||
except Exception:
|
||
return value
|
||
|
||
raw_msg_id = msg.get("Message-ID")
|
||
if raw_msg_id:
|
||
message_id = raw_msg_id
|
||
else:
|
||
message_id = f"generated-{account.id}-{folder.id}-{uid_str}"
|
||
subject = _decode_mime_header(msg.get("Subject", ""))
|
||
from_addr = _decode_mime_header(msg.get("From", ""))
|
||
to_addrs = _decode_mime_header(msg.get("To", ""))
|
||
cc_addrs = _decode_mime_header(msg.get("Cc", ""))
|
||
refs = msg.get("References", "")
|
||
in_reply_to = msg.get("In-Reply-To")
|
||
date_str = msg.get("Date", "")
|
||
|
||
received_at = datetime.now(UTC)
|
||
if date_str:
|
||
try:
|
||
from email.utils import parsedate_to_datetime
|
||
parsed = parsedate_to_datetime(date_str)
|
||
if parsed:
|
||
received_at = parsed.astimezone(UTC) if parsed.tzinfo else parsed.replace(tzinfo=UTC)
|
||
except Exception:
|
||
pass
|
||
|
||
thread_id = _compute_thread_id(message_id, refs, in_reply_to)
|
||
|
||
mail = Mail(
|
||
tenant_id=tenant_id,
|
||
account_id=account.id,
|
||
folder_id=folder.id,
|
||
imap_uid=uid_str,
|
||
message_id=message_id,
|
||
thread_id=thread_id,
|
||
in_reply_to=in_reply_to,
|
||
references_header=refs,
|
||
subject=subject,
|
||
from_address=from_addr,
|
||
to_addresses=to_addrs,
|
||
cc_addresses=cc_addrs,
|
||
body_text=body_text,
|
||
body_html=body_html,
|
||
body_html_sanitized=sanitize_html(body_html),
|
||
has_attachments=len(attachments) > 0,
|
||
size_bytes=len(raw_email),
|
||
received_at=received_at,
|
||
)
|
||
db.add(mail)
|
||
await db.flush()
|
||
synced_count += 1
|
||
|
||
# Save attachments
|
||
for att_data in attachments:
|
||
try:
|
||
raw_filename = att_data["filename"] or "attachment"
|
||
decoded_filename = _decode_mime_filename(raw_filename)
|
||
storage_path = await _save_attachment_to_storage(
|
||
mail.id, decoded_filename, att_data["content"]
|
||
)
|
||
attachment = MailAttachment(
|
||
tenant_id=tenant_id,
|
||
mail_id=mail.id,
|
||
filename=_sanitize_filename(decoded_filename),
|
||
mime_type=att_data["mime_type"],
|
||
size_bytes=att_data["size"],
|
||
storage_path=storage_path,
|
||
)
|
||
db.add(attachment)
|
||
except Exception as att_err:
|
||
logger.warning(f"Failed to save attachment for mail {mail.id}: {att_err}")
|
||
continue
|
||
|
||
# Update folder counts
|
||
total = (
|
||
await db.execute(
|
||
select(func.count()).select_from(Mail).where(
|
||
and_(
|
||
Mail.folder_id == folder_id,
|
||
Mail.tenant_id == tenant_id,
|
||
Mail.deleted_at.is_(None),
|
||
)
|
||
)
|
||
)
|
||
).scalar() or 0
|
||
unread = (
|
||
await db.execute(
|
||
select(func.count()).select_from(Mail).where(
|
||
and_(
|
||
Mail.folder_id == folder_id,
|
||
Mail.tenant_id == tenant_id,
|
||
Mail.deleted_at.is_(None),
|
||
Mail.is_seen.is_(False),
|
||
)
|
||
)
|
||
)
|
||
).scalar() or 0
|
||
folder.total_count = total
|
||
folder.unread_count = unread
|
||
|
||
await db.flush()
|
||
logger.info("imap_sync_folder: synced %d new mail(s) for folder %s", synced_count, folder.imap_name)
|
||
return {"synced": synced_count}
|
||
|
||
except Exception as exc:
|
||
logger.warning("imap_sync_folder: failed for folder %s: %s", folder_id, exc)
|
||
return {"synced": 0, "error": str(exc)}
|
||
finally:
|
||
if client is not None:
|
||
try:
|
||
await client.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
async def imap_sync_account(
|
||
db: AsyncSession,
|
||
account_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
) -> dict:
|
||
"""Sync mail folders and messages from IMAP server.
|
||
|
||
Syncs ALL folders from the IMAP server (not just INBOX).
|
||
For each folder, fetches the last MAX_EMAILS_PER_FOLDER emails
|
||
(sorted by date descending) to avoid timeouts on large mailboxes.
|
||
Creates/updates mail_folders records with German display names.
|
||
"""
|
||
account = (
|
||
await db.execute(
|
||
select(MailAccount).where(
|
||
and_(MailAccount.id == account_id, MailAccount.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not account:
|
||
return {"synced": 0, "error": "Account not found"}
|
||
|
||
if not account.is_active:
|
||
try:
|
||
await create_notification(
|
||
db, account.tenant_id, account.user_id,
|
||
"mail_account",
|
||
"Mail-Account deaktiviert",
|
||
f"Account {account.email_address} ist deaktiviert und wird nicht synchronisiert.",
|
||
)
|
||
await db.flush()
|
||
except Exception:
|
||
pass
|
||
return {"synced": 0, "error": "Account is not active"}
|
||
|
||
password = await get_account_password(account)
|
||
|
||
# ── IMAP connection ──
|
||
try:
|
||
client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
|
||
await client.wait_hello_from_server()
|
||
except Exception as e:
|
||
try:
|
||
await create_notification(
|
||
db, account.tenant_id, account.user_id,
|
||
"mail_error",
|
||
"IMAP-Verbindung fehlgeschlagen",
|
||
f"Account {account.email_address}: {e}",
|
||
)
|
||
await db.flush()
|
||
except Exception:
|
||
pass
|
||
return {"synced": 0, "error": f"IMAP connection failed: {e}"}
|
||
|
||
# ── IMAP login ──
|
||
try:
|
||
await client.login(account.username, password)
|
||
except Exception as e:
|
||
try:
|
||
await create_notification(
|
||
db, account.tenant_id, account.user_id,
|
||
"mail_auth",
|
||
"IMAP-Login fehlgeschlagen",
|
||
f"Account {account.email_address}: Passwort oder Anmeldedaten prüfen",
|
||
)
|
||
await db.flush()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
await client.logout()
|
||
except Exception:
|
||
pass
|
||
return {"synced": 0, "error": f"IMAP login failed: {e}"}
|
||
|
||
# ── Quota check (non-critical, not all servers support QUOTA) ──
|
||
try:
|
||
quota_resp = await client.getquotaroot('INBOX')
|
||
usage_pct = _parse_imap_quota_response(quota_resp)
|
||
if usage_pct is not None and usage_pct > 80:
|
||
if usage_pct > 95:
|
||
quota_title = "Postfach voll – keine neuen Mails empfangbar"
|
||
else:
|
||
quota_title = "Postfach fast voll"
|
||
try:
|
||
await create_notification(
|
||
db, account.tenant_id, account.user_id,
|
||
"mail_quota",
|
||
quota_title,
|
||
f"Account {account.email_address}: {usage_pct}% belegt",
|
||
)
|
||
await db.flush()
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
# 1) LIST all folders from IMAP server
|
||
list_response = await client.list('""', '"*"')
|
||
imap_folders, imap_delimiter = _parse_imap_list_response(list_response)
|
||
|
||
# If LIST returned nothing, fall back to standard folders (dot-delimited)
|
||
if not imap_folders:
|
||
imap_folders = [
|
||
('', 'INBOX'),
|
||
('', 'INBOX.Sent'),
|
||
('', 'INBOX.Drafts'),
|
||
('', 'INBOX.spam'),
|
||
('', 'INBOX.Trash'),
|
||
]
|
||
|
||
# 2) Load existing folders from DB for this account
|
||
existing_db_folders = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(MailFolder.account_id == account.id, MailFolder.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalars().all()
|
||
existing_by_imap: dict[str, MailFolder] = {
|
||
f.imap_name: f for f in existing_db_folders
|
||
}
|
||
|
||
# 2a) Migrate stale folder names: if DB has 'Sent' but IMAP returns
|
||
# 'INBOX.Sent', update the DB record's imap_name so it matches.
|
||
imap_names_from_server = {name for _, name in imap_folders if name}
|
||
for db_folder in existing_db_folders:
|
||
if db_folder.imap_name not in imap_names_from_server:
|
||
# Try matching by leaf component using detected delimiter
|
||
leaf = db_folder.imap_name.rsplit(imap_delimiter, 1)[-1]
|
||
for srv_name in imap_names_from_server:
|
||
srv_leaf = srv_name.rsplit(imap_delimiter, 1)[-1]
|
||
if srv_leaf.lower() == leaf.lower():
|
||
db_folder.imap_name = srv_name
|
||
existing_by_imap[srv_name] = db_folder
|
||
break
|
||
|
||
# 3) Create/update folders in DB
|
||
folder_mapping = {
|
||
"sent": account.sent_folder_imap_name,
|
||
"drafts": account.drafts_folder_imap_name,
|
||
"spam": account.spam_folder_imap_name,
|
||
"trash": account.trash_folder_imap_name,
|
||
}
|
||
db_folders = _build_folder_hierarchy(
|
||
imap_folders, account.id, tenant_id, existing_by_imap, imap_delimiter,
|
||
folder_mapping=folder_mapping,
|
||
)
|
||
for folder in db_folders:
|
||
if folder.id is None:
|
||
db.add(folder)
|
||
await db.flush()
|
||
|
||
# 3a) Re-set parent_id now that new folders have IDs after flush
|
||
folder_by_imap_post_flush = {f.imap_name: f for f in db_folders}
|
||
for folder in db_folders:
|
||
if imap_delimiter in folder.imap_name:
|
||
parts = folder.imap_name.split(imap_delimiter)
|
||
parent_imap = imap_delimiter.join(parts[:-1])
|
||
if parent_imap in folder_by_imap_post_flush:
|
||
parent = folder_by_imap_post_flush[parent_imap]
|
||
if parent.id:
|
||
folder.parent_id = parent.id
|
||
|
||
await db.flush()
|
||
|
||
# Build a map of imap_name -> folder_id for email sync
|
||
folder_by_imap = {f.imap_name: f for f in db_folders}
|
||
|
||
synced_count = 0
|
||
new_mails: list[dict] = []
|
||
|
||
# 4) Sync emails for each folder (limit to last 50 per folder)
|
||
for imap_name, folder in folder_by_imap.items():
|
||
try:
|
||
# Select the folder on the IMAP server — use the raw IMAP
|
||
# name without extra quoting (aioimaplib handles it)
|
||
select_resp = await client.select(imap_name)
|
||
if select_resp.result != 'OK':
|
||
continue
|
||
|
||
# Fetch UIDs and sort by date descending — get last 50
|
||
search_resp = await client.uid_search('ALL')
|
||
uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b''
|
||
if isinstance(uids_raw, (bytes, bytearray)):
|
||
uids = uids_raw.split()
|
||
else:
|
||
uids = []
|
||
|
||
# Limit to last MAX_EMAILS_PER_FOLDER UIDs
|
||
# UIDs are monotonically increasing, so the last ones are the newest
|
||
if len(uids) > MAX_EMAILS_PER_FOLDER:
|
||
uids = uids[-MAX_EMAILS_PER_FOLDER:]
|
||
|
||
for uid in uids:
|
||
uid_str = uid.decode() if isinstance(uid, bytes) else str(uid)
|
||
fetch_resp = await client.uid('fetch', uid_str, '(RFC822)')
|
||
raw_email = None
|
||
for line in (fetch_resp.lines if hasattr(fetch_resp, 'lines') else fetch_resp):
|
||
if isinstance(line, bytearray):
|
||
raw_email = bytes(line)
|
||
break
|
||
if raw_email is None:
|
||
continue
|
||
if isinstance(raw_email, str):
|
||
raw_email = raw_email.encode()
|
||
|
||
msg = message_from_bytes(raw_email)
|
||
body_text = ""
|
||
body_html = ""
|
||
attachments = []
|
||
|
||
if msg.is_multipart():
|
||
for part in msg.walk():
|
||
ct = part.get_content_type()
|
||
if ct == "text/plain":
|
||
payload = part.get_payload(decode=True)
|
||
if payload:
|
||
body_text = payload.decode("utf-8", errors="replace")
|
||
elif ct == "text/html":
|
||
payload = part.get_payload(decode=True)
|
||
if payload:
|
||
body_html = payload.decode("utf-8", errors="replace")
|
||
elif part.get_filename():
|
||
payload_bytes = part.get_payload(decode=True) or b""
|
||
content_id = part.get("Content-ID", None)
|
||
if content_id:
|
||
content_id = content_id.strip("<>")
|
||
attachments.append(
|
||
{
|
||
"filename": part.get_filename(),
|
||
"mime_type": ct,
|
||
"size": len(payload_bytes),
|
||
"content": payload_bytes,
|
||
"content_id": content_id,
|
||
}
|
||
)
|
||
else:
|
||
ct = msg.get_content_type()
|
||
payload = msg.get_payload(decode=True)
|
||
if payload:
|
||
decoded = payload.decode("utf-8", errors="replace")
|
||
if ct == "text/html":
|
||
body_html = decoded
|
||
else:
|
||
body_text = decoded
|
||
|
||
from email.header import decode_header, make_header
|
||
|
||
def _decode_mime_header(value: str) -> str:
|
||
"""Decode MIME encoded-words (=?UTF-8?Q?...?=) to readable text."""
|
||
if not value:
|
||
return ""
|
||
try:
|
||
return str(make_header(decode_header(value)))
|
||
except Exception:
|
||
return value
|
||
|
||
raw_msg_id = msg.get("Message-ID")
|
||
if raw_msg_id:
|
||
message_id = raw_msg_id
|
||
else:
|
||
# Deterministic ID so re-syncs don't create duplicates
|
||
message_id = f"generated-{account.id}-{folder.id}-{uid_str}"
|
||
subject = _decode_mime_header(msg.get("Subject", ""))
|
||
from_addr = _decode_mime_header(msg.get("From", ""))
|
||
to_addrs = _decode_mime_header(msg.get("To", ""))
|
||
cc_addrs = _decode_mime_header(msg.get("Cc", ""))
|
||
refs = msg.get("References", "")
|
||
in_reply_to = msg.get("In-Reply-To")
|
||
date_str = msg.get("Date", "")
|
||
|
||
# Parse date for received_at
|
||
received_at = datetime.now(UTC)
|
||
if date_str:
|
||
try:
|
||
from email.utils import parsedate_to_datetime
|
||
parsed = parsedate_to_datetime(date_str)
|
||
if parsed:
|
||
received_at = parsed.astimezone(UTC) if parsed.tzinfo else parsed.replace(tzinfo=UTC)
|
||
except Exception:
|
||
pass
|
||
|
||
# Compute thread_id from References/In-Reply-To
|
||
thread_id = _compute_thread_id(message_id, refs, in_reply_to)
|
||
|
||
# Dedup: first check by (account_id, folder_id, imap_uid),
|
||
# then fall back to (account_id, message_id) for mails synced
|
||
# before the imap_uid column existed.
|
||
existing_mail = (
|
||
await db.execute(
|
||
select(Mail).where(
|
||
and_(
|
||
Mail.account_id == account.id,
|
||
Mail.folder_id == folder.id,
|
||
Mail.imap_uid == uid_str,
|
||
Mail.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
if not existing_mail:
|
||
existing_mail = (
|
||
await db.execute(
|
||
select(Mail).where(
|
||
and_(
|
||
Mail.account_id == account.id,
|
||
Mail.message_id == message_id,
|
||
Mail.folder_id == folder.id,
|
||
Mail.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
if existing_mail:
|
||
# Bidirectional sync logic:
|
||
# - Mail has deleted_at AND is still on IMAP server
|
||
# - If pending delete in sync queue → our IMAP delete failed, DON'T restore
|
||
# - If no pending delete → another program restored it, DO restore
|
||
# - Mail has no deleted_at → normal update
|
||
if existing_mail.deleted_at is not None:
|
||
from app.plugins.builtins.mail.models import MailSyncQueue
|
||
pending_delete = (
|
||
await db.execute(
|
||
select(MailSyncQueue).where(
|
||
and_(
|
||
MailSyncQueue.mail_id == existing_mail.id,
|
||
MailSyncQueue.operation == "delete",
|
||
MailSyncQueue.status == "pending",
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if pending_delete:
|
||
logger.info("imap_sync_account: skipping mail %s (deleted, pending IMAP delete in queue)", existing_mail.id)
|
||
continue
|
||
else:
|
||
existing_mail.deleted_at = None
|
||
logger.info("imap_sync_account: restored mail %s (no pending delete, another program restored it)", existing_mail.id)
|
||
# Track folder moves: update folder_id and imap_uid if the
|
||
# mail now appears in a different IMAP folder.
|
||
if existing_mail.folder_id != folder.id:
|
||
existing_mail.folder_id = folder.id
|
||
existing_mail.imap_uid = uid_str
|
||
elif not existing_mail.imap_uid:
|
||
existing_mail.imap_uid = uid_str
|
||
|
||
# Save attachments for existing emails that have has_attachments but no records
|
||
if attachments and existing_mail.has_attachments:
|
||
existing_att_count = (
|
||
await db.execute(
|
||
select(func.count()).select_from(MailAttachment).where(
|
||
and_(
|
||
MailAttachment.mail_id == existing_mail.id,
|
||
MailAttachment.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
).scalar() or 0
|
||
if existing_att_count == 0:
|
||
for att_data in attachments:
|
||
try:
|
||
raw_filename = att_data["filename"] or "attachment"
|
||
decoded_filename = _decode_mime_filename(raw_filename)
|
||
storage_path = await _save_attachment_to_storage(
|
||
existing_mail.id, decoded_filename, att_data["content"]
|
||
)
|
||
attachment = MailAttachment(
|
||
tenant_id=tenant_id,
|
||
mail_id=existing_mail.id,
|
||
filename=_sanitize_filename(decoded_filename),
|
||
mime_type=att_data["mime_type"],
|
||
size_bytes=att_data["size"],
|
||
storage_path=storage_path,
|
||
content_id=att_data.get("content_id"),
|
||
)
|
||
db.add(attachment)
|
||
except Exception as att_err:
|
||
logger.warning(f"Failed to save attachment for existing mail {existing_mail.id}: {att_err}")
|
||
continue
|
||
await db.flush()
|
||
continue
|
||
|
||
mail = Mail(
|
||
tenant_id=tenant_id,
|
||
account_id=account.id,
|
||
folder_id=folder.id,
|
||
imap_uid=uid_str,
|
||
message_id=message_id,
|
||
thread_id=thread_id,
|
||
in_reply_to=in_reply_to,
|
||
references_header=refs,
|
||
subject=subject,
|
||
from_address=from_addr,
|
||
to_addresses=to_addrs,
|
||
cc_addresses=cc_addrs,
|
||
body_text=body_text,
|
||
body_html=body_html,
|
||
body_html_sanitized=sanitize_html(body_html),
|
||
has_attachments=len(attachments) > 0,
|
||
size_bytes=len(raw_email),
|
||
received_at=received_at,
|
||
)
|
||
db.add(mail)
|
||
await db.flush()
|
||
synced_count += 1
|
||
|
||
# Collect for new-mail notifications
|
||
new_mails.append({
|
||
"from": from_addr,
|
||
"subject": subject,
|
||
})
|
||
|
||
# Save attachments to storage and DB
|
||
for att_data in attachments:
|
||
try:
|
||
# Decode MIME-encoded filenames (e.g. =?utf-8?q?...?=)
|
||
raw_filename = att_data["filename"] or "attachment"
|
||
decoded_filename = _decode_mime_filename(raw_filename)
|
||
storage_path = await _save_attachment_to_storage(
|
||
mail.id, decoded_filename, att_data["content"]
|
||
)
|
||
attachment = MailAttachment(
|
||
tenant_id=tenant_id,
|
||
mail_id=mail.id,
|
||
filename=_sanitize_filename(decoded_filename),
|
||
mime_type=att_data["mime_type"],
|
||
size_bytes=att_data["size"],
|
||
storage_path=storage_path,
|
||
content_id=att_data.get("content_id"),
|
||
)
|
||
db.add(attachment)
|
||
except Exception as att_err:
|
||
logger.warning(f"Failed to save attachment for new mail {mail.id}: {att_err}")
|
||
continue
|
||
await db.flush()
|
||
|
||
# Update folder counts
|
||
total = (
|
||
await db.execute(
|
||
select(func.count()).select_from(Mail).where(
|
||
and_(Mail.folder_id == folder.id, Mail.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar()
|
||
unread = (
|
||
await db.execute(
|
||
select(func.count()).select_from(Mail).where(
|
||
and_(
|
||
Mail.folder_id == folder.id,
|
||
Mail.tenant_id == tenant_id,
|
||
Mail.is_seen.is_(False),
|
||
)
|
||
)
|
||
)
|
||
).scalar()
|
||
folder.total_count = total or 0
|
||
folder.unread_count = unread or 0
|
||
|
||
await db.flush()
|
||
|
||
except Exception:
|
||
# Skip folders that can't be selected (e.g. no select permission)
|
||
continue
|
||
|
||
# ── New-mail notifications (max 10, then summary) ──
|
||
if new_mails:
|
||
try:
|
||
if len(new_mails) <= 10:
|
||
for nm in new_mails:
|
||
try:
|
||
await create_notification(
|
||
db, account.tenant_id, account.user_id,
|
||
"mail_new",
|
||
f"Neue E-Mail von {nm['from']}",
|
||
nm["subject"],
|
||
)
|
||
except Exception:
|
||
pass
|
||
else:
|
||
try:
|
||
await create_notification(
|
||
db, account.tenant_id, account.user_id,
|
||
"mail_new",
|
||
f"Neue E-Mails: {len(new_mails)} neue Nachrichten",
|
||
f"Account {account.email_address} hat {len(new_mails)} neue E-Mails empfangen.",
|
||
)
|
||
except Exception:
|
||
pass
|
||
await db.flush()
|
||
except Exception:
|
||
pass
|
||
|
||
await db.flush()
|
||
await client.logout()
|
||
return {"synced": synced_count}
|
||
except Exception as e:
|
||
return {"synced": 0, "error": str(e)}
|
||
|
||
|
||
def _compute_thread_id(message_id: str, references: str, in_reply_to: str | None) -> str:
|
||
"""Compute thread ID from References/In-Reply-To headers (F-MAIL-05)."""
|
||
ref_parts: list[str] = []
|
||
if references:
|
||
ref_parts = [r.strip() for r in references.split() if r.strip()]
|
||
if in_reply_to and in_reply_to.strip() not in ref_parts:
|
||
ref_parts.append(in_reply_to.strip())
|
||
if ref_parts:
|
||
return ref_parts[0]
|
||
return message_id or str(uuid.uuid4())
|
||
|
||
|
||
# ─── SMTP Send Service (F-MAIL-02) ───
|
||
|
||
|
||
async def send_mail_via_smtp(
|
||
db: AsyncSession,
|
||
*,
|
||
tenant_id: uuid.UUID,
|
||
user_id: uuid.UUID,
|
||
account: MailAccount,
|
||
to_addrs: list[str],
|
||
cc_addrs: list[str] = None,
|
||
bcc_addrs: list[str] = None,
|
||
subject: str = "",
|
||
body_html: str = "",
|
||
body_text: str = "",
|
||
in_reply_to: str | None = None,
|
||
references_header: str | None = None,
|
||
signature: MailSignature | None = None,
|
||
attachment_paths: list[dict] | None = None,
|
||
) -> dict:
|
||
"""Send an email via SMTP using aiosmtplib.
|
||
|
||
Args:
|
||
attachment_paths: list of dicts with keys 'path', 'filename', 'mime_type'
|
||
pointing to files on disk to attach.
|
||
"""
|
||
cc_addrs = cc_addrs or []
|
||
bcc_addrs = bcc_addrs or []
|
||
attachment_paths = attachment_paths or []
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Apply signature if provided
|
||
if signature and signature.body_html:
|
||
body_html = body_html + f"<br><br>{signature.body_html}"
|
||
if body_text:
|
||
body_text = body_text + "\n\n-- \n" + _strip_html(signature.body_html)
|
||
|
||
# Build email message
|
||
msg = EmailMessage()
|
||
msg["From"] = formataddr((account.display_name or "", account.email_address))
|
||
msg["To"] = ", ".join(to_addrs)
|
||
if cc_addrs:
|
||
msg["Cc"] = ", ".join(cc_addrs)
|
||
msg["Subject"] = subject
|
||
msg["Date"] = formatdate(localtime=True)
|
||
msg_id = make_msgid(
|
||
domain=account.email_address.split("@")[-1] if "@" in account.email_address else "localhost"
|
||
)
|
||
msg["Message-ID"] = msg_id
|
||
if in_reply_to:
|
||
msg["In-Reply-To"] = in_reply_to
|
||
if references_header:
|
||
msg["References"] = references_header
|
||
|
||
if body_html:
|
||
msg.set_content(body_text or _strip_html(body_html), subtype="plain")
|
||
msg.add_alternative(body_html, subtype="html")
|
||
else:
|
||
msg.set_content(body_text, subtype="plain")
|
||
|
||
# Add attachments to the message
|
||
for att_info in attachment_paths:
|
||
file_path = att_info.get("path", "")
|
||
filename = att_info.get("filename", os.path.basename(file_path))
|
||
mime_type = att_info.get("mime_type", "application/octet-stream")
|
||
if not file_path or not os.path.exists(file_path):
|
||
continue
|
||
with open(file_path, "rb") as f: # noqa: ASYNC230
|
||
content = f.read()
|
||
# Determine maintype/subtype from mime_type
|
||
if "/" in mime_type:
|
||
maintype, subtype = mime_type.split("/", 1)
|
||
else:
|
||
maintype, subtype = "application", "octet-stream"
|
||
msg.add_attachment(
|
||
content,
|
||
maintype=maintype,
|
||
subtype=subtype,
|
||
filename=filename,
|
||
)
|
||
|
||
# Send via SMTP
|
||
password = await get_account_password(account)
|
||
try:
|
||
smtp = aiosmtplib.SMTP(
|
||
hostname=account.smtp_host,
|
||
port=account.smtp_port,
|
||
use_tls=account.smtp_tls,
|
||
)
|
||
await smtp.connect()
|
||
await smtp.login(account.username, password)
|
||
recipients = to_addrs + cc_addrs + bcc_addrs
|
||
await smtp.send_message(msg, recipients=recipients)
|
||
await smtp.quit()
|
||
|
||
# Store sent mail in Sent folder — use configured mapping if set,
|
||
# otherwise flexible lookup to handle different IMAP naming conventions
|
||
if account.sent_folder_imap_name:
|
||
sent_folder = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(
|
||
MailFolder.account_id == account.id,
|
||
MailFolder.imap_name == account.sent_folder_imap_name,
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
else:
|
||
sent_folder = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(
|
||
MailFolder.account_id == account.id,
|
||
MailFolder.imap_name.in_(
|
||
["Sent", "INBOX.Sent", "Sent Items", "Sent Mail"]
|
||
),
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
if not sent_folder:
|
||
sent_folder = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(
|
||
MailFolder.account_id == account.id,
|
||
MailFolder.is_standard == True,
|
||
MailFolder.imap_name.ilike("%sent%"),
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
if sent_folder:
|
||
thread_id = _compute_thread_id(msg_id, references_header or "", in_reply_to)
|
||
sent_mail = Mail(
|
||
tenant_id=tenant_id,
|
||
account_id=account.id,
|
||
folder_id=sent_folder.id,
|
||
message_id=msg_id,
|
||
thread_id=thread_id,
|
||
in_reply_to=in_reply_to,
|
||
references_header=references_header,
|
||
subject=subject,
|
||
from_address=account.email_address,
|
||
to_addresses=", ".join(to_addrs),
|
||
cc_addresses=", ".join(cc_addrs),
|
||
bcc_addresses=", ".join(bcc_addrs),
|
||
body_text=body_text or _strip_html(body_html),
|
||
body_html=body_html,
|
||
body_html_sanitized=sanitize_html(body_html),
|
||
is_seen=True,
|
||
is_answered=bool(in_reply_to),
|
||
sent_at=datetime.now(UTC),
|
||
)
|
||
db.add(sent_mail)
|
||
await db.flush()
|
||
|
||
# Upload sent mail to IMAP Sent folder via APPEND
|
||
try:
|
||
imap_client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
|
||
await imap_client.wait_hello_from_server()
|
||
await imap_client.login(account.username, password)
|
||
# Build raw email bytes for APPEND
|
||
raw_email_bytes = msg.as_bytes()
|
||
append_resp = await imap_client.append(
|
||
sent_folder.imap_name,
|
||
r'(\Seen)',
|
||
None,
|
||
raw_email_bytes,
|
||
)
|
||
if append_resp.result == 'OK':
|
||
logger.info("send_mail: uploaded sent mail to IMAP folder %s", sent_folder.imap_name)
|
||
else:
|
||
logger.warning("send_mail: IMAP APPEND failed for sent folder %s: %s", sent_folder.imap_name, append_resp)
|
||
await imap_client.logout()
|
||
except Exception as imap_exc:
|
||
logger.warning("send_mail: IMAP APPEND failed (non-critical): %s", imap_exc)
|
||
|
||
# ── Notification: mail sent ──
|
||
try:
|
||
await create_notification(
|
||
db, tenant_id, user_id,
|
||
"mail_sent",
|
||
"E-Mail gesendet",
|
||
subject,
|
||
)
|
||
await db.flush()
|
||
except Exception:
|
||
pass
|
||
|
||
return {"status": "sent", "message_id": msg_id}
|
||
except Exception as e:
|
||
# ── Notification: send error ──
|
||
try:
|
||
await create_notification(
|
||
db, tenant_id, user_id,
|
||
"mail_send_error",
|
||
"E-Mail konnte nicht gesendet werden",
|
||
str(e),
|
||
)
|
||
await db.flush()
|
||
except Exception:
|
||
pass
|
||
return {"status": "error", "error": str(e)}
|
||
|
||
|
||
async def reply_to_mail(
|
||
db: AsyncSession,
|
||
*,
|
||
tenant_id: uuid.UUID,
|
||
user_id: uuid.UUID,
|
||
original_mail: Mail,
|
||
account: MailAccount,
|
||
body_html: str,
|
||
body_text: str = "",
|
||
reply_to_all: bool = False,
|
||
signature: MailSignature | None = None,
|
||
) -> dict:
|
||
"""Reply to a mail, setting In-Reply-To and References headers (F-MAIL-02)."""
|
||
to_addrs = [original_mail.from_address]
|
||
if reply_to_all and original_mail.cc_addresses:
|
||
to_addrs.extend([a.strip() for a in original_mail.cc_addresses.split(",") if a.strip()])
|
||
|
||
refs = original_mail.references_header or ""
|
||
new_refs = f"{refs} {original_mail.message_id}".strip()
|
||
|
||
result = await send_mail_via_smtp(
|
||
db,
|
||
tenant_id=tenant_id,
|
||
user_id=user_id,
|
||
account=account,
|
||
to_addrs=to_addrs,
|
||
subject=f"Re: {original_mail.subject}".replace("Re: Re: ", "Re: "),
|
||
body_html=body_html,
|
||
body_text=body_text,
|
||
in_reply_to=original_mail.message_id,
|
||
references_header=new_refs,
|
||
signature=signature,
|
||
)
|
||
|
||
# Mark original as answered
|
||
original_mail.is_answered = True
|
||
await db.flush()
|
||
return result
|
||
|
||
|
||
async def forward_mail(
|
||
db: AsyncSession,
|
||
*,
|
||
tenant_id: uuid.UUID,
|
||
user_id: uuid.UUID,
|
||
original_mail: Mail,
|
||
account: MailAccount,
|
||
to_addrs: list[str],
|
||
cc_addrs: list[str] = None,
|
||
body_html: str = "",
|
||
body_text: str = "",
|
||
signature: MailSignature | None = None,
|
||
) -> dict:
|
||
"""Forward a mail with original as forwarded content (F-MAIL-02)."""
|
||
fwd_subject = f"Fwd: {original_mail.subject}".replace("Fwd: Fwd: ", "Fwd: ")
|
||
fwd_body = (
|
||
f"<br><br>----- Original Message -----<br>"
|
||
f"From: {original_mail.from_address}<br>"
|
||
f"Subject: {original_mail.subject}<br><br>"
|
||
f"{original_mail.body_html or original_mail.body_text}"
|
||
)
|
||
full_html = body_html + fwd_body
|
||
full_text = (body_text or _strip_html(body_html)) + "\n\n----- Original Message -----\n"
|
||
|
||
result = await send_mail_via_smtp(
|
||
db,
|
||
tenant_id=tenant_id,
|
||
user_id=user_id,
|
||
account=account,
|
||
to_addrs=to_addrs,
|
||
cc_addrs=cc_addrs or [],
|
||
subject=fwd_subject,
|
||
body_html=full_html,
|
||
body_text=full_text,
|
||
signature=signature,
|
||
)
|
||
|
||
original_mail.is_forwarded = True
|
||
await db.flush()
|
||
return result
|
||
|
||
|
||
# ─── Template Service (F-MAIL-06) ───
|
||
|
||
|
||
def substitute_template_vars(template_body: str, variables: dict[str, str]) -> str:
|
||
"""Replace {{placeholder}} variables in template body."""
|
||
result = template_body
|
||
for key, value in variables.items():
|
||
result = result.replace(f"{{{{{key}}}}}", value)
|
||
result = result.replace(f"{{{{{key.lower()}}}}}", value)
|
||
result = result.replace(f"{{{{{key.upper()}}}}}", value)
|
||
return result
|
||
|
||
|
||
# ─── Mail Rule Engine (F-MAIL-07) ───
|
||
|
||
|
||
def matches_condition(mail: Mail, conditions: dict) -> bool:
|
||
"""Check if a mail matches all rule conditions."""
|
||
for field, expected in conditions.items():
|
||
if field == "from_contains":
|
||
if expected.lower() not in mail.from_address.lower():
|
||
return False
|
||
elif field == "subject_contains":
|
||
if expected.lower() not in mail.subject.lower():
|
||
return False
|
||
elif field == "to_contains":
|
||
if expected.lower() not in mail.to_addresses.lower():
|
||
return False
|
||
elif field == "body_contains":
|
||
body = (mail.body_text + mail.body_html).lower()
|
||
if expected.lower() not in body:
|
||
return False
|
||
elif field == "has_attachments":
|
||
if mail.has_attachments != bool(expected):
|
||
return False
|
||
elif field == "is_flagged":
|
||
if mail.is_flagged != bool(expected):
|
||
return False
|
||
return True
|
||
|
||
|
||
async def execute_rule_actions(
|
||
db: AsyncSession, mail: Mail, actions: dict, tenant_id: uuid.UUID
|
||
) -> dict:
|
||
"""Execute rule actions on a matching mail."""
|
||
results = {}
|
||
for action, value in actions.items():
|
||
if action == "move_to_folder":
|
||
folder_id = uuid.UUID(value) if isinstance(value, str) else value
|
||
folder = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(MailFolder.id == folder_id, MailFolder.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if folder:
|
||
mail.folder_id = folder.id
|
||
results["moved"] = str(folder.id)
|
||
elif action == "label":
|
||
label_id = uuid.UUID(value) if isinstance(value, str) else value
|
||
existing = (
|
||
await db.execute(
|
||
select(MailLabelAssignment).where(
|
||
and_(
|
||
MailLabelAssignment.mail_id == mail.id,
|
||
MailLabelAssignment.label_id == label_id,
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not existing:
|
||
assignment = MailLabelAssignment(
|
||
tenant_id=tenant_id,
|
||
mail_id=mail.id,
|
||
label_id=label_id,
|
||
)
|
||
db.add(assignment)
|
||
results["labeled"] = str(label_id)
|
||
elif action == "mark_seen":
|
||
mail.is_seen = bool(value)
|
||
results["seen"] = bool(value)
|
||
elif action == "mark_flagged":
|
||
mail.is_flagged = bool(value)
|
||
results["flagged"] = bool(value)
|
||
elif action == "forward_to":
|
||
results["forward_to"] = value
|
||
await db.flush()
|
||
return results
|
||
|
||
|
||
async def apply_rules_to_mail(db: AsyncSession, mail: Mail, tenant_id: uuid.UUID) -> list[dict]:
|
||
"""Find and apply all matching rules to a mail, sorted by priority."""
|
||
rules = (
|
||
(
|
||
await db.execute(
|
||
select(MailRule)
|
||
.where(
|
||
and_(
|
||
MailRule.tenant_id == tenant_id,
|
||
MailRule.is_active,
|
||
or_(
|
||
MailRule.account_id == mail.account_id,
|
||
MailRule.account_id.is_(None),
|
||
),
|
||
)
|
||
)
|
||
.order_by(MailRule.priority)
|
||
)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
|
||
applied = []
|
||
for rule in rules:
|
||
conditions = json.loads(rule.conditions) if rule.conditions else {}
|
||
actions = json.loads(rule.actions) if rule.actions else {}
|
||
if matches_condition(mail, conditions):
|
||
result = await execute_rule_actions(db, mail, actions, tenant_id)
|
||
applied.append({"rule_id": str(rule.id), "rule_name": rule.name, "actions": result})
|
||
return applied
|
||
|
||
|
||
# ─── Vacation Auto-Reply (F-MAIL-08) ───
|
||
|
||
|
||
VACATION_DEDUP_HOURS = 24
|
||
|
||
|
||
async def should_send_vacation_reply(
|
||
db: AsyncSession,
|
||
account_id: uuid.UUID,
|
||
sender_address: str,
|
||
tenant_id: uuid.UUID,
|
||
) -> bool:
|
||
"""Check if vacation auto-reply should be sent (dedup within 24h)."""
|
||
cutoff = datetime.now(UTC) - timedelta(hours=VACATION_DEDUP_HOURS)
|
||
existing = (
|
||
await db.execute(
|
||
select(VacationSentLog).where(
|
||
and_(
|
||
VacationSentLog.account_id == account_id,
|
||
VacationSentLog.sender_address == sender_address,
|
||
VacationSentLog.sent_at >= cutoff,
|
||
VacationSentLog.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
return existing is None
|
||
|
||
|
||
async def log_vacation_sent(
|
||
db: AsyncSession,
|
||
account_id: uuid.UUID,
|
||
sender_address: str,
|
||
tenant_id: uuid.UUID,
|
||
) -> None:
|
||
"""Log that a vacation auto-reply was sent to a sender."""
|
||
log = VacationSentLog(
|
||
tenant_id=tenant_id,
|
||
account_id=account_id,
|
||
sender_address=sender_address,
|
||
sent_at=datetime.now(UTC),
|
||
)
|
||
db.add(log)
|
||
await db.flush()
|
||
|
||
|
||
# ─── PGP Service (F-MAIL-12) ───
|
||
|
||
|
||
def import_pgp_private_key(private_key_armored: str, passphrase: str = "") -> tuple[str, str]:
|
||
"""Import a PGP private key. Returns (key_id, public_key_armored)."""
|
||
key, _ = pgpy.PGPKey.from_blob(private_key_armored)
|
||
if key.is_protected:
|
||
with key.unlock(passphrase):
|
||
pub_key = key.pubkey
|
||
key_id = str(key.fingerprint).upper()[-16:]
|
||
return key_id, str(pub_key)
|
||
pub_key = key.pubkey
|
||
key_id = str(key.fingerprint).upper()[-16:]
|
||
return key_id, str(pub_key)
|
||
|
||
|
||
def import_pgp_public_key(public_key_armored: str) -> str:
|
||
"""Import a PGP public key. Returns key_id."""
|
||
key, _ = pgpy.PGPKey.from_blob(public_key_armored)
|
||
return str(key.fingerprint).upper()[-16:]
|
||
|
||
|
||
def pgp_encrypt_message(plaintext: str, recipient_public_key_armored: str) -> str:
|
||
"""Encrypt a message with recipient's public PGP key."""
|
||
pub_key, _ = pgpy.PGPKey.from_blob(recipient_public_key_armored)
|
||
msg = pgpy.PGPMessage.new(plaintext)
|
||
encrypted = pub_key.encrypt(msg)
|
||
return str(encrypted)
|
||
|
||
|
||
def pgp_decrypt_message(ciphertext: str, private_key_armored: str, passphrase: str = "") -> str:
|
||
"""Decrypt a PGP-encrypted message."""
|
||
key, _ = pgpy.PGPKey.from_blob(private_key_armored)
|
||
enc_msg = pgpy.PGPMessage.from_blob(ciphertext)
|
||
if key.is_protected:
|
||
with key.unlock(passphrase):
|
||
decrypted = key.decrypt(enc_msg)
|
||
return decrypted.message.decode("utf-8")
|
||
decrypted = key.decrypt(enc_msg)
|
||
return decrypted.message.decode("utf-8")
|
||
|
||
|
||
# ─── Contact Linking (F-MAIL-10) ───
|
||
|
||
|
||
def extract_email_addresses(text: str) -> list[str]:
|
||
"""Extract email addresses from a text string."""
|
||
if not text:
|
||
return []
|
||
return re.findall(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", text)
|
||
|
||
|
||
# ─── Utility ───
|
||
|
||
|
||
def _strip_html(html: str) -> str:
|
||
"""Simple HTML to text conversion for plain text fallback."""
|
||
if not html:
|
||
return ""
|
||
# Remove tags
|
||
text = re.sub(r"<[^>]+>", "", html)
|
||
# Replace HTML entities
|
||
text = (
|
||
text.replace(" ", " ")
|
||
.replace("&", "&")
|
||
.replace("<", "<")
|
||
.replace(">", ">")
|
||
.replace(""", '"')
|
||
)
|
||
return text.strip()
|
||
|
||
|
||
def mail_to_response(
|
||
mail: Mail,
|
||
attachments: list[MailAttachment] | None = None,
|
||
labels: list[MailLabel] | None = None,
|
||
) -> dict:
|
||
"""Convert a Mail ORM object to a response dict."""
|
||
resp = {
|
||
"id": str(mail.id),
|
||
"account_id": str(mail.account_id),
|
||
"folder_id": str(mail.folder_id),
|
||
"message_id": mail.message_id,
|
||
"thread_id": mail.thread_id,
|
||
"in_reply_to": mail.in_reply_to,
|
||
"subject": mail.subject,
|
||
"from_address": mail.from_address,
|
||
"from_name": mail.from_address.split("<")[0].strip().strip('"') if "<" in mail.from_address else mail.from_address,
|
||
"to_addresses": extract_email_addresses(mail.to_addresses) if mail.to_addresses else [],
|
||
"cc_addresses": extract_email_addresses(mail.cc_addresses) if mail.cc_addresses else [],
|
||
"bcc_addresses": extract_email_addresses(mail.bcc_addresses) if mail.bcc_addresses else [],
|
||
"body_text": mail.body_text,
|
||
"body_html": mail.body_html if mail.body_html else None,
|
||
"body_html_sanitized": mail.body_html_sanitized,
|
||
"sanitized_html": mail.body_html_sanitized,
|
||
"date": mail.received_at.isoformat() if mail.received_at else (mail.sent_at.isoformat() if mail.sent_at else None),
|
||
"is_seen": mail.is_seen,
|
||
"is_flagged": mail.is_flagged,
|
||
"flag_type": getattr(mail, 'flag_type', None),
|
||
"is_draft": mail.is_draft,
|
||
"is_answered": mail.is_answered,
|
||
"is_forwarded": mail.is_forwarded,
|
||
"has_attachments": mail.has_attachments,
|
||
"size_bytes": mail.size_bytes,
|
||
"received_at": mail.received_at,
|
||
"sent_at": mail.sent_at,
|
||
"contact_id": str(mail.contact_id) if mail.contact_id else None,
|
||
"company_id": str(mail.company_id) if mail.company_id else None,
|
||
"attachments": [],
|
||
"labels": [],
|
||
}
|
||
if attachments:
|
||
resp["attachments"] = [attachment_to_response(a) for a in attachments]
|
||
if labels:
|
||
resp["labels"] = [
|
||
{"id": str(lbl.id), "name": lbl.name, "color": lbl.color} for lbl in labels
|
||
]
|
||
return resp
|
||
|
||
|
||
def folder_to_response(folder: MailFolder) -> dict:
|
||
"""Convert MailFolder to response dict."""
|
||
return {
|
||
"id": str(folder.id),
|
||
"account_id": str(folder.account_id),
|
||
"name": folder.name,
|
||
"imap_name": folder.imap_name,
|
||
"parent_id": str(folder.parent_id) if folder.parent_id else None,
|
||
"is_standard": folder.is_standard,
|
||
"unread_count": folder.unread_count,
|
||
"total_count": folder.total_count,
|
||
}
|
||
|
||
|
||
def rule_to_response(rule: MailRule) -> dict:
|
||
"""Convert MailRule to response dict."""
|
||
return {
|
||
"id": str(rule.id),
|
||
"name": rule.name,
|
||
"account_id": str(rule.account_id) if rule.account_id else None,
|
||
"priority": rule.priority,
|
||
"is_active": rule.is_active,
|
||
"conditions": json.loads(rule.conditions) if rule.conditions else {},
|
||
"actions": json.loads(rule.actions) if rule.actions else {},
|
||
}
|
||
|
||
|
||
def template_to_response(template: MailTemplate) -> dict:
|
||
"""Convert MailTemplate to response dict."""
|
||
return {
|
||
"id": str(template.id),
|
||
"name": template.name,
|
||
"subject": template.subject,
|
||
"body_html": template.body_html,
|
||
}
|
||
|
||
|
||
def signature_to_response(sig: MailSignature) -> dict:
|
||
"""Convert MailSignature to response dict."""
|
||
return {
|
||
"id": str(sig.id),
|
||
"name": sig.name,
|
||
"body_html": sig.body_html,
|
||
"account_id": str(sig.account_id) if sig.account_id else None,
|
||
"is_default": sig.is_default,
|
||
}
|
||
|
||
|
||
def label_to_response(label: MailLabel) -> dict:
|
||
"""Convert MailLabel to response dict."""
|
||
return {
|
||
"id": str(label.id),
|
||
"name": label.name,
|
||
"color": label.color,
|
||
}
|
||
|
||
|
||
# ─── IMAP Flag Sync ───
|
||
|
||
|
||
async def imap_sync_mail_flags(
|
||
db: AsyncSession,
|
||
mail_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
) -> None:
|
||
"""Sync is_seen/is_flagged flags from DB to IMAP server.
|
||
|
||
Connects to the IMAP server, selects the mail's folder,
|
||
and uses UID STORE to set/remove \\Seen and \\Flagged flags.
|
||
Non-critical: logs warnings on failure but does not raise.
|
||
"""
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Load the mail with its folder and account
|
||
mail = (
|
||
await db.execute(
|
||
select(Mail).where(
|
||
and_(Mail.id == mail_id, Mail.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not mail:
|
||
logger.warning("imap_sync_mail_flags: mail %s not found", mail_id)
|
||
return
|
||
|
||
folder = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not folder:
|
||
logger.warning("imap_sync_mail_flags: folder %s not found for mail %s", mail.folder_id, mail_id)
|
||
return
|
||
|
||
account = (
|
||
await db.execute(
|
||
select(MailAccount).where(
|
||
and_(MailAccount.id == folder.account_id, MailAccount.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not account:
|
||
logger.warning("imap_sync_mail_flags: account not found for mail %s", mail_id)
|
||
return
|
||
|
||
if not mail.message_id:
|
||
logger.warning("imap_sync_mail_flags: mail %s has no message_id, cannot sync", mail_id)
|
||
return
|
||
|
||
password = await get_account_password(account)
|
||
client = None
|
||
|
||
try:
|
||
client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
|
||
await client.wait_hello_from_server()
|
||
await client.login(account.username, password)
|
||
|
||
# Select the folder
|
||
select_resp = await client.select(folder.imap_name)
|
||
if select_resp.result != 'OK':
|
||
logger.warning("imap_sync_mail_flags: cannot select folder %s", folder.imap_name)
|
||
return
|
||
|
||
# Find the UID by searching for the Message-ID header
|
||
search_resp = await client.uid_search(f'HEADER Message-ID "{mail.message_id}"')
|
||
uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b''
|
||
if isinstance(uids_raw, (bytes, bytearray)):
|
||
uids = uids_raw.split()
|
||
else:
|
||
uids = []
|
||
|
||
if not uids:
|
||
logger.warning("imap_sync_mail_flags: no UID found for Message-ID %s", mail.message_id)
|
||
return
|
||
|
||
uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0])
|
||
|
||
# Sync \Seen flag
|
||
if mail.is_seen:
|
||
await client.uid('store', uid_str, '+FLAGS (\\Seen)')
|
||
else:
|
||
await client.uid('store', uid_str, '-FLAGS (\\Seen)')
|
||
|
||
# Sync \Flagged flag
|
||
if mail.is_flagged:
|
||
await client.uid('store', uid_str, '+FLAGS (\\Flagged)')
|
||
else:
|
||
await client.uid('store', uid_str, '-FLAGS (\\Flagged)')
|
||
|
||
logger.info("imap_sync_mail_flags: synced flags for mail %s (UID %s)", mail_id, uid_str)
|
||
|
||
except Exception as exc:
|
||
logger.warning("imap_sync_mail_flags: failed for mail %s: %s", mail_id, exc)
|
||
finally:
|
||
if client is not None:
|
||
try:
|
||
await client.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ─── IMAP Delete ───
|
||
|
||
|
||
async def _find_trash_folder_name(
|
||
db: AsyncSession, account: MailAccount, tenant_id: uuid.UUID
|
||
) -> str | None:
|
||
"""Find the IMAP Trash folder name for an account.
|
||
|
||
Checks account.trash_folder_imap_name, then DB folders with 'trash' in imap_name,
|
||
then queries the IMAP server LIST for common Trash folder names.
|
||
"""
|
||
# 1) Explicit mapping on account
|
||
if account.trash_folder_imap_name:
|
||
return account.trash_folder_imap_name
|
||
|
||
# 2) DB folder with 'trash' in imap_name
|
||
db_folders = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(MailFolder.account_id == account.id, MailFolder.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalars().all()
|
||
for f in db_folders:
|
||
if 'trash' in f.imap_name.lower():
|
||
return f.imap_name
|
||
|
||
# 3) Query IMAP server for common Trash folder names
|
||
password = await get_account_password(account)
|
||
client = None
|
||
try:
|
||
client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
|
||
await client.wait_hello_from_server()
|
||
await client.login(account.username, password)
|
||
list_resp = await client.list('""', '"*"')
|
||
imap_folders, _ = _parse_imap_list_response(list_resp)
|
||
trash_candidates = ['Trash', 'INBOX.Trash', 'INBOX.Trash', 'Deleted', 'Deleted Items', 'Papierkorb']
|
||
for _, name in imap_folders:
|
||
if name in trash_candidates or 'trash' in name.lower():
|
||
return name
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
if client:
|
||
try:
|
||
await client.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
return None
|
||
|
||
|
||
async def imap_delete_mail(
|
||
db: AsyncSession, mail_id: uuid.UUID, tenant_id: uuid.UUID, *, permanent: bool = False
|
||
) -> None:
|
||
"""Delete mail from IMAP server.
|
||
|
||
If permanent=False (default): MOVE mail to IMAP Trash folder (like real mail clients).
|
||
If permanent=True: STORE \\Deleted + EXPUNGE (permanent delete from server).
|
||
Uses UID MOVE if supported, otherwise COPY + STORE \\Deleted + EXPUNGE.
|
||
Raises exceptions on failure so caller can queue for retry.
|
||
"""
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
mail = (
|
||
await db.execute(
|
||
select(Mail).where(
|
||
and_(Mail.id == mail_id, Mail.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not mail:
|
||
logger.warning("imap_delete_mail: mail %s not found", mail_id)
|
||
return
|
||
|
||
folder = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not folder:
|
||
logger.warning("imap_delete_mail: folder %s not found for mail %s", mail.folder_id, mail_id)
|
||
return
|
||
|
||
account = (
|
||
await db.execute(
|
||
select(MailAccount).where(
|
||
and_(MailAccount.id == folder.account_id, MailAccount.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not account:
|
||
logger.warning("imap_delete_mail: account not found for mail %s", mail_id)
|
||
return
|
||
|
||
if not mail.imap_uid and not mail.message_id:
|
||
logger.warning("imap_delete_mail: mail %s has no imap_uid or message_id, cannot sync", mail_id)
|
||
return
|
||
|
||
# Find Trash folder on IMAP server (only needed for non-permanent delete)
|
||
trash_folder_name = None
|
||
if not permanent:
|
||
trash_folder_name = await _find_trash_folder_name(db, account, tenant_id)
|
||
if not trash_folder_name:
|
||
logger.warning("imap_delete_mail: no Trash folder found for account %s, will permanent delete", account.id)
|
||
|
||
password = await get_account_password(account)
|
||
client = None
|
||
|
||
try:
|
||
client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
|
||
await client.wait_hello_from_server()
|
||
await client.login(account.username, password)
|
||
|
||
select_resp = await client.select(folder.imap_name)
|
||
if select_resp.result != 'OK':
|
||
raise RuntimeError(f"cannot select folder {folder.imap_name}")
|
||
|
||
# Use stored imap_uid directly; fall back to Message-ID search
|
||
if mail.imap_uid:
|
||
uid_str = mail.imap_uid
|
||
else:
|
||
search_resp = await client.uid_search(f'HEADER Message-ID "{mail.message_id}"')
|
||
uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b''
|
||
if isinstance(uids_raw, (bytes, bytearray)):
|
||
uids = uids_raw.split()
|
||
else:
|
||
uids = []
|
||
if not uids:
|
||
raise RuntimeError(f"no UID found for Message-ID {mail.message_id}")
|
||
uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0])
|
||
|
||
if permanent or not trash_folder_name:
|
||
# Permanent delete: STORE \Deleted + EXPUNGE
|
||
await client.uid('store', uid_str, r'+FLAGS (\\Deleted)')
|
||
await client.expunge()
|
||
logger.info("imap_delete_mail: permanent deleted mail %s (UID %s)", mail_id, uid_str)
|
||
return
|
||
|
||
# Non-permanent: MOVE to Trash folder (like Thunderbird, Outlook, etc.)
|
||
# Don't move if already in Trash
|
||
if folder.imap_name.lower() == trash_folder_name.lower():
|
||
# Already in Trash — permanent delete
|
||
await client.uid('store', uid_str, r'+FLAGS (\\Deleted)')
|
||
await client.expunge()
|
||
logger.info("imap_delete_mail: permanent deleted mail %s (already in Trash, UID %s)", mail_id, uid_str)
|
||
return
|
||
|
||
# Try UID MOVE first
|
||
try:
|
||
move_resp = await client.uid('move', uid_str, trash_folder_name)
|
||
if move_resp.result == 'OK':
|
||
logger.info("imap_delete_mail: moved mail %s (UID %s) to Trash %s", mail_id, uid_str, trash_folder_name)
|
||
return
|
||
except Exception as move_exc:
|
||
logger.info("imap_delete_mail: UID MOVE not supported, falling back: %s", move_exc)
|
||
|
||
# Fallback: COPY to Trash + STORE \Deleted + EXPUNGE
|
||
copy_resp = await client.uid('copy', uid_str, trash_folder_name)
|
||
if copy_resp.result != 'OK':
|
||
raise RuntimeError(f"COPY to Trash failed for mail {mail_id}")
|
||
await client.uid('store', uid_str, r'+FLAGS (\\Deleted)')
|
||
await client.expunge()
|
||
logger.info("imap_delete_mail: moved mail %s (UID %s) to Trash via COPY+DELETE", mail_id, uid_str)
|
||
|
||
except Exception as exc:
|
||
logger.warning("imap_delete_mail: failed for mail %s: %s", mail_id, exc)
|
||
raise
|
||
finally:
|
||
if client is not None:
|
||
try:
|
||
await client.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ─── IMAP Move ───
|
||
|
||
|
||
async def imap_move_mail(
|
||
db: AsyncSession,
|
||
mail_id: uuid.UUID,
|
||
target_folder_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
) -> None:
|
||
"""Move mail to another folder on IMAP server.
|
||
|
||
Uses stored imap_uid directly; falls back to Message-ID search if missing.
|
||
Uses UID MOVE if supported, otherwise COPY + STORE \\Deleted + EXPUNGE.
|
||
Raises exceptions on failure so caller can queue for retry.
|
||
"""
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
mail = (
|
||
await db.execute(
|
||
select(Mail).where(
|
||
and_(Mail.id == mail_id, Mail.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not mail:
|
||
logger.warning("imap_move_mail: mail %s not found", mail_id)
|
||
return
|
||
|
||
source_folder = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not source_folder:
|
||
logger.warning("imap_move_mail: source folder not found for mail %s", mail_id)
|
||
return
|
||
|
||
target_folder = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(MailFolder.id == target_folder_id, MailFolder.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not target_folder:
|
||
logger.warning("imap_move_mail: target folder %s not found", target_folder_id)
|
||
return
|
||
|
||
account = (
|
||
await db.execute(
|
||
select(MailAccount).where(
|
||
and_(MailAccount.id == source_folder.account_id, MailAccount.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not account:
|
||
logger.warning("imap_move_mail: account not found for mail %s", mail_id)
|
||
return
|
||
|
||
if not mail.imap_uid and not mail.message_id:
|
||
logger.warning("imap_move_mail: mail %s has no imap_uid or message_id, cannot sync", mail_id)
|
||
return
|
||
|
||
password = await get_account_password(account)
|
||
client = None
|
||
|
||
try:
|
||
client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
|
||
await client.wait_hello_from_server()
|
||
await client.login(account.username, password)
|
||
|
||
select_resp = await client.select(source_folder.imap_name)
|
||
if select_resp.result != 'OK':
|
||
raise RuntimeError(f"cannot select folder {source_folder.imap_name}")
|
||
|
||
# Use stored imap_uid directly; fall back to Message-ID search
|
||
if mail.imap_uid:
|
||
uid_str = mail.imap_uid
|
||
else:
|
||
search_resp = await client.uid_search(f'HEADER Message-ID "{mail.message_id}"')
|
||
uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b''
|
||
if isinstance(uids_raw, (bytes, bytearray)):
|
||
uids = uids_raw.split()
|
||
else:
|
||
uids = []
|
||
if not uids:
|
||
raise RuntimeError(f"no UID found for Message-ID {mail.message_id}")
|
||
uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0])
|
||
|
||
# Try UID MOVE first; fall back to COPY + STORE \\Deleted + EXPUNGE
|
||
try:
|
||
move_resp = await client.uid('move', uid_str, target_folder.imap_name)
|
||
if move_resp.result == 'OK':
|
||
logger.info("imap_move_mail: moved mail %s (UID %s) via UID MOVE", mail_id, uid_str)
|
||
return
|
||
except Exception as move_exc:
|
||
logger.info("imap_move_mail: UID MOVE not supported, falling back: %s", move_exc)
|
||
|
||
# Fallback: COPY + STORE \\Deleted + EXPUNGE
|
||
copy_resp = await client.uid('copy', uid_str, target_folder.imap_name)
|
||
if copy_resp.result != 'OK':
|
||
raise RuntimeError(f"COPY failed for mail {mail_id}")
|
||
|
||
await client.uid('store', uid_str, r'+FLAGS (\\Deleted)')
|
||
await client.expunge()
|
||
|
||
logger.info("imap_move_mail: moved mail %s (UID %s) via COPY+DELETE", mail_id, uid_str)
|
||
|
||
except Exception as exc:
|
||
logger.warning("imap_move_mail: failed for mail %s: %s", mail_id, exc)
|
||
raise
|
||
finally:
|
||
if client is not None:
|
||
try:
|
||
await client.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ─── Draft Save / Update ───
|
||
|
||
|
||
async def save_draft(
|
||
db: AsyncSession,
|
||
account_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
user_id: uuid.UUID,
|
||
data: dict,
|
||
) -> Mail:
|
||
"""Save a new draft mail to DB and IMAP Drafts folder."""
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 1. Find the Drafts folder (imap_name contains 'Drafts')
|
||
account = (
|
||
await db.execute(
|
||
select(MailAccount).where(
|
||
and_(MailAccount.id == account_id, MailAccount.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not account:
|
||
raise ValueError("Account not found")
|
||
|
||
folders = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(MailFolder.account_id == account_id, MailFolder.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalars().all()
|
||
|
||
drafts_folder = None
|
||
for f in folders:
|
||
if 'draft' in f.imap_name.lower():
|
||
drafts_folder = f
|
||
break
|
||
|
||
if not drafts_folder:
|
||
raise ValueError("No Drafts folder found for this account")
|
||
|
||
# 2. Create Mail record
|
||
to_str = ', '.join(data.get('to', []))
|
||
cc_str = ', '.join(data.get('cc', []))
|
||
bcc_str = ', '.join(data.get('bcc', []))
|
||
subject = data.get('subject', '')
|
||
body_text = data.get('body_text', '')
|
||
body_html = data.get('body_html', '')
|
||
|
||
msg_id = make_msgid()
|
||
now = datetime.now(UTC)
|
||
|
||
mail = Mail(
|
||
tenant_id=tenant_id,
|
||
account_id=account_id,
|
||
folder_id=drafts_folder.id,
|
||
message_id=msg_id,
|
||
thread_id=msg_id,
|
||
subject=subject,
|
||
from_address=account.email_address,
|
||
to_addresses=to_str,
|
||
cc_addresses=cc_str,
|
||
bcc_addresses=bcc_str,
|
||
body_text=body_text,
|
||
body_html=body_html,
|
||
body_html_sanitized=body_html,
|
||
is_seen=True,
|
||
is_flagged=False,
|
||
is_draft=True,
|
||
is_answered=False,
|
||
is_forwarded=False,
|
||
has_attachments=False,
|
||
size_bytes=len(body_text.encode('utf-8')),
|
||
received_at=now,
|
||
sent_at=None,
|
||
)
|
||
db.add(mail)
|
||
await db.flush()
|
||
|
||
# ── Notification: draft saved ──
|
||
try:
|
||
await create_notification(
|
||
db, tenant_id, user_id,
|
||
"mail_draft",
|
||
"Entwurf gespeichert",
|
||
subject or "Ohne Betreff",
|
||
)
|
||
await db.flush()
|
||
except Exception:
|
||
pass
|
||
|
||
# 3. Build RFC822 message and APPEND to IMAP Drafts folder
|
||
password = await get_account_password(account)
|
||
client = None
|
||
try:
|
||
client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
|
||
await client.wait_hello_from_server()
|
||
await client.login(account.username, password)
|
||
|
||
# Build RFC822 message
|
||
email_msg = EmailMessage()
|
||
email_msg['From'] = formataddr((account.display_name, account.email_address))
|
||
if to_str:
|
||
email_msg['To'] = to_str
|
||
if cc_str:
|
||
email_msg['Cc'] = cc_str
|
||
email_msg['Subject'] = subject
|
||
email_msg['Date'] = formatdate(localtime=True)
|
||
email_msg['Message-ID'] = msg_id
|
||
email_msg.set_content(body_text if body_text else '')
|
||
if body_html:
|
||
email_msg.add_alternative(body_html, subtype='html')
|
||
|
||
rfc822_bytes = email_msg.as_bytes()
|
||
|
||
# APPEND to Drafts folder
|
||
append_resp = await client.append(
|
||
drafts_folder.imap_name,
|
||
r'(\\Draft)',
|
||
str(int(now.timestamp())),
|
||
rfc822_bytes,
|
||
)
|
||
if append_resp.result != 'OK':
|
||
logger.warning("save_draft: IMAP APPEND failed for drafts folder %s", drafts_folder.imap_name)
|
||
else:
|
||
logger.info("save_draft: appended draft %s to IMAP Drafts", mail.id)
|
||
|
||
except Exception as exc:
|
||
logger.warning("save_draft: IMAP append failed (non-critical): %s", exc)
|
||
finally:
|
||
if client is not None:
|
||
try:
|
||
await client.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
return mail
|
||
|
||
|
||
async def update_draft(
|
||
db: AsyncSession,
|
||
mail_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
data: dict,
|
||
) -> Mail:
|
||
"""Update an existing draft."""
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
mail = (
|
||
await db.execute(
|
||
select(Mail).where(
|
||
and_(Mail.id == mail_id, Mail.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not mail:
|
||
raise ValueError("Mail not found")
|
||
if not mail.is_draft:
|
||
raise ValueError("Mail is not a draft")
|
||
|
||
# 2. Update fields
|
||
to_str = ', '.join(data.get('to', []))
|
||
cc_str = ', '.join(data.get('cc', []))
|
||
bcc_str = ', '.join(data.get('bcc', []))
|
||
mail.to_addresses = to_str
|
||
mail.cc_addresses = cc_str
|
||
mail.bcc_addresses = bcc_str
|
||
mail.subject = data.get('subject', '')
|
||
mail.body_text = data.get('body_text', '')
|
||
mail.body_html = data.get('body_html', '')
|
||
mail.body_html_sanitized = data.get('body_html', '')
|
||
mail.size_bytes = len(mail.body_text.encode('utf-8'))
|
||
await db.flush()
|
||
|
||
# 3. Delete old IMAP copy and APPEND new one
|
||
folder = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not folder:
|
||
return mail
|
||
|
||
account = (
|
||
await db.execute(
|
||
select(MailAccount).where(
|
||
and_(MailAccount.id == folder.account_id, MailAccount.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not account:
|
||
return mail
|
||
|
||
password = await get_account_password(account)
|
||
client = None
|
||
try:
|
||
client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
|
||
await client.wait_hello_from_server()
|
||
await client.login(account.username, password)
|
||
|
||
# Delete old copy from IMAP
|
||
select_resp = await client.select(folder.imap_name)
|
||
if select_resp.result == 'OK' and mail.message_id:
|
||
search_resp = await client.uid_search(f'HEADER Message-ID "{mail.message_id}"')
|
||
uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b''
|
||
if isinstance(uids_raw, (bytes, bytearray)):
|
||
uids = uids_raw.split()
|
||
else:
|
||
uids = []
|
||
if uids:
|
||
uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0])
|
||
await client.uid('store', uid_str, r'+FLAGS (\\Deleted)')
|
||
await client.expunge()
|
||
|
||
# APPEND new copy
|
||
email_msg = EmailMessage()
|
||
email_msg['From'] = formataddr((account.display_name, account.email_address))
|
||
if to_str:
|
||
email_msg['To'] = to_str
|
||
if cc_str:
|
||
email_msg['Cc'] = cc_str
|
||
email_msg['Subject'] = mail.subject
|
||
email_msg['Date'] = formatdate(localtime=True)
|
||
email_msg['Message-ID'] = mail.message_id
|
||
email_msg.set_content(mail.body_text if mail.body_text else '')
|
||
if mail.body_html:
|
||
email_msg.add_alternative(mail.body_html, subtype='html')
|
||
|
||
rfc822_bytes = email_msg.as_bytes()
|
||
now = datetime.now(UTC)
|
||
append_resp = await client.append(
|
||
folder.imap_name,
|
||
r'(\\Draft)',
|
||
str(int(now.timestamp())),
|
||
rfc822_bytes,
|
||
)
|
||
if append_resp.result != 'OK':
|
||
logger.warning("update_draft: IMAP APPEND failed for drafts folder %s", folder.imap_name)
|
||
else:
|
||
logger.info("update_draft: appended updated draft %s to IMAP Drafts", mail.id)
|
||
|
||
except Exception as exc:
|
||
logger.warning("update_draft: IMAP sync failed (non-critical): %s", exc)
|
||
finally:
|
||
if client is not None:
|
||
try:
|
||
await client.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
return mail
|
||
|
||
|
||
# ─── IMAP Folder Create / Delete ───
|
||
|
||
|
||
async def imap_create_folder(
|
||
db: AsyncSession, account_id: uuid.UUID, folder_name: str, tenant_id: uuid.UUID
|
||
) -> None:
|
||
"""Create folder on IMAP server.
|
||
|
||
Connects to IMAP, creates folder with CREATE command.
|
||
Non-critical: errors are logged, DB operation still succeeds.
|
||
"""
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
account = (
|
||
await db.execute(
|
||
select(MailAccount).where(
|
||
and_(MailAccount.id == account_id, MailAccount.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not account:
|
||
logger.warning("imap_create_folder: account %s not found", account_id)
|
||
return
|
||
|
||
password = await get_account_password(account)
|
||
client = None
|
||
|
||
try:
|
||
client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
|
||
await client.wait_hello_from_server()
|
||
await client.login(account.username, password)
|
||
|
||
resp = await client.create(folder_name)
|
||
if resp.result != "OK":
|
||
logger.warning(
|
||
"imap_create_folder: CREATE failed for %s: %s", folder_name, resp
|
||
)
|
||
else:
|
||
logger.info("imap_create_folder: created folder %s on IMAP", folder_name)
|
||
try:
|
||
await create_notification(
|
||
db, account.tenant_id, account.user_id,
|
||
"mail_folder",
|
||
"Ordner erstellt",
|
||
folder_name,
|
||
)
|
||
await db.flush()
|
||
except Exception:
|
||
pass
|
||
|
||
except Exception as exc:
|
||
logger.warning("imap_create_folder: failed (non-critical): %s", exc)
|
||
finally:
|
||
if client is not None:
|
||
try:
|
||
await client.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
async def imap_delete_folder(
|
||
db: AsyncSession, folder_id: uuid.UUID, tenant_id: uuid.UUID
|
||
) -> None:
|
||
"""Delete folder from IMAP server.
|
||
|
||
Connects to IMAP, deletes folder with DELETE command.
|
||
Non-critical: errors are logged, DB operation still succeeds.
|
||
"""
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
folder = (
|
||
await db.execute(
|
||
select(MailFolder).where(
|
||
and_(MailFolder.id == folder_id, MailFolder.tenant_id == tenant_id)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not folder:
|
||
logger.warning("imap_delete_folder: folder %s not found", folder_id)
|
||
return
|
||
|
||
account = (
|
||
await db.execute(
|
||
select(MailAccount).where(
|
||
and_(
|
||
MailAccount.id == folder.account_id,
|
||
MailAccount.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not account:
|
||
logger.warning("imap_delete_folder: account not found for folder %s", folder_id)
|
||
return
|
||
|
||
password = await get_account_password(account)
|
||
client = None
|
||
|
||
try:
|
||
client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
|
||
await client.wait_hello_from_server()
|
||
await client.login(account.username, password)
|
||
|
||
resp = await client.delete(folder.imap_name)
|
||
if resp.result != "OK":
|
||
logger.warning(
|
||
"imap_delete_folder: DELETE failed for %s: %s", folder.imap_name, resp
|
||
)
|
||
else:
|
||
logger.info("imap_delete_folder: deleted folder %s on IMAP", folder.imap_name)
|
||
try:
|
||
await create_notification(
|
||
db, account.tenant_id, account.user_id,
|
||
"mail_folder",
|
||
"Ordner gelöscht",
|
||
folder.imap_name,
|
||
)
|
||
await db.flush()
|
||
except Exception:
|
||
pass
|
||
|
||
except Exception as exc:
|
||
logger.warning("imap_delete_folder: failed (non-critical): %s", exc)
|
||
finally:
|
||
if client is not None:
|
||
try:
|
||
await client.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ─── Auto-Sync ───
|
||
|
||
|
||
async def auto_sync_all_accounts() -> None:
|
||
"""Auto-sync all active mail accounts.
|
||
|
||
Called periodically by the background scheduler.
|
||
Iterates all active mail accounts and syncs each one.
|
||
"""
|
||
import logging
|
||
|
||
from app.core.db import get_session_factory
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
factory = get_session_factory()
|
||
async with factory() as db:
|
||
accounts = (
|
||
await db.execute(
|
||
select(MailAccount).where(MailAccount.is_active.is_(True))
|
||
)
|
||
).scalars().all()
|
||
|
||
if not accounts:
|
||
return
|
||
|
||
logger.info("auto_sync_all_accounts: syncing %d active account(s)", len(accounts))
|
||
|
||
for account in accounts:
|
||
try:
|
||
result = await imap_sync_account(db, account.id, account.tenant_id)
|
||
logger.info(
|
||
"auto_sync_all_accounts: synced account %s (%s): %s",
|
||
account.id,
|
||
account.username,
|
||
result,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"auto_sync_all_accounts: failed for account %s: %s",
|
||
account.id,
|
||
exc,
|
||
)
|
||
try:
|
||
await create_notification(
|
||
db, account.tenant_id, account.user_id,
|
||
"mail_sync_error",
|
||
"Synchronisierung fehlgeschlagen",
|
||
f"Account {account.email_address}: {exc}",
|
||
)
|
||
except Exception:
|
||
pass
|
||
# commit per-account so partial progress is saved
|
||
try:
|
||
await db.commit()
|
||
except Exception:
|
||
await db.rollback()
|
||
|
||
|
||
async def process_sync_queue(db: AsyncSession) -> None:
|
||
"""Process pending IMAP operations from the sync queue.
|
||
|
||
Called at the start of each auto-sync loop iteration.
|
||
Retries failed delete/move operations.
|
||
"""
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
from app.plugins.builtins.mail.models import MailSyncQueue
|
||
|
||
pending = (
|
||
await db.execute(
|
||
select(MailSyncQueue).where(
|
||
and_(
|
||
MailSyncQueue.status == "pending",
|
||
MailSyncQueue.attempts < MailSyncQueue.max_attempts,
|
||
)
|
||
)
|
||
)
|
||
).scalars().all()
|
||
|
||
if not pending:
|
||
return
|
||
|
||
logger.info("process_sync_queue: processing %d pending operation(s)", len(pending))
|
||
|
||
for entry in pending:
|
||
try:
|
||
if entry.operation == "delete":
|
||
is_permanent = entry.payload.get("permanent", False) if entry.payload else False
|
||
await imap_delete_mail(db, entry.mail_id, entry.tenant_id, permanent=is_permanent)
|
||
elif entry.operation == "move":
|
||
target_folder_id = uuid.UUID(entry.payload.get("target_folder_id", ""))
|
||
await imap_move_mail(db, entry.mail_id, target_folder_id, entry.tenant_id)
|
||
else:
|
||
logger.warning("process_sync_queue: unknown operation %s", entry.operation)
|
||
entry.status = "failed"
|
||
entry.last_error = f"Unknown operation: {entry.operation}"
|
||
continue
|
||
|
||
entry.status = "completed"
|
||
entry.updated_at = datetime.now(UTC)
|
||
logger.info("process_sync_queue: completed %s for mail %s", entry.operation, entry.mail_id)
|
||
|
||
except Exception as exc:
|
||
entry.attempts += 1
|
||
entry.last_error = str(exc)
|
||
entry.updated_at = datetime.now(UTC)
|
||
if entry.attempts >= entry.max_attempts:
|
||
entry.status = "failed"
|
||
logger.warning(
|
||
"process_sync_queue: giving up on %s for mail %s after %d attempts: %s",
|
||
entry.operation, entry.mail_id, entry.attempts, exc,
|
||
)
|
||
else:
|
||
logger.info(
|
||
"process_sync_queue: retry %d/%d for %s mail %s: %s",
|
||
entry.attempts, entry.max_attempts, entry.operation, entry.mail_id, exc,
|
||
)
|
||
|
||
await db.flush()
|