Oleksandr Kozachuk
2caa18a17f
search_games used `where name ~ "<q>"*`: prefix-only, diacritic- and word-order-sensitive -- "MARVEL Tōkon" found nothing though IGDB has it. Switch to IGDB's native `search` clause (relevance-ranked, diacritic-insensitive). fetch_url and image downloads read bodies with content.read(n), which returns only the first buffered chunk (~7 KB): pages collapsed to their <title>. httpread.read_capped collects chunks up to the byte cap; image downloads read limit+1 so over-limit files are still rejected instead of cached truncated (IMG-10). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
524 lines
21 KiB
Python
524 lines
21 KiB
Python
import logging
|
|
import time
|
|
from functools import cache
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import requests
|
|
|
|
TWITCH_OAUTH_URL = "https://id.twitch.tv/oauth2/token"
|
|
# Refresh this long before Twitch expires the token (app tokens live ~60 days)
|
|
TOKEN_REFRESH_MARGIN = 86400
|
|
|
|
|
|
class IGDBQuery(object):
|
|
def __init__(self, client_id, igdb_api_key=None, client_secret=None):
|
|
self.client_id = client_id
|
|
self.igdb_api_key = igdb_api_key
|
|
self.client_secret = client_secret
|
|
# Unknown for statically configured tokens; set after each refresh
|
|
self._token_expires_at = None
|
|
|
|
def _refresh_token(self):
|
|
response = requests.post(
|
|
TWITCH_OAUTH_URL,
|
|
params={"client_id": self.client_id, "client_secret": self.client_secret, "grant_type": "client_credentials"},
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
self.igdb_api_key = data["access_token"]
|
|
self._token_expires_at = time.time() + data.get("expires_in", 0) - TOKEN_REFRESH_MARGIN
|
|
logging.info("IGDB: refreshed Twitch app access token")
|
|
|
|
def _ensure_token(self):
|
|
if not self.client_secret:
|
|
return
|
|
if not self.igdb_api_key or (self._token_expires_at is not None and time.time() >= self._token_expires_at):
|
|
self._refresh_token()
|
|
|
|
def send_igdb_request(self, endpoint, query_body):
|
|
igdb_url = f"https://api.igdb.com/v4/{endpoint}"
|
|
|
|
try:
|
|
self._ensure_token()
|
|
response = self._post_igdb(igdb_url, query_body)
|
|
if self.client_secret and response.status_code == 401:
|
|
# Token expired server-side (e.g. statically configured) — refresh and retry once
|
|
self._refresh_token()
|
|
response = self._post_igdb(igdb_url, query_body)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.RequestException as e:
|
|
print(f"Error during IGDB API request: {e}")
|
|
return None
|
|
|
|
def _post_igdb(self, igdb_url, query_body):
|
|
headers = {"Client-ID": self.client_id, "Authorization": f"Bearer {self.igdb_api_key}"}
|
|
return requests.post(igdb_url, headers=headers, data=query_body)
|
|
|
|
@staticmethod
|
|
def build_query(fields, filters=None, limit=10, offset=None, search_term=None):
|
|
query = ""
|
|
if search_term:
|
|
escaped = search_term.replace("\\", "\\\\").replace('"', '\\"')
|
|
query += f'search "{escaped}"; '
|
|
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, search_term=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, search_term)
|
|
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",
|
|
"game_type",
|
|
"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:
|
|
# IGDB native full-text search: diacritic- and word-order-insensitive,
|
|
# unlike a `name ~ "..."*` prefix filter
|
|
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",
|
|
"cover.url",
|
|
],
|
|
additional_filters={"game_type": "= 0"}, # Main games only (IGDB renamed category -> game_type)
|
|
limit=limit,
|
|
search_term=query.strip(),
|
|
)
|
|
|
|
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"],
|
|
},
|
|
},
|
|
]
|