nucleic/ml/purpose-classifier/review_data.py

#!/usr/bin/env python3
"""Check or finalize the deterministic human label-and-difficulty review."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Sequence

from prepare_data import (
    DATASET_VERSION,
    DEFAULT_CURATION_REVIEW,
    DEFAULT_FIXTURES,
    default_source_paths,
    load_reviewed_semantic_exclusions,
)
from purpose_data import (
    DataError,
    SourceRecord,
    curate_records,
    exclude_reviewed_duplicates,
    load_classifiable_fixtures,
    load_sources,
)
from review_contract import (
    DEFAULT_SAMPLE_FRACTION,
    DEFAULT_SAMPLE_SEED,
    finalize_human_review,
    inspect_review_csv,
    stratified_review_sample,
    write_review_csv,
)


SCRIPT_DIR = Path(__file__).resolve().parent
REPOSITORY_ROOT = SCRIPT_DIR.parent.parent
DEFAULT_REVIEW_CSV = SCRIPT_DIR / ".artifacts" / "human-review-v1.csv"
DEFAULT_OUTPUT = SCRIPT_DIR / "data" / "human-review-v1.json"


def _relative(path: Path) -> str:
    try:
        return str(path.resolve().relative_to(REPOSITORY_ROOT))
    except ValueError:
        return str(path.resolve())


def reviewed_population(args: argparse.Namespace) -> list[SourceRecord]:
    sources = (
        [path.resolve() for path in args.source]
        if args.source
        else default_source_paths()
    )
    fixtures = load_classifiable_fixtures(args.fixtures.resolve())
    lexical = curate_records(
        load_sources(sources),
        fixtures,
        near_duplicate_threshold=args.near_duplicate_threshold,
    )
    return exclude_reviewed_duplicates(
        lexical.records,
        load_reviewed_semantic_exclusions(args.curation_review.resolve()),
    ).records


def review_policy_status(path: Path) -> str:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
        status = value["humanLabelAndDifficultyReview"]["status"]
    except (OSError, UnicodeError, json.JSONDecodeError, KeyError, TypeError) as exc:
        raise DataError(f"{path}: cannot read human-review policy: {exc}") from exc
    if not isinstance(status, str) or not status:
        raise DataError(f"{path}: invalid human-review policy status")
    return status


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--source", action="append", type=Path)
    parser.add_argument("--fixtures", type=Path, default=DEFAULT_FIXTURES)
    parser.add_argument(
        "--curation-review",
        type=Path,
        default=DEFAULT_CURATION_REVIEW,
        help="completed semantic duplicate review used to reconstruct the sample",
    )
    parser.add_argument("--review-csv", type=Path, default=DEFAULT_REVIEW_CSV)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    parser.add_argument("--review-fraction", type=float, default=DEFAULT_SAMPLE_FRACTION)
    parser.add_argument("--review-seed", type=int, default=DEFAULT_SAMPLE_SEED)
    parser.add_argument("--near-duplicate-threshold", type=float, default=0.92)
    action = parser.add_mutually_exclusive_group()
    action.add_argument(
        "--finalize",
        action="store_true",
        help="write the versionable review ledger; fails unless every row is complete",
    )
    action.add_argument(
        "--regenerate",
        action="store_true",
        help="replace the blank review CSV from the deterministic current sample",
    )
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        population = reviewed_population(args)
        policy_status = review_policy_status(args.curation_review.resolve())
        sample = stratified_review_sample(
            population,
            fraction=args.review_fraction,
            seed=args.review_seed,
        )
        if args.regenerate:
            if args.review_csv.exists():
                existing = inspect_review_csv(
                    args.review_csv.resolve(),
                    sample,
                    source_formatter=_relative,
                )
                if existing.completed:
                    raise DataError(
                        f"{args.review_csv}: refusing to replace "
                        f"{existing.completed} completed review rows"
                    )
            write_review_csv(
                args.review_csv.resolve(),
                sample,
                source_formatter=_relative,
            )
            print(
                f"Regenerated {len(sample)} blank human-review rows at "
                f"{_relative(args.review_csv)}."
            )
            return 0
        progress = inspect_review_csv(
            args.review_csv.resolve(),
            sample,
            source_formatter=_relative,
        )
        if args.finalize:
            artifact = finalize_human_review(
                args.review_csv.resolve(),
                args.output.resolve(),
                population,
                dataset_version=DATASET_VERSION,
                fraction=args.review_fraction,
                seed=args.review_seed,
                source_formatter=_relative,
            )
            print(
                f"Finalized {artifact['sampleRecords']} human-review decisions at "
                f"{_relative(args.output)}."
            )
            return 0
    except (DataError, OSError, UnicodeError, json.JSONDecodeError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1

    print(
        f"Human review: {progress.completed}/{progress.records} complete "
        f"({progress.accepted} accept, {progress.relabeled} relabel, "
        f"{progress.rejected} reject, {progress.incomplete} remaining). "
        f"Dataset policy: {policy_status}."
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())