Spaces:
Sleeping
Sleeping
| """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"] | |
| unmatched = [i for i in items if i["reason"] == "client_unmatched"] | |
| 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 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( | |
| 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/<heat_code>/</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 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: | |
| 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>" | |