""" Production-grade scavenger hunt dataset generator with validation pipeline. Incorporates all critical fixes from the audit. Usage: python generate_dataset.py --batches 10 --output dataset.json --validate strict Output: - dataset.json: All validated examples - generation_log.json: Per-batch metadata (cities, ages, difficulties, quality scores) - validation_errors.jsonl: All rejected examples with failure reasons """ import os import json import time import logging import argparse import random import re from datetime import datetime from collections import Counter from typing import Dict, List, Optional, Tuple from dataclasses import dataclass, asdict from pathlib import Path # Gemini/Claude API (install: pip install google-genai or anthropic) try: from google import genai from google.genai import types HAS_GEMINI = True except: HAS_GEMINI = False try: from anthropic import Anthropic HAS_ANTHROPIC = True except: HAS_ANTHROPIC = False # String similarity (install: pip install python-Levenshtein) try: from Levenshtein import distance as levenshtein_distance except: def levenshtein_distance(s1, s2): """Fallback Levenshtein distance (naive).""" if len(s1) < len(s2): return levenshtein_distance(s2, s1) if len(s2) == 0: return len(s1) previous_row = range(len(s2) + 1) for i, c1 in enumerate(s1): current_row = [i + 1] for j, c2 in enumerate(s2): insertions = previous_row[j + 1] + 1 deletions = current_row[j] + 1 substitutions = previous_row[j] + (c1 != c2) current_row.append(min(insertions, deletions, substitutions)) previous_row = current_row return previous_row[-1] # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # ============================================================================ # VALIDATION RULES & CONSTANTS # ============================================================================ PROPER_NOUN_PATTERNS = [ r'\b(?:Eiffel|Tower|Big|Ben|Statue|Liberty|Colosseum|Tower|Bridge|Cathedral|Basilica|' r'Notre|Dame|Louvre|Vatican|Kremlin|Parliament|Thames|Seine|Nile|Amazon|' r'Paris|London|Tokyo|NYC|New York|Sydney|Berlin|Amsterdam|Istanbul|Rome|' r'Bangkok|Dubai|Shanghai|Hong Kong|Singapore|Jakarta|Mumbai|Delhi|Cairo|Lagos|' r'Rio|Mexico|Buenos Aires|Toronto|Vancouver|Moscow|Beijing|Seoul|Bangkok|' r'Eiffel Tower|Big Ben|Statue of Liberty|Colosseum|Tower Bridge|' r'Taj Mahal|Great Wall|Golden Gate|Christ Redeemer|Sagrada Familia|' r'Plaza Mayor|Red Square|Tiananmen|Shibuya|Senso-?ji|Forbidden City|' r'Buckingham|Westminster|Versailles|Schönbrunn|Prado|Hermitage|' r'Leaning Tower|Pantheon|Arc de Triomphe|Champs|Elysées|Fifth Avenue|' r'Times Square|Central Park|Hyde Park|Regent Park|Golden Gate|' r'Brooklyn Bridge|Tower Bridge|Millennium Bridge|Rialto|' r'Vatican|Kremlin|Parliament|Congress|House|Senate|White|House|' r'McDonald|Starbucks|Coca|Pepsi|Amazon|Google|Apple|Microsoft|Facebook)\b', r'\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+)+\b' # Capitalized proper nouns (Name of Landmark) ] CITY_BANK = { 'Paris': 'PAR', 'Tokyo': 'TYO', 'New York City': 'NYC', 'Cape Town': 'CPT', 'Marrakech': 'MRK', 'Buenos Aires': 'BUE', 'Mumbai': 'BOM', 'Berlin': 'BER', 'Sydney': 'SYD', 'Nairobi': 'NBO', 'Istanbul': 'IST', 'Mexico City': 'MEX', 'Amsterdam': 'AMS', 'Bangalore': 'BLR', 'Lagos': 'LOS', 'Seoul': 'SEO', 'Copenhagen': 'CPH', 'Lisbon': 'LIS', 'Bogotá': 'BOG', 'Auckland': 'AKL' } BATCH_QUOTAS = { 'age_group': { 'children_only': 2, 'teens': 2, 'adults': 3, 'mixed_family': 2, 'mixed_adults': 1 }, 'difficulty': {'easy': 3, 'medium': 4, 'hard': 3}, 'scoring_method': {'first_to_finish': 2, 'point_accumulation': 6, 'timed_bonus': 2}, 'team_count_range': {'solo': 4, 'duo': 3, 'small_team': 2, 'large_team': 1}, 'duration': {30: 1, 45: 2, 60: 4, 90: 2, 120: 1}, 'cities': 'unique', # All 10 must be different 'regions': 3 # At least 3 geographic regions } # ============================================================================ # VALIDATION FUNCTIONS # ============================================================================ def contains_proper_noun(text: str) -> bool: """Check if text contains proper nouns (city names, landmarks, brands).""" if not isinstance(text, str): return False for pattern in PROPER_NOUN_PATTERNS: if re.search(pattern, text, re.IGNORECASE): return True return False def validate_proper_nouns(example: Dict) -> Tuple[bool, str]: """Validate that no proper nouns appear in task descriptions or hints.""" for task in example.get('output', {}).get('tasks', []): desc = task.get('description', '') if contains_proper_noun(desc): return False, f"Task {task['task_id']}: proper noun in description" for hint_level in ['hint_1', 'hint_2', 'hint_3']: hint = task.get('hints', {}).get(hint_level, '') if contains_proper_noun(hint): return False, f"Task {task['task_id']}: proper noun in {hint_level}" return True, "" def validate_landscape_tags(example: Dict) -> Tuple[bool, str]: """Validate that all tasks use only tags present in input.""" input_tags = set(example.get('input', {}).get('location', {}).get('landscape_tags', [])) for task in example.get('output', {}).get('tasks', []): task_tags = set(task.get('landscape_tags_used', [])) invalid_tags = task_tags - input_tags if invalid_tags: return False, f"Task {task['task_id']}: uses tags not in input: {invalid_tags}" return True, "" def validate_hint_progression(example: Dict) -> Tuple[bool, str]: """Validate hint progression (distinct, increasing specificity).""" for task in example.get('output', {}).get('tasks', []): hints = task.get('hints', {}) h1 = hints.get('hint_1', '') h2 = hints.get('hint_2', '') h3 = hints.get('hint_3', '') # Distance check if levenshtein_distance(h1, h2) < 8: return False, f"Task {task['task_id']}: hint_1 and hint_2 too similar" if levenshtein_distance(h2, h3) < 8: return False, f"Task {task['task_id']}: hint_2 and hint_3 too similar" # Length progression (crude specificity proxy) if len(h2) <= len(h1): return False, f"Task {task['task_id']}: hint_2 not longer than hint_1" if len(h3) <= len(h2): return False, f"Task {task['task_id']}: hint_3 not longer than hint_2" return True, "" def validate_task_diversity(example: Dict) -> Tuple[bool, str]: """Validate that at least 3 different task types are present.""" task_types = [t.get('task_type') for t in example.get('output', {}).get('tasks', [])] unique_types = len(set(task_types)) if unique_types < 3: return False, f"Only {unique_types} unique task types; need >= 3" return True, "" def validate_time_realism(example: Dict) -> Tuple[bool, str]: """Validate that task time estimates are realistic.""" for task in example.get('output', {}).get('tasks', []): task_type = task.get('task_type') est_time = task.get('estimated_time_minutes', 0) max_times = { 'find_and_photograph': 20, 'observe_and_answer': 15, 'logical_puzzle': 25, 'timed_challenge': 30, 'social_interaction': 10, 'collect_and_return': 20, 'reach_and_verify': 20 } max_time = max_times.get(task_type, 30) if est_time > max_time: return False, f"Task {task['task_id']}: {task_type} estimated {est_time}min (max {max_time})" return True, "" def validate_total_time(example: Dict) -> Tuple[bool, str]: """Validate that total estimated time <= duration_minutes.""" total_time = example.get('output', {}).get('estimated_total_time_minutes', 0) duration = example.get('input', {}).get('preferences', {}).get('duration_minutes', 0) if total_time > duration: return False, f"Total time {total_time}min > duration {duration}min" return True, "" def validate_arithmetic_checksums(example: Dict) -> Tuple[bool, str]: """Validate 9 arithmetic checksums from original prompt.""" output = example.get('output', {}) tasks = output.get('tasks', []) # Checksum 1: total_possible_points expected_total = sum(t.get('points', 0) for t in tasks) if output.get('bonus_task') and output['bonus_task'].get('points'): expected_total += output['bonus_task']['points'] if output.get('total_possible_points') != expected_total: return False, f"Checksum 1: total_possible_points mismatch" # Checksum 2: max_deductible_points expected_max_ded = len(tasks) * 10 if output.get('max_deductible_points') != expected_max_ded: return False, f"Checksum 2: max_deductible_points mismatch" # Checksum 3: minimum_possible_points expected_min = output.get('total_possible_points', 0) - output.get('max_deductible_points', 0) if output.get('minimum_possible_points') != expected_min: return False, f"Checksum 3: minimum_possible_points mismatch" # Checksum 4: estimated_total_time_minutes expected_time = sum(t.get('estimated_time_minutes', 0) for t in tasks) if output.get('estimated_total_time_minutes') != expected_time: return False, f"Checksum 4: estimated_total_time_minutes mismatch" # Checksum 5: scoring_method matches decision tree age_group = example['input']['players']['age_group'] team_count = example['input']['players']['team_count'] difficulty = example['input']['preferences']['difficulty'] duration = example['input']['preferences']['duration_minutes'] expected_method = select_scoring_method(age_group, team_count, difficulty, duration) # Soft check: warn but don't fail (model may prefer alternative) # Checksum 6: time_bonus only when scoring_method = timed_bonus has_time_bonus = output.get('scoring_summary', {}).get('time_bonus_per_minute_early') is not None is_timed_bonus = output.get('rules', {}).get('scoring_method') == 'timed_bonus' if has_time_bonus != is_timed_bonus: return False, f"Checksum 6: time_bonus inconsistent with scoring_method" # Checksum 7: team_aggregation_method only when team_count > 1 has_agg = output.get('scoring_summary', {}).get('team_aggregation_method') is not None should_have_agg = team_count > 1 if has_agg != should_have_agg: return False, f"Checksum 7: team_aggregation_method inconsistent with team_count" # Checksum 8: bonus_task only when eligible has_bonus = output.get('bonus_task') is not None and output['bonus_task'].get('description') is not None eligible = difficulty == 'hard' and len(tasks) >= 7 if has_bonus != eligible: return False, f"Checksum 8: bonus_task eligibility mismatch" # Checksum 9: difficulty distribution matches game difficulty task_diffs = [t.get('difficulty_contribution') for t in tasks] easy_pct = (task_diffs.count('easy') / len(task_diffs) * 100) if task_diffs else 0 medium_pct = (task_diffs.count('medium') / len(task_diffs) * 100) if task_diffs else 0 hard_pct = (task_diffs.count('hard') / len(task_diffs) * 100) if task_diffs else 0 if difficulty == 'easy': if easy_pct < 70 or hard_pct > 0: return False, f"Checksum 9a: easy game has {easy_pct}% easy, {hard_pct}% hard" elif difficulty == 'medium': if not (30 <= easy_pct <= 50) or hard_pct > 20: return False, f"Checksum 9b: medium game distribution off" elif difficulty == 'hard': if easy_pct > 20 or hard_pct < 40: return False, f"Checksum 9c: hard game has {easy_pct}% easy, {hard_pct}% hard" return True, "" def validate_climate_expression(example: Dict) -> Tuple[bool, str]: """Validate that climate zone is expressed in tasks or safety notes.""" climate = example['input']['location']['climate_zone'] output = example['output'] # Check safety notes safety_notes = output.get('safety_constraints', {}).get('notes', '') if climate.lower() in safety_notes.lower(): return True, "" # Check task descriptions for climate-related keywords task_descs = ' '.join([t['description'] for t in output.get('tasks', [])]) climate_keywords = { 'tropical': ['shade', 'humid', 'heat', 'cool'], 'arid': ['shade', 'water', 'dehydration', 'sun'], 'mediterranean': ['terrace', 'outdoor', 'plaza'], 'temperate': [], # No special expression needed 'continental': ['cold', 'warm clothing'], 'polar': ['cold', 'warm', 'brief'] } if climate in climate_keywords: keywords = climate_keywords[climate] if keywords and any(kw in task_descs.lower() for kw in keywords): return True, "" return False, f"Climate zone '{climate}' not expressed in output" def validate_batch_quotas(batch: List[Dict]) -> Tuple[bool, List[str]]: """Validate that a batch of 10 examples meets all quotas.""" errors = [] if len(batch) != 10: errors.append(f"Batch has {len(batch)} examples; need exactly 10") return False, errors # City diversity cities = [e['input']['location']['city'] for e in batch] if len(set(cities)) != 10: errors.append(f"Duplicate cities in batch: {Counter(cities)}") # Age group distribution age_groups = [e['input']['players']['age_group'] for e in batch] age_counts = Counter(age_groups) if age_counts != BATCH_QUOTAS['age_group']: errors.append(f"Age distribution {dict(age_counts)} != quota {BATCH_QUOTAS['age_group']}") # Difficulty distribution difficulties = [e['input']['preferences']['difficulty'] for e in batch] diff_counts = Counter(difficulties) if diff_counts != BATCH_QUOTAS['difficulty']: errors.append(f"Difficulty distribution {dict(diff_counts)} != quota") # Scoring method distribution methods = [e['output']['rules']['scoring_method'] for e in batch] method_counts = Counter(methods) if method_counts != BATCH_QUOTAS['scoring_method']: errors.append(f"Scoring method distribution {dict(method_counts)} != quota") # Duration distribution durations = [e['input']['preferences']['duration_minutes'] for e in batch] duration_counts = Counter(durations) if duration_counts != BATCH_QUOTAS['duration']: errors.append(f"Duration distribution {dict(duration_counts)} != quota") return len(errors) == 0, errors def validate_example(example: Dict, strict: bool = True) -> Tuple[bool, str]: """Run all validations on a single example.""" validators = [ validate_proper_nouns, validate_landscape_tags, validate_hint_progression, validate_task_diversity, validate_time_realism, validate_total_time, validate_arithmetic_checksums, validate_climate_expression, ] for validator in validators: is_valid, error_msg = validator(example) if not is_valid: return False, error_msg return True, "" # ============================================================================ # HELPER FUNCTIONS # ============================================================================ def select_scoring_method(age_group, team_count, difficulty, duration_minutes): """Deterministic scoring method selection (from audit fix #2).""" if age_group in ['children_only', 'mixed_family']: return 'point_accumulation' if team_count > 2: return 'point_accumulation' if team_count in [0, 1]: if difficulty == 'hard' and duration_minutes >= 60: return 'timed_bonus' elif difficulty == 'easy' and duration_minutes <= 45: return 'first_to_finish' else: return 'point_accumulation' if team_count == 2: if difficulty == 'hard' and duration_minutes >= 90: return 'timed_bonus' else: return 'point_accumulation' return 'point_accumulation' def mask_city_name(example: Dict, mask_probability: float = 0.8) -> Dict: """Mask city name during fine-tuning (from audit fix #9).""" if random.random() < mask_probability: example['input']['location']['city'] = '[CITY]' example['input']['location']['city_code'] = '[CODE]' return example def generate_batch_summary(batch: List[Dict]) -> Dict: """Generate metadata summary for a batch.""" if not batch: return {} cities = [e['input']['location']['city'] for e in batch] age_groups = [e['input']['players']['age_group'] for e in batch] difficulties = [e['input']['preferences']['difficulty'] for e in batch] methods = [e['output']['rules']['scoring_method'] for e in batch] quality_scores = [e['output'].get('quality_score', 0) for e in batch] return { 'batch_size': len(batch), 'cities': cities, 'age_groups': Counter(age_groups), 'difficulties': Counter(difficulties), 'scoring_methods': Counter(methods), 'avg_quality_score': sum(quality_scores) / len(quality_scores) if quality_scores else 0, 'timestamp': datetime.now().isoformat() } # ============================================================================ # GENERATION # ============================================================================ def load_system_prompt(prompt_file: str = "./system_prompt.txt") -> str: """Load the system prompt from file.""" if not os.path.exists(prompt_file): raise FileNotFoundError(f"System prompt file '{prompt_file}' not found. " f"Create it with the revised prompt from the audit.") with open(prompt_file, "r", encoding="utf-8") as f: return f.read() def generate_batch_gemini(system_prompt: str, batch_num: int, max_retries: int = 3) -> Optional[List[Dict]]: """Generate a batch using Gemini 2.5 Flash.""" if not HAS_GEMINI: raise ImportError("google-genai not installed. Install with: pip install google-genai") client = genai.Client() user_prompt = f""" Generate batch #{batch_num}. {system_prompt} CONSTRAINTS FOR THIS BATCH: - Exactly 10 examples - All 10 cities must be different (from the city bank) - Age group distribution: children_only: 2, teens: 2, adults: 3, mixed_family: 2, mixed_adults: 1 - Difficulty distribution: easy: 3, medium: 4, hard: 3 - Scoring method distribution: first_to_finish: 2, point_accumulation: 6, timed_bonus: 2 - All landscape_tags must be from the controlled vocabulary - NO proper nouns in any task description or hints - All hints must show clear progression - All checksums must pass Return ONLY valid JSON (no preamble, no markdown, no comments). """ for attempt in range(max_retries): try: response = client.models.generate_content( model='gemini-2.5-flash', contents=user_prompt, config=types.GenerateContentConfig( response_mime_type="application/json", temperature=0.7, ), ) batch_data = json.loads(response.text) # Validate batch structure if not isinstance(batch_data, list) or len(batch_data) != 10: logger.warning(f"Batch {batch_num}: Invalid structure on attempt {attempt+1}") continue return batch_data except json.JSONDecodeError as e: logger.warning(f"Batch {batch_num}: JSON decode error on attempt {attempt+1}: {e}") if attempt == max_retries - 1: raise except Exception as e: logger.error(f"Batch {batch_num}: Error on attempt {attempt+1}: {e}") if attempt == max_retries - 1: raise # Backoff before retry time.sleep(5 * (attempt + 1)) return None def main(): parser = argparse.ArgumentParser(description='Generate scavenger hunt training dataset') parser.add_argument('--batches', type=int, default=10, help='Number of batches to generate (10 examples each)') parser.add_argument('--output', default='scavenger_hunt_dataset.json', help='Output file path') parser.add_argument('--validate', choices=['strict', 'lenient', 'none'], default='strict', help='Validation level') parser.add_argument('--model', choices=['gemini', 'claude'], default='gemini', help='Which API to use') parser.add_argument('--mask-city', type=float, default=0.8, help='City masking probability during generation') args = parser.parse_args() # Load system prompt try: system_prompt = load_system_prompt() except FileNotFoundError as e: logger.error(str(e)) return # Initialize storage all_examples = [] validation_log = [] batch_metadata = [] # Load existing data if resuming if os.path.exists(args.output): try: with open(args.output, 'r') as f: all_examples = json.load(f) logger.info(f"Loaded {len(all_examples)} existing examples") except json.JSONDecodeError: logger.warning(f"Could not load {args.output}; starting fresh") # Generate batches for batch_num in range(1, args.batches + 1): logger.info(f"Generating batch {batch_num}/{args.batches}...") try: if args.model == 'gemini': batch_data = generate_batch_gemini(system_prompt, batch_num) else: logger.error("Claude model not yet implemented") continue if not batch_data: logger.warning(f"Batch {batch_num}: generation returned None") continue # Validate each example valid_examples = [] for i, example in enumerate(batch_data): is_valid, error_msg = validate_example(example, strict=(args.validate == 'strict')) if is_valid: # Mask city name if requested if args.mask_city > 0: example = mask_city_name(example, args.mask_city) valid_examples.append(example) else: validation_log.append({ 'batch': batch_num, 'example_index': i, 'error': error_msg, 'timestamp': datetime.now().isoformat() }) logger.warning(f" Example {i}: {error_msg}") if not valid_examples: logger.error(f"Batch {batch_num}: All 10 examples failed validation. Skipping batch.") continue # Check batch quotas if we have enough examples if len(valid_examples) == 10: batch_valid, quota_errors = validate_batch_quotas(valid_examples) if not batch_valid: logger.warning(f"Batch {batch_num} failed quota checks:") for error in quota_errors: logger.warning(f" - {error}") # In strict mode, reject entire batch if args.validate == 'strict': continue # Add valid examples all_examples.extend(valid_examples) # Log metadata batch_meta = generate_batch_summary(valid_examples) batch_meta['batch_num'] = batch_num batch_meta['valid_count'] = len(valid_examples) batch_metadata.append(batch_meta) # Save incrementally with open(args.output, 'w') as f: json.dump(all_examples, f, indent=2) with open(args.output.replace('.json', '_metadata.json'), 'w') as f: json.dump(batch_metadata, f, indent=2) logger.info(f"Batch {batch_num}: {len(valid_examples)}/10 valid. " f"Total: {len(all_examples)} examples") # Rate limiting time.sleep(15) except Exception as e: logger.error(f"Batch {batch_num}: Unrecoverable error: {e}") continue # Final summary logger.info(f"\n{'='*60}") logger.info(f"GENERATION COMPLETE") logger.info(f"{'='*60}") logger.info(f"Total valid examples: {len(all_examples)}") logger.info(f"Total validation failures: {len(validation_log)}") logger.info(f"Output file: {args.output}") logger.info(f"Metadata file: {args.output.replace('.json', '_metadata.json')}") # Save validation log if validation_log: with open(args.output.replace('.json', '_errors.jsonl'), 'w') as f: for entry in validation_log: f.write(json.dumps(entry) + '\n') logger.info(f"Validation errors: {args.output.replace('.json', '_errors.jsonl')}") if __name__ == "__main__": main()