#!/usr/bin/env python3 """Prepare independent TELEN reviewer packets without cross-label leakage.""" from __future__ import annotations import argparse import hashlib import json from datetime import datetime, timezone from pathlib import Path import pandas as pd REPO_ROOT = Path(__file__).resolve().parents[1] EVENT_KEY = ["event_group_id", "candidate_id"] EVENT_COMMON = [ "protocol_version", *EVENT_KEY, "event_so_ky_hieu", "affected_so_ky_hieu", "official_event_url", "official_affected_url", "official_event_artifact_sha256", "official_affected_artifact_sha256", ] EVENT_LABEL_STEMS = ( "relation_class", "effect_scope", "valid_from", "valid_to", "effective_date_basis", "intermediate_instruments", "target_articles", "provision_mapping", "rationale", "reviewed_at", ) QREL_KEY = ["query_id", "snapshot_date", "candidate_version_id"] QREL_PACKET_KEY = ["query_id", "snapshot_date", "display_candidate_id"] QREL_COMMON = [ "protocol_version", "event_group_id", "query_group_id", "query_id", "snapshot_date", "display_candidate_id", "snapshot_phase", "candidate_pool_id", "candidate_pool_status", "question_status", "query_text", "candidate_text", "candidate_official_url", "candidate_text_sha256", ] QREL_LABEL_STEMS = ( "semantic_support", "temporal_status", "relevance_grade", "official_source_span", "answer_rationale", "rationale", "reviewed_at", ) QUESTION_AUTHOR_COLUMNS = [ "protocol_version", "event_group_id", "query_group_id", "pair_id", "candidate_id", "event_scope_candidate", "legal_issue_summary", "affected_question_1", "affected_question_2", "affected_question_3", "matched_unaffected_question", "matched_unaffected_expected_answer", "matched_unaffected_relevant_provision_ids", "pre_snapshot_date", "post_snapshot_date", "expected_answer_pre", "expected_answer_post", "answer_source_spans", "pre_relevant_provision_ids", "post_relevant_provision_ids", "pre_status", "post_status", "query_author", "annotation_status", ] QUESTION_REVIEW_COMMON = QUESTION_AUTHOR_COLUMNS QUESTION_LABEL_STEMS = ("question_quality", "rationale", "reviewed_at") def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def portable_path(path: Path) -> str: try: return str(path.resolve().relative_to(REPO_ROOT.resolve())) except ValueError: return str(path) def _require_columns(frame: pd.DataFrame, columns: list[str], label: str) -> None: missing = sorted(set(columns).difference(frame.columns)) if missing: raise ValueError(f"{label} is missing columns: {missing}") def _validate_unique(frame: pd.DataFrame, keys: list[str], label: str) -> None: if frame.duplicated(keys, keep=False).any(): raise ValueError(f"{label} contains duplicate keys: {keys}") def _blind_order( frame: pd.DataFrame, keys: list[str], *, seed: int, role: str ) -> pd.DataFrame: if frame.empty: return frame.reset_index(drop=True) material = frame[keys].astype(str).agg("\x1f".join, axis=1) order_key = material.map( lambda value: hashlib.sha256( f"{seed}\x1f{role}\x1f{value}".encode("utf-8") ).hexdigest() ) return ( frame.assign(_blind_order=order_key) .sort_values("_blind_order", kind="stable") .drop(columns="_blind_order") .reset_index(drop=True) ) def _role_columns( common: list[str], stems: tuple[str, ...], role_number: int ) -> list[str]: return [ *common, *(f"{stem}_annotator_{role_number}" for stem in stems), f"annotator_{role_number}", ] def _write_packet( frame: pd.DataFrame, *, columns: list[str], keys: list[str], path: Path, seed: int, role: str, ) -> dict[str, object]: _require_columns(frame, columns, role) packet = _blind_order(frame[columns].copy(), keys, seed=seed, role=role) _validate_unique(packet, keys, role) packet.to_csv(path, index=False) return { "path": portable_path(path), "sha256": sha256_file(path), "rows": len(packet), "columns": list(packet.columns), "unique_keys": True, } def prepare(args: argparse.Namespace) -> dict[str, object]: events = pd.read_csv(args.events, dtype=str).fillna("") qrels = pd.read_csv(args.qrels, dtype=str).fillna("") questions = pd.read_csv(args.questions, dtype=str).fillna("") _validate_unique(events, EVENT_KEY, "events") _validate_unique(qrels, QREL_KEY, "qrels") _validate_unique(questions, ["query_group_id"], "questions") args.output_dir.mkdir(parents=True, exist_ok=True) packets: dict[str, dict[str, object]] = {} for role_number in (1, 2): role = f"annotator_{role_number}" packets[f"events_{role}"] = _write_packet( events, columns=_role_columns(EVENT_COMMON, EVENT_LABEL_STEMS, role_number), keys=EVENT_KEY, path=args.output_dir / f"events_{role}.csv", seed=args.seed, role=f"events:{role}", ) packets[f"qrels_{role}"] = _write_packet( qrels, columns=_role_columns(QREL_COMMON, QREL_LABEL_STEMS, role_number), keys=QREL_PACKET_KEY, path=args.output_dir / f"qrels_{role}.csv", seed=args.seed, role=f"qrels:{role}", ) packets[f"questions_{role}"] = _write_packet( questions, columns=_role_columns( QUESTION_REVIEW_COMMON, QUESTION_LABEL_STEMS, role_number ), keys=["query_group_id"], path=args.output_dir / f"questions_{role}.csv", seed=args.seed, role=f"questions:{role}", ) packets["question_author"] = _write_packet( questions, columns=QUESTION_AUTHOR_COLUMNS, keys=["query_group_id"], path=args.output_dir / "question_author.csv", seed=args.seed, role="question_author", ) event_a_columns = set(packets["events_annotator_1"]["columns"]) event_b_columns = set(packets["events_annotator_2"]["columns"]) qrel_a_columns = set(packets["qrels_annotator_1"]["columns"]) qrel_b_columns = set(packets["qrels_annotator_2"]["columns"]) question_a_columns = set(packets["questions_annotator_1"]["columns"]) question_b_columns = set(packets["questions_annotator_2"]["columns"]) leakage_checks = { "event_a_has_no_b_or_adjudicated": not any( "annotator_2" in column or "adjudicated" in column for column in event_a_columns ), "event_b_has_no_a_or_adjudicated": not any( "annotator_1" in column or "adjudicated" in column for column in event_b_columns ), "qrel_a_has_no_b_or_adjudicated": not any( "annotator_2" in column or "adjudicated" in column for column in qrel_a_columns ), "qrel_b_has_no_a_or_adjudicated": not any( "annotator_1" in column or "adjudicated" in column for column in qrel_b_columns ), "qrel_packets_hide_internal_candidate_version": ( "candidate_version_id" not in qrel_a_columns and "candidate_version_id" not in qrel_b_columns ), "question_a_has_no_b_or_adjudicated": not any( "annotator_2" in column or "adjudicated" in column for column in question_a_columns ), "question_b_has_no_a_or_adjudicated": not any( "annotator_1" in column or "adjudicated" in column for column in question_b_columns ), "event_packets_hide_machine_relation_hints": not any( column in event_a_columns | event_b_columns for column in { "relation_discovery", "auto_status", "machine_verification_status", "relation_evidence_class", "expected_effective_date_official", } ), } passed = all(leakage_checks.values()) receipt = { "schema_version": 1, "artifact_type": "blind_annotation_packet_audit", "created_at": datetime.now(timezone.utc).isoformat(), "seed": args.seed, "inputs": { "events": portable_path(args.events), "events_sha256": sha256_file(args.events), "qrels": portable_path(args.qrels), "qrels_sha256": sha256_file(args.qrels), "questions": portable_path(args.questions), "questions_sha256": sha256_file(args.questions), }, "packets": packets, "leakage_checks": leakage_checks, "independent_packet_gate": { "passed": passed, "event_assignment_ready": passed and len(events) > 0, "question_review_assignment_ready": bool( passed and len(questions) > 0 and questions["annotation_status"].eq("ready_for_blind_review").all() ), "qrel_assignment_ready": bool( passed and len(qrels) > 0 and qrels["candidate_pool_status"].eq("frozen").all() and qrels["question_status"].eq("adjudicated_gold").all() and qrels["display_candidate_id"].str.match( r"^cand-[0-9a-f]{12,64}$", na=False ).all() and qrels["query_text"].str.strip().ne("").all() and qrels["candidate_text"].str.strip().ne("").all() and qrels["candidate_official_url"].str.startswith( "https://", na=False ).all() and qrels["candidate_text_sha256"].str.match( r"^[0-9a-f]{64}$", na=False ).all() ), "learned_training_authorized": False, }, } args.receipt.parent.mkdir(parents=True, exist_ok=True) args.receipt.write_text( json.dumps(receipt, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) return receipt def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) root = REPO_ROOT / "tmp/telen/temporal-pilot" parser.add_argument( "--events", type=Path, default=root / "official_verification_template.csv" ) parser.add_argument( "--qrels", type=Path, default=root / "qrel_judgment_template.csv" ) parser.add_argument( "--questions", type=Path, default=root / "expert_annotation_template.csv" ) parser.add_argument( "--output-dir", type=Path, default=REPO_ROOT / "tmp/telen/blind-annotation-packets", ) parser.add_argument( "--receipt", type=Path, default=REPO_ROOT / "research/results/telen_blind_annotation_packet_audit_2026-08-29.json", ) parser.add_argument("--seed", type=int, default=20260829) return parser.parse_args() def main() -> None: receipt = prepare(parse_args()) print(json.dumps(receipt, ensure_ascii=False, indent=2)) if not receipt["independent_packet_gate"]["passed"]: raise SystemExit(2) if __name__ == "__main__": main()