leocrm/app/plugins/builtins/mail/attachments.py

"""Attachment storage helpers for the Mail plugin.

Extracted from services.py as part of the God-object split (BUG-018 pilot).
Re-exported by ``app.plugins.builtins.mail.services``.
"""

from __future__ import annotations

import os
import re
import uuid

import aiofiles

from app.config import settings
from app.plugins.builtins.mail.models import MailAttachment

# ─── 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 base64

            def decode_q(match):
                charset, encoding, encoded = (
                    match.group(1),
                    match.group(2).upper(),
                    match.group(3),
                )
                if encoding == "B":
                    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,
    }