igdb: auto-refresh twitch token; category -> game_type filter

Static app tokens expire after ~60 days -> every lookup failed with
401. With igdb-client-secret set, the bot fetches the token via
client-credentials OAuth itself, refreshes a day before expiry and
retries once on 401; a static igdb-access-token still works.

IGDB renamed games.category to game_type: the category = 0 filter in
search_games silently matched nothing even with a valid token.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Oleksandr Kozachuk
2026-07-13 18:41:31 +02:00
parent 4166520923
commit 86e631926f
6 changed files with 155 additions and 21 deletions
+21 -8
View File
@@ -20,13 +20,6 @@ The bot now supports real-time video game information through IGDB (Internet Gam
- **Category**: Select appropriate category - **Category**: Select appropriate category
3. Note down your **Client ID** 3. Note down your **Client ID**
4. Generate a **Client Secret** 4. Generate a **Client Secret**
5. Get an access token using this curl command:
```bash
curl -X POST 'https://id.twitch.tv/oauth2/token' \
-H 'Content-Type: application/x-www-form-urlencoded' \
-d 'client_id=YOUR_CLIENT_ID&client_secret=YOUR_CLIENT_SECRET&grant_type=client_credentials'
```
6. Save the `access_token` from the response
### 2. Configure the Bot ### 2. Configure the Bot
@@ -35,6 +28,25 @@ Update your `config.toml` file:
```toml ```toml
# IGDB Configuration for game information # IGDB Configuration for game information
igdb-client-id = "your_actual_client_id_here" igdb-client-id = "your_actual_client_id_here"
igdb-client-secret = "your_actual_client_secret_here"
enable-game-info = true
```
With the client secret configured, the bot fetches an app access token from
Twitch itself and refreshes it automatically before it expires (Twitch app
tokens live ~60 days) — no manual token handling needed.
Alternatively, a static token still works (legacy setup — it expires after
~60 days and then game lookups fail with 401 until you replace it):
```bash
curl -X POST 'https://id.twitch.tv/oauth2/token' \
-H 'Content-Type: application/x-www-form-urlencoded' \
-d 'client_id=YOUR_CLIENT_ID&client_secret=YOUR_CLIENT_SECRET&grant_type=client_credentials'
```
```toml
igdb-client-id = "your_actual_client_id_here"
igdb-access-token = "your_actual_access_token_here" igdb-access-token = "your_actual_access_token_here"
enable-game-info = true enable-game-info = true
``` ```
@@ -99,7 +111,8 @@ The integration provides two OpenAI functions:
- Verify client ID and access token are set - Verify client ID and access token are set
2. **Authentication errors** 2. **Authentication errors**
- Regenerate access token (they expire) - Prefer `igdb-client-secret` — the bot then refreshes tokens itself
- With a static `igdb-access-token`: regenerate it (they expire)
- Verify client ID matches your Twitch app - Verify client ID matches your Twitch app
3. **No game results** 3. **No game results**
+5 -1
View File
@@ -14,7 +14,11 @@ system = "You are a smart AI assistant with access to real-time video game infor
# IGDB Configuration for game information # IGDB Configuration for game information
igdb-client-id = "YOUR_IGDB_CLIENT_ID" igdb-client-id = "YOUR_IGDB_CLIENT_ID"
igdb-access-token = "YOUR_IGDB_ACCESS_TOKEN" # With the Twitch app client secret set, the bot fetches and refreshes the
# access token itself (recommended). A static igdb-access-token still works
# but expires after ~60 days.
igdb-client-secret = "YOUR_IGDB_CLIENT_SECRET"
# igdb-access-token = "YOUR_IGDB_ACCESS_TOKEN"
enable-game-info = true enable-game-info = true
# --- operator / safety (SPEC-003, SPEC-006) --- # --- operator / safety (SPEC-003, SPEC-006) ---
+38 -5
View File
@@ -1,27 +1,60 @@
import logging import logging
import time
from functools import cache from functools import cache
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
import requests 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): class IGDBQuery(object):
def __init__(self, client_id, igdb_api_key): def __init__(self, client_id, igdb_api_key=None, client_secret=None):
self.client_id = client_id self.client_id = client_id
self.igdb_api_key = igdb_api_key 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): def send_igdb_request(self, endpoint, query_body):
igdb_url = f"https://api.igdb.com/v4/{endpoint}" igdb_url = f"https://api.igdb.com/v4/{endpoint}"
headers = {"Client-ID": self.client_id, "Authorization": f"Bearer {self.igdb_api_key}"}
try: try:
response = requests.post(igdb_url, headers=headers, data=query_body) 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() response.raise_for_status()
return response.json() return response.json()
except requests.RequestException as e: except requests.RequestException as e:
print(f"Error during IGDB API request: {e}") print(f"Error during IGDB API request: {e}")
return None 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 @staticmethod
def build_query(fields, filters=None, limit=10, offset=None): 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};" query = f"fields {','.join(fields) if fields is not None and len(fields) > 0 else '*'}; limit {limit};"
@@ -79,7 +112,7 @@ class IGDBQuery(object):
"id", "id",
"name", "name",
"alternative_names", "alternative_names",
"category", "game_type",
"release_dates", "release_dates",
"franchise", "franchise",
"language_supports", "language_supports",
@@ -120,7 +153,7 @@ class IGDBQuery(object):
"themes.name", "themes.name",
"cover.url", "cover.url",
], ],
additional_filters={"category": "= 0"}, # Main games only additional_filters={"game_type": "= 0"}, # Main games only (IGDB renamed category -> game_type)
limit=limit, limit=limit,
) )
+9 -5
View File
@@ -137,16 +137,20 @@ class OpenAIResponder(AIResponder, LeonardoAIDrawMixIn):
# Initialize IGDB if enabled # Initialize IGDB if enabled
self.igdb = None self.igdb = None
igdb_client_id = self.config.get("igdb-client-id")
igdb_client_secret = self.config.get("igdb-client-secret")
igdb_access_token = self.config.get("igdb-access-token")
logging.info("IGDB Configuration Check:") logging.info("IGDB Configuration Check:")
logging.info(f" enable-game-info: {self.config.get('enable-game-info', 'NOT SET')}") logging.info(f" enable-game-info: {self.config.get('enable-game-info', 'NOT SET')}")
logging.info(f" igdb-client-id: {'SET' if self.config.get('igdb-client-id') else 'NOT SET'}") logging.info(f" igdb-client-id: {'SET' if igdb_client_id else 'NOT SET'}")
logging.info(f" igdb-access-token: {'SET' if self.config.get('igdb-access-token') else 'NOT SET'}") logging.info(f" igdb-client-secret: {'SET' if igdb_client_secret else 'NOT SET'}")
logging.info(f" igdb-access-token: {'SET' if igdb_access_token else 'NOT SET'}")
if self.config.get("enable-game-info", False) and self.config.get("igdb-client-id") and self.config.get("igdb-access-token"): if self.config.get("enable-game-info", False) and igdb_client_id and (igdb_client_secret or igdb_access_token):
try: try:
self.igdb = IGDBQuery(self.config["igdb-client-id"], self.config["igdb-access-token"]) self.igdb = IGDBQuery(igdb_client_id, igdb_access_token, client_secret=igdb_client_secret)
logging.info("✅ IGDB integration SUCCESSFULLY enabled for game information") logging.info("✅ IGDB integration SUCCESSFULLY enabled for game information")
logging.info(f" Client ID: {self.config['igdb-client-id'][:8]}...") logging.info(f" Client ID: {igdb_client_id[:8]}...")
logging.info(f" Available functions: {len(self.igdb.get_openai_functions())}") logging.info(f" Available functions: {len(self.igdb.get_openai_functions())}")
except Exception as e: except Exception as e:
logging.error(f"❌ Failed to initialize IGDB: {e}") logging.error(f"❌ Failed to initialize IGDB: {e}")
+1 -1
View File
@@ -28,7 +28,7 @@ class TestIGDBIntegration(unittest.IsolatedAsyncioTestCase):
responder = OpenAIResponder(self.config_with_igdb) responder = OpenAIResponder(self.config_with_igdb)
mock_igdb.assert_called_once_with("test_client", "test_token") mock_igdb.assert_called_once_with("test_client", "test_token", client_secret=None)
self.assertEqual(responder.igdb, mock_igdb_instance) self.assertEqual(responder.igdb, mock_igdb_instance)
def test_igdb_initialization_disabled(self): def test_igdb_initialization_disabled(self):
+81 -1
View File
@@ -160,7 +160,7 @@ class TestIGDBQuery(unittest.TestCase):
"id", "id",
"name", "name",
"alternative_names", "alternative_names",
"category", "game_type",
"release_dates", "release_dates",
"franchise", "franchise",
"language_supports", "language_supports",
@@ -174,5 +174,85 @@ class TestIGDBQuery(unittest.TestCase):
self.assertEqual(result, [{"id": 1, "name": "Super Mario Bros"}]) self.assertEqual(result, [{"id": 1, "name": "Super Mario Bros"}])
class TestIGDBTokenRefresh(unittest.TestCase):
@staticmethod
def _oauth_response(token="fresh_token", expires_in=5_000_000):
response = Mock()
response.json.return_value = {"access_token": token, "expires_in": expires_in}
response.raise_for_status.return_value = None
return response
@staticmethod
def _api_response(payload, status_code=200):
response = Mock()
response.status_code = status_code
response.json.return_value = payload
response.raise_for_status.return_value = None
return response
@patch("fjerkroa_bot.igdblib.requests.post")
def test_fetches_token_when_only_secret_configured(self, mock_post):
"""Without a static token, the first request fetches one via Twitch OAuth."""
mock_post.side_effect = [self._oauth_response(), self._api_response([{"id": 1}])]
igdb = IGDBQuery("cid", client_secret="secret")
result = igdb.send_igdb_request("games", "fields name; limit 1;")
self.assertEqual(result, [{"id": 1}])
oauth_call, api_call = mock_post.call_args_list
self.assertEqual(oauth_call.args[0], "https://id.twitch.tv/oauth2/token")
self.assertEqual(
oauth_call.kwargs["params"],
{"client_id": "cid", "client_secret": "secret", "grant_type": "client_credentials"},
)
self.assertEqual(api_call.kwargs["headers"]["Authorization"], "Bearer fresh_token")
@patch("fjerkroa_bot.igdblib.requests.post")
def test_refreshes_and_retries_on_401(self, mock_post):
"""A 401 with a configured secret triggers one refresh and retry."""
mock_post.side_effect = [
self._api_response(None, status_code=401),
self._oauth_response(),
self._api_response([{"id": 2}]),
]
igdb = IGDBQuery("cid", "expired_token", client_secret="secret")
result = igdb.send_igdb_request("games", "fields name; limit 1;")
self.assertEqual(result, [{"id": 2}])
self.assertEqual(igdb.igdb_api_key, "fresh_token")
self.assertEqual(mock_post.call_args_list[2].kwargs["headers"]["Authorization"], "Bearer fresh_token")
@patch("fjerkroa_bot.igdblib.time.time")
@patch("fjerkroa_bot.igdblib.requests.post")
def test_proactive_refresh_before_expiry(self, mock_post, mock_time):
"""An expired self-fetched token is refreshed before the request."""
mock_time.return_value = 1_000_000.0
mock_post.side_effect = [self._oauth_response("token_a", expires_in=5_000_000), self._api_response([])]
igdb = IGDBQuery("cid", client_secret="secret")
igdb.send_igdb_request("games", "fields name;")
# jump past the token expiry -> next request refreshes first
mock_time.return_value = 1_000_000.0 + 5_000_000
mock_post.side_effect = [self._oauth_response("token_b"), self._api_response([])]
igdb.send_igdb_request("games", "fields name;")
self.assertEqual(igdb.igdb_api_key, "token_b")
@patch("fjerkroa_bot.igdblib.requests.post")
def test_no_refresh_without_secret(self, mock_post):
"""Static-token setups keep the old behavior: no OAuth calls, error -> None."""
response = Mock()
response.status_code = 401
response.raise_for_status.side_effect = requests.RequestException("401 Client Error")
mock_post.return_value = response
igdb = IGDBQuery("cid", "expired_token")
result = igdb.send_igdb_request("games", "fields name; limit 1;")
self.assertIsNone(result)
mock_post.assert_called_once()
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()