"""
delivery.reports.render — render a CustomerReport to Markdown / HTML.
The findings block delegates to the Stage 4a evidence renderers, so card
formatting, escaping, and severity badges stay consistent between the
per-issue view and the full report.
"""
from __future__ import annotations
import html as _html
from typing import List
from ..evidence import EvidenceView
from ..evidence.render import (
_card_html as _evidence_card_html,
)
from ..evidence.render import (
_card_markdown as _evidence_card_markdown,
)
from ..evidence.render import (
_esc,
_fmt_pct,
_fmt_score,
_severity_summary_line,
)
from .customer import CustomerReport, ReportDecision
_REPORT_TITLE = "Operational Drift Report"
# --- diff section (Stage 21) ---------------------------------------------
def _entity_pill(item: dict, *, with_from: bool = False) -> str:
"""One pill (entity id + severity badge) used in the diff lists."""
eid = _esc(item.get("entity_id", ""))
sev = item.get("severity", "")
if with_from and item.get("from_severity"):
from_sev = _esc(item["from_severity"])
return (
f'
{eid} '
f'{from_sev}'
f' → {_esc(sev)}'
)
return (
f'{eid} '
f'{_esc(sev)}'
)
def _diff_section_html(diff: dict) -> str:
"""Render the 'Since previous run' callout. Suppressed entirely when
every counter is zero — an empty diff is just visual noise."""
n_new = diff.get("n_new", 0)
n_resolved = diff.get("n_resolved", 0)
n_severity_changed = diff.get("n_severity_changed", 0)
n_unchanged = diff.get("n_unchanged", 0)
against = diff.get("against_run_id")
if not any((n_new, n_resolved, n_severity_changed)):
# nothing changed — surface that fact briefly, since "no news"
# IS the headline some days
return (
''
'Since previous run: no changes '
f'({n_unchanged} unchanged).'
'
'
)
sections = []
if against is None:
sections.append(
'This is the first run on this tenant — '
'every issue below is new.
'
)
def _list_block(title, items, with_from=False):
if not items:
return ""
pills = "".join(_entity_pill(it, with_from=with_from)
for it in items[:20])
more = "" if len(items) <= 20 else (
f' +{len(items) - 20} more'
)
return (
f'{_esc(title)} '
f'({len(items)})'
f'
{more}
'
)
sections.append(_list_block("New", diff.get("new", [])))
sections.append(_list_block("Severity changed",
diff.get("severity_changed", []),
with_from=True))
sections.append(_list_block("Resolved", diff.get("resolved", [])))
return (
''
'
Since previous run
'
'
'
f'{n_new} new · '
f'{n_severity_changed} severity changed · '
f'{n_resolved} resolved · '
f'{n_unchanged} unchanged'
'
'
+ "".join(sections)
+ '
'
)
# --- markdown -------------------------------------------------------------
def _summary_markdown(report: CustomerReport) -> List[str]:
s = report.summary
return [
"## Executive Summary",
"",
s.headline,
"",
f"- **Entities analysed:** {s.n_entities_analyzed}",
f"- **Drift issues:** {s.n_issues} ({_severity_summary_line(report.evidence)})",
f"- **Decisions raised:** {s.n_decisions}",
"",
]
def _findings_markdown(view: EvidenceView) -> List[str]:
if not view.cards:
return [
"## Findings",
"",
"No entities crossed a drift threshold on this run.",
"",
]
lines = ["## Findings", ""]
for card in view.cards:
lines += _evidence_card_markdown(card)
return lines
def _decisions_markdown(decisions: List[ReportDecision]) -> List[str]:
if not decisions:
return [
"## Decision Queue",
"",
"No decisions raised on this run.",
"",
]
lines = [
"## Decision Queue",
"",
"Ranked by urgency. Each decision links back to the issue that triggered it.",
"",
"| Urgency | Entity | Recommendation | Confidence | Issue |",
"|---:|---|---|---:|---|",
]
for d in decisions:
lines.append(
f"| {_fmt_score(d.urgency)} | `{d.entity_id}` | {d.recommendation} | "
f"{_fmt_pct(d.confidence)} | `{d.issue_id}` |"
)
lines.append("")
return lines
def _methodology_markdown() -> List[str]:
return [
"## Methodology",
"",
"Drift is detected by a generic, config-driven pipeline over a "
"per-tenant calibration derived from the tenant's own history (MAD-robust). "
"Each issue is scored from five signals (Δ, ψ, ξ, γ, κ) and mapped to a "
"severity. Decisions are recommendations grounded in the evidence of the "
"issue that triggered them — they are operational hypotheses to validate, "
"not guaranteed outcomes.",
"",
]
def render_markdown(report: CustomerReport) -> str:
lines = [
f"# {_REPORT_TITLE} — `{report.tenant_id}`",
"",
(
f"- **Run:** `{report.run_id}` "
f"\n- **Entity type:** {report.entity_type} "
f"\n- **Finished at:** {report.finished_at}"
),
"",
]
lines += _summary_markdown(report)
lines += _findings_markdown(report.evidence)
lines += _decisions_markdown(report.decisions)
lines += _methodology_markdown()
return "\n".join(lines).rstrip() + "\n"
# --- html -----------------------------------------------------------------
_HTML_HEAD = (
""
"Operational Drift Report"
""
)
_HTML_FOOT = ""
def _summary_html(report: CustomerReport) -> str:
s = report.summary
parts = [
"Executive Summary
",
f'{_esc(s.headline)}
',
"",
f"- Entities analysed: {s.n_entities_analyzed}
",
(
"- Drift issues: "
f"{s.n_issues} ({_esc(_severity_summary_line(report.evidence))})
"
),
f"- Decisions raised: {s.n_decisions}
",
"
",
]
return "".join(parts)
def _findings_html(view: EvidenceView) -> str:
if not view.cards:
return (
"Findings
"
"No entities crossed a drift threshold on this run.
"
)
parts = ["Findings
"]
for card in view.cards:
parts.append(_evidence_card_html(card))
return "".join(parts)
def _decisions_html(decisions: List[ReportDecision]) -> str:
if not decisions:
return "Decision Queue
No decisions raised on this run.
"
parts = [
"Decision Queue
",
'Ranked by urgency. Each decision links back to '
"the issue that triggered it.
",
""
"| Urgency | Entity | Recommendation | "
"Confidence | Issue | "
"
",
]
for d in decisions:
parts.append(
""
f"| {_esc(_fmt_score(d.urgency))} | "
f"{_esc(d.entity_id)} | "
f"{_esc(d.recommendation)} | "
f"{_esc(_fmt_pct(d.confidence))} | "
f"{_esc(d.issue_id)} | "
"
"
)
parts.append("
")
return "".join(parts)
def _methodology_html() -> str:
return (
"Methodology
"
"Drift is detected by a generic, config-driven pipeline over a "
"per-tenant calibration derived from the tenant's own history "
"(MAD-robust). Each issue is scored from five signals (Δ, ψ, ξ, γ, κ) "
"and mapped to a severity. Decisions are recommendations grounded in "
"the evidence of the issue that triggered them — they are "
"operational hypotheses to validate, not guaranteed outcomes.
"
)
def render_html(report: CustomerReport) -> str:
body = [
_HTML_HEAD,
f"{_html.escape(_REPORT_TITLE)} — "
f"{_esc(report.tenant_id)}
",
'',
f"Run {_esc(report.run_id)}",
f" · entity type {_esc(report.entity_type)}",
f" · finished {_esc(report.finished_at)}",
"
",
# Stage 21: "Since previous run" callout, above the summary so it
# is the first thing an operator opening the daily report sees
_diff_section_html(report.diff) if report.diff is not None else "",
_summary_html(report),
_findings_html(report.evidence),
_decisions_html(report.decisions),
_methodology_html(),
_HTML_FOOT,
]
return "".join(body)