485 lines
19 KiB
Python
485 lines
19 KiB
Python
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 get_games_by_release_date(
|
|
self, year: int, month: Optional[int] = None, platform: Optional[str] = None, limit: int = 10
|
|
) -> Optional[List[Dict[str, Any]]]:
|
|
"""
|
|
Search for games by release date, optionally filtered by platform.
|
|
"""
|
|
try:
|
|
# Calculate date range for the query
|
|
import datetime
|
|
|
|
if month:
|
|
# Specific month
|
|
start_date = datetime.datetime(year, month, 1)
|
|
if month == 12:
|
|
end_date = datetime.datetime(year + 1, 1, 1) - datetime.timedelta(seconds=1)
|
|
else:
|
|
end_date = datetime.datetime(year, month + 1, 1) - datetime.timedelta(seconds=1)
|
|
else:
|
|
# Entire year
|
|
start_date = datetime.datetime(year, 1, 1)
|
|
end_date = datetime.datetime(year + 1, 1, 1) - datetime.timedelta(seconds=1)
|
|
|
|
start_timestamp = int(start_date.timestamp())
|
|
end_timestamp = int(end_date.timestamp())
|
|
|
|
# Build query filters
|
|
additional_filters = {"first_release_date": f">= {start_timestamp} & first_release_date <= {end_timestamp}"}
|
|
|
|
# Add platform filter if specified
|
|
if platform:
|
|
# Try to map common platform names
|
|
platform_mapping = {
|
|
"ps5": "PlayStation 5",
|
|
"playstation 5": "PlayStation 5",
|
|
"xbox series x": "Xbox Series X|S",
|
|
"xbox series s": "Xbox Series X|S",
|
|
"xbox series x|s": "Xbox Series X|S",
|
|
"switch": "Nintendo Switch",
|
|
"nintendo switch": "Nintendo Switch",
|
|
"pc": "PC (Microsoft Windows)",
|
|
"windows": "PC (Microsoft Windows)",
|
|
}
|
|
platform_key = platform.lower()
|
|
if platform_key in platform_mapping:
|
|
platform = platform_mapping[platform_key]
|
|
|
|
additional_filters["platforms.name"] = f'~ "{platform}"*'
|
|
|
|
# Search games
|
|
games = self.generalized_igdb_query(
|
|
{}, # No name search
|
|
"games",
|
|
[
|
|
"id",
|
|
"name",
|
|
"summary",
|
|
"first_release_date",
|
|
"genres.name",
|
|
"platforms.name",
|
|
"involved_companies.company.name",
|
|
"cover.url",
|
|
"rating",
|
|
"aggregated_rating",
|
|
],
|
|
additional_filters=additional_filters,
|
|
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 by release date {year}/{month}: {e}")
|
|
return None
|
|
|
|
def get_games_by_platform(self, platform: str, genre: Optional[str] = None, limit: int = 10) -> Optional[List[Dict[str, Any]]]:
|
|
"""
|
|
Search for games by platform, optionally filtered by genre.
|
|
"""
|
|
try:
|
|
# Platform name mapping
|
|
platform_mapping = {
|
|
"ps5": "PlayStation 5",
|
|
"playstation 5": "PlayStation 5",
|
|
"xbox series x": "Xbox Series X|S",
|
|
"xbox series s": "Xbox Series X|S",
|
|
"xbox series x|s": "Xbox Series X|S",
|
|
"switch": "Nintendo Switch",
|
|
"nintendo switch": "Nintendo Switch",
|
|
"pc": "PC (Microsoft Windows)",
|
|
"windows": "PC (Microsoft Windows)",
|
|
}
|
|
|
|
platform_key = platform.lower()
|
|
if platform_key in platform_mapping:
|
|
platform = platform_mapping[platform_key]
|
|
|
|
# Build query filters
|
|
additional_filters = {"platforms.name": f'~ "{platform}"*'}
|
|
|
|
# Add genre filter if specified
|
|
if genre:
|
|
additional_filters["genres.name"] = f'~ "{genre}"*'
|
|
|
|
# Search games
|
|
games = self.generalized_igdb_query(
|
|
{}, # No name search
|
|
"games",
|
|
[
|
|
"id",
|
|
"name",
|
|
"summary",
|
|
"first_release_date",
|
|
"genres.name",
|
|
"platforms.name",
|
|
"involved_companies.company.name",
|
|
"cover.url",
|
|
"rating",
|
|
"aggregated_rating",
|
|
],
|
|
additional_filters=additional_filters,
|
|
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 by platform {platform}: {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 by name (e.g., 'Elden Ring', 'Call of Duty', 'Mario'). Do NOT use for release date or platform queries.",
|
|
"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_games_by_release_date",
|
|
"description": "Find games releasing in a specific time period. Use when users ask about upcoming releases, games coming out in a specific month/year, or new releases.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"year": {
|
|
"type": "integer",
|
|
"description": "Release year (e.g., 2025)",
|
|
"minimum": 2020,
|
|
"maximum": 2030,
|
|
},
|
|
"month": {
|
|
"type": "integer",
|
|
"description": "Release month (1-12). Optional, if not specified will search entire year",
|
|
"minimum": 1,
|
|
"maximum": 12,
|
|
},
|
|
"platform": {
|
|
"type": "string",
|
|
"description": "Platform name (e.g., 'PlayStation 5', 'Xbox Series X|S', 'Nintendo Switch', 'PC'). Optional, if not specified will search all platforms",
|
|
},
|
|
"limit": {
|
|
"type": "integer",
|
|
"description": "Maximum number of games to return (default: 10, max: 20)",
|
|
"minimum": 1,
|
|
"maximum": 20,
|
|
},
|
|
},
|
|
"required": ["year"],
|
|
},
|
|
},
|
|
{
|
|
"name": "get_games_by_platform",
|
|
"description": "Find games available on a specific platform. Use when users ask about games for a particular console or system.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"platform": {
|
|
"type": "string",
|
|
"description": "Platform name (e.g., 'PlayStation 5', 'Xbox Series X|S', 'Nintendo Switch', 'PC (Microsoft Windows)')",
|
|
},
|
|
"genre": {
|
|
"type": "string",
|
|
"description": "Game genre (optional) - e.g., 'Action', 'RPG', 'Sports', 'Strategy'",
|
|
},
|
|
"limit": {
|
|
"type": "integer",
|
|
"description": "Maximum number of games to return (default: 10, max: 20)",
|
|
"minimum": 1,
|
|
"maximum": 20,
|
|
},
|
|
},
|
|
"required": ["platform"],
|
|
},
|
|
},
|
|
{
|
|
"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"],
|
|
},
|
|
},
|
|
]
|