smoke_signal/scripts/08_validation_report.py

#!/usr/bin/env python3
"""
Smoke Signal — Stage 10: Validation Report
============================================
Generates a full pipeline validation report before Codex ingestion.

Checks:
  - Source manifest completeness and hash integrity
  - Page routing accuracy (spot check)
  - OCR confidence distribution
  - Review queue clearance rate
  - Export schema validity
  - Contamination risk (low-confidence text in export)
  - Rights class separation
  - Page traceability (every exported line has source evidence)

Outputs:
  - reports/validation_report_<batch>.md  — human-readable
  - reports/validation_report_<batch>.json — machine-readable

Usage:
    python scripts/08_validation_report.py
    python scripts/08_validation_report.py --batch-id SS-BATCH-001
    python scripts/08_validation_report.py --export-file exports/codex_export_20260516.jsonl
"""

import argparse
import csv
import json
import sys
from datetime import datetime
from pathlib import Path

# ── Paths ──────────────────────────────────────────────────────────────────────
ROOT          = Path(__file__).resolve().parents[1]
MANIFEST_CSV  = ROOT / "manifest" / "source_manifest.csv"
EXPORTS_DIR   = ROOT / "exports"
REGIONS_DIR   = ROOT / "regions"
OCR_RAW_DIR   = ROOT / "ocr_raw"
REVIEW_DIR    = ROOT / "review"
REPORTS_DIR   = ROOT / "reports"
LOGS_DIR      = ROOT / "logs"

REPORTS_DIR.mkdir(parents=True, exist_ok=True)
DECISIONS_CSV = REVIEW_DIR / "review_decisions.csv"
QUEUE_CSV     = REVIEW_DIR / "review_queue.csv"

CONFIG = {
    "config_version":              "ss_validate_v0.1",
    "min_page_route_accuracy":     0.95,
    "min_ocr_word_accuracy":       0.80,
    "min_review_clearance":        0.90,
    "max_contamination_risk":      0.00,
    "confidence_contamination_threshold": 0.60,
}


# ── Loaders ────────────────────────────────────────────────────────────────────
def load_manifest() -> list:
    if not MANIFEST_CSV.exists():
        return []
    with open(MANIFEST_CSV, newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def load_queue() -> list:
    if not QUEUE_CSV.exists():
        return []
    with open(QUEUE_CSV, newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def load_decisions() -> list:
    if not DECISIONS_CSV.exists():
        return []
    with open(DECISIONS_CSV, newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def load_export(export_file: Path) -> list:
    records = []
    if not export_file.exists():
        return records
    with open(export_file, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                try:
                    records.append(json.loads(line))
                except json.JSONDecodeError:
                    pass
    return records


# ── Check functions ────────────────────────────────────────────────────────────
def check_manifest(manifest: list) -> dict:
    total       = len(manifest)
    unknown_rights = sum(1 for r in manifest if r.get("rights_class") == "unknown")
    excluded    = sum(1 for r in manifest if r.get("rights_class") == "excluded")
    no_hash     = sum(1 for r in manifest if not r.get("sha256"))
    status_counts = {}
    for r in manifest:
        s = r.get("status", "unknown")
        status_counts[s] = status_counts.get(s, 0) + 1

    issues = []
    if unknown_rights > 0:
        issues.append(f"{unknown_rights} sources have unknown rights class")
    if no_hash > 0:
        issues.append(f"{no_hash} sources missing SHA-256 hash")

    return {
        "check": "source_manifest",
        "total_sources":    total,
        "unknown_rights":   unknown_rights,
        "excluded":         excluded,
        "no_hash":          no_hash,
        "status_counts":    status_counts,
        "issues":           issues,
        "passed":           len(issues) == 0,
    }


def check_review_clearance(queue: list, decisions: list) -> dict:
    total_queued  = len(queue)
    decided_ids   = {d.get("region_id") for d in decisions}
    queue_ids     = {q.get("region_id") for q in queue}
    cleared       = len(queue_ids & decided_ids)
    pending       = total_queued - cleared
    clearance_rate = cleared / max(total_queued, 1)

    approved  = sum(1 for d in decisions if d.get("status") in ("accepted", "edited"))
    rejected  = sum(1 for d in decisions if d.get("status") == "rejected")
    quarantined = sum(1 for d in decisions if d.get("status") == "quarantined")

    passed = clearance_rate >= CONFIG["min_review_clearance"]
    issues = []
    if pending > 0:
        issues.append(f"{pending} items still pending review ({clearance_rate:.0%} cleared)")

    return {
        "check":           "review_clearance",
        "total_queued":    total_queued,
        "cleared":         cleared,
        "pending":         pending,
        "clearance_rate":  round(clearance_rate, 3),
        "approved":        approved,
        "rejected":        rejected,
        "quarantined":     quarantined,
        "target":          CONFIG["min_review_clearance"],
        "issues":          issues,
        "passed":          passed,
    }


def check_export_contamination(export_records: list) -> dict:
    """Check for low-confidence text that shouldn't be in export."""
    threshold   = CONFIG["confidence_contamination_threshold"]
    risky       = []

    for rec in export_records:
        conf = float(rec.get("confidence", 1.0))
        status = rec.get("review_status", "")
        if conf < threshold and status == "accepted":
            risky.append({
                "region_id": rec.get("region_id"),
                "confidence": conf,
                "review_status": status,
            })

    contamination_risk = len(risky) / max(len(export_records), 1)
    passed = contamination_risk <= CONFIG["max_contamination_risk"]

    issues = []
    if risky:
        issues.append(f"{len(risky)} low-confidence records in export (conf < {threshold})")

    return {
        "check":               "contamination_risk",
        "total_export_records": len(export_records),
        "risky_records":       len(risky),
        "contamination_rate":  round(contamination_risk, 4),
        "threshold":           threshold,
        "risky_sample":        risky[:5],
        "issues":              issues,
        "passed":              passed,
    }


def check_traceability(export_records: list) -> dict:
    """Every exported record must have source_hash, page_number, region_id, run_id."""
    missing_trace = []
    for rec in export_records:
        missing = [f for f in ("source_hash", "page_number", "region_id", "run_id")
                   if not rec.get(f)]
        if missing:
            missing_trace.append({
                "region_id": rec.get("region_id", "unknown"),
                "missing_fields": missing
            })

    passed = len(missing_trace) == 0
    issues = []
    if missing_trace:
        issues.append(f"{len(missing_trace)} records missing traceability fields")

    return {
        "check":                "traceability",
        "total_records":        len(export_records),
        "untraceable_records":  len(missing_trace),
        "sample":               missing_trace[:5],
        "issues":               issues,
        "passed":               passed,
    }


def check_rights_separation(export_records: list, manifest: list) -> dict:
    """Confirm rights classes are not mixed in export."""
    manifest_by_id = {r.get("book_id"): r for r in manifest}
    rights_in_export = set()
    unknown_in_export = []

    for rec in export_records:
        bid = rec.get("book_id")
        m   = manifest_by_id.get(bid, {})
        rc  = m.get("rights_class", "unknown")
        rights_in_export.add(rc)
        if rc in ("unknown", "excluded"):
            unknown_in_export.append(bid)

    issues = []
    if unknown_in_export:
        issues.append(f"Unknown/excluded rights class found in export: {set(unknown_in_export)}")

    return {
        "check":                "rights_separation",
        "rights_classes_found": list(rights_in_export),
        "unknown_in_export":    list(set(unknown_in_export)),
        "issues":               issues,
        "passed":               len(unknown_in_export) == 0,
    }


def check_ocr_confidence_distribution(queue: list) -> dict:
    """Summarise OCR confidence distribution across reviewed pages."""
    confs = []
    for item in queue:
        try:
            confs.append(float(item.get("confidence", 0)))
        except (ValueError, TypeError):
            pass

    if not confs:
        return {"check": "ocr_confidence", "passed": True, "note": "no_data"}

    confs.sort()
    n       = len(confs)
    mean    = sum(confs) / n
    below60 = sum(1 for c in confs if c < 0.60)
    below80 = sum(1 for c in confs if c < 0.80)

    return {
        "check":       "ocr_confidence_distribution",
        "page_count":  n,
        "mean":        round(mean, 3),
        "min":         round(confs[0], 3),
        "max":         round(confs[-1], 3),
        "p25":         round(confs[n // 4], 3),
        "p50":         round(confs[n // 2], 3),
        "p75":         round(confs[3 * n // 4], 3),
        "below_0.60":  below60,
        "below_0.80":  below80,
        "passed":      True,
    }


# ── Report writer ─────────────────────────────────────────────────────────────
def write_report(checks: list, batch_tag: str, export_path: Path) -> Path:
    all_passed = all(c.get("passed", False) for c in checks)
    issues_all = [i for c in checks for i in c.get("issues", [])]
    verdict    = "✅ PASS — Safe to proceed to Codex ingestion" if all_passed else "❌ FAIL — Do not ingest until issues resolved"

    # ── JSON ─────────────────────────────────────────────────────────────────
    json_path = REPORTS_DIR / f"validation_report_{batch_tag}.json"
    report_data = {
        "batch_id":      batch_tag,
        "generated_at":  datetime.utcnow().isoformat() + "Z",
        "config_version": CONFIG["config_version"],
        "export_file":   str(export_path),
        "verdict":       "PASS" if all_passed else "FAIL",
        "all_issues":    issues_all,
        "checks":        checks,
    }
    with open(json_path, "w") as f:
        json.dump(report_data, f, indent=2)

    # ── Markdown ──────────────────────────────────────────────────────────────
    md_path = REPORTS_DIR / f"validation_report_{batch_tag}.md"
    lines   = [
        f"# Smoke Signal — Validation Report",
        f"**Batch:** {batch_tag}  ",
        f"**Generated:** {datetime.utcnow().isoformat()}Z  ",
        f"**Export:** `{export_path.name}`  ",
        f"\n## Verdict\n",
        f"### {verdict}\n",
    ]

    if issues_all:
        lines.append("## Issues to Resolve\n")
        for issue in issues_all:
            lines.append(f"- ⚠️ {issue}")
        lines.append("")

    lines.append("## Check Results\n")
    lines.append("| Check | Status | Notes |")
    lines.append("|-------|--------|-------|")

    for c in checks:
        icon  = "✅" if c.get("passed") else "❌"
        name  = c.get("check", "unknown").replace("_", " ").title()
        notes = "; ".join(c.get("issues", [])) or "OK"
        lines.append(f"| {name} | {icon} | {notes} |")

    lines.append("\n## Detailed Results\n")
    for c in checks:
        lines.append(f"### {c.get('check','').replace('_',' ').title()}")
        for k, v in c.items():
            if k not in ("check", "issues", "passed", "sample", "risky_sample"):
                lines.append(f"- **{k}:** {v}")
        lines.append("")

    lines.append("---")
    lines.append(f"*Smoke Signal {CONFIG['config_version']} · Generated {datetime.utcnow().date()}*")

    with open(md_path, "w", encoding="utf-8") as f:
        f.write("\n".join(lines))

    return md_path, json_path, all_passed


# ── Main ───────────────────────────────────────────────────────────────────────
def main():
    parser = argparse.ArgumentParser(description="Smoke Signal — Stage 10: Validation Report")
    parser.add_argument("--batch-id",    help="Batch ID tag")
    parser.add_argument("--export-file", help="Path to JSONL export file to validate")
    args = parser.parse_args()

    batch_tag = args.batch_id or datetime.utcnow().strftime("%Y%m%d_%H%M%S")

    print(f"\n{'='*60}")
    print(f"  Smoke Signal — Stage 10: Validation Report")
    print(f"  Batch   : {batch_tag}")
    print(f"  Config  : {CONFIG['config_version']}")
    print(f"{'='*60}\n")

    # Find export file
    if args.export_file:
        export_path = Path(args.export_file)
    else:
        jsonl_files = sorted(EXPORTS_DIR.glob("codex_export_*.jsonl"), key=lambda p: p.stat().st_mtime, reverse=True)
        if not jsonl_files:
            print("  No export files found. Run 07_codex_exporter.py first.")
            sys.exit(1)
        export_path = jsonl_files[0]
        print(f"  Using latest export: {export_path.name}\n")

    # Load data
    manifest       = load_manifest()
    queue          = load_queue()
    decisions      = load_decisions()
    export_records = load_export(export_path)

    print(f"  Sources in manifest : {len(manifest)}")
    print(f"  Review queue items  : {len(queue)}")
    print(f"  Review decisions    : {len(decisions)}")
    print(f"  Export records      : {len(export_records)}\n")

    # Run checks
    checks = [
        check_manifest(manifest),
        check_review_clearance(queue, decisions),
        check_ocr_confidence_distribution(queue),
        check_export_contamination(export_records),
        check_traceability(export_records),
        check_rights_separation(export_records, manifest),
    ]

    for c in checks:
        icon = "✅" if c.get("passed") else "❌"
        print(f"  {icon} {c['check']}")
        for issue in c.get("issues", []):
            print(f"     ⚠️  {issue}")

    # Write reports
    md_path, json_path, all_passed = write_report(checks, batch_tag, export_path)

    print(f"\n{'─'*60}")
    print(f"  Report MD   → {md_path.relative_to(ROOT)}")
    print(f"  Report JSON → {json_path.relative_to(ROOT)}")
    print(f"{'─'*60}")

    if all_passed:
        print(f"\n  ✅ ALL CHECKS PASSED — Approval Gate E cleared.")
        print(f"     Safe to ingest into Codex.\n")
    else:
        print(f"\n  ❌ CHECKS FAILED — Do not ingest until issues are resolved.")
        print(f"     Fix the issues above and re-run this report.\n")
        sys.exit(1)


if __name__ == "__main__":
    main()