| |
|
|
| import os |
| import csv |
| import json |
| import logging |
| from datetime import datetime |
| from collections import OrderedDict |
| import hashlib |
|
|
| from config import ( |
| MISTRAL_API_KEY, SEARCH_API_KEY, CSE_ID, |
| OUTPUT_DIR, KNOWLEDGE_CACHE_DIR, APP_VERSION |
| ) |
| from schemas import ( |
| TRIATHLON_SCHEMA, RUNNING_SCHEMA, SWIMMING_SCHEMA, DUATHLON_SCHEMA, |
| AQUATHLON_SCHEMA, AQUABIKE_SCHEMA, CYCLING_SCHEMA, FITNESS_RACING_SCHEMA |
| ) |
| from agent import MistralAnalystAgent, AgentInitializationError |
| from utils import ( |
| serialize_knowledge_base, deserialize_knowledge_base, format_final_row, SafeCSVWriter |
| ) |
| from supabase_client import get_supabase_client |
|
|
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') |
| logger = logging.getLogger(__name__) |
|
|
| def save_output_to_supabase(filepath: str, agent_mode: str, event_type: str = None): |
| supabase = get_supabase_client() |
| if not supabase: return |
| try: |
| with open(filepath, 'r', encoding='utf-8') as f: |
| reader = csv.DictReader(f) |
| run_data = list(reader) |
| if not run_data: return |
| payload = { |
| "agent_mode": agent_mode, |
| "event_type": event_type, |
| "filename": os.path.basename(filepath), |
| "run_data": run_data, |
| "event_count": len(run_data) |
| } |
| supabase.table('run_outputs').insert(payload).execute() |
| logger.info(f"Successfully saved {os.path.basename(filepath)} to Supabase.") |
| except Exception as e: |
| logger.error(f"Failed to save output file to Supabase: {e}") |
|
|
| def process_default_website_data(data, website): |
| configs = { |
| 'indiarunning': {'key_rename_map': {"md:grid href": "link", "line-clamp-2": "festivalName", "font-bold": "registrationCost", "hidden": "city", "font-normal": "date"}}, |
| 'townscript': {'key_rename_map': {"ls-card href": "link", "font-semibold": "festivalName", "md:text-lg": "registrationCost", "whitespace-no-wrap (2)": "city", "whitespace-no-wrap": "date"}}, |
| 'bhaagoindia': {'key_rename_map': {"hover:bi-text-blue-600 href": "link", "hover:bi-text-blue-600": "festivalName", "bi-text-sm": "date", "bi-text-sm (2)": "city"}} |
| } |
| site_key = website |
| if 'bhaago' in website or 'bhago' in website: site_key = 'bhaagoindia' |
| if site_key not in configs: return [] |
| config = configs[site_key] |
| processed_list =[] |
| for item in data: |
| processed_data = {} |
| for old_key, new_key in config['key_rename_map'].items(): |
| if item.get(old_key): |
| val = item.get(old_key) |
| if isinstance(val, str): val = val.strip() |
| processed_data[new_key] = val |
| if processed_data.get("link"): processed_list.append(processed_data) |
| return processed_list |
|
|
| def get_website_processor(filename): |
| fn_lower = filename.lower() |
| if 'indiarunning' in fn_lower: return process_default_website_data, 'indiarunning' |
| elif 'townscript' in fn_lower: return process_default_website_data, 'townscript' |
| elif 'bhaago' in fn_lower or 'bhago' in fn_lower: return process_default_website_data, 'bhaagoindia' |
| return None, None |
|
|
| def load_and_process_files(input_files): |
| all_events =[] |
| for file_path in input_files: |
| logger.info(f"Loading file: {file_path}") |
| if not os.path.exists(file_path): continue |
| try: |
| with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) |
| except Exception: continue |
| processor_func, site_key = get_website_processor(os.path.basename(file_path)) |
| if processor_func: |
| processed = processor_func(data, site_key) if site_key else processor_func(data) |
| all_events.extend(processed) |
| return all_events |
|
|
| def _get_unique_events(events_data): |
| unique_events_map = OrderedDict() |
| for item in events_data: |
| link = item.get('link') |
| if not link: continue |
| name = item.get('festivalName', 'Unknown') |
| date = item.get('date', 'Unknown') |
| composite_key = f"{link}|{name}|{date}" |
| unique_events_map[composite_key] = item |
| return list(unique_events_map.values()) |
|
|
| def _infer_event_type_from_name(event_name): |
| if not event_name: return None |
| name = event_name.lower() |
| if any(x in name for x in['triathlon', 'ironman']): return 'Triathlon' |
| if any(x in name for x in['duathlon']): return 'Duathlon' |
| if any(x in name for x in['swim', 'aquatic']): return 'Swimming' |
| if any(x in name for x in['cyclothon', 'cycling', 'ride', 'tour de', 'pedal']): return 'Cycling' |
| if any(x in name for x in['run', 'marathon', '10k', '5k', 'walkathon', 'jog']): return 'Run' |
| return None |
|
|
| def run_default_websites_mode(input_files, output_dir_override=None, max_events=None, output_filename=None, enable_fallback=False): |
| print("=" * 60) |
| print(f"MODE: Default Websites Processor | Version: {APP_VERSION}") |
| print("=" * 60) |
| agent = None |
| failed_missions =[] |
| effective_output_dir = output_dir_override or OUTPUT_DIR |
| run_timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") |
| type_to_filepath = {} |
| try: |
| events_data = load_and_process_files(input_files) |
| event_list = _get_unique_events(events_data) |
| print(f"[SUCCESS] Extracted {len(event_list)} unique event instances.") |
| events_to_process = event_list[:max_events] if max_events else event_list |
| if not events_to_process: return |
| schema_map = {"triathlon": TRIATHLON_SCHEMA, "run": RUNNING_SCHEMA, "running": RUNNING_SCHEMA, "trail running": RUNNING_SCHEMA, "swimathon": SWIMMING_SCHEMA, "swimming": SWIMMING_SCHEMA, "duathlon": DUATHLON_SCHEMA, "aquathlon": AQUATHLON_SCHEMA, "aquabike": AQUABIKE_SCHEMA, "cycling": CYCLING_SCHEMA, "fitness racing": FITNESS_RACING_SCHEMA} |
| |
| agent = MistralAnalystAgent( |
| mistral_key=MISTRAL_API_KEY, |
| search_key=SEARCH_API_KEY, cse_id=CSE_ID, schema=[], |
| enable_fallback=enable_fallback |
| ) |
| |
| for i, event_info in enumerate(events_to_process): |
| if event_info.get("link") == "BLANK_ROW": continue |
| event_url = event_info.get("link") |
| if not event_url: continue |
| print("\n" + "=" * 60) |
| print(f"[STARTING MISSION] {i + 1}/{len(events_to_process)} FOR: {event_url}") |
| festival_name = event_info.get('festivalName') or event_url |
| event_type = _infer_event_type_from_name(festival_name) |
| if not event_type: event_type = agent.determine_event_type(event_url) |
| if not event_type: |
| failed_missions.append(f"{event_url} (Type failed)") |
| continue |
| if event_type.lower() == 'running': event_type = 'Run' |
| schema = schema_map.get(event_type.lower()) |
| if not schema: continue |
| if event_type not in type_to_filepath: |
| base_name = f"{output_filename}_{event_type}" if output_filename else f"DefaultWebsites_{event_type}_{run_timestamp}" |
| type_to_filepath[event_type] = os.path.join(effective_output_dir, f"{base_name}.csv") |
| target_file = type_to_filepath[event_type] |
| race_info = {"Festival": festival_name, "Type": event_type} |
| pre_filled_data = {k: v for k, v in event_info.items() if k in schema} |
| |
| |
| agent.schema = schema |
| agent.field_instructions = agent._generate_field_instructions() |
| |
| knowledge_base = agent.run_direct(race_info, direct_urls=[event_url], pre_filled_data=pre_filled_data, pre_filled_confidence=1.0) |
| if knowledge_base: |
| url_part = hashlib.md5(event_url.encode()).hexdigest()[:8] |
| caching_key = agent.get_caching_key(festival_name, url=event_url) |
| cache_file_path = os.path.join(KNOWLEDGE_CACHE_DIR, f"{caching_key}.json") |
| with open(cache_file_path, 'w', encoding='utf-8') as f: json.dump(serialize_knowledge_base(knowledge_base), f, indent=4) |
| with SafeCSVWriter(target_file, schema) as writer: |
| for variant_name, data in knowledge_base.items(): |
| row = format_final_row(festival_name, variant_name, data, schema) |
| if row: |
| row['eventWebsite'] = event_url |
| writer.writerow(row) |
| print(f"[SUCCESS] MISSION COMPLETE FOR: {festival_name}") |
| else: failed_missions.append(f"{event_url} (No data)") |
| except AgentInitializationError as e: |
| print(f"[FATAL] Error: {e}") |
| return |
| finally: |
| if agent: agent.shutdown() |
| for event_type, filepath in type_to_filepath.items(): |
| if os.path.exists(filepath): save_output_to_supabase(filepath, agent_mode="default_websites", event_type=event_type) |
| print("\n" + "=" * 60) |
| print("Default Websites Run Complete") |
| if failed_missions: |
| print("\n[SUMMARY] Failed Missions:") |
| for event in failed_missions: print(f" - {event}") |
| else: |
| print("\n[SUCCESS] All missions completed successfully.") |
| print("=" * 60) |
