foodlinkk-command-center/email-agent/app/imap_sync.py

"""IMAP sync and CEO email approval helpers."""
from __future__ import annotations

import email
import imaplib
import json
from datetime import datetime, timezone
from email.header import decode_header
from typing import Any, Optional

import httpx

from app.config import settings
from app.db import execute, execute_returning, fetch_all, fetch_one, json_param


def _decode_header(value: Optional[str]) -> str:
    if not value:
        return ""
    parts = decode_header(value)
    out: list[str] = []
    for chunk, enc in parts:
        if isinstance(chunk, bytes):
            out.append(chunk.decode(enc or "utf-8", errors="replace"))
        else:
            out.append(str(chunk))
    return " ".join(out).strip()


def _accounts_to_sync() -> list[dict[str, Any]]:
    return fetch_all(
        """
        SELECT id, label, email_address, imap_host, imap_port, imap_user,
               imap_password, imap_use_ssl, last_sync_at
        FROM email_accounts
        WHERE is_active = TRUE AND sync_enabled = TRUE
          AND imap_host IS NOT NULL AND imap_host <> ''
        ORDER BY id
        """
    )


def sync_all_accounts() -> dict[str, Any]:
    accounts = _accounts_to_sync()
    if not accounts:
        return {"accounts": 0, "imported": 0, "skipped": 0, "message": "No IMAP accounts configured"}

    imported = skipped = 0
    details: list[dict[str, Any]] = []
    for acct in accounts:
        try:
            result = _sync_account(acct)
            imported += result["imported"]
            skipped += result["skipped"]
            details.append({"account_id": acct["id"], **result})
        except Exception as exc:  # noqa: BLE001
            details.append({"account_id": acct["id"], "error": str(exc)})

    return {"accounts": len(accounts), "imported": imported, "skipped": skipped, "details": details}


def _sync_account(acct: dict[str, Any]) -> dict[str, Any]:
    host = acct["imap_host"]
    port = int(acct.get("imap_port") or 993)
    user = acct.get("imap_user") or acct["email_address"]
    password = acct.get("imap_password") or ""
    use_ssl = acct.get("imap_use_ssl", True)

    if use_ssl:
        mail = imaplib.IMAP4_SSL(host, port)
    else:
        mail = imaplib.IMAP4(host, port)
    mail.login(user, password)
    mail.select("INBOX")

    status, data = mail.search(None, "UNSEEN")
    if status != "OK":
        mail.logout()
        return {"imported": 0, "skipped": 0}

    ids = data[0].split() if data[0] else []
    imported = skipped = 0
    for num in ids[-50:]:
        status, msg_data = mail.fetch(num, "(RFC822)")
        if status != "OK" or not msg_data or not msg_data[0]:
            continue
        raw = msg_data[0][1]
        msg = email.message_from_bytes(raw)
        message_id = (msg.get("Message-ID") or f"local-{acct['id']}-{num.decode()}").strip()
        existing = fetch_one("SELECT id FROM emails WHERE message_id = %s", (message_id,))
        if existing:
            skipped += 1
            continue

        subject = _decode_header(msg.get("Subject"))
        from_addr = _decode_header(msg.get("From"))
        to_addrs = [_decode_header(msg.get("To"))] if msg.get("To") else []
        body_text = ""
        if msg.is_multipart():
            for part in msg.walk():
                if part.get_content_type() == "text/plain" and not part.get_filename():
                    payload = part.get_payload(decode=True)
                    if payload:
                        body_text = payload.decode(part.get_content_charset() or "utf-8", errors="replace")
                    break
        else:
            payload = msg.get_payload(decode=True)
            if payload:
                body_text = payload.decode(msg.get_content_charset() or "utf-8", errors="replace")

        row = execute_returning(
            """
            INSERT INTO emails (
                message_id, direction, from_addr, to_addrs, subject, body_text,
                received_at, is_read, raw_headers
            ) VALUES (%s, 'in', %s, %s, %s, %s, %s, FALSE, %s)
            RETURNING id
            """,
            (
                message_id,
                from_addr,
                to_addrs,
                subject,
                body_text[:50000] if body_text else None,
                datetime.now(timezone.utc),
                json_param({"account_id": acct["id"], "label": acct["label"]}),
            ),
        )
        imported += 1
        if row:
            _create_review_recommendation(int(row["id"]), subject, from_addr)

    execute(
        "UPDATE email_accounts SET last_sync_at = NOW() WHERE id = %s",
        (acct["id"],),
    )
    mail.logout()
    return {"imported": imported, "skipped": skipped}


def _create_review_recommendation(email_id: int, subject: str, from_addr: str) -> None:
    execute_returning(
        """
        INSERT INTO ai_recommendations (
            recommendation_type, title, description, priority, status,
            generated_by, related_entity_type, related_entity_id, data_sources
        ) VALUES (
            'email_review', %s, %s, 'medium', 'pending',
            'email-agent', 'email', %s, %s
        )
        RETURNING id
        """,
        (
            f"Nieuwe e-mail: {subject[:200] or '(geen onderwerp)'}",
            f"Inkomend bericht van {from_addr}. Beoordelen en eventueel beantwoorden.",
            email_id,
            json_param({"email_id": email_id, "from": from_addr}),
        ),
    )


def list_pending_drafts() -> list[dict[str, Any]]:
    return fetch_all(
        """
        SELECT id, title, description, status, data_sources, created_at
        FROM ai_recommendations
        WHERE recommendation_type IN ('email_draft', 'email_review')
          AND status = 'pending'
        ORDER BY created_at DESC
        LIMIT 50
        """
    )


def create_draft(title: str, body: str, to_addr: str, subject: str) -> dict[str, Any]:
    row = execute_returning(
        """
        INSERT INTO ai_recommendations (
            recommendation_type, title, description, priority, status,
            generated_by, data_sources
        ) VALUES (
            'email_draft', %s, %s, 'high', 'pending', 'email-agent', %s
        )
        RETURNING id, title, status, created_at
        """,
        (
            title,
            body,
            json_param({"to": to_addr, "subject": subject, "body": body}),
        ),
    )
    return dict(row or {})


def approve_draft(rec_id: int) -> dict[str, Any]:
    rec = fetch_one(
        """
        SELECT id, recommendation_type, data_sources, status
        FROM ai_recommendations WHERE id = %s
        """,
        (rec_id,),
    )
    if not rec:
        raise ValueError("Recommendation not found")
    if rec["status"] != "pending":
        raise ValueError(f"Already {rec['status']}")

    ds = rec.get("data_sources") or {}
    if isinstance(ds, str):
        ds = json.loads(ds)

    if rec["recommendation_type"] == "email_draft":
        payload = {
            "to": ds.get("to", ""),
            "subject": ds.get("subject", rec.get("title", "")),
            "body": ds.get("body", ""),
        }
        with httpx.Client(timeout=30.0) as client:
            resp = client.post(f"{settings.TOOLS_API_URL}/emails/send", json=payload)
            resp.raise_for_status()
            send_result = resp.json()
    else:
        send_result = {"action": "marked_reviewed"}

    execute(
        "UPDATE ai_recommendations SET status = 'approved', updated_at = NOW() WHERE id = %s",
        (rec_id,),
    )
    return {"recommendation_id": rec_id, "send_result": send_result, "status": "approved"}