import logging from functools import cache from typing import Any, Dict, List, Optional import requests class IGDBQuery(object): def __init__(self, client_id, igdb_api_key): self.client_id = client_id self.igdb_api_key = igdb_api_key def send_igdb_request(self, endpoint, query_body): igdb_url = f"https://api.igdb.com/v4/{endpoint}" headers = {"Client-ID": self.client_id, "Authorization": f"Bearer {self.igdb_api_key}"} try: response = requests.post(igdb_url, headers=headers, data=query_body) response.raise_for_status() return response.json() except requests.RequestException as e: print(f"Error during IGDB API request: {e}") return None @staticmethod def build_query(fields, filters=None, limit=10, offset=None): query = f"fields {','.join(fields) if fields is not None and len(fields) > 0 else '*'}; limit {limit};" if offset is not None: query += f" offset {offset};" if filters: filter_statements = [f"{key} {value}" for key, value in filters.items()] query += " where " + " & ".join(filter_statements) + ";" return query def generalized_igdb_query(self, params, endpoint, fields, additional_filters=None, limit=10, offset=None): all_filters = {key: f'~ "{value}"*' for key, value in params.items() if value} if additional_filters: all_filters.update(additional_filters) query = self.build_query(fields, all_filters, limit, offset) data = self.send_igdb_request(endpoint, query) print(f"{endpoint}: {query} -> {data}") return data def create_query_function(self, name, description, parameters, endpoint, fields, additional_filters=None, limit=10): return { "name": name, "description": description, "parameters": {"type": "object", "properties": parameters}, "function": lambda params: self.generalized_igdb_query(params, endpoint, fields, additional_filters, limit), } @cache def platform_families(self): families = self.generalized_igdb_query({}, "platform_families", ["id", "name"], limit=500) return {v["id"]: v["name"] for v in families} @cache def platforms(self): platforms = self.generalized_igdb_query( {}, "platforms", ["id", "name", "alternative_name", "abbreviation", "platform_family"], limit=500 ) ret = {} for p in platforms: names = [p["name"]] if "alternative_name" in p: names.append(p["alternative_name"]) if "abbreviation" in p: names.append(p["abbreviation"]) family = self.platform_families().get(p.get("platform_family")) if "platform_family" in p else None ret[p["id"]] = {"names": names, "family": family} return ret def game_info(self, name): game_info = self.generalized_igdb_query( {"name": name}, "games", [ "id", "name", "alternative_names", "category", "release_dates", "franchise", "language_supports", "keywords", "platforms", "rating", "summary", ], limit=100, ) return game_info def search_games(self, query: str, limit: int = 5) -> Optional[List[Dict[str, Any]]]: """ Search for games with a flexible query string. Returns formatted game information suitable for AI responses. """ if not query or not query.strip(): return None try: # Search for games with fuzzy matching games = self.generalized_igdb_query( {"name": query.strip()}, "games", [ "id", "name", "summary", "storyline", "rating", "aggregated_rating", "first_release_date", "genres.name", "platforms.name", "involved_companies.company.name", "game_modes.name", "themes.name", "cover.url", ], additional_filters={"category": "= 0"}, # Main games only limit=limit, ) if not games: return None # Format games for AI consumption formatted_games = [] for game in games: formatted_game = self._format_game_for_ai(game) if formatted_game: formatted_games.append(formatted_game) return formatted_games if formatted_games else None except Exception as e: logging.error(f"Error searching games for query '{query}': {e}") return None def get_game_details(self, game_id: int) -> Optional[Dict[str, Any]]: """ Get detailed information about a specific game by ID. """ try: games = self.generalized_igdb_query( {}, "games", [ "id", "name", "summary", "storyline", "rating", "aggregated_rating", "first_release_date", "genres.name", "platforms.name", "involved_companies.company.name", "game_modes.name", "themes.name", "keywords.name", "similar_games.name", "cover.url", "screenshots.url", "videos.video_id", "release_dates.date", "release_dates.platform.name", "age_ratings.rating", ], additional_filters={"id": f"= {game_id}"}, limit=1, ) if games and len(games) > 0: return self._format_game_for_ai(games[0], detailed=True) except Exception as e: logging.error(f"Error getting game details for ID {game_id}: {e}") return None def _format_game_for_ai(self, game_data: Dict[str, Any], detailed: bool = False) -> Dict[str, Any]: """ Format game data in a way that's easy for AI to understand and present to users. """ try: formatted = {"name": game_data.get("name", "Unknown"), "summary": game_data.get("summary", "No summary available")} # Add basic info if "rating" in game_data: formatted["rating"] = f"{game_data['rating']:.1f}/100" if "aggregated_rating" in game_data: formatted["user_rating"] = f"{game_data['aggregated_rating']:.1f}/100" # Release information if "first_release_date" in game_data: import datetime release_date = datetime.datetime.fromtimestamp(game_data["first_release_date"]) formatted["release_year"] = release_date.year if detailed: formatted["release_date"] = release_date.strftime("%Y-%m-%d") # Platforms if "platforms" in game_data and game_data["platforms"]: platforms = [p.get("name", "") for p in game_data["platforms"] if p.get("name")] formatted["platforms"] = platforms[:5] # Limit to prevent overflow # Genres if "genres" in game_data and game_data["genres"]: genres = [g.get("name", "") for g in game_data["genres"] if g.get("name")] formatted["genres"] = genres # Companies (developers/publishers) if "involved_companies" in game_data and game_data["involved_companies"]: companies = [] for company_data in game_data["involved_companies"]: if "company" in company_data and "name" in company_data["company"]: companies.append(company_data["company"]["name"]) formatted["companies"] = companies[:5] # Limit for readability if detailed: # Add more detailed info for specific requests if "storyline" in game_data and game_data["storyline"]: formatted["storyline"] = game_data["storyline"] if "game_modes" in game_data and game_data["game_modes"]: modes = [m.get("name", "") for m in game_data["game_modes"] if m.get("name")] formatted["game_modes"] = modes if "themes" in game_data and game_data["themes"]: themes = [t.get("name", "") for t in game_data["themes"] if t.get("name")] formatted["themes"] = themes return formatted except Exception as e: logging.error(f"Error formatting game data: {e}") return {"name": game_data.get("name", "Unknown"), "summary": "Error retrieving game information"} def get_openai_functions(self) -> List[Dict[str, Any]]: """ Generate OpenAI function definitions for game-related queries. Returns function definitions that OpenAI can use to call IGDB API. """ return [ { "name": "search_games", "description": "Search for video games by name or title. Use when users ask about specific games, game recommendations, or want to know about games.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "The game name or search query (e.g., 'Elden Ring', 'Mario', 'Zelda Breath of the Wild')", }, "limit": { "type": "integer", "description": "Maximum number of games to return (default: 5, max: 10)", "minimum": 1, "maximum": 10, }, }, "required": ["query"], }, }, { "name": "get_game_details", "description": "Get detailed information about a specific game when you have its ID from a previous search.", "parameters": { "type": "object", "properties": {"game_id": {"type": "integer", "description": "The IGDB game ID from a previous search result"}}, "required": ["game_id"], }, }, ]