Spaces:
Sleeping
Sleeping
| 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 | |
| from .render_html_components import REQ_LABELS as _REQ_LABELS | |
| 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", | |
| } | |
| _STATUS_LABELS: dict[str, str] = { | |
| "signal_only": "⚠ Signal only", | |
| "partially_aligned": "~ Partially aligned", | |
| "aligned": "✓ Aligned", | |
| "not_detected": "✗ Not detected", | |
| } | |
| 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." | |
| def _score_boundary_short_line() -> str: | |
| return "Score reflects calculation integrity, not calibrated validity. Triage signal only." | |
| def _score_boundary_lines() -> list[str]: | |
| return [ | |
| "**What is verified:** calculation integrity. The same input produces the same score.", | |
| "**What is not verified:** calibrated measurement validity. Weights and detector scope remain bounded.", | |
| "**Use this score as a triage signal, not as certification, safety proof, or deployment approval.**", | |
| ] | |
| def _compact_labels(items: list[str], *, limit: int = 5) -> str: | |
| values = [str(item).strip() for item in items if str(item).strip()] | |
| if not values: | |
| return "" | |
| shown = values[:limit] | |
| extra = max(0, len(values) - len(shown)) | |
| text = ", ".join(shown) | |
| if extra: | |
| text += f" (+{extra} more)" | |
| return text | |
| # ── 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 []), | |
| *_classification_applied_md(result), | |
| 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']}", | |
| *_tier_lock_label_md(result), | |
| f"**About This Score:** {_score_boundary_short_line()}", | |
| *_score_boundary_lines(), | |
| "", | |
| "## 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']}{_s3_formula(result)} |", | |
| 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 _classification_applied_md(result: dict[str, Any]) -> list[str]: | |
| cls = result.get("classification", {}) | |
| ca = cls.get("ca_severity", "none") | |
| cap = cls.get("score_cap") | |
| t0 = cls.get("t0_hard_floor", False) | |
| cap_str = str(cap) if cap is not None else "none" | |
| t0_str = "active" if t0 else "clear" | |
| return [f"**Classification Applied:** ca_severity={ca} | score_cap={cap_str} | t0_floor={t0_str}"] | |
| def _tier_lock_label_md(result: dict[str, Any]) -> list[str]: | |
| cls = result.get("classification", {}) | |
| score_cap = cls.get("score_cap") | |
| if score_cap is None: | |
| return [] | |
| if cls.get("t0_hard_floor"): | |
| return [ | |
| f"**Tier Lock [T0-FLOOR]:** Score ceiling active at **39** (T0 maximum). " | |
| f"CA-DIRECT classification with insufficient code presence. " | |
| f"Resolving this condition is required before any tier advancement." | |
| ] | |
| return [ | |
| f"**Tier Lock [CA-CAP]:** Score ceiling active at **{score_cap}** (T2 maximum). " | |
| f"Clinical-adjacent surface detected without explicit non-clinical boundary. " | |
| f"Adding a non-diagnostic disclaimer resolves this lock." | |
| ] | |
| def _s3_formula(result: dict[str, Any]) -> str: | |
| s3_raw = result.get("stage_3_rubric", {}).get("stage_3_raw_total", {}) | |
| raw = s3_raw.get("score") | |
| max_val = s3_raw.get("max") | |
| if raw is None or not max_val: | |
| return "" | |
| normalized = result.get("score", {}).get("stage_3_code_bio") | |
| if isinstance(normalized, (int, float)): | |
| return f" (raw: {raw}/{max_val} -> normalized: {int(round(normalized))})" | |
| return f" (raw: {raw}/{max_val})" | |
| 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')}; >=0.80=consistent, >=0.55=mixed, <0.55=divergent), " | |
| f"uncertainty band `{uncertainty.get('status', 'unknown')}` " | |
| f"({uncertainty.get('uncertainty', 'n/a')}; <0.20=low-spread, <=0.45=review-advised, >0.45=manual-review), " | |
| 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 ( | |
| f"mirror-only in {__version__} — 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: | |
| req_id = item["requirement_id"] | |
| label = _REQ_LABELS.get(req_id, req_id) | |
| status = _STATUS_LABELS.get(item["status"], item["status"]) | |
| src_tag = "" | |
| if item.get("source_ids"): | |
| src_tag = " `[" + ", ".join(item["source_ids"]) + "]`" | |
| lines.append(f"- **{label}**{src_tag} — {status}") | |
| refs = item.get("finding_refs", []) | |
| if refs: | |
| lines.append(f" - Triggered by: {', '.join(f'`{r}`' for r in refs)}") | |
| gaps = item.get("not_assessed", []) | |
| if gaps: | |
| lines.append(f" - Not assessed: {'; '.join(gaps)}") | |
| lines.append(f" - {item['note']}") | |
| lines.append("") | |
| summary = result.get("regulatory_traceability", {}).get("summary") | |
| if summary: | |
| lines.append(f"**Traceability 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[:5]: | |
| 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) | |
| summary = " ; ".join(snippets) | |
| extra = max(0, len(details) - len(snippets)) | |
| if extra: | |
| summary += f" ; (+{extra} more mapping details)" | |
| return summary | |
| 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 {_compact_labels(secondary)}" | |
| 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: {_compact_labels(covered_by)}" | |
| 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]: | |
| rows: list[tuple[str, str, str]] = [] | |
| stage2 = result.get("stage_2r_rubric", {}) | |
| code_integrity = result.get("code_integrity", {}) | |
| if "R2R_D2_missing_clinical_use_boundary" in stage2: | |
| rows.append(( | |
| "R2R_D2 missing clinical boundary", | |
| "Add non-clinical/non-diagnostic disclaimer to README and all adjacent docs", | |
| "+20 S2R (+4 final) | unlocks tier cap", | |
| )) | |
| if "R2R_D4_unsupported_workflow_claim" in stage2: | |
| rows.append(( | |
| "R2R_D4 unsupported workflow claim", | |
| "Align README workflow/demo/CLI claims with actual local support surfaces", | |
| "+15 S2R (+3 final)", | |
| )) | |
| if code_integrity.get("C2_dependency_pinning", {}).get("status") in {"WARN", "FAIL"}: | |
| rows.append(( | |
| "C2 dependency pinning WARN", | |
| "Pin production dependencies; document external-service dependence explicitly", | |
| "C2 -> PASS (no direct score delta)", | |
| )) | |
| if code_integrity.get("C5_compliance_boundary_integrity", {}).get("status") in {"WARN", "FAIL"}: | |
| rows.append(( | |
| "C5 compliance boundary WARN", | |
| "Remove unsupported legal/compliance language or add backing governance evidence", | |
| "C5 -> PASS (no direct score delta)", | |
| )) | |
| if code_integrity.get("C6_mock_auth_or_fail_open_boundary", {}).get("status") in {"WARN", "FAIL"}: | |
| rows.append(( | |
| "C6 mock-auth boundary WARN", | |
| "Separate mock-auth/auto-login flows from production trust boundary narrative", | |
| "C6 -> PASS (no direct score delta)", | |
| )) | |
| if not rows: | |
| return [] | |
| lines = [ | |
| "", "## Remediation Roadmap", | |
| "", | |
| "| Finding | Action | Expected Impact |", | |
| "| --- | --- | --- |", | |
| ] | |
| for finding, action, impact in rows: | |
| lines.append(f"| {finding} | {action} | {impact} |") | |
| 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}") | |
| for item in items: | |
| req_id = item["requirement_id"] | |
| label = _REQ_LABELS.get(req_id, req_id) | |
| status = _STATUS_LABELS.get(item["status"], item["status"]) | |
| src = ", ".join(item.get("source_ids", [])) | |
| lines.append(f" {label}: {status}") | |
| if src: | |
| lines.append(f" source: {src}") | |
| refs = item.get("finding_refs", []) | |
| if refs: | |
| lines.append(f" triggered by: {', '.join(refs)}") | |
| gaps = item.get("not_assessed", []) | |
| if gaps: | |
| lines.append(f" not assessed: {'; '.join(gaps)}") | |
| 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) | |
| def _draw_footer(canvas: Any, _: Any) -> None: | |
| canvas.saveState() | |
| canvas.setStrokeColor(_hx(_MGRAY)) | |
| canvas.setLineWidth(0.5) | |
| canvas.line(doc.leftMargin, 13.2 * mm, A4[0] - doc.rightMargin, 13.2 * mm) | |
| canvas.setFont("Helvetica", 7) | |
| canvas.setFillColor(_hx(_DGRAY)) | |
| canvas.drawCentredString( | |
| A4[0] / 2, | |
| 8.2 * mm, | |
| f"STEM BIO-AI Local CLI Scan | {result.get('stem_ai_version', __version__)} | Deterministic surface scan — no LLM, network, or runtime execution.", | |
| ) | |
| canvas.drawCentredString( | |
| A4[0] / 2, | |
| 4.9 * mm, | |
| "Not clinical certification. Not regulatory clearance. Not medical advice.", | |
| ) | |
| canvas.restoreState() | |
| doc.build(story, onFirstPage=_draw_footer, onLaterPages=_draw_footer) | |
| # ── 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, 3 * mm)) | |
| 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'<font color="{_WHITE}"><b>STEM BIO-AI Evidence-Surface Scan v{result["stem_ai_version"]}</b></font>', | |
| _style("H1", 10.5, 13, _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), 4), | |
| ("BOTTOMPADDING", (0, 0), (-1, -1), 4), | |
| ("LEFTPADDING", (0, 0), (-1, -1), 8), | |
| ])) | |
| meta = ( | |
| f'<font color="{_DGRAY}"><b>Repository:</b> {_xt(t["name"])} | ' | |
| f'<b>Commit:</b> {commit} | ' | |
| f'<b>Branch:</b> {_xt(branch)}</font><br/>' | |
| f'<font color="{_DGRAY}"><b>Audit Date:</b> {audit_date} | ' | |
| f'<b>Mode:</b> {mode} | ' | |
| f'<b>Policy:</b> {_xt(profile_label)}</font>' | |
| ) | |
| meta_data = [[Paragraph(meta, _style("M1", 10.2, 13.8, _DGRAY))]] | |
| meta_tbl = Table(meta_data, colWidths=["100%"]) | |
| meta_tbl.setStyle(TableStyle([ | |
| ("BACKGROUND", (0, 0), (-1, -1), _hx(_MGRAY)), | |
| ("TOPPADDING", (0, 0), (-1, -1), 6), | |
| ("BOTTOMPADDING", (0, 0), (-1, -1), 6), | |
| ("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'<font color="{_NAVY}" size="22"><b>{fs}</b></font>' | |
| f'<font color="{_DGRAY}" size="11"> / 100</font>', | |
| _style("SC1", 22, 27, _NAVY, True, "CENTER"), | |
| ), | |
| Paragraph("Final Score", _style("SL1", 8, 11, _DGRAY, False, "CENTER")), | |
| ] | |
| tier_badge = [[Paragraph( | |
| f'<font color="{_WHITE}"><b>{_xt(tier)}</b></font>', | |
| _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'<font color="{_DGRAY}"><b>Use Scope:</b></font><br/>' | |
| f'<font color="{_DGRAY}" size="8">{_xt(use_scope)}</font>', | |
| _style("US1", 8, 11, _DGRAY), | |
| ), | |
| ] | |
| row_tbl = Table([[score_cell, scope_cell]], colWidths=[44 * 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)), | |
| ])) | |
| score_note_lines = "<br/>".join([ | |
| f"<b>⚠ About this score</b><br/>{_xt(_score_boundary_short_line())}", | |
| '• <b>What is verified:</b> calculation integrity. The same input produces the same score.', | |
| '• <b>What is not verified:</b> calibrated measurement validity. Weights and detector scope remain bounded.', | |
| '• <b>Use this score as a triage signal:</b> not as certification, safety proof, or deployment approval.', | |
| ]) | |
| score_note_tbl = Table( | |
| [[Paragraph(f'<font color="{_DGRAY}" size="7.5">{score_note_lines}</font>', _style("SN1", 7.5, 10, _DGRAY))]], | |
| colWidths=["100%"], | |
| ) | |
| score_note_tbl.setStyle(TableStyle([ | |
| ("BACKGROUND", (0, 0), (-1, -1), _hx("#FFF4D6")), | |
| ("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.7, _hx(_AMBER)), | |
| ])) | |
| return [row_tbl, Spacer(1, 1 * mm), score_note_tbl] | |
| 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'<font color="{_WHITE}"><b>{label}</b><br/><i>{sub}</i></font>', | |
| _style(f"CH_{label}", 8.5, 12, _WHITE, True, "CENTER"), | |
| )], | |
| [Paragraph( | |
| f'<font color="{col}" size="22"><b>{val}</b></font>' | |
| f'<font color="{_DGRAY}" size="9"> / 100</font>', | |
| _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]: | |
| score = result["score"] | |
| cls = result.get("classification", {}) | |
| risks = result.get("notable_risks", []) | |
| positive = result.get("notable_positive_evidence", []) | |
| airi = result.get("airi_risk_coverage", {}) | |
| stage_3 = result.get("stage_3_rubric", {}) | |
| stage_4 = result.get("stage_4_rubric", {}) | |
| if cls.get("t0_hard_floor"): | |
| posture = ( | |
| "Bio-governance posture is not suitable for clinical or patient-adjacent trust. " | |
| "Direct-clinical framing appears without an adequate boundary declaration." | |
| ) | |
| posture_color = _RED | |
| elif cls.get("score_cap") is not None: | |
| posture = ( | |
| "Bio-governance posture is partial and bounded. Some provenance and replication signals exist, " | |
| "but explicit non-clinical boundary and trust-control evidence remain insufficient." | |
| ) | |
| posture_color = _AMBER | |
| else: | |
| posture = ( | |
| "Bio-governance posture is structurally stronger in the reviewed repository surfaces, " | |
| "but this artifact still remains a pre-screen rather than certification." | |
| ) | |
| posture_color = _GREEN | |
| present_lines: list[str] = [] | |
| if stage_3.get("B1_data_provenance_controls", {}).get("score", 0) > 0: | |
| present_lines.append("Repository provenance surfaces are present through dependency or lock manifests.") | |
| if stage_3.get("B3_coi_funding", {}).get("score", 0) > 0: | |
| present_lines.append("Funding / COI acknowledgement language is present.") | |
| if result.get("stage_1_rubric", {}).get("S1_domain_package", {}).get("score", 0) > 0: | |
| present_lines.append("Package metadata was available for repo-local consistency checks.") | |
| if result.get("replication_score", 0) > 0: | |
| present_lines.append("Some reproducibility evidence exists, including environment lock or container surfaces.") | |
| if result.get("stage_traceability", {}).get("stage_4"): | |
| present_lines.append("Regulatory traceability mappings exist for reproducibility and record-keeping scaffolding.") | |
| for item in positive[:3]: | |
| if item not in present_lines: | |
| present_lines.append(str(item)) | |
| if not present_lines: | |
| present_lines.append("Only limited positive governance evidence surfaced in the reviewed repository sources.") | |
| missing_lines = [str(r) for r in risks[:5]] or ["No major missing-or-contradicted governance surfaces were surfaced."] | |
| reg_lines = _regulatory_bullets(result) | |
| airi_lines = "" | |
| if airi: | |
| airi_lines += ( | |
| f'• <b>Secondary risk vocabulary:</b> {airi.get("covered_count", 0)} mapped triggers across ' | |
| f'{airi.get("total_risks_in_detector_scope", 0)} in-bundle AIRI risks.<br/>' | |
| f'• <b>Meaning:</b> broadens local findings into risk language; it does not prove harm, safety, or compliance.<br/>' | |
| f'• <b>Why this matters:</b> useful when a user needs a broader risk vocabulary around governance gaps, not when they need deployment approval.' | |
| ) | |
| covered = airi.get("covered_risks", []) | |
| for idx, risk in enumerate(covered[:2], start=1): | |
| reason = _airi_reason_summary(risk) | |
| primary = _airi_primary_summary(risk) | |
| detail = f'{_xt(str(risk.get("id", "—")))} {_xt(_clip_words(str(risk.get("title", "")), 48))}' | |
| if primary: | |
| detail += f' | {_xt(primary)}' | |
| if reason: | |
| detail += f' | why: {_xt(_clip_words(reason, 92))}' | |
| airi_lines += f'<br/>• <b>Mapped theme {idx}:</b> {detail}' | |
| gaps = airi.get("known_gaps_in_bundle", []) | |
| if gaps: | |
| gap_preview = "; ".join( | |
| f"{g.get('id', '—')} {_xt(_clip_words(str(g.get('title', '')), 22))}" | |
| for g in gaps[:3] | |
| ) | |
| airi_lines += f'<br/>• <b>Still unmapped here:</b> {gap_preview}' | |
| def _summary_block(title: str, body: str, head_color: str, key: str) -> Table: | |
| tbl = Table([ | |
| [Paragraph(f'<font color="{_WHITE}"><b>{_xt(title)}</b></font>', _style(f"{key}_H", 9, 12, _WHITE, True))], | |
| [Paragraph(f'<font color="{_DGRAY}" size="8">{body}</font>', _style(f"{key}_B", 8, 11, _DGRAY))], | |
| ], colWidths=["100%"]) | |
| tbl.setStyle(TableStyle([ | |
| ("BACKGROUND", (0, 0), (0, 0), _hx(head_color)), | |
| ("BACKGROUND", (0, 1), (0, 1), _hx(_LGRAY)), | |
| ("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 | |
| left_stack: list[Any] = [ | |
| _summary_block( | |
| "What Is Actually Present", | |
| "".join(f'• {_xt(_clip_words(line, 170))}<br/>' for line in present_lines[:6]), | |
| _GREEN, | |
| "WIP", | |
| ), | |
| Spacer(1, 2 * mm), | |
| _summary_block( | |
| "What Is Missing Or Contradicted", | |
| "".join(f'• {_xt(_clip_words(line, 170))}<br/>' for line in missing_lines), | |
| _RED, | |
| "WIM", | |
| ), | |
| ] | |
| 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"), | |
| ])) | |
| right_stack: list[Any] = [ | |
| _summary_block( | |
| "Governance Posture", | |
| f'<b>{_xt(score["formal_tier"])}</b> — {_xt(score.get("use_scope", ""))}<br/>{_xt(posture)}', | |
| posture_color, | |
| "GOV", | |
| ), | |
| ] | |
| if reg_lines: | |
| right_stack += [Spacer(1, 2 * mm), _summary_block("Regulatory Traceability", reg_lines, _NAVY, "REG")] | |
| if airi_lines: | |
| right_stack += [Spacer(1, 2 * mm), _summary_block("AIRI Risk Triggers", airi_lines, _TEAL, "AIRI")] | |
| right_tbl = Table([[item] for item in right_stack], colWidths=["100%"]) | |
| right_tbl.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'<font color="{_WHITE}"><b>Bio Deterministic Diagnostics</b></font>', _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'<font color="{_DGRAY}" size="7.5"><b>{_xt(label)}</b><br/>{_xt(", ".join(status_parts) if status_parts else "no findings")}</font>', | |
| _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_pdf_text_lines(result: dict[str, Any]) -> list[str]: | |
| """Plain-text regulatory traceability page for the simple PDF fallback.""" | |
| basis = result.get("regulatory_basis", {}) | |
| traceability = result.get("stage_traceability", {}) | |
| if not basis and not traceability: | |
| return [] | |
| note = basis.get("note", {}) | |
| lines = ["Regulatory Traceability", ""] | |
| if note.get("body_line_1"): | |
| lines.append(note["body_line_1"]) | |
| if note.get("body_line_2"): | |
| lines.append(note["body_line_2"]) | |
| lines.append("") | |
| _stage_labels = { | |
| "stage_1": "Stage 1 - README Intent", | |
| "stage_2r": "Stage 2R - Repo Consistency", | |
| "stage_3": "Stage 3 - Code / Bio", | |
| "stage_4": "Stage 4 - Replication", | |
| "bio_diagnostics": "Bio Diagnostics", | |
| } | |
| 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(_stage_labels.get(stage_key, stage_key)) | |
| for item in items: | |
| label = _REQ_LABELS.get(item["requirement_id"], item["requirement_id"]) | |
| status = _STATUS_PDF_LABEL.get(item.get("status", ""), item.get("status", "")) | |
| lines.append(f"- {label}: {status}") | |
| refs = ", ".join(item.get("finding_refs", [])) | |
| if refs: | |
| lines.append(f" triggered by: {refs}") | |
| summary = result.get("regulatory_traceability", {}).get("summary") | |
| if summary: | |
| lines.append("") | |
| lines.append(f"Summary: {summary}") | |
| return lines | |
| def _regulatory_bullets(result: dict[str, Any]) -> str: | |
| """Compact actionable bullet summary of regulatory traceability for page 1. | |
| Surfaces what actually has structural alignment (partially_aligned) versus | |
| weak signal-only references, instead of restating the boilerplate basis note. | |
| """ | |
| trace = result.get("stage_traceability", {}) | |
| items: list[dict[str, Any]] = [] | |
| seen: set[str] = set() | |
| for stage_key in ("stage_1", "stage_2r", "stage_3", "stage_4", "bio_diagnostics"): | |
| for it in trace.get(stage_key, []): | |
| rid = it.get("requirement_id", "") | |
| if rid in seen: | |
| continue | |
| seen.add(rid) | |
| items.append(it) | |
| if not items: | |
| return "" | |
| framework_order = ["EU AI Act", "ICH M15", "IMDRF", "FDA"] | |
| frameworks: set[str] = set() | |
| for it in items: | |
| for sid in it.get("source_ids", []): | |
| if sid.startswith("eu_ai_act"): | |
| frameworks.add("EU AI Act") | |
| elif sid.startswith("ich_m15"): | |
| frameworks.add("ICH M15") | |
| elif sid.startswith("imdrf"): | |
| frameworks.add("IMDRF") | |
| elif sid.startswith("fda"): | |
| frameworks.add("FDA") | |
| fw_str = ", ".join(f for f in framework_order if f in frameworks) | |
| partial = [it for it in items if it.get("status") == "partially_aligned"] | |
| signal = [it for it in items if it.get("status") == "signal_only"] | |
| lines = [f'• <b>Frameworks touched:</b> {_xt(fw_str)}'] | |
| if partial: | |
| labels = "; ".join( | |
| _REQ_LABELS.get(it["requirement_id"], it["requirement_id"]) | |
| for it in partial[:3] | |
| ) | |
| lines.append(f'• <b>Structural alignment exists ({len(partial)}):</b> {_xt(labels)}') | |
| if signal: | |
| lines.append( | |
| f'• <b>Signal-only references ({len(signal)}):</b> useful for pre-audit traceability, but not strong enough to count as compliance proof' | |
| ) | |
| lines.append('• <b>Meaning:</b> repository evidence maps to governance frameworks, but the report does not establish compliance or clearance.') | |
| lines.append('• <b>Why this matters:</b> this section helps answer whether governance scaffolding is present at all before a formal audit, not whether the repository is approved for use.') | |
| return "<br/>".join(lines) | |
| 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'<font color="{_DGRAY}" size="7.5"><b>{_xt(note.get("title", "Regulatory basis note"))}</b></font>', | |
| f'<font color="{_DGRAY}" size="7.5">{_xt(note.get("body_line_1", ""))}</font>', | |
| f'<font color="{_DGRAY}" size="7.5">{_xt(note.get("body_line_2", ""))}</font>', | |
| ] | |
| if basis.get("review_required"): | |
| body_lines.append( | |
| f'<font color="{_AMBER}" size="7.2"><b>Review required:</b> {_xt(", ".join(basis.get("review_reasons", [])))}</font>' | |
| ) | |
| if summary: | |
| body_lines.append( | |
| f'<font color="{_DGRAY}" size="7.2"><b>Traceability summary:</b> {_xt(_clip_words(summary, 220))}</font>' | |
| ) | |
| panel = Table( | |
| [[Paragraph("<br/>".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] | |
| _STATUS_PDF_COLOR = { | |
| "signal_only": _AMBER, | |
| "partially_aligned": _TEAL, | |
| "aligned": _GREEN, | |
| "not_detected": _DGRAY, | |
| } | |
| _STATUS_PDF_LABEL = { | |
| "signal_only": "Signal only", | |
| "partially_aligned": "Partially aligned", | |
| "aligned": "Aligned", | |
| "not_detected": "Not detected", | |
| } | |
| def _regulatory_traceability_pdf(result: dict[str, Any]) -> list[Any]: | |
| """Detailed per-requirement regulatory traceability block for PDF pages.""" | |
| basis = result.get("regulatory_basis", {}) | |
| traceability = result.get("stage_traceability", {}) | |
| if not basis and not traceability: | |
| return [] | |
| note = basis.get("note", {}) | |
| story: list[Any] = [] | |
| story += _chapter_hdr("Chapter 3 — Regulatory Traceability", _NAVY) | |
| story.append(Paragraph( | |
| f'<font color="{_DGRAY}" size="8">' | |
| f'<b>{_xt(note.get("title", "Regulatory basis note"))}</b><br/>' | |
| f'{_xt(note.get("body_line_1", ""))}<br/>' | |
| f'<i>{_xt(note.get("body_line_2", ""))}</i>' | |
| f'</font>', | |
| _style("REGT_BASIS", 8, 11, _DGRAY), | |
| )) | |
| story.append(Spacer(1, 3 * mm)) | |
| _STAGE_LABELS_PDF = { | |
| "stage_1": "Stage 1 — README Intent", | |
| "stage_2r": "Stage 2R — Repo Consistency", | |
| "stage_3": "Stage 3 — Code / Bio", | |
| "stage_4": "Stage 4 — Replication", | |
| "bio_diagnostics": "Bio Diagnostics", | |
| } | |
| _label_col_w = _PDF_CONTENT_WIDTH - 38 * mm | |
| for stage_key in ("stage_1", "stage_2r", "stage_3", "stage_4", "bio_diagnostics"): | |
| items = traceability.get(stage_key, []) | |
| if not items: | |
| continue | |
| story.append(Paragraph( | |
| f'<font color="{_NAVY}" size="8.5"><b>{_xt(_STAGE_LABELS_PDF.get(stage_key, stage_key))}</b></font>', | |
| _style(f"REGT_S_{stage_key[:6]}", 8.5, 11, _NAVY, True), | |
| )) | |
| story.append(Spacer(1, 1 * mm)) | |
| for item in items: | |
| req_id = item["requirement_id"] | |
| label = _REQ_LABELS.get(req_id, req_id) | |
| status = item.get("status", "") | |
| status_label = _STATUS_PDF_LABEL.get(status, status) | |
| status_color = _STATUS_PDF_COLOR.get(status, _DGRAY) | |
| refs = ", ".join(item.get("finding_refs", [])) | |
| gaps = "; ".join(item.get("not_assessed", [])) | |
| detail_lines = [f'<font color="{_DGRAY}" size="7.5">{_xt(item["note"])}</font>'] | |
| if refs: | |
| detail_lines.append( | |
| f'<font color="{_DGRAY}" size="7.5">Triggered by: <i>{_xt(refs)}</i></font>' | |
| ) | |
| if gaps: | |
| detail_lines.append( | |
| f'<font color="{_DGRAY}" size="7.5">Not assessed: {_xt(gaps)}</font>' | |
| ) | |
| row = Table( | |
| [[ | |
| Paragraph( | |
| f'<font color="{_NAVY}" size="8"><b>{_xt(label)}</b></font><br/>' | |
| + "<br/>".join(detail_lines), | |
| _style(f"REGT_B_{req_id[:10]}", 8, 11, _DGRAY), | |
| ), | |
| Paragraph( | |
| f'<font color="{status_color}" size="7.5"><b>{_xt(status_label)}</b></font>', | |
| _style(f"REGT_V_{req_id[:10]}", 7.5, 10, status_color, True, "RIGHT"), | |
| ), | |
| ]], | |
| colWidths=[_label_col_w, 38 * mm], | |
| ) | |
| row.setStyle(TableStyle([ | |
| ("BACKGROUND", (0, 0), (-1, -1), _hx(_LGRAY)), | |
| ("BOX", (0, 0), (-1, -1), 0.4, _hx(_MGRAY)), | |
| ("TOPPADDING", (0, 0), (-1, -1), 4), | |
| ("BOTTOMPADDING", (0, 0), (-1, -1), 4), | |
| ("LEFTPADDING", (0, 0), (-1, -1), 6), | |
| ("RIGHTPADDING", (0, 0), (-1, -1), 5), | |
| ("VALIGN", (0, 0), (-1, -1), "TOP"), | |
| ])) | |
| story.append(row) | |
| story.append(Spacer(1, 1.5 * mm)) | |
| story.append(Spacer(1, 2 * mm)) | |
| summary = result.get("regulatory_traceability", {}).get("summary", "") | |
| if summary: | |
| story.append(Paragraph( | |
| f'<font color="{_DGRAY}" size="7.5"><i>{_xt(summary)}</i></font>', | |
| _style("REGT_SUMM", 7.5, 10, _DGRAY), | |
| )) | |
| return story | |
| def _footer_block() -> list[Any]: | |
| return [Spacer(1, 2 * mm)] | |
| 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 += _page_regulatory_traceability(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'<font color="{_WHITE}"><b>{title}</b></font>', | |
| _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 _chapter_hdr(title: str, color: str = _NAVY) -> list[Any]: | |
| return [ | |
| Paragraph( | |
| f'<font color="{color}" size="14"><b>{title}</b></font>', | |
| _style(f"CH_{title[:8]}", 14, 17, color, True), | |
| ), | |
| Spacer(1, 3.5 * mm), | |
| ] | |
| def _subsec_hdr(title: str, color: str = _NAVY) -> list[Any]: | |
| return [ | |
| Paragraph( | |
| f'<font color="{color}" size="10.5"><b>{title}</b></font>', | |
| _style(f"SUB_{title[:8]}", 10.5, 13, color, True), | |
| ), | |
| Spacer(1, 1.8 * mm), | |
| ] | |
| def _mini_score(label: str, val: int, max_val: int, col: str) -> Table: | |
| d = [ | |
| [Paragraph( | |
| f'<font color="{col}" size="20"><b>{val}</b></font>' | |
| f'<font color="{_DGRAY}" size="9"> / {max_val}</font>', | |
| _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'<font color="{_WHITE}"><b>Check</b></font>', _style(f"{id_prefix}H1", 8, 10, _WHITE, True)), | |
| Paragraph(f'<font color="{_WHITE}"><b>Points</b></font>', _style(f"{id_prefix}H2", 8, 10, _WHITE, True, "CENTER")), | |
| Paragraph(f'<font color="{_WHITE}"><b>Evidence / Finding</b></font>', _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'<b>{_xt(name)}</b>', _style(f"{uid}N", 8, 11, _DGRAY, True)), | |
| Paragraph( | |
| f'<font color="{col}"><b>{_xt(score_str)}</b></font>', | |
| _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", {}) | |
| story += _chapter_hdr("Chapter 1 — Stage Scorecard and Governance Scoring", _NAVY) | |
| # ── 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."), | |
| ] | |
| 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) | |
| # 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: <font color="{ca_col}"><b>{"YES" if ca else "NO"}</b></font>' | |
| f' ({_xt(cls["ca_severity"])}) ' | |
| f'• Explicit Disclaimer: <font color="{disc_col}"><b>{"PRESENT" if has_disc else "ABSENT"}</b></font>' | |
| f' ' | |
| f'• T0 Hard Floor: <font color="{t0_col}"><b>{"TRIGGERED" if cls["t0_hard_floor"] else "Clear"}</b></font>' | |
| ) | |
| info_tbl = Table([[Paragraph(info_text, _style("INF1", 8, 11, _DGRAY, False, "CENTER"))]], 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), | |
| ("RIGHTPADDING", (0, 0), (-1, -1), 8), | |
| ("ALIGN", (0, 0), (-1, -1), "CENTER"), | |
| ("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", "")) | |
| _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) | |
| verdict_text = ( | |
| f'• <b>Consistency Verdict:</b> <font color="{verdict_col}"><b>{_xt(verdict)}</b></font>' | |
| ) | |
| verdict_tbl = Table([[Paragraph(verdict_text, _style("S2RVERDICT", 8, 11, _DGRAY, False, "CENTER"))]], colWidths=["100%"]) | |
| verdict_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), | |
| ("RIGHTPADDING", (0, 0), (-1, -1), 8), | |
| ("ALIGN", (0, 0), (-1, -1), "CENTER"), | |
| ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), | |
| ])) | |
| story.append(verdict_tbl) | |
| 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'<font color="{_SLATE}"><b>Engineering Accountability (T-series)</b></font>', | |
| _style("TS1", 8.5, 12, _SLATE, True), | |
| ), | |
| Spacer(1, 1 * mm), | |
| _rubric_rows(t_items, "T"), | |
| Spacer(1, 3 * mm), | |
| Paragraph( | |
| f'<font color="{_SLATE}"><b>Biological Integrity (B-series)</b></font>', | |
| _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) | |
| raw_entry = rubric.get("stage_3_raw_total", {}) | |
| raw_score = raw_entry.get("score") | |
| raw_max = raw_entry.get("max") | |
| if raw_score is not None and raw_max: | |
| 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"]) | |
| story.append(Spacer(1, 2 * mm)) | |
| story.append(Paragraph( | |
| f'<font color="{_DGRAY}" size="8"><b>Normalized score:</b> ' | |
| f'T-series {t_total}/45 + B-series {b_total}/35 = raw {raw_score}/{raw_max} → {s3}/100.</font>', | |
| _style("S3NORM", 8, 11, _DGRAY), | |
| )) | |
| # 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_rows = [ | |
| ("T-series vs B-series", f"T-series (engineering) attained: {t_total} / 45 | B-series (bio integrity) attained: {b_total} / 35"), | |
| ("Local CLI scan maximum", f"{local_max} / 100 (T1+T2+T3 max 15 each; B1 max 10; B2/B3 require manual review)"), | |
| ("Gap to T3", f"{gap_t3} points needed across all stages to reach final score >= 70"), | |
| ("Gap to T4", f"{gap_t4} points needed across all stages to reach final score >= 85"), | |
| ("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"), | |
| ] | |
| gap_table_rows: list[list[Any]] = [[ | |
| Paragraph(f'<font color="{_WHITE}"><b>Stage 3 gap interpretation</b></font>', _style("S3GAPH", 8.5, 11, _WHITE, True)), | |
| "" | |
| ]] | |
| for label, detail in gap_rows: | |
| gap_table_rows.append([ | |
| Paragraph(f'<font color="{_SLATE}"><b>{_xt(label)}</b></font>', _style(f"S3GL_{label[:6]}", 8, 10, _SLATE, True)), | |
| Paragraph(f'<font color="{_DGRAY}" size="8">{_xt(detail)}</font>', _style(f"S3GD_{label[:6]}", 8, 11, _DGRAY)), | |
| ]) | |
| gap_table = Table(gap_table_rows, colWidths=[45 * mm, None]) | |
| gap_table.setStyle(TableStyle([ | |
| ("SPAN", (0, 0), (1, 0)), | |
| ("BACKGROUND", (0, 0), (1, 0), _hx(_SLATE)), | |
| ("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]), | |
| ("TOPPADDING", (0, 0), (-1, -1), 4), | |
| ("BOTTOMPADDING", (0, 0), (-1, -1), 4), | |
| ("LEFTPADDING", (0, 0), (-1, -1), 6), | |
| ("RIGHTPADDING", (0, 0), (-1, -1), 6), | |
| ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), | |
| ("GRID", (0, 1), (-1, -1), 0.3, _hx(_MGRAY)), | |
| ("VALIGN", (0, 0), (-1, -1), "TOP"), | |
| ])) | |
| story.append(gap_table) | |
| 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'<font color="{_DGRAY}" size="7.5"><i>Replication tier: {stage4_tier}. ' | |
| 'Stage 4 is reported separately and does not alter the formal score.</i></font>', | |
| _style("S4_NOTE", 7.5, 10, _DGRAY), | |
| )) | |
| story.append(Spacer(1, 3 * mm)) | |
| story += _sec_hdr("Formal Score Effect", _NAVY) | |
| final_box = Table([[ | |
| Paragraph( | |
| f'<font color="{_DGRAY}" size="8"><b>Final Score:</b> {score["final_score"]} / 100 ({_xt(score["formal_tier"])})<br/>' | |
| '<b>Formal effect:</b> Stage 4 does not raise or lower the formal repository score.<br/>' | |
| '• This page exists to show reproducibility and operational evidence posture separately from the governance score.<br/>' | |
| '• Stronger replication evidence improves trust in reproducibility posture, not the formal governance tier.</font>', | |
| _style("S4_SCOPE", 8, 11, _DGRAY), | |
| ) | |
| ]], colWidths=["100%"]) | |
| final_box.setStyle(TableStyle([ | |
| ("BACKGROUND", (0, 0), (-1, -1), _hx("#FFF7E3")), | |
| ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), | |
| ("TOPPADDING", (0, 0), (-1, -1), 6), | |
| ("BOTTOMPADDING", (0, 0), (-1, -1), 6), | |
| ("LEFTPADDING", (0, 0), (-1, -1), 7), | |
| ("RIGHTPADDING", (0, 0), (-1, -1), 7), | |
| ])) | |
| story.append(final_box) | |
| 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 += _chapter_hdr("Chapter 2 — 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'<font color="{col}"><b>[{s}] {_xt(short.get(key, key))}:</b></font>', | |
| _style(f"RG_H_{key[:4]}", 8.5, 12, col, True), | |
| )) | |
| story.append(Paragraph( | |
| f'<font color="{_DGRAY}" size="8">→ {_xt(guidance)}</font>', | |
| _style(f"RG_B_{key[:4]}", 8, 11, _DGRAY), | |
| )) | |
| story.append(Spacer(1, 2 * mm)) | |
| else: | |
| story.append(Paragraph( | |
| f'<font color="{_GREEN}">All code integrity checks PASSED. ' | |
| 'Continue monitoring with each major release.</font>', | |
| _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'<font color="{_NAVY}"><b>File Integrity (SHA-256)</b></font>', | |
| _style("FIH1", 9, 12, _NAVY, True), | |
| )) | |
| story.append(Spacer(1, 1 * mm)) | |
| hash_rows: list[list[Any]] = [[ | |
| Paragraph(f'<font color="{_WHITE}"><b>File</b></font>', _style("FHH1", 8, 10, _WHITE, True)), | |
| Paragraph(f'<font color="{_WHITE}"><b>SHA-256 Hash</b></font>', _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'<font size="6.5" color="{_DGRAY}">{h}</font>', | |
| _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'• <font color="{_DGRAY}" size="8">{_xt(risk)}</font>', | |
| _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'<font color="{_DGRAY}" size="8">' | |
| f'Covered Risks: <b>{airi.get("covered_count", 0)} / {airi.get("total_risks_in_detector_scope", 0)}</b> ' | |
| f'| Coverage Rate: <b>{airi.get("coverage_rate", 0):.3f}</b></font>', | |
| _style("AIRI_COMPACT", 8, 11, _DGRAY), | |
| )) | |
| story.append(Paragraph( | |
| f'<font color="{_DGRAY}" size="7">{_xt(_surface_compaction_note(result))}</font>', | |
| _style("AIRI_COMPACT_NOTE", 7, 10, _DGRAY), | |
| )) | |
| for risk in airi.get("covered_risks", [])[:3]: | |
| reason = _airi_reason_summary(risk) | |
| story.append(Paragraph( | |
| f'<font color="{_DGRAY}" size="8">• <b>{_xt(str(risk.get("id", "—")))}</b> ' | |
| f'{_xt(str(risk.get("title", "")))}' | |
| f'{f" — why: {_xt(reason)}" if reason else ""}</font>', | |
| _style(f"AIRC_{str(risk.get('id', 'risk'))[:8]}", 8, 10, _DGRAY), | |
| )) | |
| if airi.get("known_gaps_in_bundle"): | |
| _all_gaps = airi.get("known_gaps_in_bundle", []) | |
| gap_preview = ", ".join( | |
| f"{g.get('id', '—')} {_xt(str(g.get('title', '')))}" | |
| for g in _all_gaps[:5] | |
| ) | |
| _gap_extra = f" (+{len(_all_gaps) - 5} more)" if len(_all_gaps) > 5 else "" | |
| story.append(Paragraph( | |
| f'<font color="{_DGRAY}" size="8">Known gaps: {gap_preview}{_gap_extra}</font>', | |
| _style("AIRI_GAPC", 8, 10, _DGRAY), | |
| )) | |
| story.append(Spacer(1, 3 * mm)) | |
| story += _regulatory_basis_box(result) | |
| story.append(Spacer(1, 3 * mm)) | |
| story += _sec_hdr("Method Boundary", _DGRAY) | |
| story.append(Paragraph( | |
| f'<font color="{_DGRAY}" size="8"><b>Final score:</b> {score["final_score"]} / 100 ({_xt(score["formal_tier"])})' | |
| f'<br/><b>Method:</b> {_xt(result.get("method", ""))}</font>', | |
| _style("MBCOMPACT", 8, 11, _DGRAY), | |
| )) | |
| story += _footer_block() | |
| return _single_page_story(story, break_before=True) | |
| # ── Page 6: Remediation + AIRI + Method (7p only) ──────────────────────────── | |
| def _page6_method_airi(result: dict[str, Any]) -> list[Any]: | |
| story: list[Any] = [] | |
| risks = result.get("notable_risks", []) | |
| airi = result.get("airi_risk_coverage", {}) | |
| score = result.get("score", {}) | |
| stage2_score = int(score.get("stage_2_repo_local_consistency", 0) or 0) | |
| stage3_score = int(score.get("stage_3_code_bio", 0) or 0) | |
| story += _chapter_hdr("Chapter 4 — Remediation Actions, AIRI Risk Triggers & Method Boundary", _NAVY) | |
| story += _subsec_hdr("4.1 Decision Path", _SLATE) | |
| decision_rows = [ | |
| ("Stage 2R score", f"{stage2_score} / 100 — repo-local contradictions and missing trust boundaries are holding the report down."), | |
| ("Stage 3 score", f"{stage3_score} / 100 — accountability and bio-governance evidence remain partial rather than mature."), | |
| ] | |
| decision_table_rows: list[list[Any]] = [[ | |
| Paragraph(f'<font color="{_WHITE}"><b>Developer-facing decision path</b></font>', _style("DPDF_H", 8.5, 11, _WHITE, True)), | |
| "" | |
| ]] | |
| for label, detail in decision_rows: | |
| decision_table_rows.append([ | |
| Paragraph(f'<font color="{_SLATE}"><b>{_xt(label)}</b></font>', _style(f"DPDF_K_{label[:4]}", 8, 10, _SLATE, True)), | |
| Paragraph(f'<font color="{_DGRAY}" size="8">{_xt(detail)}</font>', _style(f"DPDF_V_{label[:4]}", 8, 11, _DGRAY)), | |
| ]) | |
| decision_table = Table(decision_table_rows, colWidths=[40 * mm, None]) | |
| decision_table.setStyle(TableStyle([ | |
| ("SPAN", (0, 0), (1, 0)), | |
| ("BACKGROUND", (0, 0), (1, 0), _hx(_SLATE)), | |
| ("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]), | |
| ("TOPPADDING", (0, 0), (-1, -1), 4), | |
| ("BOTTOMPADDING", (0, 0), (-1, -1), 4), | |
| ("LEFTPADDING", (0, 0), (-1, -1), 6), | |
| ("RIGHTPADDING", (0, 0), (-1, -1), 6), | |
| ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), | |
| ("GRID", (0, 1), (-1, -1), 0.3, _hx(_MGRAY)), | |
| ("VALIGN", (0, 0), (-1, -1), "TOP"), | |
| ])) | |
| story.append(decision_table) | |
| story.append(Spacer(1, 4 * mm)) | |
| story += _subsec_hdr("4.2 Top Remediation Actions", _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'<font color="{_GREEN}"><b>No critical risks detected by local CLI scan.</b></font><br/>' | |
| f'<font color="{_DGRAY}" size="8">A manual audit is still recommended for ' | |
| 'clinical-adjacent deployment. Local CLI cannot assess B2 (bias) or B3 (COI).</font>', | |
| _style("NR1", 8, 12, _DGRAY), | |
| )) | |
| else: | |
| remediation_rows: list[list[Any]] = [[ | |
| Paragraph(f'<font color="{_WHITE}"><b>Priority</b></font>', _style("REMD_H1", 8.5, 11, _WHITE, True)), | |
| Paragraph(f'<font color="{_WHITE}"><b>What to fix first</b></font>', _style("REMD_H2", 8.5, 11, _WHITE, True)), | |
| ]] | |
| for i, risk in enumerate(risks[:4], 1): | |
| guidance = _pri_detail.get(risk, ( | |
| "Review this finding and implement appropriate controls before supervised or clinical-adjacent deployment." | |
| )) | |
| remediation_rows.append([ | |
| Paragraph( | |
| f'<font color="{_RED if i == 1 else _AMBER}"><b>P{i}</b></font>', | |
| _style(f"REMD_P{i}", 8, 11, _DGRAY, True, "CENTER"), | |
| ), | |
| Paragraph( | |
| f'<b>{_xt(_clip_words(risk, 95))}</b><br/><font color="{_DGRAY}" size="7.5">{_xt(_clip_words(guidance, 145))}</font>', | |
| _style(f"REMD_V{i}", 7.8, 10, _DGRAY), | |
| ), | |
| ]) | |
| remediation_table = Table(remediation_rows, colWidths=[16 * mm, None]) | |
| remediation_table.setStyle(TableStyle([ | |
| ("BACKGROUND", (0, 0), (-1, 0), _hx(_RED)), | |
| ("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]), | |
| ("TOPPADDING", (0, 0), (-1, -1), 4), | |
| ("BOTTOMPADDING", (0, 0), (-1, -1), 4), | |
| ("LEFTPADDING", (0, 0), (-1, -1), 6), | |
| ("RIGHTPADDING", (0, 0), (-1, -1), 6), | |
| ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), | |
| ("GRID", (0, 0), (-1, -1), 0.3, _hx(_MGRAY)), | |
| ("VALIGN", (0, 0), (-1, -1), "TOP"), | |
| ])) | |
| story.append(remediation_table) | |
| # AIRI summary | |
| if airi: | |
| story.append(Spacer(1, 4 * mm)) | |
| story += _subsec_hdr("4.3 AIRI Risk Triggers Summary", _TEAL) | |
| airi_rows: list[list[Any]] = [[ | |
| Paragraph(f'<font color="{_WHITE}"><b>AIRI summary</b></font>', _style("AIRIPDF_H", 8.5, 11, _WHITE, True)), | |
| "" | |
| ]] | |
| airi_rows.append([ | |
| Paragraph(f'<font color="{_TEAL}"><b>Coverage</b></font>', _style("AIRIPDF_K0", 8, 10, _TEAL, True)), | |
| Paragraph( | |
| f'<font color="{_DGRAY}" size="8">Covered Risks: <b>{airi.get("covered_count", 0)} / {airi.get("total_risks_in_detector_scope", 0)}</b> ' | |
| f'| Coverage Rate: <b>{airi.get("coverage_rate", 0):.3f}</b> ' | |
| f'| Bundle Scope: <b>{_xt(str(airi.get("airi_bundle_scope", "unknown")))}</b></font>', | |
| _style("AIRIPDF_V0", 8, 11, _DGRAY), | |
| ), | |
| ]) | |
| airi_rows.append([ | |
| Paragraph(f'<font color="{_TEAL}"><b>Meaning</b></font>', _style("AIRIPDF_KM0", 8, 10, _TEAL, True)), | |
| Paragraph( | |
| f'<font color="{_DGRAY}" size="8">AIRI broadens local governance findings into a secondary risk vocabulary. ' | |
| f'It does not prove harm, compliance, or deployment readiness.</font>', | |
| _style("AIRIPDF_VM0", 8, 11, _DGRAY), | |
| ), | |
| ]) | |
| covered_risks = airi.get("covered_risks", []) | |
| for idx, risk in enumerate(covered_risks[:2], start=1): | |
| reason = _airi_reason_summary(risk) | |
| primary = _airi_primary_summary(risk) | |
| detail = f'{_xt(str(risk.get("id", "—")))} {_xt(_clip_words(str(risk.get("title", "")), 44))}' | |
| if primary: | |
| detail += f' | {_xt(primary)}' | |
| if reason: | |
| detail += f' | why: {_xt(_clip_words(reason, 80))}' | |
| airi_rows.append([ | |
| Paragraph(f'<font color="{_TEAL}"><b>Mapped theme {idx}</b></font>', _style(f"AIRIPDF_KM{idx}", 8, 10, _TEAL, True)), | |
| Paragraph(f'<font color="{_DGRAY}" size="8">{detail}</font>', _style(f"AIRIPDF_VM{idx}", 8, 11, _DGRAY)), | |
| ]) | |
| gaps = airi.get("known_gaps_in_bundle", []) | |
| if gaps: | |
| gap_preview = ", ".join( | |
| f"{g.get('id', '—')} {_xt(_clip_words(str(g.get('title', '')), 18))}" for g in gaps[:3] | |
| ) | |
| _gap_extra = f" (+{len(gaps) - 3} more)" if len(gaps) > 3 else "" | |
| airi_rows.append([ | |
| Paragraph(f'<font color="{_TEAL}"><b>Known gaps</b></font>', _style("AIRIPDF_KG", 8, 10, _TEAL, True)), | |
| Paragraph(f'<font color="{_DGRAY}" size="8">{gap_preview}{_gap_extra}</font>', _style("AIRIPDF_VG", 8, 11, _DGRAY)), | |
| ]) | |
| airi_rows.append([ | |
| Paragraph(f'<font color="{_TEAL}"><b>Why this matters</b></font>', _style("AIRIPDF_KW", 8, 10, _TEAL, True)), | |
| Paragraph( | |
| f'<font color="{_DGRAY}" size="8">This section helps a reviewer describe governance weaknesses in broader risk language, but it should remain secondary to repository boundary, traceability, and evidence posture.</font>', | |
| _style("AIRIPDF_VW", 8, 11, _DGRAY), | |
| ), | |
| ]) | |
| airi_table = Table(airi_rows, colWidths=[34 * mm, None]) | |
| airi_table.setStyle(TableStyle([ | |
| ("SPAN", (0, 0), (1, 0)), | |
| ("BACKGROUND", (0, 0), (1, 0), _hx(_TEAL)), | |
| ("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]), | |
| ("TOPPADDING", (0, 0), (-1, -1), 4), | |
| ("BOTTOMPADDING", (0, 0), (-1, -1), 4), | |
| ("LEFTPADDING", (0, 0), (-1, -1), 6), | |
| ("RIGHTPADDING", (0, 0), (-1, -1), 6), | |
| ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), | |
| ("GRID", (0, 1), (-1, -1), 0.3, _hx(_MGRAY)), | |
| ("VALIGN", (0, 0), (-1, -1), "TOP"), | |
| ])) | |
| story.append(airi_table) | |
| # Method Boundary | |
| story.append(Spacer(1, 6 * mm)) | |
| story += _subsec_hdr("4.4 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'<font color="{_AMBER}"><b>Scope boundary:</b></font> ' | |
| '<font color="#4A5568" size="8">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.</font>', | |
| _style("MBSCOPE", 8, 11, _DGRAY), | |
| )) | |
| story += _footer_block() | |
| return _single_page_story(story, break_before=True) | |
| # ── Page 7: Regulatory Traceability (dedicated, detailed packet only) ───────── | |
| def _page_regulatory_traceability(result: dict[str, Any]) -> list[Any]: | |
| block = _regulatory_traceability_pdf(result) | |
| if not block: | |
| return [] | |
| story = list(block) | |
| story += _footer_block() | |
| return _single_page_story(story, break_before=True) | |
| # ── Page 8: Report Metadata (detailed packet only) ─────────────────────────── | |
| def _page7_report_metadata(result: dict[str, Any]) -> list[Any]: | |
| story: list[Any] = [] | |
| score = result["score"] | |
| story += _chapter_hdr("Chapter 5 — 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'<font color="{_WHITE}"><b>Field</b></font>', _style("MH1", 8, 10, _WHITE, True)), | |
| Paragraph(f'<font color="{_WHITE}"><b>Value</b></font>', _style("MH2", 8, 10, _WHITE, True)), | |
| ]] | |
| for field, val in meta_items: | |
| meta_data.append([ | |
| Paragraph(f'<b>{_xt(field)}</b>', _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']}", | |
| f"About This Score: {_score_boundary_short_line()}", | |
| f"- {_score_boundary_lines()[0].replace('**', '')}", | |
| f"- {_score_boundary_lines()[1].replace('**', '')}", | |
| f"- {_score_boundary_lines()[2].replace('**', '')}", | |
| "", | |
| "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" (+{len(airi.get('covered_risks', [])) - 5} beyond preview)" | |
| if len(airi.get("covered_risks", [])) > 5 | |
| else "" | |
| ), | |
| 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", [])[:5] | |
| ], | |
| "", | |
| "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"], | |
| ])) | |
| _reg_lines = _regulatory_pdf_text_lines(result) | |
| if _reg_lines: | |
| sets.append(_fit_page(_reg_lines)) | |
| 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" | |