123 lines
4.6 KiB
Python
123 lines
4.6 KiB
Python
"""report.py - Stage 4: assemble the final report.
|
|
|
|
Produces a single JSON object (also the web API payload) and a human-readable
|
|
Markdown summary grouped by severity.
|
|
"""
|
|
|
|
from typing import List, Dict
|
|
from datetime import datetime, timezone
|
|
|
|
from backend import config
|
|
|
|
|
|
def _extraction_coverage(sheets: List[Dict]) -> Dict | None:
|
|
"""Summarize per-sheet extraction coverage for the report summary.
|
|
|
|
Sheets may carry a ``coverage`` dict (``total_lines``/``covered_lines``/
|
|
``ratio``) attached during wave-1 extraction. Older paths and scanned
|
|
pages have none; when no sheet is measured, return None so callers can
|
|
omit the key entirely.
|
|
"""
|
|
measured = [s for s in sheets if isinstance(s.get("coverage"), dict)]
|
|
if not measured:
|
|
return None
|
|
floor = getattr(config, "EXTRACT_COVERAGE_FLOOR", 0.6)
|
|
return {
|
|
"pages_measured": len(measured),
|
|
"pages_below_floor": [
|
|
s.get("page_number") for s in measured
|
|
if s["coverage"].get("ratio", 0.0) < floor
|
|
],
|
|
"fallback_pages": [
|
|
s.get("page_number") for s in measured
|
|
if any(a.get("grounding") == "text_layer_fallback"
|
|
for a in s.get("assertions", []))
|
|
],
|
|
"mean_ratio": round(
|
|
sum(s["coverage"].get("ratio", 0.0) for s in measured)
|
|
/ len(measured), 3),
|
|
}
|
|
|
|
|
|
def build_report(conflicts: List[Dict], sheets: List[Dict], clusters: List[Dict],
|
|
source: str = "") -> Dict:
|
|
"""Assemble the structured report object."""
|
|
by_sev = {"high": 0, "medium": 0, "low": 0}
|
|
by_cat: Dict[str, int] = {}
|
|
for c in conflicts:
|
|
by_sev[c["severity"]] = by_sev.get(c["severity"], 0) + 1
|
|
by_cat[c["category"]] = by_cat.get(c["category"], 0) + 1
|
|
|
|
disciplines = sorted({s["discipline"] for s in sheets if s.get("discipline")})
|
|
summary = {
|
|
"sheets_analyzed": len(sheets),
|
|
"disciplines": disciplines,
|
|
"assertions_extracted": sum(len(s.get("assertions", [])) for s in sheets),
|
|
"clusters_checked": len(clusters),
|
|
"conflicts_found": len(conflicts),
|
|
"by_severity": by_sev,
|
|
"by_category": by_cat,
|
|
}
|
|
coverage = _extraction_coverage(sheets)
|
|
if coverage is not None:
|
|
summary["extraction_coverage"] = coverage
|
|
return {
|
|
"source": source,
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"summary": summary,
|
|
"conflicts": conflicts,
|
|
"sheets": [
|
|
{
|
|
"page_number": s.get("page_number"),
|
|
"sheet_number": s.get("sheet_number"),
|
|
"discipline": s.get("discipline"),
|
|
"sheet_title": s.get("sheet_title"),
|
|
"assertion_count": len(s.get("assertions", [])),
|
|
}
|
|
for s in sheets
|
|
],
|
|
}
|
|
|
|
|
|
def to_markdown(report: Dict) -> str:
|
|
s = report["summary"]
|
|
lines = [
|
|
f"# Conflict Check Report",
|
|
"",
|
|
f"- Source: `{report.get('source','')}`",
|
|
f"- Generated: {report.get('generated_at','')}",
|
|
f"- Sheets analyzed: {s['sheets_analyzed']} ({', '.join(s['disciplines']) or 'none'})",
|
|
f"- Assertions extracted: {s['assertions_extracted']}",
|
|
f"- Clusters checked: {s['clusters_checked']}",
|
|
f"- **Conflicts found: {s['conflicts_found']}** "
|
|
f"(high {s['by_severity']['high']}, medium {s['by_severity']['medium']}, low {s['by_severity']['low']})",
|
|
"",
|
|
]
|
|
|
|
if not report["conflicts"]:
|
|
lines += ["No cross-discipline conflicts detected.", ""]
|
|
return "\n".join(lines)
|
|
|
|
for sev in ("high", "medium", "low"):
|
|
group = [c for c in report["conflicts"] if c["severity"] == sev]
|
|
if not group:
|
|
continue
|
|
lines += [f"## {sev.capitalize()} severity ({len(group)})", ""]
|
|
for i, c in enumerate(group, 1):
|
|
lines += [
|
|
f"### {i}. [{c['category']}] {c['location']}",
|
|
f"- Disciplines: {', '.join(c.get('disciplines', []))}",
|
|
f"- Sheets: {', '.join(c.get('sheets', []))}",
|
|
f"- {c['description']}",
|
|
]
|
|
for ev in c.get("evidence", []):
|
|
lines.append(
|
|
f" - {ev.get('discipline','?')} ({ev.get('sheet','?')}): "
|
|
f"\"{ev.get('source_text','')}\""
|
|
)
|
|
if c.get("recommended_resolution"):
|
|
lines.append(f"- Resolution: {c['recommended_resolution']}")
|
|
lines.append(f"- Confidence: {c.get('confidence','')}")
|
|
lines.append("")
|
|
return "\n".join(lines)
|