59 lines
1.7 KiB
Python
59 lines
1.7 KiB
Python
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] 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}")
|