"""Report generation: grouping, HTML rendering, plain-text digest."""
from __future__ import annotations
import socket
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from jinja2 import Environment, FileSystemLoader, select_autoescape
from ..models import Finding, RunSummary
TEMPLATE_DIR = Path(__file__).resolve().parent.parent / "templates"
@dataclass
class Groups:
"""Findings bucketed the way a person reads them."""
avoid: list[Finding] = field(default_factory=list)
wait: list[Finding] = field(default_factory=list)
go_behind: list[Finding] = field(default_factory=list)
manual: list[Finding] = field(default_factory=list)
errors: list[Finding] = field(default_factory=list)
current: list[Finding] = field(default_factory=list)
ahead: list[Finding] = field(default_factory=list)
# Latest version found, but we have no record of what is installed, so
# no comparison is possible. Distinct from `manual`, where we could not
# even find the latest version.
no_baseline: list[Finding] = field(default_factory=list)
# Deliberately held at their current version. Not news, so kept out of
# the "do not install" section it would otherwise dominate.
frozen: list[Finding] = field(default_factory=list)
# AVWatch knows how to track these but the capability is switched off
# (usually a missing API key). Distinct from `manual`, where no version
# source exists at all -- conflating the two makes a config gap look
# like the gear is untracked.
needs_setup: list[Finding] = field(default_factory=list)
# Items whose installed versions already violate a dependency. These are
# broken right now, not a future risk, so they lead the report.
mismatched: list[Finding] = field(default_factory=list)
@property
def actionable(self) -> list[Finding]:
return self.avoid + self.wait + self.go_behind
def group(findings: list[Finding]) -> Groups:
g = Groups()
for f in findings:
rec = f.verdict.recommendation if f.verdict else "unknown"
# A live mismatch is a present-tense fault and gets its own section
# at the top. Skip the normal buckets so the item is not listed twice
# with two different framings.
if any(c.severity == "blocking" and "MISMATCH NOW" in c.message
for c in f.coupling):
g.mismatched.append(f)
continue
if f.item.freeze:
g.frozen.append(f)
elif f.check.status == "error":
g.errors.append(f)
elif f.check.status == "unconfigured":
g.needs_setup.append(f)
elif f.check.status == "manual":
g.manual.append(f)
elif f.check.latest and not f.item.installed:
g.no_baseline.append(f)
elif f.comparison == "current":
g.current.append(f)
elif f.comparison == "ahead":
g.ahead.append(f)
elif f.comparison == "behind":
if rec == "avoid":
g.avoid.append(f)
elif rec == "wait":
g.wait.append(f)
elif rec == "go":
g.go_behind.append(f)
else:
g.manual.append(f)
else:
g.manual.append(f)
return g
def engine_label(findings: list[Finding]) -> str:
engines = {f.verdict.engine for f in findings if f.verdict}
claude = sorted(e for e in engines if e.startswith("claude"))
if claude:
model = claude[0].split("(")[-1].rstrip(")") if "(" in claude[0] else "Claude"
return f"Claude ({model}) with rule-based cross-check"
return "rule-based heuristics only (no API key configured)"
def _env() -> Environment:
return Environment(
loader=FileSystemLoader(str(TEMPLATE_DIR)),
autoescape=select_autoescape(["html", "jinja"]),
trim_blocks=True,
lstrip_blocks=True,
)
def _tiles(run: RunSummary, g: Groups) -> list[tuple[str, int, str]]:
return [
("Behind", run.updates_found, "#8a6100"),
("Do not install", len(g.avoid), "#a3231d"),
("Cleared", len(g.go_behind), "#1f7a3d"),
("New", run.new_since_last, "#2b5c8a"),
("By hand", len(g.manual), "#4a5878"),
]
def summary_line(run: RunSummary, g: Groups) -> str:
"""The one sentence someone reads on their phone."""
# A live version mismatch outranks everything: it is already broken.
if g.mismatched:
names = ", ".join(f.item.name for f in g.mismatched[:3])
return (
f"{len(g.mismatched)} item(s) are running versions that do not "
f"match gear they are required to match ({names}). That is a "
f"fault now, not a future risk -- fix it before considering any "
f"other update."
)
# Lead with this: if most of the fleet could not be checked, every other
# number in the report is measuring a fraction of it.
if g.needs_setup and len(g.needs_setup) > run.items_checked / 3:
return (
f"{len(g.needs_setup)} of {run.items_checked} tracked items could "
f"not be checked this run because the version lookup is not "
f"configured. They are in the inventory -- AVWatch just cannot "
f"read their versions yet. Everything below covers only the "
f"{run.items_checked - len(g.needs_setup)} items it could reach."
)
# Without baselines the rest of the numbers mean little either.
if g.no_baseline and not g.actionable:
return (
f"AVWatch found current versions for {len(g.no_baseline)} item(s) "
"but has no record of what you actually have installed, so it "
"cannot tell you whether you are behind. Record the installed "
"versions to make future briefings useful."
)
if g.avoid:
worst = g.avoid[0]
return (
f"{len(g.avoid)} update(s) look actively dangerous this week — "
f"most urgently {worst.item.name}. "
f"{len(g.go_behind)} other update(s) are cleared to install."
)
if g.wait and not g.go_behind:
return (
f"Nothing is cleared to install this week. {len(g.wait)} update(s) "
"are available but should soak longer before you touch them."
)
if g.go_behind:
return (
f"{len(g.go_behind)} update(s) are cleared to install, "
f"{len(g.wait)} should wait, and nothing looks dangerous."
)
if run.updates_found == 0:
return "No new updates across the fleet this week. Nothing to do."
return f"{run.updates_found} update(s) available; see the detail below."
def render_html(findings: list[Finding], run: RunSummary,
out_path: Path | None = None) -> str:
g = group(findings)
html = _env().get_template("report.html.jinja").render(
groups=g,
run=run,
generated_local=datetime.now().strftime("%A %d %B %Y, %H:%M"),
engine_label=engine_label(findings),
)
if out_path:
out_path = Path(out_path)
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(html, encoding="utf-8")
return html
def render_email(findings: list[Finding], run: RunSummary,
report_path: Path | None = None) -> str:
g = group(findings)
return _env().get_template("email.html.jinja").render(
groups=g,
run=run,
tiles=_tiles(run, g),
summary_line=summary_line(run, g),
generated_local=datetime.now().strftime("%A %d %B %Y, %H:%M"),
engine_label=engine_label(findings),
report_path=str(report_path) if report_path else None,
hostname=socket.gethostname(),
)
def render_text(findings: list[Finding], run: RunSummary) -> str:
"""Plain-text alternative part, and what ``avwatch report --text`` prints."""
g = group(findings)
lines = [
"AVWATCH BRIEFING",
datetime.now().strftime("%A %d %B %Y, %H:%M"),
"=" * 62,
"",
summary_line(run, g),
"",
f"Tracked {run.items_checked} | behind {run.updates_found} | "
f"new {run.new_since_last} | errors {run.errors}",
"",
]
if g.mismatched:
title = f"VERSION MISMATCH -- BROKEN NOW ({len(g.mismatched)})"
lines += [title, "=" * len(title)]
for f in g.mismatched:
lines.append(f"* {f.item.name}: running {f.item.installed}")
for c in f.coupling:
if c.severity == "blocking":
lines.append(f" {c.message}")
lines.append("")
lines.append("")
for title, bucket in (
("DO NOT INSTALL", g.avoid),
("HOLD OFF FOR NOW", g.wait),
("CLEARED TO UPDATE", g.go_behind),
):
if not bucket:
continue
lines += [title, "-" * len(title)]
for f in bucket:
v = f.verdict
flag = " [NEW]" if f.is_new_since_last_run else ""
lines.append(f"* {f.item.name} ({f.item.vendor}){flag}")
lines.append(
f" {f.item.installed or 'unknown'} -> {f.check.latest}"
+ (f" ({f.bump}, {f.check.age_days:.0f}d old)"
if f.check.age_days is not None else "")
)
if v:
if v.headline:
lines.append(f" {v.headline}")
for r in v.reasons[:4]:
lines.append(f" - {r}")
lines.append(
f" risk {v.risk_score:.0f}/100, rule-based "
f"{v.heuristic_score:.0f}/100 ({v.heuristic_recommendation})"
)
for e in f.evidence[:3]:
lines.append(f" [{e.severity}] {e.title[:70]}")
if e.url:
lines.append(f" {e.url}")
lines.append("")
lines.append("")
if g.no_baseline:
title = "NO INSTALLED VERSION ON RECORD"
lines += [title, "-" * len(title),
"Latest version was found, but AVWatch does not know what you",
"are running, so it cannot say whether you are behind.",
"Record each one to fix this permanently:", ""]
for f in g.no_baseline:
rel = (f" (released {f.check.released.date().isoformat()})"
if f.check.released else "")
lines.append(f"* {f.item.name}: latest is {f.check.latest}{rel}")
lines.append(f" python -m avwatch set-installed {f.item.id} ")
lines.append("")
if g.needs_setup:
title = f"COULD NOT BE CHECKED ({len(g.needs_setup)})"
reasons: dict[str, list[str]] = {}
for f in g.needs_setup:
reasons.setdefault(f.check.detail or "unknown reason", []).append(
f.item.name)
lines += [title, "-" * len(title),
"These ARE in the inventory. AVWatch could not read their",
"current versions this run, for the reasons below.", ""]
for reason, names in sorted(reasons.items(), key=lambda kv: -len(kv[1])):
lines.append(f" {reason}")
lines.append(f" affects {len(names)} item(s): "
f"{', '.join(sorted(names)[:6])}"
+ (f" and {len(names) - 6} more"
if len(names) > 6 else ""))
lines.append("")
lines.append("")
if g.frozen:
title = "FROZEN BY POLICY (no action wanted)"
lines += [title, "-" * len(title)]
for f in g.frozen:
newer = (f" -- {f.check.latest} is available"
if f.comparison == "behind" else "")
lines.append(f"* {f.item.name}: held at "
f"{f.item.installed or 'current version'}{newer}")
lines.append("")
if g.manual:
lines += ["CHECK BY HAND", "-" * 13]
for f in g.manual:
lines.append(f"* {f.item.name} ({f.item.vendor}) - {f.item.check_url or 'no link set'}")
if f.check.detail:
lines.append(f" {f.check.detail}")
lines.append("")
if g.errors:
lines += ["SOURCE ERRORS", "-" * 13]
for f in g.errors:
lines.append(f"* {f.item.name}: {f.check.detail}")
lines.append("")
lines += [
"-" * 62,
"Community posts are unverified writing by strangers. AVWatch ranks",
"and summarises them; it does not vouch for them. Have a rollback",
"plan before touching show-critical gear.",
]
return "\n".join(lines)