feat: live IMAP sync on folder select + 60s auto-sync interval
- New imap_sync_folder function: syncs single folder quickly for live use
- New POST /mail/folders/{folder_id}/sync route
- Frontend: syncFolder API + called on every folder selection
- Auto-sync interval reduced from 300s to 60s
- Sent mail IMAP APPEND already uploads full body via msg.as_bytes()
This commit is contained in:
Agent Zero
@@ -536,6 +536,226 @@ def _build_folder_hierarchy(
|
||||
return result
|
||||
|
||||
|
||||
async def imap_sync_folder(
|
||||
db: AsyncSession,
|
||||
folder_id: uuid.UUID,
|
||||
tenant_id: uuid.UUID,
|
||||
) -> dict:
|
||||
"""Sync a single folder from IMAP server — fast, for live use when selecting a folder."""
|
||||
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"}
|
||||
|
||||
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}"}
|
||||
|
||||
# Fetch UIDs — get last MAX_EMAILS_PER_FOLDER
|
||||
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 = []
|
||||
|
||||
if 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)
|
||||
|
||||
# Check if we already have this UID
|
||||
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 existing_mail:
|
||||
# Restore soft-deleted mails if they still exist on IMAP
|
||||
if existing_mail.deleted_at is not None:
|
||||
existing_mail.deleted_at = None
|
||||
logger.info("imap_sync_folder: restored soft-deleted mail %s", existing_mail.id)
|
||||
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
|
||||
|
||||
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,
|
||||
|
||||
Reference in New Issue
Block a user