Spaces:
Sleeping
Sleeping
File size: 2,710 Bytes
14ea8cd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | #!/usr/bin/env python3
"""Create a non-identifying descriptive aggregate from a private review export.
The input export must remain outside the repository. The output contains only
summary statistics and instrument metadata; raw answers, notes, participant
codes, signatures and withdrawal secrets are never copied.
"""
from __future__ import annotations
import argparse
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from review_portal.analysis import summarize
_FORBIDDEN_KEYS = {
"answers",
"notes",
"participant_code",
"withdrawal_token",
"invitation_code",
"signature",
"email",
"name",
}
def _load_records(path: Path) -> list[dict[str, Any]]:
payload: object = json.loads(path.read_text(encoding="utf-8"))
if isinstance(payload, dict):
payload = payload.get("records", [])
if not isinstance(payload, list) or not all(
isinstance(row, dict) for row in payload
):
raise ValueError("Input must be a JSON list or an admin export with records.")
return [dict(row) for row in payload]
def _assert_safe(value: object) -> None:
if isinstance(value, dict):
for key, child in value.items():
if str(key).lower() in _FORBIDDEN_KEYS:
raise ValueError(f"Unsafe field in aggregate: {key}")
_assert_safe(child)
elif isinstance(value, list):
for child in value:
_assert_safe(child)
def build_aggregate(records: list[dict[str, Any]]) -> dict[str, Any]:
aggregate = {
"schema_version": 1,
"classification": "PRIVATE_HUMAN_REVIEW_DESCRIPTIVES_NOT_EVIDENCE",
"generated_at": datetime.now(timezone.utc).isoformat(),
"source_record_count": len(records),
"automatic_evidence_promotion": False,
"raw_answers_included": False,
"summary": summarize(records),
}
_assert_safe(aggregate)
return aggregate
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"input", type=Path, help="Private admin export JSON outside the repository"
)
parser.add_argument("output", type=Path, help="Sanitized aggregate JSON")
args = parser.parse_args()
records = _load_records(args.input)
aggregate = build_aggregate(records)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(
json.dumps(aggregate, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
print(f"Wrote sanitized aggregate for {len(records)} records to {args.output}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|