RcmEmailAutomation / app /lib /utils /brief_html.py
Avinash Nalla
Phase 2+4: PO/Invoice client folder routing and subject-line MTR trigger
c55abce
Raw
History Blame Contribute Delete
6.06 kB
"""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/&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 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>"