Oleksandr Kozachuk
fbec05dfe9
- Fix infinite retry loop in ai_responder.py that caused test_fix1 to hang - Add missing picture_edit parameter to all AIResponse constructor calls - Set up complete development toolchain with Black, isort, Bandit, and MyPy - Create comprehensive Makefile for development workflows - Add pre-commit hooks with formatting, linting, security, and type checking - Update test mocking to provide contextual responses for different scenarios - Configure all tools for 140 character line length and strict type checking - Add DEVELOPMENT.md with setup instructions and workflow documentation 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
from functools import cache
|
|
|
|
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()[p["id"]] 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},
|
|
[
|
|
"id",
|
|
"name",
|
|
"alternative_names",
|
|
"category",
|
|
"release_dates",
|
|
"franchise",
|
|
"language_supports",
|
|
"keywords",
|
|
"platforms",
|
|
"rating",
|
|
"summary",
|
|
],
|
|
limit=100,
|
|
)
|
|
return game_info
|