| |
| """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() |
|
|
