import json import re from typing import Dict, List, Any, Optional, Union import logging logger = logging.getLogger(__name__) class JSONParser: """Centralized JSON parsing utilities for challenge data""" @staticmethod def extract_json_from_response(response_str: str) -> Optional[Dict]: """ Extract JSON from various response formats Args: response_str: The response string that may contain JSON Returns: Parsed JSON dictionary or None if parsing fails """ try: # Convert response to string if it's not already response_str = str(response_str) # Try to extract JSON between custom tags first json_match = re.search(r'\s*(.*?)\s*', response_str, re.DOTALL) if json_match: json_str = json_match.group(1).strip() return json.loads(json_str) # Try to extract JSON from markdown code blocks json_match = re.search(r'```json\s*(.*?)\s*```', response_str, re.DOTALL) if json_match: json_str = json_match.group(1).strip() return json.loads(json_str) # Try to find JSON object in the string json_match = re.search(r'\{.*\}', response_str, re.DOTALL) if json_match: json_str = json_match.group(0) return json.loads(json_str) # If no structured format found, try parsing the entire string return json.loads(response_str.strip()) except json.JSONDecodeError as e: logger.warning(f"JSON decode error: {e}") return None except Exception as e: logger.error(f"Unexpected error parsing JSON: {e}") return None @staticmethod def normalize_challenges(challenges_data: Union[Dict, List]) -> List[Dict]: """ Normalize challenge data to a consistent format Args: challenges_data: Raw challenge data in various formats Returns: List of normalized challenge dictionaries """ try: # If it's already a list, return it if isinstance(challenges_data, list): return [JSONParser._normalize_single_challenge(challenge) for challenge in challenges_data] # If it's a dict with challenges key if isinstance(challenges_data, dict): if "challenges" in challenges_data: challenges = challenges_data["challenges"] if isinstance(challenges, list): return [JSONParser._normalize_single_challenge(challenge) for challenge in challenges] else: return [JSONParser._normalize_single_challenge(challenges)] else: # Treat the entire dict as a single challenge return [JSONParser._normalize_single_challenge(challenges_data)] return [] except Exception as e: logger.error(f"Error normalizing challenges: {e}") return [] @staticmethod def _normalize_single_challenge(challenge: Dict) -> Dict: """Normalize a single challenge to consistent format""" normalized = { "name": challenge.get("name", challenge.get("title", "Unknown Challenge")), "description": challenge.get("description", challenge.get("desc", "")), "prize": challenge.get("prize", challenge.get("reward", challenge.get("amount", "$0"))) } # Add optional fields if available optional_fields = ["id", "url", "track", "deadline", "type", "technologies"] for field in optional_fields: if field in challenge: normalized[field] = challenge[field] return normalized @staticmethod def create_error_response(raw_response: Any, error_message: str) -> Dict: """Create standardized error response with raw data""" return { "error": error_message, "raw_response": str(raw_response), "challenges": [] } @staticmethod def validate_challenge_response(response_data: Dict) -> bool: """ Validate that the response contains valid challenge data Args: response_data: The parsed response data Returns: True if valid, False otherwise """ if not isinstance(response_data, dict): return False # Check if it has challenges if "challenges" in response_data: challenges = response_data["challenges"] if not isinstance(challenges, list): return False # Validate each challenge has required fields for challenge in challenges: if not isinstance(challenge, dict): return False if not all(field in challenge for field in ["name", "description", "prize"]): return False return True