""" validator.py — scavenger_hunt Validates one training example. Returns (is_valid: bool, errors: list[str]). auto_fix() recomputes arithmetic fields in-place — never reject for math the model shouldn't be trusted with. """ import os import re as _re import sys from Levenshtein import distance as lev HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.join(HERE, "..", "common")) from cq_common import ( VALID_TAGS, nouns_in as _nouns_in, soft_leaks_in, is_verbal_only_proof, has_risk_indicators, specific_flag_count, HINT_LIMITS, hint_spoilers_in, fix_text_leaks, ) POINTS = {"easy": 10, "medium": 20, "hard": 30} def validate_example(ex: dict) -> tuple[bool, list[str]]: errors = [] # ── structural guard ────────────────────────────────────────────────────── try: inp = ex["input"] out = ex["output"] loc = inp["location"] tasks = out["tasks"] except (KeyError, TypeError) as e: return False, [f"STRUCTURE: missing key {e}"] if not tasks: return False, ["STRUCTURE: tasks array is empty"] input_tags = set(loc.get("landscape_tags", [])) # ── vocabulary ──────────────────────────────────────────────────────────── bad = input_tags - VALID_TAGS if bad: errors.append(f"VOCAB: unknown tags {sorted(bad)}") # ── per-task ────────────────────────────────────────────────────────────── types_seen, total_pts, total_time, diff_counts = [], 0, 0, {"easy":0,"medium":0,"hard":0} for t in tasks: tid = t.get("task_id", "T??") # proper nouns in description found = _nouns_in(t.get("description","")) if found: errors.append(f"NOUN: {tid} description contains {set(found)}") # proper nouns in hints for hk in ("hint_1","hint_2","hint_3"): found = _nouns_in(t.get("hints",{}).get(hk,"")) if found: errors.append(f"NOUN: {tid} {hk} contains {set(found)}") # soft city-leakage words (audit §1) — description + hints leaks = soft_leaks_in(t.get("description","")) if leaks: errors.append(f"LEAK: {tid} description contains city-knowledge words {set(leaks)}") for hk in ("hint_1","hint_2","hint_3"): leaks = soft_leaks_in(t.get("hints",{}).get(hk,"")) if leaks: errors.append(f"LEAK: {tid} {hk} contains city-knowledge words {set(leaks)}") # hint length limits + anti-spoiler check (audit §4) for hk, limit in HINT_LIMITS.items(): h = t.get("hints",{}).get(hk,"") if len(h) > limit: errors.append(f"HINTLEN: {tid} {hk} is {len(h)} chars (max {limit})") spoilers = hint_spoilers_in(t.get("hints",{}).get("hint_3","")) if spoilers: errors.append(f"HINTSPOILER: {tid} hint_3 contains direction give-aways {spoilers}") # verifiable proof for social_interaction tasks (audit §2) if t.get("task_type") == "social_interaction" and is_verbal_only_proof(t.get("completion_proof","")): errors.append(f"PROOF: {tid} social_interaction completion_proof is verbal-only, needs photo/receipt") # safety flag specificity (audit §3) risky_text = " ".join([t.get("description",""), t.get("title","")]) if has_risk_indicators(risky_text): if specific_flag_count(t.get("safety_flags")) < 2: errors.append(f"SAFETY: {tid} touches water/elevation/alley/crowd/dark but has <2 specific safety_flags") # tags used must be subset of input tags used = set(t.get("landscape_tags_used",[])) extra = used - input_tags if extra: errors.append(f"TAGS: {tid} uses tags not in input {sorted(extra)}") # points must match difficulty d, p = t.get("difficulty_contribution"), t.get("points") if d in POINTS and p != POINTS[d]: errors.append(f"POINTS: {tid} is {d} but has {p} pts (want {POINTS[d]})") if d in diff_counts: diff_counts[d] += 1 # hint progression — distinct + hint_3 longer than hint_1 h1 = t.get("hints",{}).get("hint_1","") h2 = t.get("hints",{}).get("hint_2","") h3 = t.get("hints",{}).get("hint_3","") if h1 and h2 and lev(h1.lower(), h2.lower()) < 8: errors.append(f"HINTS: {tid} hint_1 ≈ hint_2 (too similar)") if h2 and h3 and lev(h2.lower(), h3.lower()) < 8: errors.append(f"HINTS: {tid} hint_2 ≈ hint_3 (too similar)") if h1 and h3 and len(h3) <= len(h1): errors.append(f"HINTS: {tid} hint_3 not more detailed than hint_1") types_seen.append(t.get("task_type")) total_pts += p if isinstance(p, int) else 0 total_time += t.get("estimated_time_minutes", 0) # ── task-type diversity ─────────────────────────────────────────────────── needed = 3 if len(tasks) >= 4 else 2 if len(set(types_seen)) < needed: errors.append(f"DIVERSITY: {len(set(types_seen))} task types, need {needed}") # ── time budget ─────────────────────────────────────────────────────────── duration = inp.get("preferences",{}).get("duration_minutes", 0) if total_time > duration: errors.append(f"TIME: tasks sum {total_time}min > duration {duration}min") # ── arithmetic checksums (informational — auto_fix handles these) ───────── bonus_pts = (out.get("bonus_task") or {}).get("points") or 0 declared = out.get("total_possible_points") expected = total_pts + bonus_pts if declared not in (expected, total_pts): errors.append(f"CHECKSUM: total_possible_points={declared}, computed={expected}") if out.get("task_count") != len(tasks): errors.append(f"CHECKSUM: task_count={out.get('task_count')} but {len(tasks)} tasks") # ── scoring safety (hard rules only) ───────────────────────────────────── age = inp.get("players",{}).get("age_group","") method = out.get("rules",{}).get("scoring_method","") tc = inp.get("players",{}).get("team_count", 1) if age in ("children_only","mixed_family") and method == "timed_bonus": errors.append(f"SCORING: {age} must not use timed_bonus") if tc > 2 and method == "first_to_finish": errors.append(f"SCORING: first_to_finish invalid with team_count={tc}") # time_bonus field must match scoring_method tb = out.get("scoring_summary",{}).get("time_bonus_per_minute_early") if (tb is not None) != (method == "timed_bonus"): errors.append(f"CHECKSUM: time_bonus_per_minute_early={tb} inconsistent with {method}") # ── difficulty mix ──────────────────────────────────────────────────────── game_diff = inp.get("preferences",{}).get("difficulty","") n = len(tasks) if game_diff == "easy" and diff_counts["hard"] > 0: errors.append(f"MIX: easy game has {diff_counts['hard']} hard tasks") if game_diff == "hard" and n >= 4: hard_pct = diff_counts["hard"] / n * 100 if hard_pct < 35: errors.append(f"MIX: hard game only {hard_pct:.0f}% hard tasks (need ≥35%)") return len(errors) == 0, errors _LEAK_REPLACEMENTS = { "most famous": "most distinctive", "the famous": "the notable", "best-known": "most noticeable", "best known": "most noticeable", "central": "the", "downtown": "this area", "largest": "biggest", "main": "the", } _SPOILER_PHRASES = [ "just before", "next to", "to the left of", "to the right of", "near the corner of", "across from", "adjacent to", "directly behind", "right after", "just past", ] def _strip_leaks(text: str) -> str: if not text: return text for phrase, repl in _LEAK_REPLACEMENTS.items(): text = _re.sub(r"\b" + _re.escape(phrase) + r"\b", repl, text, flags=_re.IGNORECASE) text = _re.sub(r"\biconic\b(?!\s+landmark)", "notable", text, flags=_re.IGNORECASE) text = _re.sub(r"\bthe the\b", "the", text, flags=_re.IGNORECASE) return _re.sub(r"\s{2,}", " ", text).strip() def _strip_spoilers(text: str) -> str: if not text: return text for phrase in _SPOILER_PHRASES: text = _re.sub(_re.escape(phrase), "near", text, flags=_re.IGNORECASE) return _re.sub(r"\s{2,}", " ", text).strip() def _truncate(text: str, limit: int) -> str: if not text or len(text) <= limit: return text cut = text[:limit] if " " in cut: cut = cut[: cut.rfind(" ")] return cut.rstrip(" ,.;:") def auto_fix(ex: dict) -> dict: """Recompute all arithmetic output fields in-place. Call before validate_example.""" try: loc = ex["input"]["location"] ex["output"]["game_title"] = fix_text_leaks(ex["output"].get("game_title", ""), [loc.get("city"), loc.get("country")]) tasks = ex["output"]["tasks"] for t in tasks: d = t.get("difficulty_contribution") if d in POINTS: t["points"] = POINTS[d] # strip soft city-leakage words from description (audit §1) t["description"] = _strip_leaks(t.get("description", "")) # clean + enforce hint limits / anti-spoiler (audit §4) hints = t.get("hints") if isinstance(hints, dict): for hk in ("hint_1", "hint_2", "hint_3"): if hk in hints and isinstance(hints[hk], str): hints[hk] = _strip_leaks(hints[hk]) if "hint_3" in hints and isinstance(hints["hint_3"], str): hints["hint_3"] = _strip_spoilers(hints["hint_3"]) for hk, limit in HINT_LIMITS.items(): if hk in hints and isinstance(hints[hk], str): hints[hk] = _truncate(hints[hk], limit) # ensure ≥2 specific safety_flags for risky tasks (audit §3) risky_text = " ".join([t.get("description", ""), t.get("title", "")]) if has_risk_indicators(risky_text): flags = t.get("safety_flags") or [] if specific_flag_count(flags) < 2: extra = [ "watch your footing — surfaces near here can be uneven or slippery", "stay alert for blind corners and limited visibility around this spot", ] for e in extra: if specific_flag_count(flags) >= 2: break flags.append(e) t["safety_flags"] = flags base = sum(t["points"] for t in tasks) bonus_pts = (ex["output"].get("bonus_task") or {}).get("points") or 0 ex["output"]["task_count"] = len(tasks) ex["output"]["total_possible_points"] = base + bonus_pts ex["output"]["max_deductible_points"] = len(tasks) * 10 ex["output"]["minimum_possible_points"] = max(0, base + bonus_pts - len(tasks) * 10) ex["output"]["estimated_total_time_minutes"] = sum( t.get("estimated_time_minutes", 0) for t in tasks) ex["output"]["scoring_summary"]["base_points_available"] = base + bonus_pts except (KeyError, TypeError): pass return ex