""" 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)}

    ', "", ] 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.

    ", "" "" "" "", ] for d in decisions: parts.append( "" f"" f"" f"" f"" f"" "" ) parts.append("
    UrgencyEntityRecommendationConfidenceIssue
    {_esc(_fmt_score(d.urgency))}{_esc(d.entity_id)}{_esc(d.recommendation)}{_esc(_fmt_pct(d.confidence))}{_esc(d.issue_id)}
    ") 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)