init: initial commit

stats/twitch_stats.py

@@ -0,0 +1,60 @@
+from prometheus_client import Gauge
+import requests
+import os
+import time
+
+# Load env vars
+TWITCH_CLIENT_ID = os.getenv("TWITCH_CLIENT_ID")
+TWITCH_CLIENT_SECRET = os.getenv("TWITCH_CLIENT_SECRET")
+CHANNELS = os.getenv("TWITCH_CHANNELS", "").split(",")
+
+# Metric definition
+live_status = Gauge('twitch_stream_live', 'Is the stream live?', ['channel'])
+
+_token_cache = {
+    "access_token": None,
+    "expires_at": 0
+}
+
+print("[twitch] Client ID:", TWITCH_CLIENT_ID)
+print("[twitch] Client Secret:", TWITCH_CLIENT_SECRET)
+print("[twitch] Channels:", CHANNELS)
+
+def get_token():
+    if time.time() < _token_cache["expires_at"] - 60:
+        return _token_cache["access_token"]
+
+    response = requests.post("https://id.twitch.tv/oauth2/token", data={
+        "client_id": TWITCH_CLIENT_ID,
+        "client_secret": TWITCH_CLIENT_SECRET,
+        "grant_type": "client_credentials"
+    })
+
+    # Print the response for debugging
+    print(f"[twitch] Response: {response.json()}")
+
+    data = response.json()
+    
+    # Check for errors in the response
+    if "access_token" not in data:
+        raise ValueError(f"Error getting token: {data}")
+    
+    _token_cache["access_token"] = data["access_token"]
+    _token_cache["expires_at"] = time.time() + data["expires_in"]
+    return _token_cache["access_token"]
+
+headers = {
+    'Client-ID': TWITCH_CLIENT_ID,
+    'Authorization': f'Bearer {get_token()}'
+}
+
+def update():
+    for channel in CHANNELS:
+        url = f'https://api.twitch.tv/helix/streams?user_login={channel}'
+        try:
+            response = requests.get(url, headers=headers)
+            data = response.json()
+            is_live = int(bool(data.get('data')))
+            live_status.labels(channel=channel).set(is_live)
+        except Exception as e:
+            print(f"[twitch] Error checking {channel}: {e}")