Phase 1: heat code extraction and multi-folder MTR routing

app/lib/utils/brief_html.py

@@ -0,0 +1,126 @@
+"""Daily Brief HTML rendering.
+
+Pure-stdlib module — has no dependencies on the rest of the app, so it can
+be imported by preview scripts, tests, and the scheduler alike. The shape
+of the items list matches a SELECT * from the `daily_brief` table.
+"""
+from __future__ import annotations
+
+import json
+
+
+def build_brief_html(items: list[dict]) -> str:
+    """Render the Daily Brief email body from `daily_brief` rows.
+
+    Sections rendered (in order, only if non-empty):
+        1. Ambiguous Routings — `reason == "ambiguous"`
+        2. MTRs Missing Heat Code — `reason == "mtr_no_heat_code"`
+        3. Processing Errors — `reason == "error"`
+
+    If all sections are empty, returns a short "nothing to report" body.
+    """
+    ambiguous = [i for i in items if i["reason"] == "ambiguous"]
+    no_heat   = [i for i in items if i["reason"] == "mtr_no_heat_code"]
+    errors    = [i for i in items if i["reason"] == "error"]
+
+    sections: list[str] = []
+
+    # One-line summary at top so ops doesn't need to count rows.
+    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 errors:    summary_bits.append(f"{len(errors)} processing error(s)")
+    if summary_bits:
+        sections.append(
+            f"<p>Today's automation summary: {'; '.join(summary_bits)}.</p>"
+        )
+
+    if ambiguous:
+        rows = ""
+        for item in ambiguous:
+            scores: dict = (
+                json.loads(item["confidence_scores"])
+                if item.get("confidence_scores") else {}
+            )
+            score_cols = "".join(
+                f"<td>{doc_type}: {score:.2%}</td>"
+                for doc_type, score in sorted(scores.items())
+            )
+            rows += (
+                f"<tr>"
+                f"<td>{item.get('sender_email', '')}</td>"
+                f"<td>{item.get('subject', '')}</td>"
+                f"<td>{item.get('attachment_filename', '')}</td>"
+                f"{score_cols}"
+                f"</tr>"
+            )
+        first_scores = (
+            json.loads(ambiguous[0]["confidence_scores"])
+            if ambiguous[0].get("confidence_scores") else {}
+        )
+        score_headers = "".join(
+            f"<th>{doc_type}</th>" for doc_type in sorted(first_scores.keys())
+        )
+        sections.append(
+            "<h2>Ambiguous Routings</h2>"
+            "<p>These attachments scored above the confidence threshold for "
+            "more than one document type and were not auto-routed. Review and "
+            "file manually.</p>"
+            "<table border='1' cellpadding='4' cellspacing='0'>"
+            f"<tr><th>Sender</th><th>Subject</th><th>Attachment</th>{score_headers}</tr>"
+            f"{rows}"
+            "</table>"
+        )
+
+    if no_heat:
+        rows = ""
+        for item in no_heat:
+            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>MTRs Missing Heat Code</h2>"
+            "<p>These MTRs were uploaded to <code>MTR/_no_heat_code/</code> "
+            "because no heat code could be extracted from the document. "
+            "Please review the file in Dropbox, identify the heat code(s), "
+            "and move/copy under the correct <code>MTR/&lt;heat_code&gt;/</code> "
+            "folder.</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:
+            rows += (
+                f"<tr>"
+                f"<td>{item.get('sender_email', '')}</td>"
+                f"<td>{item.get('subject', '')}</td>"
+                f"<td>{item.get('attachment_filename', '')}</td>"
+                f"<td>{item.get('error_message', '')}</td>"
+                f"</tr>"
+            )
+        sections.append(
+            "<h2>Processing Errors</h2>"
+            "<p>These attachments raised an exception during processing. "
+            "Stack traces are in the processor log files.</p>"
+            "<table border='1' cellpadding='4' cellspacing='0'>"
+            "<tr><th>Sender</th><th>Subject</th><th>Attachment</th><th>Error</th></tr>"
+            f"{rows}"
+            "</table>"
+        )
+
+    if not sections:
+        return (
+            "<html><body>"
+            "<p>No items to report today — all attachments routed cleanly.</p>"
+            "</body></html>"
+        )
+
+    return f"<html><body>{''.join(sections)}</body></html>"

app/lib/utils/heat_code.py

@@ -0,0 +1,427 @@
+"""Heat code extraction for MTR (Material Test Report) PDFs.
+
+Public API:
+    extract_heat_codes(file_bytes, filename=None) -> list[str]
+        Returns a deduplicated, order-preserving list of heat codes found
+        in the document. Empty list if none found — caller's responsibility
+        to route to the no-heat-code fallback.
+
+Strategy (in order; results unioned, then filtered + deduped):
+    1. pdfplumber table extraction — look for columns whose header matches
+       a heat-code label and pull the column's non-empty values.
+    2. Label-anchored regex on extracted text — handles inline "Heat No.: ABC123"
+       and similar patterns.
+    3. Filename hint — if the filename contains a plausible heat-code-shaped
+       token AND that token also appears in document text, include it.
+    4. OCR fallback for image-only PDFs — runs at 300 DPI with psm 6, then
+       reruns strategies 2+3 on the OCR text.
+
+Filtering rejects:
+    - ASTM/SA/EN/ISO/ASME spec numbers ("A105", "SA-312", "EN 10204", "ASME B16.5")
+    - Standalone 4-digit years (2020–2030)
+    - Common ALL-CAPS technical words (MAX, MIN, PASS, FAIL, etc.)
+    - Template placeholders ("__", "___", "—", "-")
+    - Single characters or 1–3 char tokens (too short to be heat codes)
+
+Designed to drop into `app/lib/utils/heat_code.py` in the email automation
+repo with no changes — public API and imports are repo-compatible.
+"""
+from __future__ import annotations
+
+import io
+import logging
+import re
+from typing import Iterable
+
+logger = logging.getLogger(__name__)
+
+
+# ─── Label patterns ────────────────────────────────────────────────────────────
+# Synonyms for "heat code" that may appear as labels in MTRs. Case-insensitive.
+# Order matters only for logging; matching is union.
+_LABEL_PATTERNS: tuple[str, ...] = (
+    r"heat\s*code",
+    r"heat\s*no\.?",
+    r"heat\s*nr\.?",
+    r"heat\s*#",
+    r"heat\s*number",
+    r"lot\s*no\.?",      # Bonney Forge and other small-mill MTRs
+    r"lot\s*number",
+)
+
+# Matches a single label occurrence, capturing label and trailing token.
+# Heat code token: 4-15 chars, alphanumeric with optional dashes.
+_LABEL_ANCHORED_RE = re.compile(
+    r"(?i)(?P<label>" + "|".join(_LABEL_PATTERNS) + r")"
+    r"\s*[:\-]?\s*"
+    r"(?P<code>[A-Z0-9][A-Z0-9\-]{2,14})",
+)
+
+# Standalone heat-code shape (used for filename/table validation).
+# Must start with alphanumeric, 4-15 chars total, alphanumeric + dashes.
+_CODE_SHAPE_RE = re.compile(r"^[A-Z0-9][A-Z0-9\-]{3,14}$", re.IGNORECASE)
+
+# Pipe-spec patterns to reject (e.g., "40S", "304L", "80SS" — digit-led + short).
+# Heat codes like "1ZZWN" survive because they have multiple letters in a row.
+_PIPE_SPEC_RE = re.compile(r"^\d{1,4}[A-Z]{1,2}$", re.IGNORECASE)
+
+# Tokens to reject regardless of context.
+_SPEC_PREFIXES: tuple[str, ...] = (
+    "ASTM", "ASME", "SA-", "EN", "ISO", "DIN", "API", "AISI", "JIS", "NACE",
+    "MSS", "MR0",
+)
+
+# All-caps words that look code-like but are technical jargon.
+_REJECT_WORDS: frozenset[str] = frozenset({
+    "MAX", "MIN", "AVG", "STD", "NA", "N/A", "TBD", "TBC", "ACCEPT", "REJECT",
+    "PASS", "FAIL", "PASSED", "FAILED", "OK", "ACCEPTED",
+    "REMARK", "REMARKS", "NOTE", "NOTES", "TOTAL", "VALUE", "RESULT",
+    "TYPE", "GRADE", "SIZE", "DESCRIPTION", "QTY", "QUANTITY",
+    "TEMP", "SPEC", "SPECIFICATION", "REV", "DATE", "PAGE", "BLANK", "NONE",
+    "LADLE", "PRODUCT", "MILL", "NORMALIZED", "ANNEALED", "TEMPERED",
+    "CUSTOMER", "ITEM", "ORDER", "PO", "INVOICE", "QUANTITY",
+    "HEAT", "TREATMENT", "COMPOSITION", "ANALYSIS",
+    "BUNDLE", "DRAWING", "MFG", "MFGID", "POR", "TAG",
+    "DIMENSION", "DIMENSIONAL", "VISUAL", "CHEMICAL", "MECHANICAL", "HARDNESS",
+    "TENSILE", "YIELD", "ELONG", "ELONGATION", "REDUCTION",
+    "RESULT", "RESULTS", "EXAMINATION", "INSPECTION", "MATERIAL",
+    "TENSION", "LOCATION", "ABBREVIATION", "WELDED", "SEAMLESS",
+    "MARKING", "CERTIFIED", "CERTIFICATE", "REPORT",
+})
+
+# Spec number patterns to reject (e.g., A312, SA-105, A105-21).
+_SPEC_NUM_RE = re.compile(r"^[A-Z]{1,4}-?\d{2,5}(-\d{1,3})?$", re.IGNORECASE)
+
+# Year-like standalone tokens (2020-2030).
+_YEAR_RE = re.compile(r"^(20[2-3]\d)$")
+
+
+# ─── Public API ────────────────────────────────────────────────────────────────
+
+def extract_heat_codes(file_bytes: bytes, filename: str | None = None) -> list[str]:
+    """Extract heat codes from an MTR PDF.
+
+    Returns a deduplicated, ordered list. Empty if no codes can be confidently
+    extracted — caller files to a `_no_heat_code/` fallback path.
+    """
+    candidates: list[tuple[str, str]] = []   # (code, source) for logging
+
+    # Strategy 1: pdfplumber tables (text-layer PDFs)
+    try:
+        for code in _from_pdfplumber_tables(file_bytes):
+            candidates.append((code, "table"))
+    except Exception as e:
+        logger.debug("pdfplumber table extraction failed: %s", e)
+
+    # Strategy 2: column-position aware extraction (text-layer)
+    # Handles MTRs where the heat label is a column header and the value sits
+    # directly below it in the same X-range (DSI, 24L0604, A8633, KB419-1).
+    try:
+        for code in _from_column_position(file_bytes):
+            candidates.append((code, "column"))
+    except Exception as e:
+        logger.debug("column-position extraction failed: %s", e)
+
+    # Strategy 3: label-anchored regex on text-layer extract
+    text_layer = _text_layer(file_bytes)
+    for code in _from_label_anchored(text_layer):
+        candidates.append((code, "text_label"))
+
+    # Strategy 4: filename hint
+    if filename:
+        for code in _from_filename(filename, text_layer):
+            candidates.append((code, "filename"))
+
+    # Strategy 5: OCR fallback if we found nothing or text layer is thin
+    if not candidates or len(text_layer) < 500:
+        try:
+            ocr_text = _ocr_text(file_bytes)
+            for code in _from_label_anchored(ocr_text):
+                candidates.append((code, "ocr_label"))
+            for code in _from_ocr_column_heuristic(ocr_text):
+                candidates.append((code, "ocr_column"))
+            if filename:
+                for code in _from_filename(filename, ocr_text):
+                    candidates.append((code, "ocr_filename"))
+        except Exception as e:
+            logger.debug("OCR fallback failed: %s", e)
+
+    # Filter + dedupe (case-insensitive on code; preserve first-seen casing)
+    filtered = _filter_and_dedupe(candidates)
+
+    if logger.isEnabledFor(logging.DEBUG):
+        logger.debug(
+            "Heat code extraction: %d raw candidates → %d after filter. Final: %s",
+            len(candidates), len(filtered), filtered,
+        )
+
+    return filtered
+
+
+# ─── Strategy implementations ──────────────────────────────────────────────────
+
+def _text_layer(file_bytes: bytes) -> str:
+    """Extract embedded text layer via pdfplumber. '' if image-only."""
+    try:
+        import pdfplumber
+        with pdfplumber.open(io.BytesIO(file_bytes)) as pdf:
+            return "\n".join((page.extract_text() or "") for page in pdf.pages[:3])
+    except Exception:
+        return ""
+
+
+def _ocr_text(file_bytes: bytes) -> str:
+    """OCR first 3 pages at 300 DPI with multiple page-segmentation modes
+    to maximize chances of catching small/isolated heat-code text.
+
+    psm 6  — uniform block of text. Good for regular tables.
+    psm 11 — sparse text. Good for small isolated values in form boxes
+             (e.g., MTR W000009810's "1ZZWN" in the small HEAT CODE cell).
+    psm 4  — single column. Good for column-aligned tables (Flotite).
+
+    Outputs are concatenated; downstream code is order-insensitive and
+    dedupes candidates.
+    """
+    from pdf2image import convert_from_bytes
+    import pytesseract
+    imgs = convert_from_bytes(file_bytes, dpi=300, first_page=1, last_page=3)
+    parts: list[str] = []
+    for img in imgs:
+        for psm in ("6", "11", "4"):
+            try:
+                parts.append(pytesseract.image_to_string(img, config=f"--psm {psm}"))
+            except Exception as e:
+                logger.debug("OCR psm=%s failed: %s", psm, e)
+    return "\n".join(parts)
+
+
+def _from_pdfplumber_tables(file_bytes: bytes) -> Iterable[str]:
+    """Find tables whose header column matches a heat label; yield column values."""
+    import pdfplumber
+
+    label_re = re.compile(r"(?i)" + "|".join(_LABEL_PATTERNS))
+
+    with pdfplumber.open(io.BytesIO(file_bytes)) as pdf:
+        for page in pdf.pages[:3]:
+            for table in page.extract_tables() or []:
+                if not table or len(table) < 2:
+                    continue
+
+                # Find heat-label column index by scanning header rows.
+                heat_col_idx: int | None = None
+                for row in table[: min(3, len(table))]:
+                    for idx, cell in enumerate(row):
+                        if cell and label_re.search(str(cell)):
+                            heat_col_idx = idx
+                            break
+                    if heat_col_idx is not None:
+                        break
+
+                if heat_col_idx is None:
+                    continue
+
+                # Pull values from data rows in that column.
+                for row in table:
+                    if heat_col_idx >= len(row):
+                        continue
+                    cell = row[heat_col_idx]
+                    if not cell:
+                        continue
+                    for token in _tokenize_cell(str(cell)):
+                        if _CODE_SHAPE_RE.match(token):
+                            yield token
+
+
+def _from_column_position(file_bytes: bytes) -> Iterable[str]:
+    """Find the X-position of a heat-label word, then yield tokens whose
+    x-center sits within ±half the label's width on the lines directly below.
+
+    Handles MTRs where the heat label is a column header (in a wide
+    whitespace-aligned table) and the data row sits below, e.g. DSI:
+        Part   Specif.   HEAT No   C   Si   Mn   ...
+        BODY   ASTM A216 WCB-2021    VC9387   0.213 ...
+    """
+    import pdfplumber
+
+    label_re = re.compile(r"(?i)" + "|".join(_LABEL_PATTERNS))
+
+    with pdfplumber.open(io.BytesIO(file_bytes)) as pdf:
+        for page in pdf.pages[:3]:
+            words = page.extract_words(use_text_flow=True) or []
+            if not words:
+                continue
+
+            # Group words by approximate line (round y0 to nearest 2pt).
+            lines: dict[int, list[dict]] = {}
+            for w in words:
+                key = round(w["top"] / 2)
+                lines.setdefault(key, []).append(w)
+
+            sorted_keys = sorted(lines.keys())
+
+            # Look for a label word; capture its x-range; scan lines below.
+            for i, key in enumerate(sorted_keys):
+                line_words = lines[key]
+                # Reconstruct the line text to test label patterns against
+                # multi-word labels like "HEAT No" or "Heat Code".
+                # Match individual words first, then consider their pairings.
+                for j, word in enumerate(line_words):
+                    pair_text = word["text"]
+                    if j + 1 < len(line_words):
+                        pair_text = pair_text + " " + line_words[j + 1]["text"]
+
+                    if not label_re.search(pair_text):
+                        continue
+
+                    # X-range of the label (use bbox of just this word, or
+                    # span both words if the match was over the pair).
+                    x0 = word["x0"]
+                    x1 = line_words[j + 1]["x1"] if (
+                        j + 1 < len(line_words) and label_re.search(pair_text)
+                        and not label_re.search(word["text"])
+                    ) else word["x1"]
+                    label_center = (x0 + x1) / 2
+                    half_width = max((x1 - x0) / 2, 20.0)   # min 20pt to be forgiving
+
+                    # Scan up to 6 lines below for tokens whose center is within
+                    # half-width of the label.
+                    for next_key in sorted_keys[i + 1 : i + 7]:
+                        for nw in lines[next_key]:
+                            nw_center = (nw["x0"] + nw["x1"]) / 2
+                            if abs(nw_center - label_center) <= half_width:
+                                tok = nw["text"].strip().strip(",;|")
+                                if _CODE_SHAPE_RE.match(tok):
+                                    yield tok
+
+
+def _from_ocr_column_heuristic(ocr_text: str) -> Iterable[str]:
+    """For OCR text: when a line contains a heat label, yield code-shaped
+    tokens from the next 3 lines. Cheap-and-pretty-good for the cases where
+    the OCR preserves line breaks between header and data rows.
+    """
+    if not ocr_text:
+        return
+    label_re = re.compile(r"(?i)" + "|".join(_LABEL_PATTERNS))
+    lines = ocr_text.splitlines()
+    for i, line in enumerate(lines):
+        if not label_re.search(line):
+            continue
+        # Look at the next 3 non-empty lines.
+        looked = 0
+        for j in range(i + 1, min(i + 8, len(lines))):
+            nxt = lines[j].strip()
+            if not nxt:
+                continue
+            looked += 1
+            if looked > 3:
+                break
+            for tok in _tokenize_cell(nxt):
+                if _CODE_SHAPE_RE.match(tok):
+                    yield tok
+
+
+def _from_label_anchored(text: str) -> Iterable[str]:
+    """Yield candidates following 'Heat No.', 'Lot No.', etc."""
+    if not text:
+        return
+    for m in _LABEL_ANCHORED_RE.finditer(text):
+        yield m.group("code")
+
+
+def _from_filename(filename: str, doc_text: str) -> Iterable[str]:
+    """If filename contains a heat-code-shaped token that appears (exactly or
+    fuzzily) in document text, yield it.
+
+    Fuzzy match handles OCR-mangled docs where e.g. "SD55822" in the filename
+    shows up as "S055822" in the OCR text (D→0 confusion). Uses simple
+    same-length / one-edit-away check (good enough for OCR drift).
+    """
+    # Split filename on whitespace and underscore ONLY — preserve dashes so
+    # codes like "Y211213D40-1" and "KB419-1" survive intact.
+    base = filename.rsplit(".", 1)[0]   # drop .pdf
+    tokens = re.split(r"[\s_]+", base)
+    text_upper = doc_text.upper() if doc_text else ""
+
+    for tok in tokens:
+        if not _CODE_SHAPE_RE.match(tok):
+            continue
+        tok_upper = tok.upper()
+        if tok_upper in text_upper:
+            yield tok
+        elif _fuzzy_token_in_text(tok_upper, text_upper, max_edits=2):
+            yield tok
+
+
+def _fuzzy_token_in_text(token: str, text: str, max_edits: int = 2) -> bool:
+    """True if `text` contains a same-length window within `max_edits` of `token`.
+
+    Cheap-and-good-enough: scans every same-length substring and counts char
+    differences. O(len(text) * len(token)) but n is small (n ≤ 15) and we
+    only call it as a fallback.
+    """
+    n = len(token)
+    if n == 0 or len(text) < n:
+        return False
+    for i in range(len(text) - n + 1):
+        window = text[i : i + n]
+        diff = sum(1 for a, b in zip(token, window) if a != b)
+        if diff <= max_edits:
+            return True
+    return False
+
+
+def _tokenize_cell(cell_text: str) -> list[str]:
+    """Split a table cell into candidate tokens, preserving dashes."""
+    return [t.strip() for t in re.split(r"[\s,;|]+", cell_text) if t.strip()]
+
+
+def _filter_and_dedupe(candidates: list[tuple[str, str]]) -> list[str]:
+    """Apply rejection rules + dedupe case-insensitively, preserving order.
+    Also drops substring duplicates: if 'KB419' and 'KB419-1' are both present,
+    keep only the longer one (avoid over-counting when codes get split)."""
+    seen: dict[str, str] = {}   # upper -> original-cased
+
+    for code, _source in candidates:
+        code = code.strip()
+        upper = code.upper()
+
+        if upper in seen:
+            continue
+        if not _is_plausible_heat_code(code):
+            continue
+        seen[upper] = code
+
+    # Substring dedup: drop any code that is a strict prefix of another.
+    uppers = list(seen.keys())
+    keep = set(uppers)
+    for u in uppers:
+        for other in uppers:
+            if u != other and other.startswith(u) and len(other) > len(u):
+                keep.discard(u)
+                break
+    return [seen[u] for u in uppers if u in keep]
+
+
+def _is_plausible_heat_code(code: str) -> bool:
+    upper = code.upper()
+
+    if not _CODE_SHAPE_RE.match(code):
+        return False
+    if upper in _REJECT_WORDS:
+        return False
+    if _YEAR_RE.match(upper):
+        return False
+    # Pipe-spec rejection (40S, 304L, 80SS-style schedule/grade designators).
+    if _PIPE_SPEC_RE.match(upper):
+        return False
+    # Spec numbers — only reject when the token explicitly starts with a known
+    # spec prefix (ASTM, ASME, SA-, EN, ISO, etc.). Pure shape ("D460", "KB419-1")
+    # overlaps with real heat codes; we don't reject those on shape alone.
+    if any(upper.startswith(p) for p in _SPEC_PREFIXES):
+        return False
+    # Reject all-letter tokens regardless of length — real heat codes have at
+    # least one digit. (None of our 14 confirmed samples are pure-alpha.)
+    if upper.isalpha():
+        return False
+
+    return True

app/processor.py

@@ -26,6 +26,7 @@ from app.database import (
 from app.lib.utils.classifier import ClassifyResult, ScoringWeights, classify
 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(
@@ -68,6 +69,19 @@ def _date_prefix(received_dt: str) -> str:
         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"
+
+
 def _scoring_weights() -> ScoringWeights:
     return ScoringWeights(
         filename=settings.score_weight_filename,
@@ -182,18 +196,98 @@ async def _process_notification(dbx: DropboxClient, notif: dict) -> None:
                 else:
                     # result.reason == "routed"
                     month_folder = received_dt[:7] if received_dt else "unknown"
-                    dest_path = f"{_DEST_MAP[result.doc_type]}/{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)
-                    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,
+                    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>",
                         )
-                    uploaded_count += 1
-                    final_status = "routed"
+
+                        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 — 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)
+                        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"
 
         except Exception:
             # processing_error_handler already logged and wrote to DB; continue to next attachment.

app/scheduler.py

@@ -13,6 +13,7 @@ from app.database import (
     mark_brief_items_sent,
     run_retention_cleanup,
 )
+from app.lib.utils.brief_html import build_brief_html
 from app.lib.utils.notifier import _build_sendmail_payload, send_developer_alert
 
 logger = logging.getLogger(__name__)
@@ -121,7 +122,7 @@ class RenewalScheduler:
             logger.warning("Daily brief: BRIEF_RECIPIENTS not configured, skipping send")
             return
 
-        html = _build_brief_html(items)
+        html = build_brief_html(items)
         path, payload = _build_sendmail_payload(
             recipients=recipients,
             subject=f"RCM Email Automation — Daily Brief {datetime.now(timezone.utc).strftime('%Y-%m-%d')}",
@@ -173,65 +174,6 @@ class RenewalScheduler:
             self._scheduler.shutdown(wait=False)
 
 
-def _build_brief_html(items: list[dict]) -> str:
-    ambiguous = [i for i in items if i["reason"] == "ambiguous"]
-    errors = [i for i in items if i["reason"] == "error"]
-
-    sections: list[str] = []
-
-    if ambiguous:
-        rows = ""
-        for item in ambiguous:
-            scores: dict = json.loads(item["confidence_scores"]) if item["confidence_scores"] else {}
-            score_cols = "".join(
-                f"<td>{doc_type}: {score:.2%}</td>"
-                for doc_type, score in sorted(scores.items())
-            )
-            rows += (
-                f"<tr>"
-                f"<td>{item.get('sender_email', '')}</td>"
-                f"<td>{item.get('subject', '')}</td>"
-                f"<td>{item.get('attachment_filename', '')}</td>"
-                f"{score_cols}"
-                f"</tr>"
-            )
-        score_headers = "".join(
-            f"<th>{doc_type}</th>"
-            for doc_type in sorted(
-                json.loads(ambiguous[0]["confidence_scores"]).keys()
-                if ambiguous[0]["confidence_scores"] else []
-            )
-        )
-        sections.append(
-            f"<h2>Ambiguous Routings</h2>"
-            f"<table border='1' cellpadding='4' cellspacing='0'>"
-            f"<tr><th>Sender</th><th>Subject</th><th>Attachment</th>{score_headers}</tr>"
-            f"{rows}"
-            f"</table>"
-        )
-
-    if errors:
-        rows = ""
-        for item in errors:
-            rows += (
-                f"<tr>"
-                f"<td>{item.get('sender_email', '')}</td>"
-                f"<td>{item.get('subject', '')}</td>"
-                f"<td>{item.get('attachment_filename', '')}</td>"
-                f"<td>{item.get('error_message', '')}</td>"
-                f"</tr>"
-            )
-        sections.append(
-            f"<h2>Processing Errors</h2>"
-            f"<table border='1' cellpadding='4' cellspacing='0'>"
-            f"<tr><th>Sender</th><th>Subject</th><th>Attachment</th><th>Error</th></tr>"
-            f"{rows}"
-            f"</table>"
-        )
-
-    return f"<html><body>{''.join(sections)}</body></html>"
-
-
 def _cleanup_log_files(log_path: Path) -> None:
     cutoff = datetime.now(timezone.utc).replace(tzinfo=None)
     for log_file in log_path.glob("processor.log.*"):

scripts/preview_brief.py

@@ -0,0 +1,98 @@
+"""Render a sample Daily Brief email for client sign-off.
+
+Runs the production `_build_brief_html()` with realistic synthetic data
+covering all three section types: ambiguous routings, MTRs missing heat
+codes, and processing errors.
+
+Usage:  python -m scripts.preview_brief [output_path]
+
+If output_path is omitted, writes to ./daily_brief_preview.html relative
+to the repo root.
+"""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+# Make repo root importable when run via `python scripts/preview_brief.py`
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from app.lib.utils.brief_html import build_brief_html
+
+
+# Synthetic items modeled on real `daily_brief` rows. Confidence scores are
+# JSON-encoded strings because that's how they're stored in SQLite.
+SAMPLE_ITEMS: list[dict] = [
+    # Two ambiguous routings — multiple doc types crossed the threshold.
+    {
+        "id": 1,
+        "sender_email": "ap@westlake.com",
+        "subject": "RCM - Westlake PO 4502953154 and quote",
+        "attachment_filename": "Westlake_PO_4502953154.pdf",
+        "reason": "ambiguous",
+        "confidence_scores": json.dumps({
+            "po": 0.81, "invoice": 0.12, "mtr": 0.03, "quote": 0.74,
+        }),
+    },
+    {
+        "id": 2,
+        "sender_email": "billing@halliburton.com",
+        "subject": "RCM - Halliburton invoice 759085",
+        "attachment_filename": "Halliburton_invoice_759085.pdf",
+        "reason": "ambiguous",
+        "confidence_scores": json.dumps({
+            "po": 0.71, "invoice": 0.78, "mtr": 0.01, "quote": 0.05,
+        }),
+    },
+
+    # Two MTRs where heat code couldn't be extracted.
+    {
+        "id": 3,
+        "sender_email": "qa@woi-houston.com",
+        "subject": "RCM - MTR for WOI order W000009810",
+        "attachment_filename": "MTR W000009810.pdf",
+        "reason": "mtr_no_heat_code",
+        "confidence_scores": json.dumps({
+            "po": 0.04, "invoice": 0.02, "mtr": 0.82, "quote": 0.00,
+        }),
+    },
+    {
+        "id": 4,
+        "sender_email": "certs@flotite.com",
+        "subject": "RCM - Material Test Report S3000CSPGGL015MP",
+        "attachment_filename": "0.5_Full Port_Part No S3000CSPGGL015MP.pdf",
+        "reason": "mtr_no_heat_code",
+        "confidence_scores": json.dumps({
+            "po": 0.03, "invoice": 0.01, "mtr": 0.79, "quote": 0.00,
+        }),
+    },
+
+    # One processing error.
+    {
+        "id": 5,
+        "sender_email": "noreply@example-vendor.com",
+        "subject": "RCM - corrupted attachment",
+        "attachment_filename": "scan_2026-05-10_corrupted.pdf",
+        "reason": "error",
+        "error_message": (
+            "PDFSyntaxError: No /Root object! - Is this really a PDF? "
+            "(stack trace in /logs/processor.log.2026-05-10)"
+        ),
+        "confidence_scores": None,
+    },
+]
+
+
+def main() -> None:
+    out = Path(sys.argv[1]) if len(sys.argv) > 1 else (
+        Path(__file__).resolve().parent.parent / "daily_brief_preview.html"
+    )
+    html = build_brief_html(SAMPLE_ITEMS)
+    out.write_text(html, encoding="utf-8")
+    print(f"Wrote {len(html)} bytes to {out}")
+    print("Open the file in a browser to preview.")
+
+
+if __name__ == "__main__":
+    main()