81 lines
3.4 KiB
Python
81 lines
3.4 KiB
Python
import requests
|
|
from functools import cache
|
|
|
|
|
|
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
|