from __future__ import annotations
import json
import re
import textwrap
from pathlib import Path
from typing import Any
from . import __version__
from .redaction import redact_object, sanitize_artifact_text
from .render_html import render_html
try:
from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import ParagraphStyle
from reportlab.lib.units import mm
from reportlab.platypus import (
HRFlowable,
KeepInFrame,
PageBreak,
Paragraph,
SimpleDocTemplate,
Spacer,
Table,
TableStyle,
)
_RL = True
except ImportError:
_RL = False
# ── palette ───────────────────────────────────────────────────────────────────
_NAVY = "#1B2A4A"
_TEAL = "#2E86AB"
_PURPLE = "#6B4D8E"
_SLATE = "#3D5A7A"
_GREEN = "#27A560"
_AMBER = "#C97B10"
_RED = "#C0392B"
_ORANGE = "#C96010"
_LGRAY = "#F5F7FA"
_MGRAY = "#E2E8F0"
_DGRAY = "#4A5568"
_WHITE = "#FFFFFF"
_TIER_COLOR = {"T0": _RED, "T1": _RED, "T2": _ORANGE, "T3": _TEAL, "T4": _GREEN}
_PDF_CONTENT_WIDTH = A4[0] - (28 * mm)
_PDF_CONTENT_HEIGHT = A4[1] - (24 * mm)
_BIO_DETECTOR_LABELS = {
"BIO_smiles_surface_integrity": "SMILES Surface Integrity",
"BIO_smiles_rdkit_validation": "SMILES RDKit Validation",
"BIO_smiles_parser_guard": "SMILES Parser Guard",
"BIO_silent_mock_fallback": "Silent Mock Fallback",
"BIO_trace_manifest": "Traceability Manifest Surface",
"BIO_run_trace": "Bio Subprocess Run Trace",
}
def _hx(h: str) -> Any:
return colors.HexColor(h)
def _tier_hex(tier: str) -> str:
for k, v in _TIER_COLOR.items():
if k in tier:
return v
return _DGRAY
def _status_hex(s: str) -> str:
return {"PASS": _GREEN, "FAIL": _RED, "WARN": _AMBER}.get(s.upper(), _DGRAY)
def _xt(t: str) -> str:
"""Escape text for use in reportlab XML markup."""
return str(t).replace("&", "&").replace("<", "<").replace(">", ">")
def _clip_words(text: str, limit: int) -> str:
"""Trim table text without cutting through a word."""
value = " ".join(str(text).split())
if len(value) <= limit:
return value
trimmed = value[: max(0, limit - 1)].rsplit(" ", 1)[0].rstrip(".,;:")
return f"{trimmed}..."
# ── public API ────────────────────────────────────────────────────────────────
def write_outputs(
result: dict[str, Any],
output_dir: Path,
mode: str,
pages: int,
fmt: str,
explain: bool = False,
) -> list[Path]:
output_dir.mkdir(parents=True, exist_ok=True)
stem = _safe_name(result["target"]["name"])
created: list[Path] = []
safe_result = redact_object(result)
if fmt in {"json", "all"}:
p = output_dir / f"{stem}_experiment_results.json"
payload = json.dumps(safe_result, indent=2)
p.write_text(payload, encoding="utf-8")
created.append(p)
if "ai_advisory_input" in safe_result:
p = output_dir / f"{stem}_advisory_input.json"
payload = json.dumps(safe_result["ai_advisory_input"], indent=2)
p.write_text(payload, encoding="utf-8")
created.append(p)
md = render_markdown(safe_result, mode, pages)
if fmt in {"md", "all"}:
p = output_dir / f"{stem}_report.md"
md, _ = sanitize_artifact_text(md)
p.write_text(md, encoding="utf-8")
created.append(p)
if fmt in {"html", "all"}:
p = output_dir / f"{stem}_report.html"
p.write_text(render_html(safe_result), encoding="utf-8")
created.append(p)
if fmt in {"pdf", "all"}:
p = output_dir / f"{stem}_{mode}_{pages}p.pdf"
if _RL:
_write_rl_pdf(p, safe_result, mode, pages)
else:
write_simple_pdf(p, render_pdf_pages(safe_result, mode, pages))
created.append(p)
if explain:
p = output_dir / f"{stem}_explain.txt"
explain_text, _ = sanitize_artifact_text(render_explain(safe_result))
p.write_text(explain_text, encoding="utf-8")
created.append(p)
return created
def _surface_compaction_note(result: dict[str, Any]) -> str:
notes = result.get("artifact_surface_notes", {})
if isinstance(notes, dict):
text = notes.get("human_readable_compaction")
if isinstance(text, str) and text.strip():
return text.strip()
return "Human-readable surfaces may compact repeated same-file evidence; JSON remains the canonical full-fidelity artifact."
# ── markdown ──────────────────────────────────────────────────────────────────
def render_markdown(result: dict[str, Any], mode: str, pages: int) -> str:
score = result["score"]
ast_note = _ast_scope_note(result)
calibration = result.get("calibration_profile", {})
calibration_effect = _calibration_effect_note(calibration)
lines = [
"# STEM BIO-AI Local Audit Report",
"",
f"**Target:** `{result['target']['name']}`",
f"**Execution Mode:** `{result['execution_mode']}`",
f"**Calibration Profile:** `{calibration.get('profile_name', 'unknown')}` "
f"(`{calibration.get('policy_version', 'unknown')}`, "
f"`{calibration.get('profile_read_mode', 'unknown')}`, "
f"`{calibration.get('profile_status', 'unknown')}`)",
*([f"**Calibration Effect:** {calibration_effect}"] if calibration_effect else []),
f"**Final Score:** **{score['final_score']} / 100**",
f"**Formal Tier:** **{score['formal_tier']}**",
f"**Tier Meaning:** {score['formal_tier']} = {score['use_scope']}",
f"**Use Scope:** {score['use_scope']}",
"",
"## Score Matrix",
"",
"| Stage | Weight | Score |",
"| --- | ---: | ---: |",
f"| Stage 1 README Evidence Signal | 0.40 | {score['stage_1_readme_intent']} |",
f"| Stage 2R Repo-Local Consistency | 0.20 | {score['stage_2_repo_local_consistency']} |",
f"| Stage 3 Code/Bio Responsibility | 0.40 | {score['stage_3_code_bio']} |",
f"| Risk Penalty | -- | {score['risk_penalty']} |",
"",
"## Replication Evidence Lane",
"",
f"**Stage 4 Replication Score:** **{result.get('replication_score', 0)} / 100**",
f"**Replication Tier:** **{result.get('replication_tier', 'R0')}**",
"**Interpretation:** Stage 4 is a separate replication lane. It improves inspectability and reproducibility review, but it does not currently change the formal tier.",
"",
*_markdown_freshness_section(result.get("audit_freshness", {})),
"## Reasoning Diagnostics",
"",
_markdown_reasoning_summary(result.get("reasoning_model", {})),
*_markdown_reasoning_interpretation(result.get("reasoning_model", {})),
"",
*_markdown_advisory_section(result.get("ai_advisory")),
*_markdown_regulatory_section(result),
*_markdown_airi_section(result.get("airi_risk_coverage", {})),
"## Code Integrity",
]
for key, item in result["code_integrity"].items():
lines.append(f"- **{key}:** {item['status']} — {item['evidence'][0]}")
if item["status"] in {"WARN", "FAIL"}:
for detail in item.get("evidence", [])[1:4]:
lines.append(f" - {detail}")
if ast_note:
lines.append(f"- **AST analysis scope:** {ast_note}")
lines.extend(_markdown_bio_section(result))
lines.extend(["", "## Top Risks"])
for risk in result["notable_risks"][:5]:
lines.append(f"- {risk}")
lines.extend(_markdown_remediation_targets(result))
if mode == "detailed":
lines.extend(["", "## Stage 1 Evidence"])
for key, item in result.get("stage_1_rubric", {}).items():
if isinstance(item, dict):
score_value = item.get("score", "")
lines.append(
f"- **{key}:** {score_value} — {item.get('evidence', '')}"
f"{_stage1_semantics_suffix(key, item)}"
)
lines.extend(["", "## Stage 2R Evidence"])
for key, item in result["stage_2r_rubric"].items():
if isinstance(item, dict):
lines.append(
f"- **{key}:** {item.get('score', '')} — {item.get('evidence', '')}"
f"{_rubric_trace_suffix(item)}"
)
lines.extend(["", "## Stage 3 Evidence"])
for key, item in result["stage_3_rubric"].items():
lines.append(
f"- **{key}:** {item['score']} / {item['max']} — {item['evidence']}"
f"{_rubric_trace_suffix(item)}"
)
lines.extend(["", "## Stage 4 Replication Evidence"])
for key, item in result.get("stage_4_rubric", {}).items():
lines.append(f"- **{key}:** {item['score']} / {item['max']} — {item['evidence']}")
lines.extend(["", "## Method Boundary", result["method"]])
lines.extend([
"",
"## Disclaimer",
"This is an evidence-surface pre-screen, not clinical certification, "
"regulatory clearance, or medical advice.",
])
return "\n".join(lines) + "\n"
# ── explain text report ───────────────────────────────────────────────────────
_EXPLAIN_SEP = "=" * 72
_EXPLAIN_META_SKIP = frozenset({"file_count", "max_ast_files", "max_file_size_bytes"})
def render_explain(result: dict[str, Any]) -> str:
"""Return a human-readable plain-text explain report grouped by detector."""
ledger: list[dict[str, Any]] = result.get("evidence_ledger", [])
score = result["score"]
grouped: dict[str, list[dict[str, Any]]] = {}
for finding in ledger:
grouped.setdefault(finding["detector"], []).append(finding)
calibration = result.get("calibration_profile", {})
calibration_effect = _calibration_effect_note(calibration)
out: list[str] = [
"STEM BIO-AI Explain Report",
f"Target : {result['target']['name']}",
(
"Policy : "
f"{calibration.get('profile_name', 'unknown')} "
f"[{calibration.get('policy_version', 'unknown')}; "
f"{calibration.get('profile_read_mode', 'unknown')}; "
f"{calibration.get('profile_status', 'unknown')}]"
),
*([f"Policy Mode: {calibration_effect}"] if calibration_effect else []),
f"Score : {score['final_score']} / 100 ({score['formal_tier']})",
f"Replic : {result.get('replication_score', 0)} / 100"
f" ({result.get('replication_tier', 'R0')})",
"Surface : repeated same-file evidence may be compacted in narrative output; JSON remains canonical.",
_EXPLAIN_SEP, "",
]
for detector, findings in grouped.items():
out += _explain_detector_group(detector, findings)
out += _explain_bio_section(result)
out += _explain_regulatory_section(result)
out += _explain_airi_section(result.get("airi_risk_coverage", {}))
out += _explain_freshness_section(result.get("audit_freshness", {}))
out += _explain_ast_section(result.get("ast_signal_summary", {}))
out += _explain_s4_section(result.get("stage_4_rubric", {}))
out += _explain_reasoning_section(result.get("reasoning_model", {}))
out += _explain_advisory_section(result.get("ai_advisory"))
out += [_EXPLAIN_SEP,
"DISCLAIMER: Evidence-surface pre-screen only.",
"Not clinical certification, regulatory clearance, or medical advice."]
return "\n".join(out) + "\n"
def _markdown_reasoning_summary(reasoning: dict[str, Any]) -> str:
if not reasoning:
return "Reasoning diagnostics are not available."
coherence = reasoning.get("lane_coherence", {})
uncertainty = reasoning.get("uncertainty_budget", {})
gate = reasoning.get("evidence_risk_gate", {})
envelope = reasoning.get("confidence_envelope", {})
policy = reasoning.get("policy", {})
return (
f"Diagnostic-only heuristic `{reasoning.get('version', 'unknown')}` "
f"({policy.get('weights', 'uncalibrated')}); "
f"lane consistency `{coherence.get('status', 'unknown')}` "
f"({coherence.get('overall', 'n/a')}), "
f"uncertainty band `{uncertainty.get('status', 'unknown')}` "
f"({uncertainty.get('uncertainty', 'n/a')}), "
f"risk heuristic `{gate.get('status', 'unknown')}` "
f"({gate.get('evidence_risk', 'n/a')}), "
f"confidence envelope {envelope.get('lower', 'n/a')}-"
f"{envelope.get('upper', 'n/a')}. "
"This heuristic layer does not override the final score."
)
def _markdown_reasoning_interpretation(reasoning: dict[str, Any]) -> list[str]:
if not reasoning:
return []
coherence = reasoning.get("lane_coherence", {})
uncertainty = reasoning.get("uncertainty_budget", {})
notes: list[str] = []
if coherence.get("status") in {"heuristic_mixed", "heuristic_divergent"}:
notes.append(
"- **Interpretation:** lane coherence is mixed, which means README-facing intent and code/accountability signals do not move together cleanly. Review Stage 2R and Stage 3 evidence before treating the score as stable."
)
if uncertainty.get("status") == "review_advised":
notes.append(
"- **Interpretation:** the uncertainty band is elevated enough that manual review is recommended, especially for boundary claims, workflow support, and governance surfaces."
)
return notes
def _calibration_effect_note(calibration: dict[str, Any]) -> str | None:
if calibration.get("profile_read_mode") != "mirror_only":
return None
return (
"mirror-only in 1.8.0 — selected profile metadata is surfaced in artifacts, "
"but authoritative scan scoring still follows deterministic runtime constants. "
"Preview-only posture changes, including Stage 4 replication emphasis, do not "
"change the formal score until a future read-through phase. "
"Use `stem policy simulate` to preview governed score deltas and posture changes."
)
def _markdown_advisory_section(advisory: dict[str, Any] | None) -> list[str]:
if not advisory:
return []
return [
"## AI Advisory Contract",
"",
f"**Status:** `{advisory.get('status', 'unknown')}`",
f"**Provider:** `{advisory.get('provider', 'none')}`",
f"**Mode:** `{advisory.get('mode', 'unknown')}`",
f"**Invalid Citations:** {len(advisory.get('invalid_citations', []))}",
"",
]
def _markdown_freshness_section(freshness: dict[str, Any]) -> list[str]:
if not freshness:
return []
triggers = ", ".join(freshness.get("change_triggers", [])[:4])
reasons = ", ".join(freshness.get("change_triggered_reaudit_reasons", [])) or "none"
return [
"## Audit Freshness",
"",
f"**Review After:** **{freshness.get('review_after_days', 'n/a')} days**",
f"**Expires On:** `{freshness.get('expires_on', 'unknown')}`",
f"**Change-triggered re-audit recommended now:** `{freshness.get('change_triggered_reaudit_recommended_now', False)}`",
f"**Current re-audit reasons:** `{reasons}`",
f"**Trigger examples:** `{triggers}`",
"",
]
def _markdown_bio_section(result: dict[str, Any]) -> list[str]:
rows = _bio_detector_rows(result)
if not rows:
return []
lines = ["", "## Bio Deterministic Diagnostics", ""]
for detector, label, counts in rows:
parts = []
for status in ("detected", "warn", "error", "not_detected", "not_applicable", "absent"):
value = counts.get(status)
if value:
parts.append(f"{status}={value}")
findings = [
f for f in result.get("evidence_ledger", [])
if f.get("detector") == detector and f.get("status") == "detected"
]
note = findings[0].get("explanation", "") if findings else _detector_scope_note(result, detector)
lines.append(f"- **{label}:** {', '.join(parts) if parts else 'no findings'} — {note}")
return lines
def _markdown_regulatory_section(result: dict[str, Any]) -> list[str]:
basis = result.get("regulatory_basis", {})
traceability = result.get("stage_traceability", {})
if not basis and not traceability:
return []
note = basis.get("note", {})
lines = [
"## Regulatory Traceability Assistant",
"",
f"> **{note.get('title', 'Regulatory basis note')}**",
f"> {note.get('body_line_1', '')}",
f"> {note.get('body_line_2', '')}",
"",
]
if basis.get("review_required"):
reasons = ", ".join(basis.get("review_reasons", []))
lines.append(f"> Review required: `{reasons}`")
lines.append("")
for stage_key in ("stage_1", "stage_2r", "stage_3", "stage_4", "bio_diagnostics"):
items = traceability.get(stage_key, [])
if not items:
continue
lines.append(f"### {stage_key.replace('_', ' ').title()}")
for item in items:
lines.append(
f"- **{item['requirement_id']}** — {item['status']} "
f"(mapping confidence: {item['mapping_confidence']}, evidence strength: {item['evidence_strength']})"
)
lines.append(f" - {item['note']}")
lines.append("")
summary = result.get("regulatory_traceability", {}).get("summary")
if summary:
lines.append(f"**Summary:** {summary}")
lines.append("")
return lines
def _markdown_airi_section(airi: dict[str, Any]) -> list[str]:
if not airi:
return []
covered = airi.get("covered_count", 0)
total = airi.get("total_risks_in_detector_scope", 0)
rate = airi.get("coverage_rate", 0)
bundle_scope = airi.get("airi_bundle_scope", "unknown")
snapshot = airi.get("airi_upstream_snapshot_date", "unknown")
lines = [
"## AIRI Risk Triggers",
"",
f"**Covered Risks:** **{covered} / {total}**",
f"**Coverage Rate:** `{rate:.3f}`",
f"**Bundle Scope:** `{bundle_scope}`",
f"**Upstream Snapshot:** `{snapshot}`",
"**Interpretation:** This is detector-mapped AIRI coverage inside the current runtime bundle, not a claim that unmapped risks are absent.",
"**Surface Note:** repeated same-file evidence may be compacted in human-readable surfaces; canonical per-finding rows remain in JSON.",
]
covered_risks = airi.get("covered_risks", [])
if covered_risks:
lines.append("")
lines.append("**Examples of Covered AIRI Risks**")
for risk in covered_risks[:3]:
reason = _airi_reason_summary(risk)
primary = _airi_primary_summary(risk)
lines.append(
f"- `{risk.get('id', 'unknown')}` — {risk.get('title', 'unknown')} "
f"({primary}; why: {reason})"
)
gaps = airi.get("known_gaps_in_bundle", [])
if gaps:
lines.append("")
lines.append("**Known Gaps In Bundle**")
for gap in gaps:
lines.append(f"- `{gap.get('id', 'unknown')}` — {gap.get('title', 'unknown')}")
return lines
def _explain_detector_group(detector: str, findings: list[dict[str, Any]]) -> list[str]:
compact_findings = _compact_explain_findings(findings)
label = _explain_status_label({f["status"] for f in findings})
noun = "finding" if len(findings) == 1 else "findings"
compact_noun = "row" if len(compact_findings) == 1 else "rows"
heading = f"{detector} [{label}] ({len(findings)} {noun})"
if len(compact_findings) != len(findings):
heading += f" -> compacted to {len(compact_findings)} {compact_noun}"
lines = [heading]
for f in compact_findings:
lines += _explain_finding_lines(f)
lines.append("")
return lines
def _compact_explain_findings(findings: list[dict[str, Any]]) -> list[dict[str, Any]]:
compact: list[dict[str, Any]] = []
grouped: dict[tuple[str, str, str, str], dict[str, Any]] = {}
grouped_order: list[tuple[str, str, str, str]] = []
for finding in findings:
status = str(finding.get("status", "unknown"))
file_path = str(finding.get("file", ""))
line = int(finding.get("line", 0) or 0)
reason = str(finding.get("explanation") or finding.get("message") or "").strip()
if status in {"detected", "warn", "pass"} and file_path not in {"", "."} and reason:
key = (status, file_path, reason, str(finding.get("pattern_id", "")))
if key not in grouped:
clone = dict(finding)
meta = dict(clone.get("metadata") or {})
meta["aggregate_count"] = 1
meta["aggregate_lines"] = [line] if line else []
meta["aggregate_surface"] = "explain_same_file_reason"
clone["metadata"] = meta
grouped[key] = clone
grouped_order.append(key)
else:
grouped[key]["metadata"]["aggregate_count"] += 1
if line and line not in grouped[key]["metadata"]["aggregate_lines"]:
grouped[key]["metadata"]["aggregate_lines"].append(line)
continue
compact.append(finding)
for key in grouped_order:
group = grouped[key]
count = int(group.get("metadata", {}).get("aggregate_count", 1))
lines = sorted(group.get("metadata", {}).get("aggregate_lines", []))
if count > 1:
if lines:
preview = ", ".join(str(n) for n in lines[:6])
if len(lines) > 6:
preview += ", ..."
group["explanation"] = f"{group.get('explanation', '')} Aggregated {count} similar findings from one file (lines: {preview}).".strip()
else:
group["explanation"] = f"{group.get('explanation', '')} Aggregated {count} similar findings from one file.".strip()
group["snippet"] = ""
compact.append(group)
return compact
def _explain_finding_lines(f: dict[str, Any]) -> list[str]:
occ = f["finding_id"].rsplit(":", 1)[-1]
file_str = "(repository)" if f["file"] == "." else (
f"{f['file']}:{f['line']}" if f["line"] else f["file"]
)
lines = [f" [{occ}] {file_str}", f" finding_id: {f['finding_id']}"]
if f.get("pattern_id"):
lines.append(f" pattern : {f['pattern_id']}")
if f.get("evidence_status"):
lines.append(f" evidence: {f['evidence_status']}")
if f.get("confidence"):
lines.append(f" conf : {f['confidence']}")
if f.get("snippet"):
lines.append(f" snippet : \"{f['snippet']}\"")
if f.get("explanation"):
lines.append(f" reason : {f['explanation']}")
for k, v in (f.get("metadata") or {}).items():
if k not in _EXPLAIN_META_SKIP:
lines.append(f" {k} : {v}")
return lines
def _explain_ast_section(ast: dict[str, Any]) -> list[str]:
if not ast:
return []
lines = [_EXPLAIN_SEP, "AST Signal Summary", ""]
lines += [f" {k:<34} {v}" for k, v in ast.items() if v is not None]
lines.append("")
return lines
def _explain_airi_section(airi: dict[str, Any]) -> list[str]:
if not airi:
return []
lines = [
_EXPLAIN_SEP,
"AIRI Risk Triggers",
"",
f" covered_count {airi.get('covered_count', 0)}",
f" detector_scope_total {airi.get('total_risks_in_detector_scope', 0)}",
f" coverage_rate {airi.get('coverage_rate', 0)}",
f" bundle_scope {airi.get('airi_bundle_scope', 'unknown')}",
f" upstream_snapshot {airi.get('airi_upstream_snapshot_date', 'unknown')}",
]
covered_risks = airi.get("covered_risks", [])
if covered_risks:
lines.append("")
lines.append(" covered examples:")
for risk in covered_risks[:3]:
reason = _airi_reason_summary(risk)
primary = _airi_primary_summary(risk)
lines.append(
f" - {risk.get('id', 'unknown')} | {risk.get('title', 'unknown')} "
f"| {primary} | why={reason}"
)
gaps = airi.get("known_gaps_in_bundle", [])
if gaps:
lines.append("")
lines.append(" known gaps in bundle:")
for gap in gaps:
lines.append(
f" - {gap.get('id', 'unknown')} | {gap.get('title', 'unknown')}"
)
lines.append("")
return lines
def _rubric_trace_suffix(item: dict[str, Any]) -> str:
detector = str(item.get("detector_id", "")).strip()
basis = str(item.get("decision_basis", "")).strip()
tier_impact = str(item.get("tier_impact", "")).strip()
parts: list[str] = []
if detector:
parts.append(f"detector={detector}")
if basis:
parts.append(f"basis={basis}")
if tier_impact:
parts.append(f"tier-impact={tier_impact}")
if not parts:
return ""
return f" `[{' | '.join(parts)}]`"
def _airi_reason_summary(risk: dict[str, Any]) -> str:
details = risk.get("mapping_details", [])
if not details:
return "bounded detector-to-risk mapping"
snippets: list[str] = []
for detail in details[:2]:
detector = str(detail.get("detector_id", "")).strip()
trigger = str(detail.get("trigger_reason", "")).strip()
justification = str(detail.get("mapping_justification", "")).strip()
reason = trigger or justification or "bounded detector-to-risk mapping"
snippets.append(f"{detector}: {reason}" if detector else reason)
return " ; ".join(snippets)
def _airi_primary_summary(risk: dict[str, Any]) -> str:
primary = str(risk.get("primary_detector_id", "")).strip()
secondary = [str(det).strip() for det in risk.get("secondary_detector_ids", []) if str(det).strip()]
if primary and secondary:
return f"primary: {primary}; also linked by {', '.join(secondary[:2])}"
if primary:
return f"primary: {primary}"
covered_by = [str(det).strip() for det in risk.get("covered_by", []) if str(det).strip()]
if covered_by:
return f"covered by: {', '.join(covered_by[:2])}"
return "bounded detector-to-risk mapping"
def _stage1_semantics_suffix(key: str, item: dict[str, Any]) -> str:
if key != "R2_regulatory_framework":
return ""
score = item.get("score")
ladder = "+15 strong framework | +5 weak self-asserted compliance | -5 CA-INDIRECT missing framework | -10 CA-DIRECT missing framework"
return f" `[partial-credit ladder={ladder}; current={score}]`"
def _markdown_remediation_targets(result: dict[str, Any]) -> list[str]:
targets: list[str] = []
stage2 = result.get("stage_2r_rubric", {})
code_integrity = result.get("code_integrity", {})
if "R2R_D2_missing_clinical_use_boundary" in stage2:
targets.append("Add an explicit non-clinical/non-diagnostic boundary to README and adjacent docs before treating the repository as review-ready.")
if "R2R_D4_unsupported_workflow_claim" in stage2:
targets.append("Bring workflow, demo, and CLI claims into line with actual local support surfaces, or narrow the README claims.")
if code_integrity.get("C2_dependency_pinning", {}).get("status") in {"WARN", "FAIL"}:
targets.append("Pin production dependencies and document external-service dependence explicitly when local or self-host claims are part of the positioning.")
if code_integrity.get("C5_compliance_boundary_integrity", {}).get("status") in {"WARN", "FAIL"}:
targets.append("Remove unsupported legal/compliance language or add the governance and security evidence needed to defend it.")
if code_integrity.get("C6_mock_auth_or_fail_open_boundary", {}).get("status") in {"WARN", "FAIL"}:
targets.append("Separate mock-auth or auto-login convenience flows from any production, privacy, or self-host trust boundary narrative.")
if not targets:
return []
lines = ["", "## Remediation Targets"]
for target in targets:
lines.append(f"- {target}")
return lines
def _explain_freshness_section(freshness: dict[str, Any]) -> list[str]:
if not freshness:
return []
lines = [_EXPLAIN_SEP, "Audit Freshness", ""]
lines.append(f" review_after_days {freshness.get('review_after_days', 'n/a')}")
lines.append(f" expires_on {freshness.get('expires_on', 'unknown')}")
lines.append(f" freshness_basis {freshness.get('freshness_basis', 'unknown')}")
lines.append(
f" change_triggered_reaudit_now {freshness.get('change_triggered_reaudit_recommended_now', False)}"
)
lines.append(
f" change_triggered_reasons {', '.join(freshness.get('change_triggered_reaudit_reasons', [])) or 'none'}"
)
lines.append(
f" trigger_examples {', '.join(freshness.get('change_triggers', [])[:4])}"
)
lines.append("")
return lines
def _ast_scope_note(result: dict[str, Any]) -> str | None:
ast = result.get("ast_signal_summary", {})
if not ast or not ast.get("file_limit_exceeded"):
return None
considered = ast.get("files_considered", "unknown")
total = ast.get("files_total", "unknown")
return (
f"AST analysis capped at {considered} of {total} Python files; "
"remaining files were excluded from C1/C4 AST-backed analysis."
)
def _explain_s4_section(s4: dict[str, Any]) -> list[str]:
if not s4:
return []
lines = [_EXPLAIN_SEP, "Stage 4 Replication Rubric", ""]
for key, item in s4.items():
sc, mx, ev = item.get("score", 0), item.get("max", 0), item.get("evidence", "")
lines.append(f" {key:<42} {sc:>3} / {mx:<3} {ev}")
lines.append("")
return lines
def _explain_bio_section(result: dict[str, Any]) -> list[str]:
rows = _bio_detector_rows(result)
if not rows:
return []
lines = [_EXPLAIN_SEP, "Bio Deterministic Diagnostics", ""]
ledger = result.get("evidence_ledger", [])
for detector, label, counts in rows:
parts = [f"{status}={counts[status]}" for status in ("detected", "warn", "error", "not_detected", "not_applicable", "absent") if counts.get(status)]
lines.append(f" {label:<34} {', '.join(parts) if parts else 'no findings'}")
first = next((f for f in ledger if f.get("detector") == detector and f.get("status") == "detected"), None)
if first:
lines.append(f" first finding: {first.get('finding_id', 'n/a')}")
lines.append(f" reason : {first.get('explanation', '')}")
lines.append("")
return lines
def _explain_reasoning_section(reasoning: dict[str, Any]) -> list[str]:
if not reasoning:
return []
lines = [_EXPLAIN_SEP, "Reasoning Diagnostics", ""]
lines.append(f" version {reasoning.get('version', 'unknown')}")
policy = reasoning.get("policy", {})
lines.append(f" mode {policy.get('mode', 'unknown')}")
lines.append(f" final_score_override {policy.get('final_score_override', False)}")
lines.append(f" weights {policy.get('weights', 'unknown')}")
for key in ("evidence_budget", "confidence_envelope", "lane_coherence",
"uncertainty_budget", "evidence_risk_gate"):
item = reasoning.get(key, {})
status = item.get("status", "unknown")
lines.append(f" {key:<31} {status}")
if reasoning.get("lane_coherence", {}).get("status") in {"heuristic_mixed", "heuristic_divergent"}:
lines.append(" interpretation mixed lane coherence; review Stage 2R and Stage 3 evidence manually")
if reasoning.get("uncertainty_budget", {}).get("status") == "review_advised":
lines.append(" review_note uncertainty band elevated; manual review advised")
lines.append("")
return lines
def _explain_advisory_section(advisory: dict[str, Any] | None) -> list[str]:
if not advisory:
return []
lines = [_EXPLAIN_SEP, "AI Advisory Contract", ""]
lines.append(f" schema_version {advisory.get('schema_version', 'unknown')}")
lines.append(f" provider {advisory.get('provider', 'none')}")
lines.append(f" mode {advisory.get('mode', 'unknown')}")
lines.append(f" status {advisory.get('status', 'unknown')}")
lines.append(f" final_score_override {advisory.get('policy', {}).get('final_score_override', False)}")
lines.append(f" invalid_citations {len(advisory.get('invalid_citations', []))}")
lines.append("")
return lines
def _explain_regulatory_section(result: dict[str, Any]) -> list[str]:
basis = result.get("regulatory_basis", {})
traceability = result.get("stage_traceability", {})
if not basis and not traceability:
return []
note = basis.get("note", {})
lines = [_EXPLAIN_SEP, "Regulatory Traceability Assistant", ""]
lines.append(f" {note.get('title', 'Regulatory basis note')}")
lines.append(f" {note.get('body_line_1', '')}")
lines.append(f" {note.get('body_line_2', '')}")
if basis.get("review_required"):
lines.append(f" review_required {', '.join(basis.get('review_reasons', []))}")
lines.append("")
for stage_key in ("stage_1", "stage_2r", "stage_3", "stage_4", "bio_diagnostics"):
items = traceability.get(stage_key, [])
if not items:
continue
lines.append(f" {stage_key:<31}")
for item in items:
lines.append(
f" {item['requirement_id']}: {item['status']} "
f"(mapping={item['mapping_confidence']}, evidence={item['evidence_strength']})"
)
lines.append(f" note: {item['note']}")
summary = result.get("regulatory_traceability", {}).get("summary")
if summary:
lines.append("")
lines.append(f" summary {summary}")
lines.append("")
return lines
def _explain_status_label(statuses: set[str]) -> str:
for candidate in ("error", "detected", "not_detected", "absent", "not_applicable"):
if candidate in statuses:
return candidate.upper()
return next(iter(statuses), "UNKNOWN").upper()
def _bio_detector_rows(result: dict[str, Any]) -> list[tuple[str, str, dict[str, int]]]:
summary = result.get("detector_summary", {}).get("by_detector", {})
rows: list[tuple[str, str, dict[str, int]]] = []
for detector, label in _BIO_DETECTOR_LABELS.items():
counts = summary.get(detector)
if counts:
rows.append((detector, label, counts))
return rows
def _detector_scope_note(result: dict[str, Any], detector: str) -> str:
ledger = result.get("evidence_ledger", [])
for status in ("error", "detected", "not_detected", "not_applicable", "absent"):
finding = next(
(item for item in ledger if item.get("detector") == detector and item.get("status") == status),
None,
)
if finding and finding.get("explanation"):
return str(finding["explanation"])
return "No findings were emitted under current detector scope."
# ── reportlab: document entry point ──────────────────────────────────────────
def _write_rl_pdf(path: Path, result: dict[str, Any], mode: str, pages: int) -> None:
doc = SimpleDocTemplate(
str(path),
pagesize=A4,
topMargin=10 * mm,
bottomMargin=12 * mm,
leftMargin=14 * mm,
rightMargin=14 * mm,
)
story: list[Any] = []
story += _page1_executive(result, mode, pages)
if mode == "detailed":
story += _detail_pages(result, pages)
doc.build(story)
# ── style factory ─────────────────────────────────────────────────────────────
_style_cache: dict[str, Any] = {}
_STYLE_CACHE_LIMIT = 256
def _style(name: str, size: int = 9, leading: int = 12, color: str = _DGRAY,
bold: bool = False, align: str = "LEFT") -> Any:
key = f"{name}_{size}_{leading}_{color}_{bold}_{align}"
if key not in _style_cache:
if len(_style_cache) >= _STYLE_CACHE_LIMIT:
_style_cache.clear()
_style_cache[key] = ParagraphStyle(
key,
fontSize=size,
leading=leading,
textColor=_hx(color),
fontName="Helvetica-Bold" if bold else "Helvetica",
alignment={"LEFT": 0, "CENTER": 1, "RIGHT": 2}.get(align, 0),
)
return _style_cache[key]
# ── Page 1: Executive Dashboard (brief + detailed) ────────────────────────────
def _page1_executive(result: dict[str, Any], mode: str, pages: int) -> list[Any]:
story: list[Any] = []
story += _header_block(result)
story += _score_row(result)
story.append(Spacer(1, 3 * mm))
story += _stage_cards(result)
story.append(Spacer(1, 3 * mm))
story += _integrity_and_risks(result)
story.append(Spacer(1, 2 * mm))
story += _regulatory_basis_box(result)
story += _footer_block()
return _single_page_story(story)
def _header_block(result: dict[str, Any]) -> list[Any]:
t = result["target"]
commit = (t.get("commit") or "")[:12] or "—"
branch = t.get("branch") or "—"
audit_date = result.get("generated_at_local", "—")
mode = result.get("execution_mode", "—")
calibration = result.get("calibration_profile", {})
profile_label = (
f"{calibration.get('profile_name', 'unknown')} "
f"({calibration.get('profile_read_mode', 'unknown')})"
)
header_data = [[Paragraph(
f'STEM BIO-AI Evidence-Surface Scan v{result["stem_ai_version"]}',
_style("H1", 14, 18, _WHITE, True),
)]]
header_tbl = Table(header_data, colWidths=["100%"])
header_tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, -1), _hx(_NAVY)),
("TOPPADDING", (0, 0), (-1, -1), 7),
("BOTTOMPADDING", (0, 0), (-1, -1), 7),
("LEFTPADDING", (0, 0), (-1, -1), 8),
]))
meta = (
f'Repository: {_xt(t["name"])} | '
f'Commit: {commit} | '
f'Branch: {_xt(branch)} | '
f'Audit Date: {audit_date} | '
f'Mode: {mode} | '
f'Policy: {_xt(profile_label)}'
)
meta_data = [[Paragraph(meta, _style("M1", 7, 10, _DGRAY))]]
meta_tbl = Table(meta_data, colWidths=["100%"])
meta_tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, -1), _hx(_MGRAY)),
("TOPPADDING", (0, 0), (-1, -1), 4),
("BOTTOMPADDING", (0, 0), (-1, -1), 4),
("LEFTPADDING", (0, 0), (-1, -1), 8),
]))
return [header_tbl, meta_tbl, Spacer(1, 3 * mm)]
def _score_row(result: dict[str, Any]) -> list[Any]:
score = result["score"]
fs = score["final_score"]
tier = score["formal_tier"]
tier_hex = _tier_hex(tier)
use_scope = score.get("use_scope", "")
score_cell = [
Paragraph(
f'{fs}'
f' / 100',
_style("SC1", 28, 34, _NAVY, True, "CENTER"),
),
Paragraph("Final Score", _style("SL1", 8, 11, _DGRAY, False, "CENTER")),
]
tier_badge = [[Paragraph(
f'{_xt(tier)}',
_style("TB1", 12, 16, _WHITE, True, "CENTER"),
)]]
tier_tbl = Table(tier_badge, colWidths=[60 * mm])
tier_tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, -1), _hx(tier_hex)),
("TOPPADDING", (0, 0), (-1, -1), 5),
("BOTTOMPADDING", (0, 0), (-1, -1), 5),
]))
scope_cell = [
tier_tbl,
Spacer(1, 2 * mm),
Paragraph(
f'Use Scope:
'
f'{_xt(use_scope)}',
_style("US1", 8, 11, _DGRAY),
),
]
weight_note = (
f'Weighted model: '
f'Stage 1 x 0.40 + Stage 2R x 0.20 + Stage 3 x 0.40 '
f'- Risk Penalty = {fs}'
)
row_tbl = Table([[score_cell, scope_cell]], colWidths=[50 * mm, None])
row_tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (0, 0), _hx(_LGRAY)),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("TOPPADDING", (0, 0), (-1, -1), 5),
("BOTTOMPADDING", (0, 0), (-1, -1), 5),
("LEFTPADDING", (0, 0), (-1, -1), 8),
("RIGHTPADDING", (0, 0), (-1, -1), 8),
("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)),
]))
return [row_tbl, Spacer(1, 1 * mm), Paragraph(weight_note, _style("WN1", 7.5, 10, _DGRAY))]
def _stage_cards(result: dict[str, Any]) -> list[Any]:
score = result["score"]
stages = [
("Stage 1", "README Evidence", score["stage_1_readme_intent"], _TEAL),
("Stage 2R", "Repo-Local Consistency", score["stage_2_repo_local_consistency"] or 0, _PURPLE),
("Stage 3", "Code / Bio Responsibility", score["stage_3_code_bio"], _SLATE),
("Stage 4", "Replication Evidence", result.get("replication_score", 0), _GREEN),
]
cells = []
for label, sub, val, col in stages:
card = [
[Paragraph(
f'{label}
{sub}',
_style(f"CH_{label}", 8.5, 12, _WHITE, True, "CENTER"),
)],
[Paragraph(
f'{val}'
f' / 100',
_style(f"CV_{label}", 22, 26, col, True, "CENTER"),
)],
]
t = Table(card, colWidths=["100%"])
t.setStyle(TableStyle([
("BACKGROUND", (0, 0), (0, 0), _hx(col)),
("BACKGROUND", (0, 1), (0, 1), _hx(_LGRAY)),
("TOPPADDING", (0, 0), (-1, -1), 5),
("BOTTOMPADDING", (0, 0), (-1, -1), 6),
("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)),
]))
cells.append(t)
row = Table([cells], colWidths=["25%", "25%", "25%", "25%"])
row.setStyle(TableStyle([
("LEFTPADDING", (0, 0), (-1, -1), 3),
("RIGHTPADDING", (0, 0), (-1, -1), 3),
]))
return [row]
def _integrity_and_risks(result: dict[str, Any]) -> list[Any]:
ci = result["code_integrity"]
risks = result["notable_risks"]
positive = result.get("notable_positive_evidence", [])
airi = result.get("airi_risk_coverage", {})
ci_rows = [[Paragraph(
f'Code Integrity',
_style("CIH1", 9, 12, _WHITE, True),
)]]
labels = {
"C1_hardcoded_credentials": "C1 Credentials",
"C2_dependency_pinning": "C2 Dependency Pinning",
"C3_dead_or_deprecated_patient_adjacent_paths": "C3 Deprecated Paths",
"C4_exception_handling_clinical_adjacent_paths": "C4 Exception Handling",
"C5_compliance_boundary_integrity": "C5 Compliance Boundary",
"C6_mock_auth_or_fail_open_boundary": "C6 Mock Auth Boundary",
}
for key, item in ci.items():
s = item["status"]
sc = _status_hex(s)
ev = _clip_words(item["evidence"][0] if item["evidence"] else "", 92)
badge = [[Paragraph(
f'{s}',
_style(f"B_{key[:4]}", 7, 9, _WHITE, True, "CENTER"),
)]]
bt = Table(badge, colWidths=[15 * mm])
bt.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, -1), _hx(sc)),
("TOPPADDING", (0, 0), (-1, -1), 1),
("BOTTOMPADDING", (0, 0), (-1, -1), 1),
]))
short_key = labels.get(key, key)
ci_rows.append([[
bt,
Paragraph(
f'{short_key}
{_xt(ev)}',
_style(f"CI_{key[:4]}", 7.5, 10, _DGRAY),
),
]])
ci_tbl = Table(ci_rows, colWidths=["100%"])
ci_tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (0, 0), _hx(_NAVY)),
("ROWBACKGROUNDS",(0, 1), (0, -1), [_hx(_LGRAY), _hx(_WHITE)]),
("TOPPADDING", (0, 0), (-1, -1), 4),
("BOTTOMPADDING", (0, 0), (-1, -1), 4),
("LEFTPADDING", (0, 0), (-1, -1), 6),
("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)),
]))
risk_lines = "".join(f'• {_xt(r[:110])}
' for r in risks[:4])
pos_lines = "".join(f'• {_xt(p[:110])}
' for p in positive[:3])
airi_lines = ""
if airi:
airi_lines += (
f'• Covered Risks: {airi.get("covered_count", 0)} / {airi.get("total_risks_in_detector_scope", 0)}'
f' | Rate: {airi.get("coverage_rate", 0):.3f}
'
)
covered = airi.get("covered_risks", [])
if covered:
first = covered[0]
reason = _airi_reason_summary(first)
airi_lines += (
f'• {_xt(str(first.get("id", "—")))} {_xt(_clip_words(str(first.get("title", "")), 9))}'
f'{f"
why: {_xt(_clip_words(reason, 16))}" if reason else ""}'
)
right_rows = [
[Paragraph(f'Remediation Targets', _style("RH1", 9, 12, _WHITE, True))],
[Paragraph(f'{risk_lines}', _style("RL1", 8, 11, _DGRAY))],
[Paragraph(f'Positive Evidence', _style("PH1", 9, 12, _WHITE, True))],
[Paragraph(f'{pos_lines}', _style("PL1", 8, 11, _DGRAY))],
]
if airi_lines:
right_rows.extend([
[Paragraph(f'AIRI Risk Triggers', _style("AH1", 9, 12, _WHITE, True))],
[Paragraph(f'{airi_lines}', _style("AL1", 7.8, 10.5, _DGRAY))],
])
right_tbl = Table(right_rows, colWidths=["100%"])
right_style = [
("BACKGROUND", (0, 0), (0, 0), _hx(_RED)),
("BACKGROUND", (0, 1), (0, 1), _hx(_LGRAY)),
("BACKGROUND", (0, 2), (0, 2), _hx(_GREEN)),
("BACKGROUND", (0, 3), (0, 3), _hx(_WHITE)),
("TOPPADDING", (0, 0), (-1, -1), 4),
("BOTTOMPADDING", (0, 0), (-1, -1), 4),
("LEFTPADDING", (0, 0), (-1, -1), 6),
("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)),
]
if airi_lines:
right_style.extend([
("BACKGROUND", (0, 4), (0, 4), _hx(_TEAL)),
("BACKGROUND", (0, 5), (0, 5), _hx(_LGRAY)),
])
right_tbl.setStyle(TableStyle(right_style))
left_stack = [ci_tbl]
bio_tbl = _bio_diagnostics_pdf_table(result)
if bio_tbl is not None:
left_stack += [Spacer(1, 2 * mm), bio_tbl]
left_col = Table([[item] for item in left_stack], colWidths=["100%"])
left_col.setStyle(TableStyle([
("LEFTPADDING", (0, 0), (-1, -1), 0),
("RIGHTPADDING", (0, 0), (-1, -1), 0),
("TOPPADDING", (0, 0), (-1, -1), 0),
("BOTTOMPADDING",(0, 0), (-1, -1), 0),
("VALIGN", (0, 0), (-1, -1), "TOP"),
]))
two_col = Table([[left_col, right_tbl]], colWidths=["48%", "52%"])
two_col.setStyle(TableStyle([
("LEFTPADDING", (0, 0), (-1, -1), 3),
("RIGHTPADDING", (0, 0), (-1, -1), 3),
("VALIGN", (0, 0), (-1, -1), "TOP"),
]))
return [two_col]
def _bio_diagnostics_pdf_table(result: dict[str, Any]) -> Table | None:
rows = _bio_detector_rows(result)
if not rows:
return None
table_rows: list[list[Any]] = [[
Paragraph(f'Bio Deterministic Diagnostics', _style("BDH1", 8.5, 11, _WHITE, True)),
]]
for _, label, counts in rows:
status_parts = [f"{status}={counts[status]}" for status in ("detected", "not_detected", "not_applicable", "warn", "error", "absent") if counts.get(status)]
table_rows.append([
Paragraph(
f'{_xt(label)}
{_xt(", ".join(status_parts) if status_parts else "no findings")}',
_style(f"BD_{label[:6]}", 7.5, 10, _DGRAY),
)
])
tbl = Table(table_rows, colWidths=["100%"])
tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (0, 0), _hx(_PURPLE)),
("ROWBACKGROUNDS",(0, 1), (0, -1), [_hx(_LGRAY), _hx(_WHITE)]),
("TOPPADDING", (0, 0), (-1, -1), 4),
("BOTTOMPADDING", (0, 0), (-1, -1), 4),
("LEFTPADDING", (0, 0), (-1, -1), 6),
("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)),
]))
return tbl
def _regulatory_basis_box(result: dict[str, Any]) -> list[Any]:
basis = result.get("regulatory_basis", {})
note = basis.get("note", {})
summary = result.get("regulatory_traceability", {}).get("summary", "")
if not note:
return []
body_lines = [
f'{_xt(note.get("title", "Regulatory basis note"))}',
f'{_xt(note.get("body_line_1", ""))}',
f'{_xt(note.get("body_line_2", ""))}',
]
if basis.get("review_required"):
body_lines.append(
f'Review required: {_xt(", ".join(basis.get("review_reasons", [])))}'
)
if summary:
body_lines.append(
f'Traceability summary: {_xt(_clip_words(summary, 220))}'
)
panel = Table(
[[Paragraph("
".join(body_lines), _style("RGB_NOTE", 7.5, 9, _DGRAY))]],
colWidths=["100%"],
)
panel.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, -1), _hx(_LGRAY)),
("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)),
("TOPPADDING", (0, 0), (-1, -1), 5),
("BOTTOMPADDING", (0, 0), (-1, -1), 5),
("LEFTPADDING", (0, 0), (-1, -1), 7),
("RIGHTPADDING", (0, 0), (-1, -1), 7),
]))
return [panel]
def _footer_block() -> list[Any]:
return [
Spacer(1, 4 * mm),
HRFlowable(width="100%", thickness=0.5, color=_hx(_MGRAY)),
Spacer(1, 1.5 * mm),
Paragraph(
f'Independent audit summary — STEM BIO-AI v{__version__} | '
"Not clinical certification. Not regulatory clearance. Not medical advice.",
_style("FT1", 7, 9, _DGRAY, False, "CENTER"),
),
]
def _single_page_story(flowables: list[Any], *, break_before: bool = False) -> list[Any]:
wrapped = KeepInFrame(
_PDF_CONTENT_WIDTH,
_PDF_CONTENT_HEIGHT,
flowables,
mode="shrink",
)
return ([PageBreak()] if break_before else []) + [wrapped]
# ── Detail page dispatcher ────────────────────────────────────────────────────
def _detail_pages(result: dict[str, Any], pages: int) -> list[Any]:
story: list[Any] = []
story += _page2_stage_analysis(result)
story += _page3_stage3_analysis(result)
if pages >= 5:
story += _page4_stage4_replication(result)
if pages == 5:
story += _page5_compact_closure(result)
elif pages >= 7:
story += _page4_integrity_deep(result)
story += _page6_method_airi(result)
story += _page7_report_metadata(result)
return story
# ── Shared detail helpers ─────────────────────────────────────────────────────
def _sec_hdr(title: str, color: str = _NAVY) -> list[Any]:
tbl = Table([[Paragraph(
f'{title}',
_style(f"SH_{title[:8]}", 10, 14, _WHITE, True),
)]], colWidths=["100%"])
tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, -1), _hx(color)),
("TOPPADDING", (0, 0), (-1, -1), 5),
("BOTTOMPADDING", (0, 0), (-1, -1), 5),
("LEFTPADDING", (0, 0), (-1, -1), 8),
]))
return [tbl, Spacer(1, 2 * mm)]
def _mini_score(label: str, val: int, max_val: int, col: str) -> Table:
d = [
[Paragraph(
f'{val}'
f' / {max_val}',
_style(f"MS_{label[:6]}", 20, 24, col, True, "CENTER"),
)],
[Paragraph(label, _style(f"ML_{label[:6]}", 7, 9, _DGRAY, False, "CENTER"))],
]
t = Table(d, colWidths=[34 * mm])
t.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, -1), _hx(_LGRAY)),
("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)),
("TOPPADDING", (0, 0), (-1, -1), 4),
("BOTTOMPADDING", (0, 0), (-1, -1), 3),
("LEFTPADDING", (0, 0), (-1, -1), 3),
("RIGHTPADDING", (0, 0), (-1, -1), 3),
]))
return t
def _rubric_rows(items: list[tuple[str, str, str, str]], id_prefix: str = "R") -> Table:
"""items: (name, score_str, color_hex, evidence_text)"""
header = [
Paragraph(f'Check', _style(f"{id_prefix}H1", 8, 10, _WHITE, True)),
Paragraph(f'Points', _style(f"{id_prefix}H2", 8, 10, _WHITE, True, "CENTER")),
Paragraph(f'Evidence / Finding', _style(f"{id_prefix}H3", 8, 10, _WHITE, True)),
]
rows = [header]
for i, (name, score_str, col, ev) in enumerate(items):
uid = f"{id_prefix}_{i}"
rows.append([
Paragraph(f'{_xt(name)}', _style(f"{uid}N", 8, 11, _DGRAY, True)),
Paragraph(
f'{_xt(score_str)}',
_style(f"{uid}S", 8, 11, col, True, "CENTER"),
),
Paragraph(_xt(_clip_words(ev, 175)), _style(f"{uid}E", 7.5, 10, _DGRAY)),
])
t = Table(rows, colWidths=[52 * mm, 18 * mm, None])
t.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, 0), _hx(_NAVY)),
("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]),
("TOPPADDING", (0, 0), (-1, -1), 3),
("BOTTOMPADDING", (0, 0), (-1, -1), 3),
("LEFTPADDING", (0, 0), (-1, -1), 5),
("RIGHTPADDING", (0, 0), (-1, -1), 5),
("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)),
("LINEBELOW", (0, 0), (-1, 0), 0.5, _hx(_MGRAY)),
("GRID", (0, 0), (-1, -1), 0.3, _hx(_MGRAY)),
]))
return t
# ── Page 2: Stage 1 + Stage 2R Analysis ──────────────────────────────────────
def _page2_stage_analysis(result: dict[str, Any]) -> list[Any]:
story: list[Any] = []
score = result["score"]
cls = result["classification"]
s1 = score["stage_1_readme_intent"]
s2r = score["stage_2_repo_local_consistency"] or 0
ca = cls["clinical_adjacent"]
has_disc = cls["has_explicit_clinical_boundary"]
readme_present = "README.md" in result.get("file_hashes_sha256", {})
# ── Stage 1 ──────────────────────────────────────────────────────────────
story += _sec_hdr("Stage 1 — README Evidence Signal | Weight: 0.40", _TEAL)
s1_rubric = result.get("stage_1_rubric", {})
s1_order = [
("baseline", "Baseline"),
("S1_missing_readme", "README present"),
("S1_domain_readme", "BIO/medical terms in README"),
("S1_domain_package", "BIO/medical terms in package"),
("H1_clinical_certainty_hype", "H1: Clinical Certainty Hype"),
("H2_regulatory_approval_hype", "H2: Regulatory Approval Hype"),
("H3_autonomous_replacement_hype", "H3: Autonomous Replacement Hype"),
("H4_breakthrough_marketing_hype", "H4: Marketing Hype"),
("H5_universal_generalization_hype", "H5: Universal Generalization"),
("H6_perfect_accuracy_hype", "H6: Perfect Accuracy Claim"),
("R1_limitations_section", "R1: Limitations Section"),
("R2_regulatory_framework", "R2: Regulatory Framework"),
("R3_clinical_disclaimer", "R3: Clinical Boundary"),
("R4_demographic_bias_boundary", "R4: Bias / Subgroup Boundary"),
("R5_reproducibility_provisions", "R5: Reproducibility Provisions"),
]
s1_items: list[tuple[str, str, str, str]] = []
for key, label in s1_order:
item = s1_rubric.get(key)
if not item:
continue
pts = item.get("score", 0)
col = _RED if pts < 0 else _GREEN if pts > 0 else _DGRAY
evidence = item.get("evidence", "")
if key == "R2_regulatory_framework":
evidence = (
f"{evidence} "
"[partial-credit ladder: +15 strong framework | +5 weak self-asserted compliance | "
"-5 CA-INDIRECT missing framework | -10 CA-DIRECT missing framework]"
)
s1_items.append((label, f"{pts:+d}", col, evidence))
if not s1_items:
s1_items = [
("Baseline", "+60", _DGRAY, "All non-nascent repositories start at 60."),
("README present", "+0" if readme_present else "-20", _GREEN if readme_present else _RED,
"README.md detected in repository root." if readme_present else "No README found — major deduction applied."),
]
calc_note = s1_rubric.get("calculation", f"Stage 1 evidence score = {s1} / 100")
chip1 = _mini_score("S1 Score", s1, 100, _TEAL)
tbl1 = _rubric_rows(s1_items, "S1")
combined1 = Table([[chip1, tbl1]], colWidths=[38 * mm, None])
combined1.setStyle(TableStyle([
("VALIGN", (0, 0), (-1, -1), "TOP"),
("LEFTPADDING", (0, 0), (-1, -1), 2),
("RIGHTPADDING", (0, 0), (-1, -1), 2),
]))
story.append(combined1)
story.append(Spacer(1, 1 * mm))
story.append(Paragraph(
f'Calculation: {_xt(calc_note)}',
_style("S1CL", 7.5, 10, _DGRAY),
))
# Classification info bar
ca_col = _ORANGE if ca else _GREEN
disc_col = _GREEN if has_disc else _RED
t0_col = _RED if cls["t0_hard_floor"] else _GREEN
info_text = (
f'• Clinical-Adjacent: {"YES" if ca else "NO"}'
f' ({_xt(cls["ca_severity"])}) '
f'• Explicit Disclaimer: {"PRESENT" if has_disc else "ABSENT"}'
f' '
f'• T0 Hard Floor: {"TRIGGERED" if cls["t0_hard_floor"] else "Clear"}'
)
info_tbl = Table([[Paragraph(info_text, _style("INF1", 8, 11, _DGRAY))]], colWidths=["100%"])
info_tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, -1), _hx(_LGRAY)),
("TOPPADDING", (0, 0), (-1, -1), 5),
("BOTTOMPADDING", (0, 0), (-1, -1), 5),
("LEFTPADDING", (0, 0), (-1, -1), 8),
("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)),
]))
story.append(Spacer(1, 2 * mm))
story.append(info_tbl)
# ── Stage 2R ─────────────────────────────────────────────────────────────
story.append(Spacer(1, 5 * mm))
story += _sec_hdr("Stage 2R — Repo-Local Consistency | Weight: 0.20", _PURPLE)
rubric = result.get("stage_2r_rubric", {})
verdict = str(rubric.get("verdict", ""))
calculation = str(rubric.get("calculation", f"= {s2r}"))
_label_map = {
"baseline": "Baseline",
"R2R_1_readme_package_code_alignment": "R2R-1: README / Package Alignment",
"R2R_2_readme_docs_alignment": "R2R-2: README / Docs Alignment",
"R2R_3_readme_test_ci_alignment": "R2R-3: README / Test-CI Alignment",
"R2R_4_limitation_repetition": "R2R-4: Limitation Repetition",
"R2R_D1_internal_clinical_boundary_contradiction": "R2R-D1: Internal Clinical Boundary Contradiction (PENALTY)",
"R2R_D2_missing_clinical_use_boundary": "R2R-D2: Missing Clinical Boundary (PENALTY)",
"R2R_D3_stale_metadata": "R2R-D3: Stale Metadata (PENALTY)",
"R2R_D4_unsupported_workflow_claim": "R2R-D4: Unsupported Workflow Claim (PENALTY)",
}
_ev_tooltip = {
"baseline": "Every repository that is not nascent starts at 60. "
"This baseline accounts for basic structural maturity.",
"R2R_1_readme_package_code_alignment": "README and package metadata share bio-domain vocabulary, "
"indicating claim-to-implementation alignment.",
"R2R_2_readme_docs_alignment": "README and docs/ share domain vocabulary, "
"indicating consistent external communication.",
"R2R_3_readme_test_ci_alignment": "Test and CI surfaces are present and reference the same "
"domain as the README.",
"R2R_4_limitation_repetition": "Limitation or validation-boundary language repeats across more than one repository surface.",
"R2R_D1_internal_clinical_boundary_contradiction": "A non-clinical boundary is declared, but clinical deployment/support claims still appear elsewhere.",
"R2R_D2_missing_clinical_use_boundary": "Clinical-adjacent repository lacks an explicit "
"'research use only' or 'not for diagnostic use' boundary — high review risk.",
"R2R_D3_stale_metadata": "Version metadata appears inconsistent across README and package surfaces.",
"R2R_D4_unsupported_workflow_claim": "README or docs describe runnable workflow support that local tests, workflows, or entrypoints do not substantiate.",
}
s2r_items: list[tuple[str, str, str, str]] = []
for key in ("baseline", "R2R_1_readme_package_code_alignment",
"R2R_2_readme_docs_alignment", "R2R_3_readme_test_ci_alignment",
"R2R_4_limitation_repetition", "R2R_D1_internal_clinical_boundary_contradiction",
"R2R_D2_missing_clinical_use_boundary", "R2R_D3_stale_metadata",
"R2R_D4_unsupported_workflow_claim"):
item = rubric.get(key)
if item is None or not isinstance(item, dict):
continue
sc = item.get("score", 0)
ev_raw = item.get("evidence", "")
ev_ext = _ev_tooltip.get(key, "")
trace = _rubric_trace_suffix(item).strip(" `")
combined_ev = f"{ev_raw} — {ev_ext}" if ev_ext else ev_raw
if trace:
combined_ev = f"{combined_ev} — {trace}"
if key == "baseline":
col = _DGRAY
sc_str = f"+{sc}"
elif key.startswith("R2R_D"):
col = _RED if sc < 0 else _DGRAY
sc_str = str(sc)
else:
col = _TEAL if sc > 0 else _DGRAY
sc_str = f"+{sc}" if sc > 0 else "0 (not detected)"
s2r_items.append((_label_map.get(key, key), sc_str, col, combined_ev))
chip2 = _mini_score("S2R Score", s2r, 100, _PURPLE)
tbl2 = _rubric_rows(s2r_items, "S2R")
combined2 = Table([[chip2, tbl2]], colWidths=[38 * mm, None])
combined2.setStyle(TableStyle([
("VALIGN", (0, 0), (-1, -1), "TOP"),
("LEFTPADDING", (0, 0), (-1, -1), 2),
("RIGHTPADDING", (0, 0), (-1, -1), 2),
]))
story.append(combined2)
story.append(Spacer(1, 1 * mm))
verdict_col = _GREEN if "Strong" in verdict else (_AMBER if "Mixed" in verdict else _RED)
story.append(Paragraph(
f'{_xt(calculation)}'
f' {_xt(verdict)}',
_style("S2RVERDICT", 7.5, 10, _DGRAY),
))
story += _footer_block()
return _single_page_story(story, break_before=True)
# ── Page 3: Stage 3 Full Breakdown ───────────────────────────────────────────
def _page3_stage3_analysis(result: dict[str, Any]) -> list[Any]:
story: list[Any] = []
score = result["score"]
s3 = score["stage_3_code_bio"]
rubric = result.get("stage_3_rubric", {})
story += _sec_hdr("Stage 3 — Code & Bio Responsibility | Weight: 0.40", _SLATE)
_ev_ext = {
"T1_CI_CD": "CI/CD workflows (GitHub Actions, GitLab CI, CircleCI) verify "
"that commits do not silently break the pipeline. Full credit (15) requires workflow files present.",
"T2_domain_tests": "Domain-specific tests verify biological outputs — e.g., "
"sequencing pipeline correctness, variant call validation, or genomic data integrity. "
"Full credit (15) requires BIO-term presence in test files. Partial (8) if tests exist but are generic.",
"T3_changelog_release_hygiene": "A CHANGELOG tracks which version fixed which defect — "
"essential for regulatory traceability and reproducibility audits. "
"CHANGELOG.md, CHANGELOG, or NEWS.md all qualify.",
"B1_data_provenance_controls": "Dependency manifests "
"(requirements.txt, pyproject.toml, environment.yml) establish reproducibility context. "
"Score 10 if manifest detected; max 15 requires data-source, dataset-citation, or IRB language.",
"B2_bias_limitations": "Documentation of algorithmic bias, limitations, "
"or model boundary conditions. Score 8 for boundary language; max 15 requires "
"measurement evidence such as subgroup analysis, calibration, or test coverage.",
"B3_coi_funding": "Conflict of interest and funding disclosure in README or FUNDING.md. "
"Required for institutional review context and detected by local text scan.",
}
t_items: list[tuple[str, str, str, str]] = []
for key, label in [
("T1_CI_CD", "T1: CI/CD Workflow"),
("T2_domain_tests", "T2: Domain-Specific Tests"),
("T3_changelog_release_hygiene", "T3: Changelog & Release Hygiene"),
]:
item = rubric.get(key, {})
sc = item.get("score", 0)
mx = item.get("max", 15)
ev = item.get("evidence", "")
ext = _ev_ext.get(key, "")
trace = _rubric_trace_suffix(item).strip(" `")
col = _GREEN if sc == mx else (_AMBER if sc > 0 else _RED)
combined = f"{ev} — {ext}" if ext else ev
if trace:
combined = f"{combined} — {trace}"
t_items.append((label, f"{sc} / {mx}", col, combined))
b_items: list[tuple[str, str, str, str]] = []
for key, label in [
("B1_data_provenance_controls", "B1: Data Provenance Controls"),
("B2_bias_limitations", "B2: Bias / Limitations Documentation"),
("B3_coi_funding", "B3: COI & Funding Disclosure"),
]:
item = rubric.get(key, {})
sc = item.get("score", 0)
mx = item.get("max", 15)
ev = item.get("evidence", "")
ext = _ev_ext.get(key, "")
trace = _rubric_trace_suffix(item).strip(" `")
not_detectable = "local CLI scan" in ev
col = (_GREEN if sc == mx else (_AMBER if sc > 0 else
(_DGRAY if not_detectable else _RED)))
note = " [Manual review required]" if not_detectable else ""
combined = f"{ev} — {ext}" if ext else ev
if trace:
combined = f"{combined} — {trace}"
b_items.append((label, f"{sc} / {mx}{note}", col, combined))
chip3 = _mini_score("S3 Score", s3, 100, _SLATE)
body_items: list[Any] = [
Paragraph(
f'Engineering Accountability (T-series)',
_style("TS1", 8.5, 12, _SLATE, True),
),
Spacer(1, 1 * mm),
_rubric_rows(t_items, "T"),
Spacer(1, 3 * mm),
Paragraph(
f'Biological Integrity (B-series)',
_style("BS1", 8.5, 12, _SLATE, True),
),
Spacer(1, 1 * mm),
_rubric_rows(b_items, "B"),
]
main_row = Table([[chip3, body_items]], colWidths=[38 * mm, None])
main_row.setStyle(TableStyle([
("VALIGN", (0, 0), (-1, -1), "TOP"),
("LEFTPADDING", (0, 0), (-1, -1), 2),
("RIGHTPADDING", (0, 0), (-1, -1), 2),
]))
story.append(main_row)
# Gap analysis
story.append(Spacer(1, 5 * mm))
story += _sec_hdr("Stage 3 Gap Analysis — Path to Next Tier", _DGRAY)
local_max = 55
fs = score["final_score"]
gap_t3 = max(0, 70 - fs)
gap_t4 = max(0, 85 - fs)
t_total = sum(rubric.get(k, {}).get("score", 0) for k in ["T1_CI_CD", "T2_domain_tests", "T3_changelog_release_hygiene"])
b_total = sum(rubric.get(k, {}).get("score", 0) for k in ["B1_data_provenance_controls", "B2_bias_limitations", "B3_coi_funding"])
gap_lines = [
f'• T-series (engineering) attained: {t_total} / 45 '
f'B-series (bio integrity) attained: {b_total} / 35',
f'• Local CLI scan maximum: {local_max} / 100 '
f'(T1+T2+T3 max 15 each; B1 max 10; B2/B3 require manual review)',
f'• Gap to T3 (final score >= 70): {gap_t3} points needed across all stages',
f'• Gap to T4 (final score >= 85): {gap_t4} points needed across all stages',
'• B2 Bias/Limitations: Not detectable — requires manual audit of README, '
'model card, or supplementary documentation for validation boundaries and algorithmic limitations',
'• B3 COI/Funding: Not detectable — requires inspection of README or FUNDING.md '
'for conflict of interest and funding source disclosure',
]
for line in gap_lines:
story.append(Paragraph(
f'{line}',
_style(f"GL_{line[:6]}", 8, 13, _DGRAY),
))
story += _footer_block()
return _single_page_story(story, break_before=True)
# ── Page 4: Stage 4 Replication Deep Dive (5p/7p) ────────────────────────────
def _page4_stage4_replication(result: dict[str, Any]) -> list[Any]:
story: list[Any] = []
score = result["score"]
stage4_score = result.get("replication_score", 0)
stage4_tier = result.get("replication_tier", "R0")
rubric = result.get("stage_4_rubric", {})
story += _sec_hdr("Stage 4 — Replication Evidence Lane | Separate lane", _GREEN)
label_map = {
"S4_container_environment": "S4: Container / Runtime Environment",
"S4_make_reproduce_target": "S4: Reproduce Target",
"S4_environment_lock_evidence": "S4: Environment Lock Evidence",
"S4_exact_dependency_pins_or_hashes": "S4: Exact Dependency Pins / Hashes",
"S4_readme_reproducibility_section": "S4: Reproducibility Section",
"S4_checksum_files": "S4: Checksums / Integrity Files",
"S4_dataset_url": "S4: Dataset / Data Source URL",
"S4_model_weight_url_or_checksum": "S4: Model Artifact URL / Checksum",
"S4_citation_cff": "S4: CITATION.cff",
"S4_license_restriction": "S4: License / Use Restriction",
"S4_cli_entrypoint": "S4: CLI Entrypoint",
"S4_seed_setting": "S4: Deterministic Seed Setting",
"S4_runnable_examples": "S4: Runnable Examples",
}
items: list[tuple[str, str, str, str]] = []
for key, item in rubric.items():
if not isinstance(item, dict) or "score" not in item or "max" not in item:
continue
sc = item.get("score", 0)
mx = item.get("max", 0)
color = _GREEN if sc == mx and mx else (_AMBER if sc > 0 else _RED)
evidence = item.get("evidence", "")
items.append((label_map.get(key, key), f"{sc} / {mx}", color, evidence))
chip = _mini_score("S4 Score", stage4_score, 100, _GREEN)
tbl = _rubric_rows(items, "S4")
combined = Table([[chip, tbl]], colWidths=[38 * mm, None])
combined.setStyle(TableStyle([
("VALIGN", (0, 0), (-1, -1), "TOP"),
("LEFTPADDING", (0, 0), (-1, -1), 2),
("RIGHTPADDING", (0, 0), (-1, -1), 2),
]))
story.append(combined)
story.append(Spacer(1, 2 * mm))
story.append(Paragraph(
f'Replication tier: {stage4_tier}. '
'Stage 4 is reported separately and does not alter the formal score.',
_style("S4_NOTE", 7.5, 10, _DGRAY),
))
story.append(Spacer(1, 3 * mm))
story.append(Paragraph(
f''
f'Final score remains {score["final_score"]} / 100 ({_xt(score["formal_tier"])}) even when Stage 4 moves. '
'This lane exists to show reproducibility and operational evidence posture separately from the formal repository score.'
f'',
_style("S4_SCOPE", 8, 11, _DGRAY),
))
story += _footer_block()
return _single_page_story(story, break_before=True)
# ── Page 5: Code Integrity Deep Dive + Classification (7p only) ──────────────
def _page4_integrity_deep(result: dict[str, Any]) -> list[Any]:
story: list[Any] = []
ci = result["code_integrity"]
cls = result["classification"]
hashes = result.get("file_hashes_sha256", {})
tgt = result["target"]
ast_note = _ast_scope_note(result)
story += _sec_hdr("Code Integrity — Deep Analysis", _NAVY)
_remediation = {
"C1_hardcoded_credentials":
"CRITICAL: Rotate all exposed credentials immediately. Remove from git history "
"using git-filter-repo. Use environment variables or a secrets manager (AWS Secrets Manager, "
"HashiCorp Vault, Azure Key Vault). Add a pre-commit hook with detect-secrets.",
"C2_dependency_pinning":
"Pin all dependencies to exact versions (== for pip, hash-pinning for conda). "
"Run pip-audit or safety regularly. Consider pip-compile for reproducible lock files. "
"Unpinned ranges in clinical-adjacent pipelines create silent regression risk.",
"C3_dead_or_deprecated_patient_adjacent_paths":
"Audit deprecated/ directories for patient-adjacent metadata patterns. "
"If clinical data was processed, verify data destruction or anonymization logs. "
"Dead code with patient metadata patterns must be purged or explicitly annotated as test fixtures.",
"C4_exception_handling_clinical_adjacent_paths":
"Replace broad 'except Exception: pass' or 'except: return True' patterns with "
"specific error types and explicit failure logging. In clinical-adjacent code paths, "
"any silent failure is a patient safety risk. Fail closed, not open.",
"C5_compliance_boundary_integrity":
"Treat privacy, legal, or clinical-adjacent claims as governance obligations. "
"If README or product text invokes HIPAA, compliance, or self-hosted clinical safety, "
"surface supporting controls, operating boundaries, and deployment constraints explicitly.",
"C6_mock_auth_or_fail_open_boundary":
"Do not present self-host, local-mode, or privacy-sensitive flows as production-like if they rely on "
"mock authentication, auto-login, or no-auth convenience boundaries. Separate demo convenience from trust posture.",
}
_desc = {
"C1_hardcoded_credentials":
"Scans for AWS access keys (AKIA*), OpenAI keys (sk-*), GitHub tokens (ghp_*), "
"and api_key = '...' patterns in all text files.",
"C2_dependency_pinning":
"Checks whether requirements.txt / pyproject.toml / environment.yml use "
"exact version pins (==, sha256 hash) or loose ranges (>=, no pin).",
"C3_dead_or_deprecated_patient_adjacent_paths":
"Scans deprecated/ directories for patient metadata patterns: "
"patient_id, patient_age, patient_sex, sample_id, collection_date, lab_id, etc.",
"C4_exception_handling_clinical_adjacent_paths":
"Detects fail-open exception patterns: 'except Exception: pass' or "
"'except: return True' in code — these silently ignore errors that could corrupt clinical outputs.",
"C5_compliance_boundary_integrity":
"Detects unsupported legal/compliance claims or clinical-boundary weaknesses in reviewed repository "
"sources, including self-asserted privacy/compliance language without visible governance grounding.",
"C6_mock_auth_or_fail_open_boundary":
"Detects mock-auth, auto-login, or no-auth local/self-host boundary patterns in README, config, and code "
"when trust-boundary language suggests a stronger operational posture than the reviewed sources support.",
}
short = {
"C1_hardcoded_credentials": "C1: Hardcoded Credentials",
"C2_dependency_pinning": "C2: Dependency Pinning",
"C3_dead_or_deprecated_patient_adjacent_paths": "C3: Deprecated Patient Paths",
"C4_exception_handling_clinical_adjacent_paths": "C4: Fail-Open Exceptions",
"C5_compliance_boundary_integrity": "C5: Compliance Boundary Integrity",
"C6_mock_auth_or_fail_open_boundary": "C6: Mock Auth / Fail-Open Boundary",
}
ci_items: list[tuple[str, str, str, str]] = []
for key, cfg in ci.items():
s = cfg["status"]
col = _status_hex(s)
ev_raw = cfg["evidence"][0] if cfg["evidence"] else ""
ev_full = f"{ev_raw} | Scan: {_desc.get(key, '')}"
ci_items.append((short.get(key, key), s, col, ev_full))
story.append(_rubric_rows(ci_items, "CI"))
story.append(Spacer(1, 3 * mm))
# Remediation guidance
fail_warn = [(k, v) for k, v in ci.items() if v["status"] != "PASS"]
if fail_warn:
story += _sec_hdr("Remediation Guidance", _RED)
for key, v in fail_warn:
s = v["status"]
col = _RED if s == "FAIL" else _AMBER
guidance = _remediation.get(key, "Review and remediate before clinical-adjacent deployment.")
story.append(Paragraph(
f'[{s}] {_xt(short.get(key, key))}:',
_style(f"RG_H_{key[:4]}", 8.5, 12, col, True),
))
story.append(Paragraph(
f'→ {_xt(guidance)}',
_style(f"RG_B_{key[:4]}", 8, 11, _DGRAY),
))
story.append(Spacer(1, 2 * mm))
else:
story.append(Paragraph(
f'All code integrity checks PASSED. '
'Continue monitoring with each major release.',
_style("CIOK", 8, 11, _GREEN),
))
# Classification analysis
story.append(Spacer(1, 3 * mm))
story += _sec_hdr("Classification & Repository Analysis", _SLATE)
cls_items: list[tuple[str, str, str, str]] = [
("Clinical Adjacent", "YES" if cls["clinical_adjacent"] else "NO",
_ORANGE if cls["clinical_adjacent"] else _GREEN,
f'Severity: {cls["ca_severity"]}. '
'Triggered by BIO/CLINICAL_OUTPUT term regex match across README, docs, and code.'),
("T0 Hard Floor", "TRIGGERED" if cls["t0_hard_floor"] else "Clear",
_RED if cls["t0_hard_floor"] else _GREEN,
"Score forced to 0 regardless of rubric performance — e.g., AGI claim in clinical context."
if cls["t0_hard_floor"] else "No T0_HARD_FLOOR condition detected."),
("Explicit Disclaimer", "PRESENT" if cls["has_explicit_clinical_boundary"] else "ABSENT",
_GREEN if cls["has_explicit_clinical_boundary"] else _AMBER,
"Regex: 'not for clinical|not for diagnostic|research use only|not medical advice' "
"in README + docs surface." if cls["has_explicit_clinical_boundary"] else
"Disclaimer pattern not found in README or docs. High impact on Stage 1 and Stage 2R scores."),
("Files Scanned", str(tgt.get("file_count", "—")), _TEAL,
"Total files indexed by recursive walk. Text files only for content analysis; "
"binary files counted but not read."),
("Execution Mode", result.get("execution_mode", "—"), _DGRAY,
"No LLM calls. No network access. No runtime execution. "
"Deterministic regex + file-system scan only."),
]
if ast_note:
cls_items.append((
"AST Analysis Scope",
"CAPPED",
_AMBER,
ast_note,
))
story.append(_rubric_rows(cls_items, "CLS"))
# File hashes
if hashes:
story.append(Spacer(1, 3 * mm))
story.append(Paragraph(
f'File Integrity (SHA-256)',
_style("FIH1", 9, 12, _NAVY, True),
))
story.append(Spacer(1, 1 * mm))
hash_rows: list[list[Any]] = [[
Paragraph(f'File', _style("FHH1", 8, 10, _WHITE, True)),
Paragraph(f'SHA-256 Hash', _style("FHH2", 8, 10, _WHITE, True)),
]]
for fname, h in hashes.items():
hash_rows.append([
Paragraph(_xt(fname), _style(f"FN_{fname[:4]}", 8, 11, _DGRAY)),
Paragraph(f'{h}',
_style(f"FV_{fname[:4]}", 6.5, 8, _DGRAY)),
])
hash_tbl = Table(hash_rows, colWidths=[40 * mm, None])
hash_tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, 0), _hx(_NAVY)),
("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]),
("TOPPADDING", (0, 0), (-1, -1), 3),
("BOTTOMPADDING", (0, 0), (-1, -1), 3),
("LEFTPADDING", (0, 0), (-1, -1), 5),
("RIGHTPADDING", (0, 0), (-1, -1), 5),
("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)),
("GRID", (0, 0), (-1, -1), 0.3, _hx(_MGRAY)),
]))
story.append(hash_tbl)
story += _footer_block()
return _single_page_story(story, break_before=True)
# ── Page 5: Compact Closeout (5p standard packet only) ───────────────────────
def _page5_compact_closure(result: dict[str, Any]) -> list[Any]:
story: list[Any] = []
score = result["score"]
risks = result.get("notable_risks", [])
airi = result.get("airi_risk_coverage", {})
ci = result["code_integrity"]
story += _sec_hdr("Closeout Summary", _NAVY)
ci_rows: list[tuple[str, str, str, str]] = []
short = {
"C1_hardcoded_credentials": "C1: Hardcoded Credentials",
"C2_dependency_pinning": "C2: Dependency Pinning",
"C3_dead_or_deprecated_patient_adjacent_paths": "C3: Deprecated Patient Paths",
"C4_exception_handling_clinical_adjacent_paths": "C4: Fail-Open Exceptions",
"C5_compliance_boundary_integrity": "C5: Compliance Boundary Integrity",
"C6_mock_auth_or_fail_open_boundary": "C6: Mock Auth / Fail-Open Boundary",
}
for key, cfg in ci.items():
status = cfg["status"]
color = _status_hex(status)
evidence = cfg["evidence"][0] if cfg.get("evidence") else ""
ci_rows.append((short.get(key, key), status, color, evidence))
story.append(_rubric_rows(ci_rows, "CC"))
story.append(Spacer(1, 3 * mm))
if risks:
story += _sec_hdr("Top Risks", _RED)
for risk in risks[:4]:
story.append(Paragraph(
f'• {_xt(risk)}',
_style(f"TR_{risk[:6]}", 8, 11, _DGRAY),
))
if airi:
story.append(Spacer(1, 3 * mm))
story += _sec_hdr("AIRI Risk Triggers Summary", _TEAL)
story.append(Paragraph(
f''
f'Covered Risks: {airi.get("covered_count", 0)} / {airi.get("total_risks_in_detector_scope", 0)} '
f'| Coverage Rate: {airi.get("coverage_rate", 0):.3f}',
_style("AIRI_COMPACT", 8, 11, _DGRAY),
))
story.append(Paragraph(
f'{_xt(_surface_compaction_note(result))}',
_style("AIRI_COMPACT_NOTE", 7, 10, _DGRAY),
))
for risk in airi.get("covered_risks", [])[:3]:
reason = _airi_reason_summary(risk)
story.append(Paragraph(
f'• {_xt(str(risk.get("id", "—")))} '
f'{_xt(str(risk.get("title", "")))}'
f'{f" — why: {_xt(reason)}" if reason else ""}',
_style(f"AIRC_{str(risk.get('id', 'risk'))[:8]}", 8, 10, _DGRAY),
))
if airi.get("known_gaps_in_bundle"):
gap_preview = ", ".join(
f"{g.get('id', '—')} {_xt(str(g.get('title', '')))}"
for g in airi.get("known_gaps_in_bundle", [])[:2]
)
story.append(Paragraph(
f'Known gaps preview: {gap_preview}',
_style("AIRI_GAPC", 8, 10, _DGRAY),
))
story.append(Spacer(1, 3 * mm))
story += _sec_hdr("Method Boundary", _DGRAY)
story.append(Paragraph(
f'Final score: {score["final_score"]} / 100 ({_xt(score["formal_tier"])})'
f'
Method: {_xt(result.get("method", ""))}',
_style("MBCOMPACT", 8, 11, _DGRAY),
))
story += _footer_block()
return _single_page_story(story, break_before=True)
# ── Page 6: Priority Improvements + AIRI + Method (7p only) ─────────────────
def _page6_method_airi(result: dict[str, Any]) -> list[Any]:
story: list[Any] = []
risks = result.get("notable_risks", [])
positive = result.get("notable_positive_evidence", [])
airi = result.get("airi_risk_coverage", {})
# Priority Improvements
story += _sec_hdr("Priority Improvement Roadmap", _RED)
_pri_detail: dict[str, str] = {
"Clinical-adjacent surfaces exist without an explicit non-diagnostic/non-clinical boundary.":
"Add a prominent 'Research Use Only — Not for Clinical or Diagnostic Use' disclaimer "
"to README H1 or H2 section. Reference applicable frameworks: FDA SaMD guidance, "
"EU AI Act Article 6, or IRB oversight requirements for your deployment context.",
"C1_hardcoded_credentials: FAIL":
"CRITICAL: Rotate all exposed credentials immediately. Remove from git history "
"using git-filter-repo. Implement pre-commit secrets detection. "
"Use environment variables or a secrets manager for all future credential handling.",
"C2_dependency_pinning: WARN":
"Pin all production dependencies to exact versions (== for pip). "
"Add pip-audit or safety to CI pipeline for vulnerability scanning. "
"Consider pip-compile for deterministic lock files.",
"C3_dead_or_deprecated_patient_adjacent_paths: WARN":
"Audit deprecated/ directories for patient-adjacent metadata patterns. "
"If clinical data was processed historically, verify destruction or anonymization logs. "
"If patterns are from test fixtures, annotate clearly with # noqa comments.",
"C4_exception_handling_clinical_adjacent_paths: WARN":
"Replace broad exception handlers with specific error types and explicit logging. "
"In any clinical-adjacent code path: fail closed, not open. "
"Never silently return True or pass on exception.",
"C5_compliance_boundary_integrity: WARN":
"Do not rely on unsupported legal, privacy, or clinical-boundary claims. "
"Add explicit deployment boundaries, governance controls, and operational evidence before using such language.",
"C6_mock_auth_or_fail_open_boundary: WARN":
"Do not treat mock-auth, auto-login, or no-auth self-host flows as production-ready trust boundaries. "
"Separate convenience development paths from privacy, security, and compliance posture claims.",
}
no_major_risks = not risks or risks == ["No major local risks detected by the CLI scan."]
if no_major_risks:
story.append(Paragraph(
f'No critical risks detected by local CLI scan.
'
f'A manual audit is still recommended for '
'clinical-adjacent deployment. Local CLI cannot assess B2 (bias) or B3 (COI).',
_style("NR1", 8, 12, _DGRAY),
))
else:
for i, risk in enumerate(risks, 1):
guidance = _pri_detail.get(risk, (
"Review this finding and implement appropriate controls "
"before supervised or clinical-adjacent deployment."
))
ri_col = _RED if ("FAIL" in risk or "Clinical-adjacent" in risk) else _AMBER
story.append(Paragraph(
f'Priority {i}: {_xt(risk)}',
_style(f"PRI_{i}", 8.5, 12, ri_col, True),
))
story.append(Paragraph(
f'→ {_xt(guidance)}',
_style(f"PRG_{i}", 8, 11, _DGRAY),
))
story.append(Spacer(1, 2.5 * mm))
# Positive Evidence
story.append(Spacer(1, 2 * mm))
story += _sec_hdr("Positive Evidence Summary", _GREEN)
for ev in positive:
story.append(Paragraph(
f'• {_xt(ev)}',
_style(f"PE_{ev[:4]}", 8, 12, _DGRAY),
))
# AIRI summary
if airi:
story.append(Spacer(1, 3 * mm))
story += _sec_hdr("AIRI Risk Triggers Summary", _TEAL)
story.append(Paragraph(
f''
f'Covered Risks: {airi.get("covered_count", 0)} / {airi.get("total_risks_in_detector_scope", 0)} '
f'| Coverage Rate: {airi.get("coverage_rate", 0):.3f} '
f'| Bundle Scope: {_xt(str(airi.get("airi_bundle_scope", "unknown")))}'
f'',
_style("AIRIPDF_SUMMARY", 8, 12, _DGRAY),
))
story.append(Paragraph(
f'{_xt(_surface_compaction_note(result))}',
_style("AIRIPDF_NOTE", 7, 10, _DGRAY),
))
covered_risks = airi.get("covered_risks", [])
if covered_risks:
for risk in covered_risks[:3]:
reason = _airi_reason_summary(risk)
primary = _airi_primary_summary(risk)
story.append(Paragraph(
f'• {_xt(str(risk.get("id", "—")))} '
f'{_xt(str(risk.get("title", "")))}'
f'{f" — { _xt(primary) }" if primary else ""}'
f'{f" — why: {_xt(reason)}" if reason else ""}',
_style(f"AIRIPDF_{str(risk.get('id', 'risk'))[:8]}", 8, 11, _DGRAY),
))
gaps = airi.get("known_gaps_in_bundle", [])
if gaps:
gap_preview = ", ".join(
f"{g.get('id', '—')} {_xt(str(g.get('title', '')))}" for g in gaps[:2]
)
story.append(Paragraph(
f'Known gaps preview: {gap_preview}',
_style("AIRIPDF_GAPS", 8, 11, _DGRAY),
))
# Method Boundary
story.append(Spacer(1, 4 * mm))
story += _sec_hdr("Method Boundary", _DGRAY)
story.append(Paragraph(
_xt(result.get("method", "")),
_style("MB2", 8, 12, _DGRAY),
))
story.append(Spacer(1, 1 * mm))
story.append(Paragraph(
f'Scope boundary: '
'Runtime behavior, model output correctness, '
'dynamic validation, wet-lab reproducibility, and clinical validation are '
'outside the scope of this local CLI scan. This report assesses structural signals only.',
_style("MBSCOPE", 8, 11, _DGRAY),
))
story += _footer_block()
return _single_page_story(story, break_before=True)
# ── Page 7: Report Metadata (7p only) ────────────────────────────────────────
def _page7_report_metadata(result: dict[str, Any]) -> list[Any]:
story: list[Any] = []
score = result["score"]
story += _sec_hdr("Report Metadata", _NAVY)
tgt = result["target"]
meta_items = [
("Schema Version", result.get("schema_version", "—")),
("STEM BIO-AI Version", result.get("stem_ai_version", "—")),
("Generated (local date)", result.get("generated_at_local", "—")),
("Report Validity", "180 days from audit date"),
("Execution Mode", result.get("execution_mode", "—")),
("Repository", tgt["name"]),
("Remote URL", (tgt.get("remote") or "—")[:70]),
("Branch", tgt.get("branch") or "—"),
("Commit (HEAD)", (tgt.get("commit") or "—")[:40]),
("Files Scanned", str(tgt.get("file_count", "—"))),
("Final Score / Tier", f'{score["final_score"]} / 100 — {score["formal_tier"]}'),
]
meta_data: list[list[Any]] = [[
Paragraph(f'Field', _style("MH1", 8, 10, _WHITE, True)),
Paragraph(f'Value', _style("MH2", 8, 10, _WHITE, True)),
]]
for field, val in meta_items:
meta_data.append([
Paragraph(f'{_xt(field)}', _style(f"MF_{field[:4]}", 8, 11, _DGRAY, True)),
Paragraph(_xt(str(val)), _style(f"MV_{field[:4]}", 8, 11, _DGRAY)),
])
meta_tbl = Table(meta_data, colWidths=[55 * mm, None])
meta_tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, 0), _hx(_NAVY)),
("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]),
("TOPPADDING", (0, 0), (-1, -1), 3),
("BOTTOMPADDING", (0, 0), (-1, -1), 3),
("LEFTPADDING", (0, 0), (-1, -1), 5),
("RIGHTPADDING", (0, 0), (-1, -1), 5),
("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)),
("GRID", (0, 0), (-1, -1), 0.3, _hx(_MGRAY)),
]))
story.append(meta_tbl)
story += _footer_block()
return _single_page_story(story, break_before=True)
# ── plain-text PDF fallback (no reportlab) ───────────────────────────────────
def render_pdf_pages(result: dict[str, Any], mode: str, pages: int) -> list[list[str]]:
score = result["score"]
ast_note = _ast_scope_note(result)
airi = result.get("airi_risk_coverage", {})
airi_brief = [
"",
"AIRI Risk Triggers Summary",
f"- Covered Risks: {airi.get('covered_count', 0)} / {airi.get('total_risks_in_detector_scope', 0)}",
f"- Coverage Rate: {airi.get('coverage_rate', 0):.3f}",
f"- Surface Note: {_surface_compaction_note(result)}",
]
covered_risks = airi.get("covered_risks", [])
if covered_risks:
risk = covered_risks[0]
reason = _airi_reason_summary(risk)
primary = _airi_primary_summary(risk)
line = f"- {risk.get('id', '—')}: {risk.get('title', '')}"
if primary:
line += f" | {primary}"
if reason:
line += f" | why: {reason}"
airi_brief.append(line)
brief = [
"STEM BIO-AI Local Audit Brief",
f"Target: {result['target']['name']}",
f"Final Score: {score['final_score']} / 100",
f"Formal Tier: {score['formal_tier']}",
f"Use Scope: {score['use_scope']}",
"",
"Stage Scores",
f"- Stage 1 README Evidence Signal: {score['stage_1_readme_intent']} / 100",
f"- Stage 2R Repo-Local Consistency: {score['stage_2_repo_local_consistency']} / 100",
f"- Stage 3 Code/Bio Responsibility: {score['stage_3_code_bio']} / 100",
f"- Stage 4 Replication Evidence: {result.get('replication_score', 0)} / 100 ({result.get('replication_tier', 'R0')})",
"",
"Code Integrity",
*[f"- {k}: {v['status']}" for k, v in result["code_integrity"].items()],
*([f"- AST analysis scope: {ast_note}"] if ast_note else []),
"",
"Top Risks",
*[f"- {r}" for r in result["notable_risks"][:4]],
*airi_brief,
"",
"Not clinical certification. Not regulatory clearance. Not medical advice.",
]
if mode == "brief":
return [_fit_page(brief)]
p2 = _fit_page(["Stage 2R Evidence", *[
f"- {k}: {v.get('score','')} {v.get('evidence','')}"
for k, v in result["stage_2r_rubric"].items() if isinstance(v, dict)
]])
p3 = _fit_page(["Stage 3 Evidence", *[
f"- {k}: {v['score']} / {v['max']} {v['evidence']}"
for k, v in result["stage_3_rubric"].items()
], "", "Stage 4 Replication Evidence", *[
f"- {k}: {v['score']} / {v['max']} {v['evidence']}"
for k, v in result.get("stage_4_rubric", {}).items()
]])
p4 = _fit_page([
"Stage 4 Replication Evidence",
f"- Stage 4 Replication Score: {result.get('replication_score', 0)} / 100 ({result.get('replication_tier', 'R0')})",
*[
f"- {k}: {v['score']} / {v['max']} {v['evidence']}"
for k, v in result.get("stage_4_rubric", {}).items()
],
])
sets = [_fit_page(brief), p2, _fit_page([
"Stage 3 Evidence",
*[f"- {k}: {v['score']} / {v['max']} {v['evidence']}" for k, v in result["stage_3_rubric"].items()]
])]
if pages == 5:
sets.append(p4)
sets.append(_fit_page([
"Closeout Summary",
"Code Integrity",
*[f"- {k}: {v['status']} {v['evidence'][0]}" for k, v in result["code_integrity"].items()],
*([f"- AST analysis scope: {ast_note}"] if ast_note else []),
"",
"AIRI Risk Triggers Summary",
f"- Covered Risks: {airi.get('covered_count', 0)} / {airi.get('total_risks_in_detector_scope', 0)}",
f"- Coverage Rate: {airi.get('coverage_rate', 0):.3f}",
f"- Surface Note: {_surface_compaction_note(result)}",
*[
f"- {risk.get('id', '—')}: {risk.get('title', '')}"
+ (f" | {_airi_primary_summary(risk)}" if _airi_primary_summary(risk) else "")
+ (f" | why: {_airi_reason_summary(risk)}" if _airi_reason_summary(risk) else "")
for risk in airi.get("covered_risks", [])[:2]
],
"",
"Method Boundary",
result["method"],
]))
elif pages >= 7:
sets.append(p4)
sets.append(_fit_page([
"Code Integrity",
*[f"- {k}: {v['status']} {v['evidence'][0]}" for k, v in result["code_integrity"].items()],
*([f"- AST analysis scope: {ast_note}"] if ast_note else []),
]))
sets.append(_fit_page([
"Priority Improvement Roadmap",
*[f"- {r}" for r in result.get("notable_risks", [])[:4]],
"",
"AIRI Risk Triggers Summary",
f"- Covered Risks: {airi.get('covered_count', 0)} / {airi.get('total_risks_in_detector_scope', 0)}",
f"- Coverage Rate: {airi.get('coverage_rate', 0):.3f}",
f"- Bundle Scope: {airi.get('airi_bundle_scope', 'unknown')}",
f"- Surface Note: {_surface_compaction_note(result)}",
*[
f"- {risk.get('id', '—')}: {risk.get('title', '')}"
+ (f" | {_airi_primary_summary(risk)}" if _airi_primary_summary(risk) else "")
+ (f" | why: {_airi_reason_summary(risk)}" if _airi_reason_summary(risk) else "")
for risk in airi.get("covered_risks", [])[:3]
],
"",
"Method Boundary",
result["method"],
]))
sets.append(_fit_page([
"Report Metadata",
f"- Schema Version: {result.get('schema_version', '—')}",
f"- STEM BIO-AI Version: {result.get('stem_ai_version', '—')}",
f"- Generated (local date): {result.get('generated_at_local', '—')}",
f"- Repository: {result['target']['name']}",
f"- Branch: {result['target'].get('branch') or '—'}",
f"- Commit (HEAD): {(result['target'].get('commit') or '—')[:40]}",
f"- Files Scanned: {result['target'].get('file_count', '—')}",
f"- Final Score / Tier: {score['final_score']} / 100 — {score['formal_tier']}",
]))
return sets[:pages]
def write_simple_pdf(path: Path, pages: list[list[str]]) -> None:
objects: list[bytes] = []
def add(obj: str) -> int:
objects.append(obj.encode("latin-1", errors="replace"))
return len(objects)
font_id = add("<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>")
page_ids: list[int] = []
content_ids: list[int] = []
for page in pages:
stream = _page_stream(page)
content_ids.append(add(f"<< /Length {len(stream)} >>\nstream\n{stream}\nendstream"))
page_ids.append(0)
kids = []
pages_id_placeholder = len(objects) + len(pages) + 1
for idx, _ in enumerate(pages):
pid = add(
f"<< /Type /Page /Parent {pages_id_placeholder} 0 R /MediaBox [0 0 595 842] "
f"/Resources << /Font << /F1 {font_id} 0 R >> >> /Contents {content_ids[idx]} 0 R >>"
)
page_ids[idx] = pid
kids.append(f"{pid} 0 R")
pages_id = add(f"<< /Type /Pages /Kids [{' '.join(kids)}] /Count {len(page_ids)} >>")
if pages_id != pages_id_placeholder:
for idx, pid in enumerate(page_ids):
objects[pid - 1] = (
f"<< /Type /Page /Parent {pages_id} 0 R /MediaBox [0 0 595 842] "
f"/Resources << /Font << /F1 {font_id} 0 R >> >> /Contents {content_ids[idx]} 0 R >>"
).encode("latin-1", errors="replace")
catalog_id = add(f"<< /Type /Catalog /Pages {pages_id} 0 R >>")
out = bytearray(b"%PDF-1.4\n")
offsets = [0]
for idx, obj in enumerate(objects, start=1):
offsets.append(len(out))
out.extend(f"{idx} 0 obj\n".encode("ascii"))
out.extend(obj)
out.extend(b"\nendobj\n")
xref = len(out)
out.extend(f"xref\n0 {len(objects) + 1}\n".encode("ascii"))
out.extend(b"0000000000 65535 f \n")
for offset in offsets[1:]:
out.extend(f"{offset:010d} 00000 n \n".encode("ascii"))
out.extend(
f"trailer\n<< /Size {len(objects) + 1} /Root {catalog_id} 0 R >>\n"
f"startxref\n{xref}\n%%EOF\n".encode("ascii")
)
path.write_bytes(out)
def _page_stream(lines: list[str]) -> str:
chunks = ["BT", "/F1 11 Tf", "50 800 Td"]
y = 800
first = True
for line in lines:
for wrapped in textwrap.wrap(_ascii(line), width=88) or [""]:
overflow = _emit_page_chunk(chunks, wrapped, y, first)
if overflow:
return overflow
if not first:
y -= 16
first = False
chunks.append("ET")
return "\n".join(chunks)
def _emit_page_chunk(chunks: list[str], wrapped: str, y: int, first: bool) -> str:
if not first:
chunks.append("0 -16 Td")
if y - 16 < 60:
chunks.append("ET")
return "\n".join(chunks)
chunks.append(f"({_escape_pdf(wrapped)}) Tj")
return ""
def _fit_page(lines: list[str], max_lines: int = 44) -> list[str]:
fitted: list[str] = []
for line in lines:
fitted.extend(textwrap.wrap(_ascii(line), width=88) or [""])
if len(fitted) >= max_lines:
return fitted[:max_lines]
return fitted
def _escape_pdf(t: str) -> str:
return t.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
def _ascii(t: str) -> str:
return t.encode("latin-1", errors="replace").decode("latin-1")
def _safe_name(name: str) -> str:
return re.sub(r"[^A-Za-z0-9_.-]+", "_", name).strip("_") or "stem_audit"