Agent Zero
6702d69f7c
IMAP Sync Block (F-MAIL-01) aus services.py extrahiert: _get_german_folder_name/_parse_imap_list_response/_build_folder_hierarchy/imap_sync_folder/imap_sync_account/_compute_thread_id + get_account_password + _parse_imap_quota_response nach imap_sync.py (1148 Z.). services.py jetzt ~1680 Z. (von 3087). Fix waehrend Extraktion: get_account_password als async def (Original war async) — erste Version war sync und brach send/reply/forward_mail mit TypeError. Beweis: mail+sig_label_routes 51/51 passed in 106.60s; ruff clean.
1149 lines
49 KiB
Python
1149 lines
49 KiB
Python
"""IMAP sync service for the Mail plugin (F-MAIL-01).
|
||
|
||
Extracted from services.py as part of the God-object split (BUG-018 pilot).
|
||
Re-exported by ``app.plugins.builtins.mail.services``.
|
||
|
||
Contains folder sync, account sync, thread-id computation and the German
|
||
folder name mapping for standard IMAP folders.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
import uuid
|
||
from datetime import UTC, datetime
|
||
from email import message_from_bytes
|
||
|
||
import aioimaplib
|
||
from sqlalchemy import and_, func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.notifications import create_notification
|
||
from app.plugins.builtins.mail.attachments import (
|
||
_decode_mime_filename,
|
||
_sanitize_filename,
|
||
_save_attachment_to_storage,
|
||
)
|
||
from app.plugins.builtins.mail.models import (
|
||
Mail,
|
||
MailAccount,
|
||
MailAttachment,
|
||
MailFolder,
|
||
)
|
||
from app.plugins.builtins.mail.sanitize import sanitize_html
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# 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
|
||
|
||
|
||
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:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
return None
|
||
|
||
|
||
async def get_account_password(account: MailAccount) -> str:
|
||
"""Decrypt and return the account password (internal use only).
|
||
|
||
Uses per-account salt if available, falls back to legacy salt for old accounts.
|
||
"""
|
||
from app.plugins.builtins.mail.crypto import decrypt_password
|
||
|
||
return decrypt_password(account.encrypted_password, account.password_salt or None)
|
||
|
||
|
||
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:
|
||
# 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:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
|
||
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
|
||
|
||
# Publish mail.received event
|
||
from app.core.event_bus import get_event_bus
|
||
event_bus = get_event_bus()
|
||
await event_bus.publish('mail.received', {
|
||
'mail_id': str(mail.id),
|
||
'tenant_id': str(tenant_id),
|
||
'account_id': str(account.id),
|
||
'folder_id': str(folder.id),
|
||
'subject': subject,
|
||
'from_address': from_addr,
|
||
})
|
||
|
||
# Lifecycle hook: mail.after_receive
|
||
from app.core.hooks import do_action
|
||
await do_action("mail.after_receive", {'mail_id': str(mail.id), 'tenant_id': str(tenant_id), 'account_id': str(account.id), 'folder_id': str(folder.id), 'subject': subject, 'from_address': from_addr}, db=db, tenant_id=tenant_id)
|
||
await do_action("mail.after_create", {'mail_id': str(mail.id), 'tenant_id': str(tenant_id), 'subject': subject, 'from_address': from_addr}, db=db, tenant_id=tenant_id)
|
||
|
||
# Outbox event: mail.received
|
||
from app.core.outbox import enqueue_outbox_event
|
||
await enqueue_outbox_event(db, tenant_id, 'mail.received', {'mail_id': str(mail.id), 'tenant_id': str(tenant_id), 'account_id': str(account.id), 'folder_id': str(folder.id), 'subject': subject, 'from_address': from_addr}, aggregate_type='mail', aggregate_id=mail.id)
|
||
|
||
# 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
|
||
|
||
# Vanished-UID check: delete DB mails whose UID no longer exists on IMAP server
|
||
try:
|
||
vanished_search = await client.uid_search('ALL')
|
||
vanished_raw = vanished_search[1][0] if vanished_search[1] and vanished_search[1][0] else b''
|
||
if isinstance(vanished_raw, (bytes, bytearray)):
|
||
imap_uids_set = {u.decode() if isinstance(u, bytes) else str(u) for u in vanished_raw.split()}
|
||
else:
|
||
imap_uids_set = set()
|
||
db_mails = (
|
||
await db.execute(
|
||
select(Mail).where(
|
||
and_(
|
||
Mail.folder_id == folder_id,
|
||
Mail.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
).scalars().all()
|
||
for db_mail in db_mails:
|
||
if db_mail.imap_uid:
|
||
if db_mail.imap_uid not in imap_uids_set:
|
||
logger.info("imap_sync_folder: deleting vanished mail %s (UID %s no longer on server)", db_mail.id, db_mail.imap_uid)
|
||
await db.delete(db_mail)
|
||
elif db_mail.message_id:
|
||
# Mail has no UID — try to find it on IMAP by Message-ID
|
||
try:
|
||
mid_search = await client.uid_search(f'HEADER Message-ID "{db_mail.message_id}"')
|
||
mid_raw = mid_search[1][0] if mid_search[1] and mid_search[1][0] else b''
|
||
if isinstance(mid_raw, (bytes, bytearray)) and mid_raw:
|
||
found_uids = mid_raw.decode().split()
|
||
if found_uids:
|
||
db_mail.imap_uid = found_uids[0]
|
||
logger.info("imap_sync_folder: found UID %s for mail %s via Message-ID", found_uids[0], db_mail.id)
|
||
else:
|
||
logger.info("imap_sync_folder: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id)
|
||
await db.delete(db_mail)
|
||
else:
|
||
logger.info("imap_sync_folder: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id)
|
||
await db.delete(db_mail)
|
||
except Exception as mid_exc:
|
||
logger.warning("imap_sync_folder: Message-ID search failed for mail %s: %s", db_mail.id, mid_exc)
|
||
await db.flush()
|
||
except Exception as vanished_exc:
|
||
logger.warning("imap_sync_folder: vanished-UID check failed for folder %s: %s", folder_id, vanished_exc)
|
||
|
||
# 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() 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.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:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
|
||
|
||
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:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
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:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
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:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
try:
|
||
await client.logout()
|
||
except Exception:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
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:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
except Exception:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
|
||
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:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
|
||
# 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:
|
||
# 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()
|
||
|
||
# Vanished-UID check: delete DB mails whose UID no longer exists on IMAP server
|
||
try:
|
||
vanished_search = await client.uid_search('ALL')
|
||
vanished_raw = vanished_search[1][0] if vanished_search[1] and vanished_search[1][0] else b''
|
||
if isinstance(vanished_raw, (bytes, bytearray)):
|
||
imap_uids_set = {u.decode() if isinstance(u, bytes) else str(u) for u in vanished_raw.split()}
|
||
else:
|
||
imap_uids_set = set()
|
||
db_mails = (
|
||
await db.execute(
|
||
select(Mail).where(
|
||
and_(
|
||
Mail.folder_id == folder.id,
|
||
Mail.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
).scalars().all()
|
||
for db_mail in db_mails:
|
||
if db_mail.imap_uid:
|
||
if db_mail.imap_uid not in imap_uids_set:
|
||
logger.info("imap_sync_account: deleting vanished mail %s (UID %s no longer on server)", db_mail.id, db_mail.imap_uid)
|
||
await db.delete(db_mail)
|
||
elif db_mail.message_id:
|
||
# Mail has no UID — try to find it on IMAP by Message-ID
|
||
try:
|
||
mid_search = await client.uid_search(f'HEADER Message-ID "{db_mail.message_id}"')
|
||
mid_raw = mid_search[1][0] if mid_search[1] and mid_search[1][0] else b''
|
||
if isinstance(mid_raw, (bytes, bytearray)) and mid_raw:
|
||
found_uids = mid_raw.decode().split()
|
||
if found_uids:
|
||
db_mail.imap_uid = found_uids[0]
|
||
logger.info("imap_sync_account: found UID %s for mail %s via Message-ID", found_uids[0], db_mail.id)
|
||
else:
|
||
logger.info("imap_sync_account: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id)
|
||
await db.delete(db_mail)
|
||
else:
|
||
logger.info("imap_sync_account: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id)
|
||
await db.delete(db_mail)
|
||
except Exception as mid_exc:
|
||
logger.warning("imap_sync_account: Message-ID search failed for mail %s: %s", db_mail.id, mid_exc)
|
||
await db.flush()
|
||
except Exception as vanished_exc:
|
||
logger.warning("imap_sync_account: vanished-UID check failed for folder %s: %s", folder.imap_name, vanished_exc)
|
||
|
||
# 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:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
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:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
await db.flush()
|
||
except Exception:
|
||
logger.debug("Ignored exception in mail service", exc_info=True)
|
||
|
||
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) ───
|