#!/usr/bin/env python3 """Build an auditable temporal-event pilot package from discovery candidates. The generated records are annotation inputs, never gold labels. They retain the source snapshot hashes, exact event spans, automated rejection reasons, and blank official/manual verification fields. """ from __future__ import annotations import argparse import hashlib import json import sys from collections import Counter from datetime import timedelta from pathlib import Path import pandas as pd import pyarrow.parquet as pq REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "src")) from telen.bootstrap import ( # noqa: E402 contains_document_number, extract_articles, extract_event_mentions, html_to_legal_text, ) ANNOTATION_PROTOCOL_VERSION = "pilot-1.0" QREL_TEMPLATE_COLUMNS = [ "protocol_version", "event_group_id", "query_group_id", "query_id", "snapshot_phase", "snapshot_date", "candidate_version_id", "display_candidate_id", "candidate_pool_id", "candidate_pool_status", "question_status", "query_text", "candidate_text", "candidate_official_url", "candidate_text_sha256", "semantic_support_annotator_1", "semantic_support_annotator_2", "temporal_status_annotator_1", "temporal_status_annotator_2", "relevance_grade_annotator_1", "relevance_grade_annotator_2", "official_source_span_annotator_1", "official_source_span_annotator_2", "answer_rationale_annotator_1", "answer_rationale_annotator_2", "rationale_annotator_1", "rationale_annotator_2", "reviewed_at_annotator_1", "reviewed_at_annotator_2", "semantic_support_adjudicated", "temporal_status_adjudicated", "relevance_grade_adjudicated", "official_source_span", "answer_rationale", "annotator_1", "annotator_2", "adjudicator", "judgment_status", ] def sha256_text(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() 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 parse_event_date(row: pd.Series) -> pd.Timestamp | None: for field in ("event_ngay_co_hieu_luc", "event_ngay_ban_hanh"): parsed = pd.to_datetime(row.get(field), format="%d/%m/%Y", errors="coerce") if not pd.isna(parsed): return parsed return None def _target_reference_spans(text: str, document_number: str) -> list[str]: """Return short source lines that explicitly name the target instrument.""" spans = [] for line in text.splitlines(): if contains_document_number(line, document_number): spans.append(line.strip()) return spans[:10] def _write_jsonl(path: Path, records: list[dict]) -> None: with path.open("w", encoding="utf-8") as handle: for record in records: handle.write(json.dumps(record, ensure_ascii=False) + "\n") def _load_official_prefill( registry_path: Path | None, audit_path: Path | None, ) -> dict[str, dict[str, str]]: """Join registry URLs to machine-fetched hashes without creating gold labels.""" if registry_path is None or audit_path is None: return {} if not registry_path.exists() or not audit_path.exists(): return {} registry = pd.read_csv(registry_path, dtype=str).fillna("") audit = pd.read_csv(audit_path, dtype=str).fillna("") required_registry = { "candidate_id", "official_event_page_url", "official_event_pdf_url", "official_affected_page_url", "official_affected_pdf_url", "relation_evidence_class", "expected_effective_date", } required_audit = { "candidate_id", "event_page_fetch", "event_page_sha256", "event_pdf_magic_valid", "event_pdf_sha256", "affected_page_fetch", "affected_page_sha256", "affected_pdf_magic_valid", "affected_pdf_sha256", "machine_verification_status", } missing_registry = required_registry.difference(registry.columns) missing_audit = required_audit.difference(audit.columns) if missing_registry: raise ValueError( f"official registry is missing columns: {sorted(missing_registry)}" ) if missing_audit: raise ValueError( f"official audit is missing columns: {sorted(missing_audit)}" ) if registry["candidate_id"].duplicated().any(): raise ValueError("official registry contains duplicate candidate_id values") if audit["candidate_id"].duplicated().any(): raise ValueError("official audit contains duplicate candidate_id values") joined = registry.merge( audit, on="candidate_id", how="inner", validate="one_to_one", suffixes=("_registry", "_audit"), ) prefill: dict[str, dict[str, str]] = {} for _, row in joined.iterrows(): event_pdf_valid = str(row["event_pdf_magic_valid"]).casefold() == "true" affected_pdf_valid = ( str(row["affected_pdf_magic_valid"]).casefold() == "true" ) event_page_valid = row["event_page_fetch"] == "fetched" affected_page_valid = row["affected_page_fetch"] == "fetched" prefill[str(row["candidate_id"])] = { "official_event_url": ( row["official_event_pdf_url"] if event_pdf_valid else row["official_event_page_url"] if event_page_valid else "" ), "official_event_artifact_sha256": ( row["event_pdf_sha256"] if event_pdf_valid else row["event_page_sha256"] if event_page_valid else "" ), "official_affected_url": ( row["official_affected_pdf_url"] if affected_pdf_valid else row["official_affected_page_url"] if affected_page_valid else "" ), "official_affected_artifact_sha256": ( row["affected_pdf_sha256"] if affected_pdf_valid else row["affected_page_sha256"] if affected_page_valid else "" ), "machine_verification_status": row["machine_verification_status"], "relation_evidence_class": row["relation_evidence_class_registry"], "expected_effective_date": row["expected_effective_date"], } return prefill def build(args: argparse.Namespace) -> dict: candidates = pd.read_csv(args.candidates, dtype=str).fillna("") required = { "candidate_id", "relation", "event_doc_id", "event_so_ky_hieu", "affected_doc_id", "affected_so_ky_hieu", "verification_status", } missing = required.difference(candidates.columns) if missing: raise ValueError(f"candidate manifest is missing columns: {sorted(missing)}") document_ids = sorted( set(candidates["event_doc_id"]) | set(candidates["affected_doc_id"]) ) contents = ( pq.read_table(args.content, filters=[("id", "in", document_ids)]) .to_pandas() .set_index("id")["content_html"] .to_dict() ) official_registry = getattr(args, "official_registry", None) official_audit = getattr(args, "official_audit", None) official_prefill = _load_official_prefill(official_registry, official_audit) audit_rows: list[dict] = [] provision_pairs: list[dict] = [] whole_document_events: list[dict] = [] verification_rows: list[dict] = [] annotation_rows: list[dict] = [] for _, row in candidates.iterrows(): source_prefill = official_prefill.get(row["candidate_id"], {}) event_group_id = f"event:{row['event_doc_id']}" event_html = str(contents.get(row["event_doc_id"], "")) affected_html = str(contents.get(row["affected_doc_id"], "")) event_text = html_to_legal_text(event_html) affected_text = html_to_legal_text(affected_html) event_articles = extract_articles(event_html) affected_articles = extract_articles(affected_html) affected_by_number: dict[str, list] = {} for article in affected_articles: affected_by_number.setdefault(article.article_number.casefold(), []).append( article ) event_by_number: dict[str, list] = {} for article in event_articles: event_by_number.setdefault(article.article_number.casefold(), []).append( article ) target_reference_found = contains_document_number( event_text, row["affected_so_ky_hieu"] ) mentions = extract_event_mentions( event_html, row["affected_so_ky_hieu"] ) referenced_articles = [] matched_articles = [] unmatched_articles = [] for mention in mentions: for number in mention.target_article_numbers: if number not in referenced_articles: referenced_articles.append(number) if number.casefold() in affected_by_number: if number not in matched_articles: matched_articles.append(number) elif number not in unmatched_articles: unmatched_articles.append(number) rejection_reasons = [] if not target_reference_found: rejection_reasons.append("target_document_not_named_in_event_content") if len(event_text) < args.minimum_text_chars: rejection_reasons.append("event_content_too_short") if len(affected_text) < args.minimum_text_chars: rejection_reasons.append("affected_content_too_short") if not event_articles: rejection_reasons.append("event_articles_not_parseable") if not affected_articles: rejection_reasons.append("affected_articles_not_parseable") if row["relation"] == "AMENDS": if not mentions: rejection_reasons.append("no_amendment_clause_extracted") if not matched_articles: rejection_reasons.append("no_target_article_resolved") auto_status = ( "eligible_for_manual_verification" if not rejection_reasons else "rejected_by_automated_checks" ) audit_rows.append( { "candidate_id": row["candidate_id"], "relation": row["relation"], "year_band": row.get("year_band", ""), "event_doc_id": row["event_doc_id"], "affected_doc_id": row["affected_doc_id"], "target_reference_found": target_reference_found, "event_text_chars": len(event_text), "affected_text_chars": len(affected_text), "event_article_count": len(event_articles), "affected_article_count": len(affected_articles), "event_mention_count": len(mentions), "referenced_target_articles": " | ".join(referenced_articles), "resolved_target_articles": " | ".join(matched_articles), "unresolved_target_articles": " | ".join(unmatched_articles), "event_content_sha256": sha256_text(event_html), "affected_content_sha256": sha256_text(affected_html), "auto_status": auto_status, "rejection_reasons": " | ".join(rejection_reasons), } ) verification_rows.append( { "protocol_version": ANNOTATION_PROTOCOL_VERSION, "event_group_id": event_group_id, "candidate_id": row["candidate_id"], "relation_discovery": row["relation"], "event_so_ky_hieu": row["event_so_ky_hieu"], "affected_so_ky_hieu": row["affected_so_ky_hieu"], "event_source_url_discovery": row.get("event_source_url", ""), "affected_source_url_discovery": row.get( "affected_source_url", "" ), "auto_status": auto_status, "machine_verification_status": source_prefill.get( "machine_verification_status", "not_machine_audited" ), "relation_evidence_class": source_prefill.get( "relation_evidence_class", "unregistered" ), "expected_effective_date_official": source_prefill.get( "expected_effective_date", "" ), "official_event_url": source_prefill.get( "official_event_url", "" ), "official_affected_url": source_prefill.get( "official_affected_url", "" ), "official_checked_at": "", "official_event_artifact_sha256": source_prefill.get( "official_event_artifact_sha256", "" ), "official_affected_artifact_sha256": source_prefill.get( "official_affected_artifact_sha256", "" ), "relation_class_annotator_1": "", "relation_class_annotator_2": "", "relation_class_adjudicated": "", "effect_scope_annotator_1": "", "effect_scope_annotator_2": "", "effect_scope_adjudicated": "", "valid_from_annotator_1": "", "valid_from_annotator_2": "", "valid_to_annotator_1": "", "valid_to_annotator_2": "", "valid_from_adjudicated": "", "valid_to_adjudicated": "", "effective_date_basis_annotator_1": "", "effective_date_basis_annotator_2": "", "effective_date_basis": "", "intermediate_instruments_annotator_1": "", "intermediate_instruments_annotator_2": "", "intermediate_instruments": "", "target_articles_annotator_1": "", "target_articles_annotator_2": "", "target_articles_adjudicated": "", "provision_mapping_annotator_1": "", "provision_mapping_annotator_2": "", "rationale_annotator_1": "", "rationale_annotator_2": "", "reviewed_at_annotator_1": "", "reviewed_at_annotator_2": "", "provision_mapping_adjudicated": "", "annotator_1": "", "annotator_2": "", "adjudicator": "", "adjudication_rationale": "", "verification_status": "pending_manual_verification", } ) if auto_status != "eligible_for_manual_verification": continue official_effective_date = source_prefill.get("expected_effective_date", "") event_date = ( pd.Timestamp(official_effective_date) if official_effective_date else parse_event_date(row) ) event_date_iso = event_date.date().isoformat() if event_date is not None else "" pre_date_iso = ( (event_date.date() - timedelta(days=1)).isoformat() if event_date is not None else "" ) if row["relation"] in {"REPLACES", "REPEALS"}: pair_id = f"{row['candidate_id']}:document" whole_document_events.append( { "schema_version": 1, "pair_id": pair_id, "candidate_id": row["candidate_id"], "discovery_relation": row["relation"], "event_effective_date": event_date_iso, "pre_snapshot_date": pre_date_iso, "post_snapshot_date": event_date_iso, "event_document_id": row["event_doc_id"], "event_document_number": row["event_so_ky_hieu"], "event_document_title": row.get("event_title", ""), "affected_document_id": row["affected_doc_id"], "affected_document_number": row["affected_so_ky_hieu"], "affected_document_title": row.get("affected_title", ""), "event_source_url_discovery": row.get("event_source_url", ""), "affected_source_url_discovery": row.get( "affected_source_url", "" ), "event_content_sha256": sha256_text(event_html), "affected_content_sha256": sha256_text(affected_html), "event_article_count": len(event_articles), "affected_article_count": len(affected_articles), "target_reference_spans": _target_reference_spans( event_text, row["affected_so_ky_hieu"] ), "post_semantics_candidate": ( "replacement_instrument_requires_provision_mapping" if row["relation"] == "REPLACES" else "repealed_without_successor_requires_abstention_review" ), "label_status": "automated_candidate_not_gold", } ) annotation_rows.append( { "protocol_version": ANNOTATION_PROTOCOL_VERSION, "event_group_id": event_group_id, "query_group_id": f"qg:{pair_id}", "pair_id": pair_id, "candidate_id": row["candidate_id"], "event_scope_candidate": "whole_document", "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": pre_date_iso, "post_snapshot_date": event_date_iso, "expected_answer_pre": "", "expected_answer_post": "", "answer_source_spans": "", "pre_relevant_provision_ids": "", "post_relevant_provision_ids": "", "pre_status": "", "post_status": "", "question_quality_annotator_1": "", "question_quality_annotator_2": "", "question_quality_adjudicated": "", "rationale_annotator_1": "", "rationale_annotator_2": "", "reviewed_at_annotator_1": "", "reviewed_at_annotator_2": "", "annotator_1": "", "annotator_2": "", "adjudicator": "", "query_author": "", "adjudication_rationale": "", "annotation_status": "not_started", } ) for mention_index, mention in enumerate(mentions, start=1): source_articles = event_by_number.get( mention.source_article_number.casefold(), [] ) if len(source_articles) != 1: continue source_article = source_articles[0] for target_number in mention.target_article_numbers: target_articles = affected_by_number.get(target_number.casefold(), []) if len(target_articles) != 1: continue target_article = target_articles[0] pair_id = ( f"{row['candidate_id']}:s{mention.source_article_number}:" f"t{target_number}:m{mention_index}" ) record = { "schema_version": 1, "pair_id": pair_id, "candidate_id": row["candidate_id"], "discovery_relation": row["relation"], "extracted_action": mention.action, "event_effective_date": event_date_iso, "pre_snapshot_date": pre_date_iso, "post_snapshot_date": event_date_iso, "event_document_id": row["event_doc_id"], "event_document_number": row["event_so_ky_hieu"], "event_source_article_number": mention.source_article_number, "event_source_article_text": source_article.text, "affected_document_id": row["affected_doc_id"], "affected_document_number": row["affected_so_ky_hieu"], "affected_article_number": target_number, "old_provision_text": target_article.text, "extraction_evidence": mention.evidence_text, "event_source_url_discovery": row.get("event_source_url", ""), "affected_source_url_discovery": row.get( "affected_source_url", "" ), "event_content_sha256": sha256_text(event_html), "affected_content_sha256": sha256_text(affected_html), "label_status": "automated_candidate_not_gold", } provision_pairs.append(record) annotation_rows.append( { "protocol_version": ANNOTATION_PROTOCOL_VERSION, "event_group_id": event_group_id, "query_group_id": f"qg:{pair_id}", "pair_id": pair_id, "candidate_id": row["candidate_id"], "event_scope_candidate": "provision", "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": pre_date_iso, "post_snapshot_date": event_date_iso, "expected_answer_pre": "", "expected_answer_post": "", "answer_source_spans": "", "pre_relevant_provision_ids": "", "post_relevant_provision_ids": "", "pre_status": "", "post_status": "", "question_quality_annotator_1": "", "question_quality_annotator_2": "", "question_quality_adjudicated": "", "rationale_annotator_1": "", "rationale_annotator_2": "", "reviewed_at_annotator_1": "", "reviewed_at_annotator_2": "", "annotator_1": "", "annotator_2": "", "adjudicator": "", "query_author": "", "adjudication_rationale": "", "annotation_status": "not_started", } ) args.output_dir.mkdir(parents=True, exist_ok=True) audit = pd.DataFrame(audit_rows) verification = pd.DataFrame(verification_rows) annotation = pd.DataFrame(annotation_rows) qrel_template = pd.DataFrame(columns=QREL_TEMPLATE_COLUMNS) audit_path = args.output_dir / "candidate_content_audit.csv" verification_path = args.output_dir / "official_verification_template.csv" annotation_path = args.output_dir / "expert_annotation_template.csv" qrel_path = args.output_dir / "qrel_judgment_template.csv" pairs_path = args.output_dir / "provision_event_pairs.jsonl" whole_document_path = args.output_dir / "whole_document_events.jsonl" audit.to_csv(audit_path, index=False) verification.to_csv(verification_path, index=False) annotation.to_csv(annotation_path, index=False) qrel_template.to_csv(qrel_path, index=False) _write_jsonl(pairs_path, provision_pairs) _write_jsonl(whole_document_path, whole_document_events) eligible = audit["auto_status"].eq("eligible_for_manual_verification") manifest = { "schema_version": 1, "artifact_type": "temporal_pilot_annotation_input", "annotation_protocol_version": ANNOTATION_PROTOCOL_VERSION, "source_candidates": str(args.candidates), "source_content_snapshot": str(args.content), "source_content_file_sha256": sha256_file(args.content), "official_registry": str(official_registry) if official_registry else "", "official_audit": str(official_audit) if official_audit else "", "candidate_count": int(len(audit)), "eligible_for_manual_verification": int(eligible.sum()), "rejected_by_automated_checks": int((~eligible).sum()), "eligible_relation_counts": dict( Counter(audit.loc[eligible, "relation"].astype(str)) ), "rejection_reason_counts": dict( Counter( reason for reasons in audit.loc[~eligible, "rejection_reasons"] for reason in str(reasons).split(" | ") if reason ) ), "provision_event_pair_count": len(provision_pairs), "whole_document_event_count": len(whole_document_events), "annotation_group_count": len(annotation_rows), "official_source_prefill": { "registered_candidate_count": len(official_prefill), "event_artifact_count": int( verification["official_event_artifact_sha256"].ne("").sum() ), "affected_artifact_count": int( verification["official_affected_artifact_sha256"].ne("").sum() ), "complete_pair_count": int( ( verification["official_event_artifact_sha256"].ne("") & verification["official_affected_artifact_sha256"].ne("") ).sum() ), "warning": ( "Prefilled URLs and hashes prove fetched-artifact provenance only; " "all legal judgments and verification_status remain blank/pending." ), }, "semantic_status": ( "All records are discovery/annotation inputs. No relation, legal status, " "relevance judgment, or official-source verification is gold yet." ), "files": { "candidate_content_audit": str(audit_path), "official_verification_template": str(verification_path), "expert_annotation_template": str(annotation_path), "qrel_judgment_template": str(qrel_path), "provision_event_pairs": str(pairs_path), "whole_document_events": str(whole_document_path), }, } manifest_path = args.output_dir / "manifest.json" manifest_path.write_text( json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) receipt = { "schema_version": 1, "artifact_type": "temporal_pilot_handoff_receipt", "generated_at": str(pd.Timestamp.now(tz="UTC").date()), "annotation_protocol_version": ANNOTATION_PROTOCOL_VERSION, "source_hashes": { "discovery_candidates_sha256": sha256_file(args.candidates), "content_snapshot_sha256": sha256_file(args.content), "official_registry_sha256": sha256_file(official_registry) if official_registry and official_registry.exists() else "", "official_audit_sha256": sha256_file(official_audit) if official_audit and official_audit.exists() else "", }, "candidate_count": manifest["candidate_count"], "eligible_for_manual_verification": manifest[ "eligible_for_manual_verification" ], "eligible_relation_counts": manifest["eligible_relation_counts"], "provision_event_pair_count": manifest["provision_event_pair_count"], "whole_document_event_count": manifest["whole_document_event_count"], "annotation_group_count": manifest["annotation_group_count"], "official_source_prefill": manifest["official_source_prefill"], "handoff_file_sha256": { "candidate_content_audit": sha256_file(audit_path), "official_verification_template": sha256_file(verification_path), "expert_annotation_template": sha256_file(annotation_path), "qrel_judgment_template": sha256_file(qrel_path), "provision_event_pairs": sha256_file(pairs_path), "whole_document_events": sha256_file(whole_document_path), }, "gold_rows": 0, "learned_residual_authorized": False, "warning": ( "This receipt inventories non-gold annotation inputs. It does not " "promote machine-confirmed provenance to legal or relevance labels." ), } 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 manifest def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument( "--candidates", type=Path, default=REPO_ROOT / "tmp/telen/bootstrap-audit/discovery_candidates.csv", ) parser.add_argument( "--content", type=Path, default=( REPO_ROOT / "tmp/deep-research/telen/vn-legal-data/data/content.parquet" ), ) parser.add_argument( "--output-dir", type=Path, default=REPO_ROOT / "tmp/telen/temporal-pilot", ) parser.add_argument( "--official-registry", type=Path, default=( REPO_ROOT / "research/data/telen_official_source_registry_2026-08-29.csv" ), ) parser.add_argument( "--official-audit", type=Path, default=( REPO_ROOT / "tmp/telen/official-source-audit/verification_audit.csv" ), ) parser.add_argument( "--receipt", type=Path, default=( REPO_ROOT / "research/results/telen_temporal_pilot_handoff_2026-08-29.json" ), ) parser.add_argument("--minimum-text-chars", type=int, default=500) return parser.parse_args() def main() -> None: args = parse_args() if args.minimum_text_chars < 1: raise ValueError("--minimum-text-chars must be positive") manifest = build(args) print(json.dumps(manifest, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()