122 lines
5.3 KiB
Python
122 lines
5.3 KiB
Python
"""
|
|
Shared helpers for set-channel.py and search-game.py.
|
|
|
|
Stdlib-only on purpose — the bot dir intentionally avoids a venv. If a third
|
|
caller appears that needs richer behavior (retries with backoff, async, etc.),
|
|
revisit; until then minimal urllib is fine.
|
|
|
|
Auth model: every Helix call goes through `auth_call`, which runs the request,
|
|
refreshes the access token via twitchtokengenerator on 401, persists the new
|
|
tokens to .env.ophi118, mutates the in-memory env dict, and retries once.
|
|
Callers stay agnostic about whether a refresh happened.
|
|
"""
|
|
|
|
import json
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
HELIX = "https://api.twitch.tv/helix"
|
|
REFRESH_URL = "https://twitchtokengenerator.com/api/refresh/{refresh}"
|
|
|
|
# Default broadcaster env file. Looks for `.env.broadcaster` (the documented
|
|
# OSS-style filename) first, then falls back to the legacy `.env.ophi118`
|
|
# kept around for the original ophi118 rig that birthed this code. Whichever
|
|
# exists wins; if neither exists yet, ENV_FILE points at `.env.broadcaster`
|
|
# so error messages tell the user the right file to create.
|
|
_DIR = Path(__file__).resolve().parent
|
|
ENV_FILE = _DIR / ".env.broadcaster"
|
|
if not ENV_FILE.exists() and (_DIR / ".env.ophi118").exists():
|
|
ENV_FILE = _DIR / ".env.ophi118"
|
|
|
|
|
|
# ─── env file I/O ───────────────────────────────────────────────────────────
|
|
|
|
def load_env(path: Path = ENV_FILE) -> dict:
|
|
out = {}
|
|
if not path.exists():
|
|
return out
|
|
for raw in path.read_text().splitlines():
|
|
line = raw.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
k, _, v = line.partition("=")
|
|
out[k.strip()] = v.strip().strip('"').strip("'")
|
|
return out
|
|
|
|
|
|
def update_env(updates: dict, path: Path = ENV_FILE) -> None:
|
|
"""Atomic in-place rewrite of KEY=VALUE lines, comments preserved."""
|
|
lines = path.read_text().splitlines() if path.exists() else []
|
|
seen, out_lines = set(), []
|
|
for line in lines:
|
|
s = line.strip()
|
|
if s and not s.startswith("#") and "=" in s:
|
|
k = s.split("=", 1)[0].strip()
|
|
if k in updates:
|
|
out_lines.append(f"{k}={updates[k]}")
|
|
seen.add(k)
|
|
continue
|
|
out_lines.append(line)
|
|
for k, v in updates.items():
|
|
if k not in seen:
|
|
out_lines.append(f"{k}={v}")
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
tmp.write_text("\n".join(out_lines) + "\n")
|
|
tmp.replace(path)
|
|
|
|
|
|
# ─── HTTP ───────────────────────────────────────────────────────────────────
|
|
|
|
def http(method: str, url: str, *, headers=None, body=None):
|
|
"""Returns (status, parsed_json_or_text). Never raises on HTTP errors —
|
|
surfaces them as the status code so callers can branch on 401 cleanly."""
|
|
data = body.encode("utf-8") if isinstance(body, str) else body
|
|
req = urllib.request.Request(url, method=method, headers=headers or {}, data=data)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
raw = r.read().decode("utf-8")
|
|
try: return r.status, (json.loads(raw) if raw else None)
|
|
except json.JSONDecodeError: return r.status, raw
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read().decode("utf-8", errors="replace")
|
|
try: return e.code, (json.loads(raw) if raw else None)
|
|
except json.JSONDecodeError: return e.code, raw
|
|
|
|
|
|
# ─── auth ───────────────────────────────────────────────────────────────────
|
|
|
|
def refresh_token(env: dict) -> dict:
|
|
"""Mint new tokens via twitchtokengenerator. Returns the merge dict for
|
|
`update_env` — caller persists. Raises on refresh failure (no point
|
|
retrying — the refresh token is dead, the user must re-generate)."""
|
|
refresh = env["TWITCH_REFRESH_TOKEN"]
|
|
url = REFRESH_URL.format(refresh=urllib.parse.quote(refresh, safe=""))
|
|
status, body = http("GET", url)
|
|
if status != 200 or not isinstance(body, dict) or not body.get("success"):
|
|
raise RuntimeError(f"token refresh failed (status={status}): {body}")
|
|
return {
|
|
"TWITCH_ACCESS_TOKEN": body["token"],
|
|
"TWITCH_REFRESH_TOKEN": body["refresh"],
|
|
}
|
|
|
|
|
|
def auth_call(method: str, url: str, env: dict, *, body=None):
|
|
"""Authenticated Helix request with 401 → refresh → retry. Mutates `env`
|
|
in place on refresh so subsequent calls in the same process see the new
|
|
token without another reload."""
|
|
def headers_for(tok):
|
|
h = {"Authorization": f"Bearer {tok}", "Client-Id": env["TWITCH_CLIENT_ID"]}
|
|
if body is not None:
|
|
h["Content-Type"] = "application/json"
|
|
return h
|
|
|
|
status, resp = http(method, url, headers=headers_for(env["TWITCH_ACCESS_TOKEN"]), body=body)
|
|
if status == 401:
|
|
new = refresh_token(env)
|
|
update_env(new)
|
|
env.update(new)
|
|
status, resp = http(method, url, headers=headers_for(env["TWITCH_ACCESS_TOKEN"]), body=body)
|
|
return status, resp
|