Scripts, scenes, bridges, bots and more

twitch-bot/_twitch.py

@@ -0,0 +1,112 @@
+"""
+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}"
+ENV_FILE     = Path(__file__).resolve().parent / ".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

twitch-bot/search-game.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+"""
+Interactive Twitch category search. Used by loading.sh to resolve a free-form
+query into an exact directory entry.
+
+  ./search-game.py "Heroes"
+    stderr: numbered list of up to 20 matches + "Pick: " prompt
+    stdout: a single line "<game_id>\\t<canonical_name>"  (only on success)
+    exit:   0 success | 1 no-match / cancel / error | 130 ctrl-c
+
+Auto-picks when there's exactly one result. Tempting to also auto-pick on an
+exact-name match within a larger result set (e.g. typing "Doom Eternal" and
+having that exact entry near the top), but a real test surfaced the failure:
+typing "Heroes" returns "Heroes" (TV show) plus all the Heroes-* games, and
+auto-picking the standalone match silently steals from the user — who is
+*using search* precisely because the query is ambiguous. So: only one rule.
+
+Endpoint: /helix/search/categories — same one Twitch's dashboard autocomplete
+uses. Accepts any user/app token; we reuse the broadcaster token from
+.env.ophi118 because it's already there and refreshable.
+"""
+
+import sys
+import urllib.parse
+
+import _twitch as tw
+
+MAX_RESULTS = 20
+
+
+def err(msg=""):
+    print(msg, file=sys.stderr, flush=True)
+
+
+def search(env: dict, query: str) -> list:
+    qs = urllib.parse.urlencode({"query": query, "first": MAX_RESULTS})
+    status, body = tw.auth_call("GET", f"{tw.HELIX}/search/categories?{qs}", env)
+    if status != 200:
+        raise RuntimeError(f"search failed (status={status}): {body}")
+    return (body or {}).get("data") or []
+
+
+def pick(results: list, query: str) -> dict:
+    if not results:
+        raise RuntimeError(f"no Twitch categories match {query!r}")
+
+    if len(results) == 1:
+        err(f"  ✓ only match: {results[0]['name']}")
+        return results[0]
+
+    err(f"── {len(results)} matches for {query!r} ──")
+    for i, g in enumerate(results, 1):
+        err(f"  {i:2}) {g['name']}")
+    err("   0) cancel")
+
+    while True:
+        # input()'s prompt arg writes to stdout — fatal here because bash
+        # captures stdout for the resolved <id>\t<name> line. Print to stderr
+        # explicitly, then call input() with no prompt.
+        print("Pick: ", end="", file=sys.stderr, flush=True)
+        try:
+            raw = input().strip()
+        except EOFError:
+            raise RuntimeError("cancelled (EOF)")
+        if not raw.isdigit():
+            err("  ! enter a number")
+            continue
+        n = int(raw)
+        if n == 0:
+            raise RuntimeError("cancelled")
+        if 1 <= n <= len(results):
+            return results[n - 1]
+        err(f"  ! must be 0..{len(results)}")
+
+
+def main():
+    if len(sys.argv) != 2 or not sys.argv[1].strip():
+        print("usage: search-game.py <query>", file=sys.stderr)
+        sys.exit(2)
+    query = sys.argv[1].strip()
+
+    env = tw.load_env()
+    needed = ("TWITCH_ACCESS_TOKEN", "TWITCH_REFRESH_TOKEN", "TWITCH_CLIENT_ID")
+    missing = [k for k in needed if not env.get(k)]
+    if missing:
+        err(f"[err] missing in {tw.ENV_FILE.name}: {', '.join(missing)}")
+        sys.exit(1)
+
+    try:
+        results = search(env, query)
+        chosen  = pick(results, query)
+    except Exception as e:
+        err(f"[err] {type(e).__name__}: {e}")
+        sys.exit(1)
+
+    # ONLY the resolved tuple goes to stdout — bash captures this.
+    print(f"{chosen['id']}\t{chosen['name']}")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except KeyboardInterrupt:
+        sys.exit(130)

twitch-bot/set-channel.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+"""
+Update Twitch channel info (game + title) for the broadcaster account.
+
+  ./set-channel.py --game-id 517520 "Slayer is back — Nightmare run"
+  ./set-channel.py "Doom Eternal" "..."          # exact-name lookup, fails if no exact hit
+  ./set-channel.py --dry-run --game-id 517520 "..."
+
+For free-form / fuzzy game queries, use search-game.py first to resolve the
+id (loading.sh does this automatically). The /helix/games?name= fallback in
+this script is exact-and-case-sensitive.
+
+Loads `.env.ophi118` — see _twitch.py for the env contract. The token MUST
+belong to the broadcaster (Twitch enforces broadcaster_id == token user_id
+on PATCH /helix/channels), so this script intentionally targets that file
+and not `.env` (which holds the chat-bot's account).
+"""
+
+import json
+import sys
+import urllib.parse
+
+import _twitch as tw
+
+TITLE_LIMIT = 140
+
+
+def lookup_game_id(env: dict, name: str) -> str:
+    qs = urllib.parse.urlencode({"name": name})
+    status, body = tw.auth_call("GET", f"{tw.HELIX}/games?{qs}", env)
+    if status != 200:
+        raise RuntimeError(f"games lookup failed (status={status}): {body}")
+    data = (body or {}).get("data") or []
+    if not data:
+        raise RuntimeError(
+            f"no Twitch game matches {name!r} exactly. /helix/games?name= is "
+            f"case-sensitive — use search-game.py for fuzzy lookup."
+        )
+    return data[0]["id"]
+
+
+def patch_channel(env: dict, game_id: str, title: str) -> None:
+    qs = urllib.parse.urlencode({"broadcaster_id": env["TWITCH_BROADCASTER_ID"]})
+    payload = json.dumps({"game_id": game_id, "title": title})
+    status, body = tw.auth_call("PATCH", f"{tw.HELIX}/channels?{qs}", env, body=payload)
+    if status not in (200, 204):
+        raise RuntimeError(f"channel update failed (status={status}): {body}")
+
+
+def usage_and_die():
+    print("usage: set-channel.py [--dry-run] (--game-id <id> | <game-name>) <title>",
+          file=sys.stderr)
+    sys.exit(2)
+
+
+def main():
+    args = sys.argv[1:]
+    dry  = False
+    if args and args[0] == "--dry-run":
+        dry, args = True, args[1:]
+
+    game_id = None
+    if args and args[0] == "--game-id":
+        if len(args) < 3:
+            usage_and_die()
+        game_id, args = args[1], args[2:]
+
+    expected = 1 if game_id else 2
+    if len(args) != expected:
+        usage_and_die()
+
+    if game_id:
+        title = args[0][:TITLE_LIMIT]
+    else:
+        game_name = args[0]
+        title     = args[1][:TITLE_LIMIT]
+
+    env = tw.load_env()
+    needed = ("TWITCH_ACCESS_TOKEN", "TWITCH_REFRESH_TOKEN",
+              "TWITCH_CLIENT_ID", "TWITCH_BROADCASTER_ID")
+    missing = [k for k in needed if not env.get(k)]
+    if missing:
+        print(f"[err] missing in {tw.ENV_FILE.name}: {', '.join(missing)}", file=sys.stderr)
+        sys.exit(1)
+
+    if game_id is None:
+        game_id = lookup_game_id(env, game_name)
+
+    if dry:
+        print(f"  ✓ dry-run: would set game_id={game_id}, title={title!r} (no PATCH issued)")
+        return
+
+    patch_channel(env, game_id, title)
+    print(f"  ✓ Twitch channel updated → game_id={game_id}, title={title!r}")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as e:
+        print(f"[err] {type(e).__name__}: {e}", file=sys.stderr)
+        sys.exit(1)