Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 86e631926f |
+21
-8
@@ -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
@@ -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
@@ -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,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -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}")
|
||||||
|
|||||||
@@ -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
@@ -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()
|
||||||
|
|||||||
Reference in New Issue
Block a user