Phase 2+4: PO/Invoice client folder routing and subject-line MTR trigger

Phase 2 — Client folder routing:
- New table client_domains (domain primary key, client_name, audit fields).
- New module app/lib/utils/client_routing.py with resolve_client_name()
and add_or_update_client_domain() helpers.
- processor.py: PO/Invoice uploads now insert {client_name}/ between the
type path and the YYYY-MM folder. Unknown sender domains route to
{type_path}/_unmatched/ and add a daily_brief row with reason
'client_unmatched' so ops can add the mapping without redeploy.
- Daily brief gets a new 'Unmatched Sender Domains' section.
- Alembic migration 0002 + matching DDL in database.py.
- Seed script: scripts/seed_client_domains.{json,py} for initial mappings.

Phase 4 — Subject-line MTR trigger:
- Per project brief: any PDF attachment whose subject or filename contains
'MTR' (case-insensitive, word-boundary) routes as MTR without keyword
scoring. Heat code extraction still runs.

alembic/versions/0002_add_client_domains.py

@@ -0,0 +1,39 @@
+"""Add client_domains table — sender-domain to client-folder-name mapping.
+
+Used by Phase 2 client routing: when a PO or Invoice is classified,
+we look up the sender's email domain in this table to decide which
+client-name subfolder to place the file under in Dropbox.
+
+Domains are stored lowercase as the primary key. Unmatched senders are
+recorded in the daily_brief with reason='client_unmatched' so ops can
+add new mappings over time without a code change or redeploy.
+
+Revision ID: 0002
+Revises: 0001
+Create Date: 2026-05-17
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0002"
+down_revision: Union[str, None] = "0001"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS client_domains (
+            domain        TEXT PRIMARY KEY,
+            client_name   TEXT NOT NULL,
+            created_at    TEXT NOT NULL DEFAULT (datetime('now')),
+            last_seen_at  TEXT,
+            notes         TEXT
+        )
+    """)
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE IF EXISTS client_domains")

app/database.py

@@ -62,6 +62,14 @@ CREATE TABLE IF NOT EXISTS daily_brief (
     included_in_report  INTEGER NOT NULL DEFAULT 0,
     report_sent_at      TEXT
 );
+
+CREATE TABLE IF NOT EXISTS client_domains (
+    domain        TEXT PRIMARY KEY,
+    client_name   TEXT NOT NULL,
+    created_at    TEXT NOT NULL DEFAULT (datetime('now')),
+    last_seen_at  TEXT,
+    notes         TEXT
+);
 """
 
 

app/lib/utils/brief_html.py

@@ -21,6 +21,7 @@ def build_brief_html(items: list[dict]) -> str:
     """
     ambiguous = [i for i in items if i["reason"] == "ambiguous"]
     no_heat   = [i for i in items if i["reason"] == "mtr_no_heat_code"]
+    unmatched = [i for i in items if i["reason"] == "client_unmatched"]
     errors    = [i for i in items if i["reason"] == "error"]
 
     sections: list[str] = []
@@ -29,6 +30,7 @@ def build_brief_html(items: list[dict]) -> str:
     summary_bits = []
     if ambiguous: summary_bits.append(f"{len(ambiguous)} ambiguous routing(s)")
     if no_heat:   summary_bits.append(f"{len(no_heat)} MTR(s) without a heat code")
+    if unmatched: summary_bits.append(f"{len(unmatched)} unmatched sender domain(s)")
     if errors:    summary_bits.append(f"{len(errors)} processing error(s)")
     if summary_bits:
         sections.append(
@@ -95,6 +97,31 @@ def build_brief_html(items: list[dict]) -> str:
             "</table>"
         )
 
+    if unmatched:
+        rows = ""
+        for item in unmatched:
+            rows += (
+                f"<tr>"
+                f"<td>{item.get('sender_email', '')}</td>"
+                f"<td>{item.get('subject', '')}</td>"
+                f"<td>{item.get('attachment_filename', '')}</td>"
+                f"</tr>"
+            )
+        sections.append(
+            "<h2>Unmatched Sender Domains</h2>"
+            "<p>These POs/Invoices were uploaded to "
+            "<code>POs/_unmatched/</code> or <code>Invoices/_unmatched/</code> "
+            "because the sender's email domain isn't in the client mapping. "
+            "Move the file under the right client folder in Dropbox, then "
+            "add the domain → client mapping to "
+            "<code>scripts/seed_client_domains.json</code> and re-run the "
+            "seeder so future emails route correctly.</p>"
+            "<table border='1' cellpadding='4' cellspacing='0'>"
+            "<tr><th>Sender</th><th>Subject</th><th>Attachment</th></tr>"
+            f"{rows}"
+            "</table>"
+        )
+
     if errors:
         rows = ""
         for item in errors:

app/lib/utils/client_routing.py

@@ -0,0 +1,108 @@
+"""Resolve a sender's email domain to the Dropbox client-folder name.
+
+Used for routing Purchase Order and Invoice attachments under a per-client
+subfolder. Unknown domains route to `_unmatched` and are surfaced in the
+daily brief so ops can add a mapping without a redeploy.
+
+The mapping table is seeded via `scripts/seed_client_domains.py` and can
+be edited at runtime by inserting rows directly into the `client_domains`
+table (no server restart needed — lookups happen per request).
+"""
+from __future__ import annotations
+
+import logging
+import re
+from typing import Optional
+
+import aiosqlite
+
+logger = logging.getLogger(__name__)
+
+
+# Conservative domain extraction. We accept lowercase letters, digits, dot,
+# and hyphen — the standard DNS character set. Anything else gets rejected
+# so we don't pollute the table with junk like "foo@bar baz.com".
+_DOMAIN_RE = re.compile(r"^[a-z0-9.\-]+$")
+
+
+def _extract_domain(sender_email: str) -> Optional[str]:
+    """Return the lowercased, validated domain portion of an email, or None.
+
+    Strips angle brackets and surrounding whitespace; accepts "<a@b.com>",
+    "  a@b.com  ", "FOO@BAR.COM". Rejects malformed addresses outright.
+    """
+    if not sender_email:
+        return None
+    s = sender_email.strip().strip("<>").lower()
+    if "@" not in s:
+        return None
+    _local, _, domain = s.rpartition("@")
+    domain = domain.strip()
+    if not domain or not _DOMAIN_RE.match(domain):
+        return None
+    return domain
+
+
+async def resolve_client_name(
+    conn: aiosqlite.Connection,
+    sender_email: str,
+) -> Optional[str]:
+    """Look up the client-folder name for a sender's email domain.
+
+    Returns the client name string if a mapping exists, otherwise None.
+    Updates `last_seen_at` on the matched row so ops can see which mappings
+    are actively being used.
+
+    The conn is expected to be opened with `aiosqlite.Row` as row factory
+    (the standard `get_db` context manager in app.database does this).
+    """
+    domain = _extract_domain(sender_email)
+    if domain is None:
+        logger.debug("resolve_client_name: rejected sender %r", sender_email)
+        return None
+
+    cursor = await conn.execute(
+        "SELECT client_name FROM client_domains WHERE domain = ?",
+        (domain,),
+    )
+    row = await cursor.fetchone()
+    if row is None:
+        return None
+
+    # Touch last_seen_at so ops can see mapping activity.
+    await conn.execute(
+        "UPDATE client_domains SET last_seen_at = datetime('now') WHERE domain = ?",
+        (domain,),
+    )
+    await conn.commit()
+
+    return row["client_name"]
+
+
+async def add_or_update_client_domain(
+    conn: aiosqlite.Connection,
+    domain: str,
+    client_name: str,
+    notes: Optional[str] = None,
+) -> None:
+    """Upsert a domain→client mapping.
+
+    Used by the seed script and by any future admin tool. `domain` is
+    normalized lowercase; `client_name` is stored as-is (so case can be
+    preserved in the Dropbox folder name, e.g. "Westlake").
+    """
+    domain_norm = (domain or "").strip().lower()
+    if not _DOMAIN_RE.match(domain_norm):
+        raise ValueError(f"Invalid domain: {domain!r}")
+
+    await conn.execute(
+        """
+        INSERT INTO client_domains (domain, client_name, notes)
+        VALUES (?, ?, ?)
+        ON CONFLICT(domain) DO UPDATE SET
+            client_name = excluded.client_name,
+            notes       = excluded.notes
+        """,
+        (domain_norm, client_name.strip(), notes),
+    )
+    await conn.commit()

app/processor.py

@@ -24,6 +24,7 @@ from app.database import (
     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
@@ -82,6 +83,21 @@ def _sanitize_path_segment(s: str) -> str:
     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,
@@ -132,12 +148,31 @@ async def _process_notification(dbx: DropboxClient, notif: dict) -> None:
                 sender_email, subject, filename,
             ):
                 file_bytes = att_path.read_bytes()
-                result: ClassifyResult = 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,
-                )
+                # 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}:"]
@@ -269,25 +304,71 @@ async def _process_notification(dbx: DropboxClient, notif: dict) -> None:
                             final_status = "brief_pending"
 
                     else:
-                        # po / invoice — single upload at the type root.
-                        # (Phase 2 will insert a {client} subfolder here based
-                        # on the sender's email domain.)
-                        dest_path = (
-                            f"{type_path}/{month_folder}/{date_prefix}_{filename}"
-                        )
-                        actual_path = await asyncio.to_thread(
-                            dbx.upload, file_bytes, dest_path
-                        )
-                        logger.info("Uploaded %s → %s", filename, actual_path)
+                        # 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:
-                            await log_routing_result(
-                                conn, notif_id, filename, result.doc_type, "uploaded",
-                                email_identifier=email_identifier,
-                                destination_path=actual_path,
-                                confidence_scores=result.scores,
+                            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,
                             )
-                        uploaded_count += 1
-                        final_status = "routed"
+                            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.

scripts/preview_brief.py

@@ -68,9 +68,31 @@ SAMPLE_ITEMS: list[dict] = [
         }),
     },
 
-    # One processing error.
+    # Two unmatched sender domains.
     {
         "id": 5,
+        "sender_email": "purchasing@newvendor-inc.com",
+        "subject": "RCM - PO from a vendor we haven't seen before",
+        "attachment_filename": "NewVendor_PO_2026-04-22.pdf",
+        "reason": "client_unmatched",
+        "confidence_scores": json.dumps({
+            "po": 0.86, "invoice": 0.08, "mtr": 0.02, "quote": 0.04,
+        }),
+    },
+    {
+        "id": 6,
+        "sender_email": "billing@small-shop.com",
+        "subject": "RCM - Invoice March 2026",
+        "attachment_filename": "Invoice_March_2026.pdf",
+        "reason": "client_unmatched",
+        "confidence_scores": json.dumps({
+            "po": 0.11, "invoice": 0.84, "mtr": 0.01, "quote": 0.03,
+        }),
+    },
+
+    # One processing error.
+    {
+        "id": 7,
         "sender_email": "noreply@example-vendor.com",
         "subject": "RCM - corrupted attachment",
         "attachment_filename": "scan_2026-05-10_corrupted.pdf",

scripts/seed_client_domains.json

@@ -0,0 +1,6 @@
+{
+  "_README": "Initial seed mapping for client_domains table. Add new entries here, then run scripts/seed_client_domains.py to upsert them into the DB. Domain keys are lowercase. client_name is the exact Dropbox folder name to use.",
+
+  "westlake.com": "Westlake",
+  "halliburton.com": "Halliburton"
+}

scripts/seed_client_domains.py

@@ -0,0 +1,45 @@
+"""Seed the client_domains table from scripts/seed_client_domains.json.
+
+Idempotent — running this multiple times updates existing rows in place
+instead of erroring. Safe to run after deploys or whenever you've added
+new mappings to the JSON file.
+
+Usage:  python -m scripts.seed_client_domains
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+# Make repo root importable when invoked as `python scripts/seed_client_domains.py`
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from app.config import settings
+from app.database import get_db
+from app.lib.utils.client_routing import add_or_update_client_domain
+
+
+SEED_FILE = Path(__file__).resolve().parent / "seed_client_domains.json"
+
+
+async def main() -> None:
+    data = json.loads(SEED_FILE.read_text())
+    # Strip the README-style key if present.
+    data.pop("_README", None)
+
+    if not data:
+        print("Nothing to seed — JSON file has no mappings.")
+        return
+
+    async with get_db(settings.database_path) as conn:
+        for domain, client_name in data.items():
+            await add_or_update_client_domain(conn, domain, client_name)
+            print(f"  {domain:40s} → {client_name}")
+
+    print(f"\nSeeded {len(data)} domain mapping(s) into {settings.database_path}.")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())