app/processor.py

"""Standalone attachment routing process.

Run with:  python -m app.processor

Polls the SQLite DB for notifications with processing_status='fetched', classifies
each attachment via OCR, and uploads routable documents to Dropbox.
"""
from __future__ import annotations

import asyncio
import json
import logging
import logging.handlers
import signal
import traceback
from datetime import datetime
from pathlib import Path

from app.config import settings
from app.database import (
    get_db,
    get_fetched_notifications,
    log_routing_result,
    set_notification_routing_status,
)
from app.lib.utils.classifier import ClassifyResult, ScoringWeights, classify
from app.lib.utils.client_routing import resolve_client_name
from app.lib.utils.dropbox_client import DropboxClient
from app.lib.utils.error_handler import processing_error_handler
from app.lib.utils.heat_code import extract_heat_codes
from app.lib.utils.notifier import send_developer_alert

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [processor] %(levelname)s %(message)s",
)
logger = logging.getLogger(__name__)

_DEST_MAP = {
    "po": settings.dropbox_po_path,
    "invoice": settings.dropbox_invoice_path,
    "mtr": settings.dropbox_mtr_path,
}

_shutdown = asyncio.Event()


def _setup_log_file() -> None:
    settings.log_path.mkdir(parents=True, exist_ok=True)
    handler = logging.handlers.TimedRotatingFileHandler(
        filename=settings.log_path / "processor.log",
        when="midnight",
        interval=1,
        backupCount=0,
        encoding="utf-8",
    )
    handler.setFormatter(logging.Formatter("%(asctime)s [processor] %(levelname)s %(message)s"))
    logging.getLogger().addHandler(handler)


def _handle_signal(*_: object) -> None:
    logger.info("Shutdown signal received")
    _shutdown.set()


def _date_prefix(received_dt: str) -> str:
    try:
        return datetime.fromisoformat(received_dt).strftime("%Y%m%d")
    except Exception:
        return "00000000"


def _sanitize_path_segment(s: str) -> str:
    """Make a string safe for use as a Dropbox path segment.

    Strips surrounding whitespace and replaces any character that isn't an
    ASCII letter, digit, dash, dot, or underscore with an underscore. Returns
    "unknown" if the result would be empty.
    """
    import re
    s = (s or "").strip()
    cleaned = re.sub(r"[^A-Za-z0-9._-]", "_", s)
    return cleaned or "unknown"


# Phase 4 — Project brief says any attachment whose subject OR filename
# contains "MTR" (case-insensitive, word-boundary) should be routed as MTR
# regardless of OCR scoring. The classifier scores filename+OCR keywords but
# never looks at the subject, so we add this pre-classifier shortcut.
_MTR_TRIGGER_RE = __import__("re").compile(r"\bmtr\b", __import__("re").IGNORECASE)


def _is_mtr_by_subject_or_filename(subject: str, filename: str) -> bool:
    if subject and _MTR_TRIGGER_RE.search(subject):
        return True
    if filename and _MTR_TRIGGER_RE.search(filename):
        return True
    return False


def _scoring_weights() -> ScoringWeights:
    return ScoringWeights(
        filename=settings.score_weight_filename,
        ocr=settings.score_weight_ocr,
        freq_multiplier=settings.score_freq_multiplier,
        min_threshold=settings.score_min_threshold,
    )


async def _process_notification(dbx: DropboxClient, notif: dict) -> None:
    notif_id: int = notif["id"]
    email_identifier: str = notif.get("email_identifier") or notif.get("message_id") or ""
    storage_path = notif.get("storage_path")

    # Notification-level setup: load metadata before we know filename/sender.
    # On failure, error_handler logs + writes to DB, then re-raises — we catch
    # that here to mark the notification as routing_error and return early.
    try:
        async with processing_error_handler(
            settings.database_path, notif_id, email_identifier
        ):
            if not storage_path:
                raise ValueError("no storage_path recorded")
            meta = json.loads(Path(storage_path).joinpath("metadata.json").read_text())
    except Exception:
        async with get_db(settings.database_path) as conn:
            await set_notification_routing_status(conn, notif_id, "routing_error")
        return

    sender_email: str = meta.get("from", {}).get("address", "")
    subject: str = meta.get("subject", "")
    received_dt: str = meta.get("received_datetime", "")
    date_prefix = _date_prefix(received_dt)
    attachments: list[dict] = meta.get("attachments", [])
    weights = _scoring_weights()

    uploaded_count = 0
    final_status = "routing_skipped"

    for att in attachments:
        filename: str = att.get("filename", "attachment")
        content_type: str = att.get("content_type", "")
        att_path = Path(storage_path) / "attachments" / filename

        try:
            async with processing_error_handler(
                settings.database_path, notif_id, email_identifier,
                sender_email, subject, filename,
            ):
                file_bytes = att_path.read_bytes()

                # Phase 4: subject-line / filename MTR trigger from the
                # project brief. Bypasses keyword scoring if "MTR" appears
                # as a word in either the email subject or attachment
                # filename. Only applied to PDFs — non-PDFs still go
                # through the classifier and get skipped on content type.
                if (
                    content_type.startswith("application/pdf")
                    and _is_mtr_by_subject_or_filename(subject, filename)
                ):
                    logger.info(
                        "Notification %d / %s → mtr (subject/filename trigger)",
                        notif_id, filename,
                    )
                    result: ClassifyResult = ClassifyResult(
                        doc_type="mtr",
                        reason="routed",
                        scores={"mtr": 1.0},
                    )
                else:
                    result = classify(filename, file_bytes, content_type, weights)
                    logger.info(
                        "Notification %d / %s → %s (reason=%s, scores=%s)",
                        notif_id, filename, result.doc_type, result.reason, result.scores,
                    )

                if result.breakdown:
                    lines = [f"  Score breakdown — {filename}:"]
                    for doc_type, matches in sorted(
                        result.breakdown.items(),
                        key=lambda kv: result.scores.get(kv[0], 0.0),
                        reverse=True,
                    ):
                        lines.append(f"    {doc_type}  total={result.scores[doc_type]:.4f}")
                        for m in matches:
                            parts = []
                            if m.filename_hits:
                                parts.append(f"filename={m.filename_hits}× (+{m.filename_contrib:.4f})")
                            if m.ocr_hits:
                                parts.append(f"ocr={m.ocr_hits}× (+{m.ocr_contrib:.4f})")
                            lines.append(
                                f"      \"{m.keyword}\"  weight={m.weight:.2f}  {'  '.join(parts)}"
                            )
                    logger.info("\n".join(lines))

                if result.reason == "non_pdf":
                    async with get_db(settings.database_path) as conn:
                        await log_routing_result(
                            conn, notif_id, filename, "skipped", "skipped",
                            email_identifier=email_identifier,
                        )

                elif result.reason == "ambiguous":
                    async with get_db(settings.database_path) as conn:
                        from app.database import insert_daily_brief
                        await log_routing_result(
                            conn, notif_id, filename, "ambiguous", "brief_pending",
                            email_identifier=email_identifier,
                            confidence_scores=result.scores,
                        )
                        await insert_daily_brief(
                            conn,
                            email_identifier=email_identifier,
                            notification_id=notif_id,
                            sender_email=sender_email,
                            subject=subject,
                            filename=filename,
                            reason="ambiguous",
                            confidence_scores=result.scores,
                        )
                    final_status = "brief_pending"

                elif result.reason in ("below_threshold", "not_routable"):
                    async with get_db(settings.database_path) as conn:
                        await log_routing_result(
                            conn, notif_id, filename, result.doc_type, "skipped",
                            email_identifier=email_identifier,
                            confidence_scores=result.scores,
                        )

                else:
                    # result.reason == "routed"
                    month_folder = received_dt[:7] if received_dt else "unknown"
                    type_path = _DEST_MAP[result.doc_type]

                    if result.doc_type == "mtr":
                        # MTRs: extract heat code(s) and upload one copy per
                        # code under {mtr_path}/{heat_code}/{YYYY-MM}/...
                        # If no heat code can be extracted, fall back to
                        # {mtr_path}/_no_heat_code/{YYYY-MM}/... and flag
                        # the row in the daily brief for ops to manually re-file.
                        heat_codes = extract_heat_codes(file_bytes, filename=filename)
                        logger.info(
                            "MTR %s heat codes: %s",
                            filename, heat_codes or "<none — fallback>",
                        )

                        if heat_codes:
                            for heat_code in heat_codes:
                                safe_code = _sanitize_path_segment(heat_code)
                                dest_path = (
                                    f"{type_path}/{safe_code}/{month_folder}/"
                                    f"{date_prefix}_{filename}"
                                )
                                actual_path = await asyncio.to_thread(
                                    dbx.upload, file_bytes, dest_path
                                )
                                logger.info(
                                    "Uploaded MTR %s (heat=%s) → %s",
                                    filename, heat_code, actual_path,
                                )
                                async with get_db(settings.database_path) as conn:
                                    await log_routing_result(
                                        conn, notif_id, filename, "mtr", "uploaded",
                                        email_identifier=email_identifier,
                                        destination_path=actual_path,
                                        confidence_scores=result.scores,
                                    )
                                uploaded_count += 1
                            final_status = "routed"
                        else:
                            # No heat code → fallback location + daily brief flag.
                            dest_path = (
                                f"{type_path}/_no_heat_code/{month_folder}/"
                                f"{date_prefix}_{filename}"
                            )
                            actual_path = await asyncio.to_thread(
                                dbx.upload, file_bytes, dest_path
                            )
                            logger.info(
                                "Uploaded MTR %s (no heat code extracted) → %s",
                                filename, actual_path,
                            )
                            async with get_db(settings.database_path) as conn:
                                from app.database import insert_daily_brief
                                await log_routing_result(
                                    conn, notif_id, filename, "mtr",
                                    "uploaded_no_heat_code",
                                    email_identifier=email_identifier,
                                    destination_path=actual_path,
                                    confidence_scores=result.scores,
                                )
                                await insert_daily_brief(
                                    conn,
                                    email_identifier=email_identifier,
                                    notification_id=notif_id,
                                    sender_email=sender_email,
                                    subject=subject,
                                    filename=filename,
                                    reason="mtr_no_heat_code",
                                    confidence_scores=result.scores,
                                )
                            uploaded_count += 1
                            final_status = "brief_pending"

                    else:
                        # PO / Invoice — Phase 2 client folder routing.
                        # Look up the sender's email domain in client_domains.
                        # If a mapping exists, insert {client}/ between the
                        # type root and the YYYY-MM month folder. Otherwise,
                        # route to {type_path}/_unmatched/ and flag in the
                        # daily brief so ops can add the mapping.
                        async with get_db(settings.database_path) as conn:
                            client_name = await resolve_client_name(conn, sender_email)

                        if client_name:
                            safe_client = _sanitize_path_segment(client_name)
                            dest_path = (
                                f"{type_path}/{safe_client}/{month_folder}/"
                                f"{date_prefix}_{filename}"
                            )
                            actual_path = await asyncio.to_thread(
                                dbx.upload, file_bytes, dest_path
                            )
                            logger.info(
                                "Uploaded %s (client=%s) → %s",
                                filename, client_name, actual_path,
                            )
                            async with get_db(settings.database_path) as conn:
                                await log_routing_result(
                                    conn, notif_id, filename, result.doc_type, "uploaded",
                                    email_identifier=email_identifier,
                                    destination_path=actual_path,
                                    confidence_scores=result.scores,
                                )
                            uploaded_count += 1
                            final_status = "routed"
                        else:
                            # Unknown sender domain — fallback + daily brief.
                            dest_path = (
                                f"{type_path}/_unmatched/{month_folder}/"
                                f"{date_prefix}_{filename}"
                            )
                            actual_path = await asyncio.to_thread(
                                dbx.upload, file_bytes, dest_path
                            )
                            logger.info(
                                "Uploaded %s (unmatched sender=%s) → %s",
                                filename, sender_email, actual_path,
                            )
                            async with get_db(settings.database_path) as conn:
                                from app.database import insert_daily_brief
                                await log_routing_result(
                                    conn, notif_id, filename, result.doc_type,
                                    "uploaded_unmatched_client",
                                    email_identifier=email_identifier,
                                    destination_path=actual_path,
                                    confidence_scores=result.scores,
                                )
                                await insert_daily_brief(
                                    conn,
                                    email_identifier=email_identifier,
                                    notification_id=notif_id,
                                    sender_email=sender_email,
                                    subject=subject,
                                    filename=filename,
                                    reason="client_unmatched",
                                    confidence_scores=result.scores,
                                )
                            uploaded_count += 1
                            final_status = "brief_pending"

        except Exception:
            # processing_error_handler already logged and wrote to DB; continue to next attachment.
            final_status = "routing_error"

    async with get_db(settings.database_path) as conn:
        await set_notification_routing_status(conn, notif_id, final_status)
    logger.info("Notification %d → %s", notif_id, final_status)


async def run() -> None:
    dbx = DropboxClient(
        app_key=settings.dropbox_app_key,
        app_secret=settings.dropbox_app_secret.get_secret_value(),
        refresh_token=settings.dropbox_refresh_token.get_secret_value(),
    )
    logger.info("Processor started; polling every %ds", settings.processor_poll_interval)

    while not _shutdown.is_set():
        try:
            async with get_db(settings.database_path) as conn:
                pending = await get_fetched_notifications(conn)

            if pending:
                logger.info("Found %d notification(s) to process", len(pending))
                for notif in pending:
                    async with get_db(settings.database_path) as conn:
                        await set_notification_routing_status(conn, notif["id"], "routing_queued")

                for notif in pending:
                    await _process_notification(dbx, notif)
        except Exception:
            tb = traceback.format_exc()
            logger.exception("Unhandled error in processor loop")
            await send_developer_alert(
                settings,
                subject="[RCM] Processor loop unhandled exception",
                body=(
                    f"An unhandled exception was caught in the processor poll loop. "
                    f"Processing will resume after the next poll interval "
                    f"({settings.processor_poll_interval}s).\n\n"
                    f"{tb}"
                ),
            )

        try:
            await asyncio.wait_for(_shutdown.wait(), timeout=settings.processor_poll_interval)
        except asyncio.TimeoutError:
            pass

    logger.info("Processor stopped")


def main() -> None:
    _setup_log_file()
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, _handle_signal)

    try:
        loop.run_until_complete(run())
    finally:
        loop.close()


if __name__ == "__main__":
    main()