#!/usr/bin/env python3 """Validate Phase 3 pilot evidence artifacts without modifying source data.""" from __future__ import annotations import argparse import csv import hashlib import json import re import sys from dataclasses import dataclass from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parents[1] API_ROOT = REPO_ROOT / "apps" / "api" sys.path.insert(0, str(API_ROOT)) from app.services.phase3_kpi_contract import ( # noqa: E402 KPI_METRIC_REQUIRED_KEYS, KPI_METRIC_STATUSES, KPI_REPORT_PATH, PHASE3_KPI_METRICS, PREPOST_CSV_PATH, ) FORBIDDEN_HEADER_TERMS = { "address", "api_key", "birthdate", "cookie", "date_of_birth", "dob", "email", "full_name", "guardian_name", "identity_map", "national_id", "participant_name", "phone", "raw_audio", "raw_source_case", "raw_voice", "secret", "ssn", "student_id", "student_name", "token", } EMAIL_RE = re.compile(r"[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}", re.IGNORECASE) PHONE_RE = re.compile(r"(?:\+?\d[\d .()-]{7,}\d)") APPROVED_EXPORT_STATUS = "approved_for_recursive_learning_seed" CSV_ALLOWED_VALUES = { "00-intake/pilot_roster.csv": { "withdrawal_state": { "active", "withdrawn_before_data_use", "withdrawn_after_data_use", "excluded_by_operator", }, }, "01-sessions/session_completion.csv": { "completion_state": { "completed", "abandoned", "operator_cancelled", "excluded_from_analysis", }, }, "02-measures/prepost_measures.csv": { "timepoint": {"pre", "post"}, }, "04-privacy/withdrawal_log.csv": { "scope": { "future_sessions_only", "exclude_from_analysis", "exclude_from_export", "delete_where_policy_allows", }, "status": { "received", "in_progress", "completed", "rejected_by_policy", "needs_legal_review", }, }, } @dataclass(frozen=True) class CsvSpec: path: str headers: tuple[str, ...] CSV_SPECS = ( CsvSpec( "00-intake/pilot_roster.csv", ( "participant_id", "cohort_id", "consent_version", "consent_signed_at", "withdrawal_state", "enrolled_at", ), ), CsvSpec( "00-intake/consent_receipts.csv", ("participant_id", "consent_version", "signed_at", "signer_role", "receipt_id"), ), CsvSpec( "01-sessions/session_completion.csv", ( "participant_id", "session_id", "persona_id", "started_at", "ended_at", "completion_state", "turns_count", "supervisor_reviewed_at", ), ), CsvSpec( PREPOST_CSV_PATH, ("participant_id", "measure_name", "timepoint", "score", "collected_at"), ), CsvSpec( "02-measures/sus_responses.csv", ("participant_id", "item", "response", "collected_at"), ), CsvSpec( "04-privacy/withdrawal_log.csv", ("participant_id", "requested_at", "effective_at", "scope", "status", "attestation_path"), ), ) REQUIRED_MARKDOWN = ("04-privacy/privacy_audit.md",) KPI_REPORT_KEYS = { "cohort_size", "exclusions", "generated_at", "metrics", "open_schema_gaps", "pilot_id", "review", "source_window", } MANIFEST_KEYS = { "agreement", "anonymization", "approvals", "consent_scope", "created_at", "dataset_name", "export_id", "export_status", "files", "known_limitations", "pii_scan", "purpose", "selection_criteria", "source_tables", "source_window", } class Report: def __init__(self) -> None: self.errors: list[str] = [] self.warnings: list[str] = [] self.info: list[str] = [] def error(self, message: str) -> None: self.errors.append(message) def warn(self, message: str) -> None: self.warnings.append(message) def note(self, message: str) -> None: self.info.append(message) def to_dict(self, evidence_root: Path) -> dict[str, Any]: return { "evidence_root": str(evidence_root), "status": "pass" if not self.errors else "fail", "errors": self.errors, "warnings": self.warnings, "info": self.info, } def read_csv_header(path: Path, report: Report) -> list[str] | None: try: with path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.reader(handle) header = next(reader, None) if not header: report.error(f"{path}: empty CSV or missing header") return None return [cell.strip() for cell in header] except UnicodeDecodeError: report.error(f"{path}: not valid UTF-8/UTF-8-SIG") except OSError as exc: report.error(f"{path}: cannot read CSV: {exc}") return None def scan_csv_values(path: Path, report: Report, max_rows: int) -> None: try: with path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) for row_number, row in enumerate(reader, start=2): if row_number > max_rows + 1: report.note(f"{path}: scanned first {max_rows} data rows for identifier patterns") return for column, value in row.items(): if not value: continue if EMAIL_RE.search(value): report.error(f"{path}:{row_number}: possible email in column '{column}'") elif column and column.lower() not in {"turns_count", "score", "response", "item"}: if PHONE_RE.fullmatch(value.strip()): report.warn(f"{path}:{row_number}: possible phone-like value in column '{column}'") except UnicodeDecodeError: report.error(f"{path}: not valid UTF-8/UTF-8-SIG") except OSError as exc: report.error(f"{path}: cannot scan CSV: {exc}") def validate_csv_allowed_values(root: Path, spec: CsvSpec, report: Report) -> None: allowed_by_column = CSV_ALLOWED_VALUES.get(spec.path) if not allowed_by_column: return path = root / spec.path try: with path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) for row_number, row in enumerate(reader, start=2): for column, allowed_values in allowed_by_column.items(): value = (row.get(column) or "").strip() if value and value not in allowed_values: expected = ", ".join(sorted(allowed_values)) report.error( f"{spec.path}:{row_number}: invalid {column} '{value}' " f"(expected one of: {expected})" ) except UnicodeDecodeError: report.error(f"{path}: not valid UTF-8/UTF-8-SIG") except OSError as exc: report.error(f"{path}: cannot validate CSV values: {exc}") def validate_csv(root: Path, spec: CsvSpec, report: Report, max_scan_rows: int) -> None: path = root / spec.path if not path.exists(): report.error(f"missing required file: {spec.path}") return header = read_csv_header(path, report) if header is None: return missing = [name for name in spec.headers if name not in header] if missing: report.error(f"{spec.path}: missing required headers: {', '.join(missing)}") forbidden = sorted( column for column in header for term in FORBIDDEN_HEADER_TERMS if term in column.lower() ) if forbidden: report.error(f"{spec.path}: forbidden identifier/secret-like headers: {', '.join(forbidden)}") scan_csv_values(path, report, max_rows=max_scan_rows) validate_csv_allowed_values(root, spec, report) def read_json(path: Path, report: Report) -> dict[str, Any] | None: try: with path.open("r", encoding="utf-8") as handle: data = json.load(handle) except json.JSONDecodeError as exc: report.error(f"{path}: invalid JSON: {exc}") return None except UnicodeDecodeError: report.error(f"{path}: not valid UTF-8") return None except OSError as exc: report.error(f"{path}: cannot read JSON: {exc}") return None if not isinstance(data, dict): report.error(f"{path}: top-level JSON must be an object") return None return data def validate_kpi_report(root: Path, report: Report) -> None: rel_path = KPI_REPORT_PATH path = root / rel_path if not path.exists(): report.error(f"missing required file: {rel_path}") return data = read_json(path, report) if data is None: return missing_top_level = sorted(KPI_REPORT_KEYS - set(data)) if missing_top_level: report.error(f"{rel_path}: missing keys: {', '.join(missing_top_level)}") metrics = data.get("metrics") if not isinstance(metrics, dict): report.error(f"{rel_path}: missing object key 'metrics'") return missing = sorted(set(PHASE3_KPI_METRICS) - set(metrics)) if missing: report.error(f"{rel_path}: missing metric keys: {', '.join(missing)}") for metric_name, metric in metrics.items(): if not isinstance(metric, dict): report.error(f"{rel_path}: metric '{metric_name}' must be an object") continue for key in sorted(KPI_METRIC_REQUIRED_KEYS): if key not in metric: report.error(f"{rel_path}: metric '{metric_name}' missing '{key}'") if "source_files" in metric and not isinstance(metric["source_files"], list): report.error(f"{rel_path}: metric '{metric_name}' source_files must be a list") status = metric.get("status") if status is not None and status not in KPI_METRIC_STATUSES: expected = ", ".join(sorted(KPI_METRIC_STATUSES)) report.error( f"{rel_path}: metric '{metric_name}' status '{status}' is invalid " f"(expected one of: {expected})" ) def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def numeric_value(value: Any) -> float | None: if isinstance(value, bool): return None if isinstance(value, (int, float)): return float(value) return None def validate_manifest_file_entry( root: Path, rel_path: str, item: dict[str, Any], report: Report, *, approved: bool, ) -> None: required_keys = ("path", "rows", "sha256", "schema") for key in required_keys: if key not in item: message = f"{rel_path}: file entry missing '{key}'" if approved: report.error(message) else: report.warn(message) file_path = item.get("path") expected_sha = item.get("sha256") if not isinstance(file_path, str) or not file_path.strip(): if approved: report.error(f"{rel_path}: approved file entry requires non-empty path") return try: root_resolved = root.resolve() candidate = (root / file_path).resolve() candidate.relative_to(root_resolved) except ValueError: report.error(f"{rel_path}: file path escapes evidence root: {file_path}") return if not approved: return if not candidate.exists(): report.error(f"{rel_path}: approved file missing: {file_path}") return if not candidate.is_file(): report.error(f"{rel_path}: approved file path is not a file: {file_path}") return if isinstance(expected_sha, str) and expected_sha: actual_sha = sha256_file(candidate) if actual_sha != expected_sha: report.error(f"{rel_path}: sha256 mismatch for {file_path}") def validate_manifest(root: Path, report: Report) -> None: rel_path = "03-export/export_manifest.json" path = root / rel_path if not path.exists(): report.error(f"missing required file: {rel_path}") return data = read_json(path, report) if data is None: return missing = sorted(MANIFEST_KEYS - set(data)) if missing: report.error(f"{rel_path}: missing keys: {', '.join(missing)}") status = data.get("export_status") approved = status == APPROVED_EXPORT_STATUS if approved: approvals = data.get("approvals") if not isinstance(approvals, dict): report.error(f"{rel_path}: approved export requires approvals object") else: for key in ("data_steward", "legal_or_privacy_reviewer", "technical_operator", "approved_at"): if not approvals.get(key): report.error(f"{rel_path}: approved export missing approval '{key}'") pii_scan = data.get("pii_scan") if not isinstance(pii_scan, dict): report.error(f"{rel_path}: approved export requires pii_scan object") elif pii_scan.get("status") != "pass": report.error(f"{rel_path}: approved export requires pii_scan.status='pass'") agreement = data.get("agreement") if not isinstance(agreement, dict): report.error(f"{rel_path}: approved export requires agreement object") else: kappa = numeric_value(agreement.get("kappa")) icc = numeric_value(agreement.get("icc")) if kappa is None or kappa < 0.70: report.error(f"{rel_path}: approved export requires agreement.kappa >= 0.70") if icc is None or icc < 0.75: report.error(f"{rel_path}: approved export requires agreement.icc >= 0.75") selection_criteria = data.get("selection_criteria") if not isinstance(selection_criteria, dict): report.error(f"{rel_path}: approved export requires selection_criteria object") elif selection_criteria.get("include_withdrawn") is not False: report.error(f"{rel_path}: approved export requires selection_criteria.include_withdrawn=false") consent_scope = data.get("consent_scope") allowed_uses = consent_scope.get("allowed_uses") if isinstance(consent_scope, dict) else None if not isinstance(allowed_uses, list) or "recursive_learning_seed" not in allowed_uses: report.error(f"{rel_path}: approved export requires recursive_learning_seed consent scope") files = data.get("files") if isinstance(files, list): if approved and not files: report.error(f"{rel_path}: approved export requires at least one file entry") for item in files: if not isinstance(item, dict): report.error(f"{rel_path}: files[] entries must be objects") continue validate_manifest_file_entry(root, rel_path, item, report, approved=approved) elif files is not None: report.error(f"{rel_path}: 'files' must be a list") def validate_markdown(root: Path, report: Report) -> None: for rel_path in REQUIRED_MARKDOWN: path = root / rel_path if not path.exists(): report.error(f"missing required file: {rel_path}") continue try: content = path.read_text(encoding="utf-8") except UnicodeDecodeError: report.error(f"{rel_path}: not valid UTF-8") continue except OSError as exc: report.error(f"{rel_path}: cannot read Markdown: {exc}") continue if not content.strip(): report.error(f"{rel_path}: file is empty") if "Legal/privacy reviewer:" not in content: report.warn(f"{rel_path}: expected legal/privacy reviewer field") def validate(root: Path, max_scan_rows: int) -> Report: report = Report() if not root.exists(): report.error(f"evidence root does not exist: {root}") return report if not root.is_dir(): report.error(f"evidence root is not a directory: {root}") return report for spec in CSV_SPECS: validate_csv(root, spec, report, max_scan_rows=max_scan_rows) validate_kpi_report(root, report) validate_manifest(root, report) validate_markdown(root, report) return report def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Read-only Phase 3 evidence checker. Writes only when --output is provided." ) parser.add_argument( "--check", action="store_true", help="Run validation checks. This is also the default behavior.", ) parser.add_argument( "--dry-run", action="store_true", help="Alias for read-only validation; kept for operator clarity.", ) parser.add_argument( "--evidence-root", type=Path, default=Path("evidence/phase3"), help="Directory containing Phase 3 evidence files.", ) parser.add_argument( "--output", type=Path, help="Optional JSON report path. No files are written unless this is set.", ) parser.add_argument( "--json", action="store_true", help="Print the validation report as JSON instead of text.", ) parser.add_argument( "--max-scan-rows", type=int, default=1000, help="Maximum data rows per CSV to scan for obvious identifier patterns.", ) return parser.parse_args(argv) def print_text_report(data: dict[str, Any]) -> None: print(f"Phase 3 artifact check: {data['status']}") print(f"Evidence root: {data['evidence_root']}") for label in ("errors", "warnings", "info"): items = data[label] if not items: continue print(f"\n{label.upper()}:") for item in items: print(f"- {item}") def main(argv: list[str]) -> int: args = parse_args(argv) report = validate(args.evidence_root, max_scan_rows=max(0, args.max_scan_rows)) data = report.to_dict(args.evidence_root) if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(data, indent=2, ensure_ascii=True) + "\n", encoding="utf-8") if args.json: print(json.dumps(data, indent=2, ensure_ascii=True)) else: print_text_report(data) return 0 if not report.errors else 1 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))