scripts/45_validate_eval_sets.py

"""Validate development/gold separation and menu labels."""

from __future__ import annotations

import argparse
import json
import sys
from collections import Counter, defaultdict
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent))

from config import DATA_DIR, RAW_DIR
from core.query_data import normalize_query_text


DEV_PATH = DATA_DIR / "eval_queries_dev.json"
GOLD_PATH = DATA_DIR / "eval_queries_gold.json"
NO_MATCH_DEV_PATH = DATA_DIR / "eval_queries_no_match_dev.json"
QUERY_POOL_PATH = DATA_DIR / "query_index_pool.json"


def load_json(path: Path) -> list:
    with path.open("r", encoding="utf-8") as f:
        return json.load(f)


def answer_ids(item: dict) -> set[str]:
    values = item.get("expected_ids")
    if values is None and item.get("menu_id"):
        values = [item["menu_id"]]
    return {str(value).strip() for value in (values or []) if str(value).strip()}


def validate_rows(name: str, rows: list, valid_menu_ids: set[str]) -> list[str]:
    errors = []
    labels_by_query: dict[str, set[tuple[str, ...]]] = defaultdict(set)
    seen_rows = set()

    for index, item in enumerate(rows, 1):
        query = normalize_query_text(item.get("query", ""))
        ids = answer_ids(item)
        no_match = bool(item.get("no_match", False))
        if not query:
            errors.append(f"{name}[{index}]: empty query")
            continue
        if no_match and ids:
            errors.append(f"{name}[{index}]: no_match row has expected_ids")
        if not no_match and not ids:
            errors.append(f"{name}[{index}]: positive row has no expected_ids")
        unknown = ids - valid_menu_ids
        if unknown:
            errors.append(f"{name}[{index}]: unknown menu IDs {sorted(unknown)}")

        label = ("__NO_MATCH__",) if no_match else tuple(sorted(ids))
        labels_by_query[query].add(label)
        row_key = (query, label)
        if row_key in seen_rows:
            errors.append(f"{name}[{index}]: duplicate row '{query}'")
        seen_rows.add(row_key)

    for query, labels in labels_by_query.items():
        if len(labels) > 1:
            errors.append(
                f"{name}: conflicting labels for '{query}' -> {sorted(labels)}"
            )
    return errors


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--require-gold", action="store_true")
    args = parser.parse_args()

    menus = load_json(RAW_DIR / "real_menus.json")
    valid_menu_ids = {str(menu["menu_id"]) for menu in menus}
    dev = load_json(DEV_PATH)
    errors = validate_rows("dev", dev, valid_menu_ids)

    print(f"dev: {len(dev)} rows")
    print(f"dev sources: {dict(Counter(x.get('source', 'unknown') for x in dev))}")

    dev_queries = {normalize_query_text(x.get("query", "")) for x in dev}
    if NO_MATCH_DEV_PATH.exists():
        no_match_dev = load_json(NO_MATCH_DEV_PATH)
        errors.extend(validate_rows("no-match-dev", no_match_dev, valid_menu_ids))
        invalid_rows = [
            index
            for index, item in enumerate(no_match_dev, 1)
            if not item.get("no_match", False) or answer_ids(item)
        ]
        if invalid_rows:
            errors.append(
                "no-match-dev contains positive rows at indexes "
                f"{invalid_rows}"
            )
        no_match_queries = {
            normalize_query_text(x.get("query", "")) for x in no_match_dev
        }
        overlap = (dev_queries & no_match_queries) - {""}
        if overlap:
            errors.append(f"dev/no-match-dev query overlap: {len(overlap)}")
        print(f"no-match-dev: {len(no_match_dev)} rows")
    else:
        print(f"no-match-dev: not created ({NO_MATCH_DEV_PATH.name})")

    if GOLD_PATH.exists():
        gold = load_json(GOLD_PATH)
        errors.extend(validate_rows("gold", gold, valid_menu_ids))
        gold_queries = {normalize_query_text(x.get("query", "")) for x in gold}
        overlap = (dev_queries & gold_queries) - {""}
        if overlap:
            errors.append(f"dev/gold query overlap: {len(overlap)}")
        if NO_MATCH_DEV_PATH.exists():
            overlap = (no_match_queries & gold_queries) - {""}
            if overlap:
                errors.append(f"no-match-dev/gold query overlap: {len(overlap)}")

        if QUERY_POOL_PATH.exists():
            pool_queries = {
                normalize_query_text(x.get("query", ""))
                for x in load_json(QUERY_POOL_PATH)
            }
            leaked = (gold_queries & pool_queries) - {""}
            if leaked:
                errors.append(f"gold/query-index pool overlap: {len(leaked)}")

        llm_rows = sum(
            str(item.get("source", "")).lower() == "llm"
            for item in gold
        )
        if llm_rows:
            errors.append(f"gold contains LLM-authored rows: {llm_rows}")
        print(f"gold: {len(gold)} rows")
        print(
            "gold sources: "
            f"{dict(Counter(x.get('source', 'unknown') for x in gold))}"
        )
    elif args.require_gold:
        errors.append(f"gold set missing: {GOLD_PATH}")
    else:
        print(f"gold: not created yet ({GOLD_PATH.name})")

    if errors:
        print("\nVALIDATION FAILED")
        for error in errors:
            print(f"- {error}")
        raise SystemExit(1)
    print("\nVALIDATION PASSED")


if __name__ == "__main__":
    main()