import urllib.parse import re from datetime import datetime import requests from .exceptions import BotAPIError class BotAPI: def __init__(self, api_url: str, channel: str, token: str): self._base = f"{api_url.rstrip('/')}/channels/{channel}" self._session = requests.Session() self._session.headers.update({ "Authorization": f"Bearer {token}", "Content-Type": "application/json", }) def _request(self, method: str, path: str, **kwargs): url = f"{self._base}{path}" resp = self._session.request(method, url, **kwargs) if not resp.ok: try: error = resp.json().get("error", f"HTTP {resp.status_code}") except Exception: error = f"HTTP {resp.status_code}" raise BotAPIError(error, status_code=resp.status_code) if not resp.content: return None return resp.json() # ── Playlist ────────────────────────────────────────────────────────────── def get_playlist(self) -> dict: """Returns { items, currentIndex, locked }. Requires channel active.""" return self._request("GET", "/playlist") def add_to_playlist(self, id: str, type: str, pos: str = "end") -> None: """Queue media. pos is 'end' or 'next'. Requires rank >= MOD.""" self._request("POST", "/playlist", json={"id": id, "type": type, "pos": pos}) def delete_playlist_item(self, uid: int) -> None: """Remove a playlist item by uid. Requires rank >= ADMIN.""" self._request("DELETE", f"/playlist/{uid}") def skip_to(self, uid: int) -> None: """Jump to a specific playlist item by uid. Requires rank >= MOD.""" self._request("PUT", "/playlist/playing", json={"uid": uid}) def shuffle_playlist(self) -> None: """Shuffle the playlist. Requires rank >= ADMIN.""" self._request("POST", "/playlist/shuffle") def clear_playlist(self) -> None: """Clear the entire playlist. Requires rank >= ADMIN.""" self._request("POST", "/playlist/clear") # ── Emotes ──────────────────────────────────────────────────────────────── def get_emotes(self) -> list: """Returns list of { name, image, source }. Works offline.""" return self._request("GET", "/emotes") def add_emote(self, name: str, image: str) -> None: """Add a new emote. Requires rank >= OWNER. Raises BotAPIError(409) if name taken.""" self._request("POST", "/emotes", json={"name": name, "image": image}) def update_emote(self, name: str, image: str | None = None, new_name: str | None = None) -> None: """Update an emote's image or rename it. Requires rank >= OWNER.""" body: dict = {} if image is not None: body["image"] = image if new_name is not None: body["newName"] = new_name self._request("PUT", f"/emotes/{urllib.parse.quote(name)}", json=body) def delete_emote(self, name: str) -> None: """Delete an emote by name. Requires rank >= OWNER.""" self._request("DELETE", f"/emotes/{urllib.parse.quote(name)}") # ── Users / Moderation ──────────────────────────────────────────────────── def get_users(self) -> list: """List connected users. Returns list of { name, rank, afk, is_bot }.""" return self._request("GET", "/users") def kick_user(self, name: str, reason: str = "Kicked by bot") -> None: """Kick a user. Requires rank >= MOD.""" self._request( "POST", f"/users/{urllib.parse.quote(name)}/kick", json={"reason": reason}, ) def ban_user(self, name: str, reason: str = "Banned by bot") -> None: """Ban a user. Requires rank >= ADMIN.""" self._request( "POST", f"/users/{urllib.parse.quote(name)}/ban", json={"reason": reason}, ) def unban_user(self, name: str) -> None: """Remove a ban. Requires rank >= ADMIN.""" self._request("DELETE", f"/users/{urllib.parse.quote(name)}/ban") def set_user_rank(self, name: str, rank: int) -> None: """Change a user's channel rank. Requires rank >= OWNER.""" self._request( "PUT", f"/users/{urllib.parse.quote(name)}/rank", json={"rank": rank}, ) # ── Settings ────────────────────────────────────────────────────────────── def get_settings(self) -> dict: """Get channel settings. Requires rank >= OWNER and channel active.""" return self._request("GET", "/settings") def update_settings(self, **kwargs) -> None: """Update one or more channel settings. Requires rank >= OWNER and channel active.""" self._request("PUT", "/settings", json=kwargs) # ── Shows ───────────────────────────────────────────────────────────────── _SHOW_COLOR_RE = re.compile(r"^#[0-9A-Fa-f]{6}$") @staticmethod def _coerce_timestamp_ms(value): if isinstance(value, (int, float)): return float(value) if isinstance(value, str): text = value.strip() if not text: return None if text.endswith("Z"): text = f"{text[:-1]}+00:00" try: return datetime.fromisoformat(text).timestamp() * 1000.0 except ValueError: return None return None def _validate_show_payload(self, payload: dict) -> None: name = payload.get("name") if not isinstance(name, str) or not name.strip(): raise ValueError("name is required and must be a non-empty string") if len(name) > 100: raise ValueError("name must be at most 100 characters") timezone = payload.get("timezone") if not isinstance(timezone, str) or not timezone.strip(): raise ValueError("timezone is required and must be a non-empty IANA timezone string") scheduled_for = payload.get("scheduled_for") if scheduled_for is None or (isinstance(scheduled_for, str) and not scheduled_for.strip()): raise ValueError("scheduled_for is required and must be a date string or unix timestamp (ms)") if not isinstance(scheduled_for, (str, int, float)): raise ValueError("scheduled_for must be a date string or unix timestamp (ms)") estimated_end_at = payload.get("estimated_end_at") if estimated_end_at is not None and not isinstance(estimated_end_at, (str, int, float)): raise ValueError("estimated_end_at must be a date string or unix timestamp (ms)") start_ts = self._coerce_timestamp_ms(scheduled_for) end_ts = self._coerce_timestamp_ms(estimated_end_at) if estimated_end_at is not None and start_ts is not None and end_ts is not None and end_ts <= start_ts: raise ValueError("estimated_end_at must be later than scheduled_for") recurrence = payload.get("recurrence") if recurrence is not None and recurrence not in {"none", "daily", "weekly"}: raise ValueError("recurrence must be one of: none, daily, weekly") fill_mode = payload.get("fill_mode") if fill_mode is not None and fill_mode not in {"append", "replace"}: raise ValueError("fill_mode must be one of: append, replace") conflict_mode = payload.get("conflict_mode") if conflict_mode is not None and conflict_mode not in {"force", "skip"}: raise ValueError("conflict_mode must be one of: force, skip") status = payload.get("status") if status is not None and status not in {"draft", "scheduled", "paused", "running", "completed", "failed", "canceled"}: raise ValueError("status must be one of: draft, scheduled, paused, running, completed, failed, canceled") notes = payload.get("notes") if notes is not None: if not isinstance(notes, str): raise ValueError("notes must be a string or null") if len(notes) > 20000: raise ValueError("notes must be at most 20000 characters") color = payload.get("color") if color is not None: if not isinstance(color, str): raise ValueError("color must be a string or null") if not self._SHOW_COLOR_RE.fullmatch(color): raise ValueError("color must match /^#[0-9A-Fa-f]{6}$/") playlist = payload.get("playlist") if not isinstance(playlist, list) or not playlist: raise ValueError("playlist is required and must be a non-empty list") for index, item in enumerate(playlist): if not isinstance(item, dict): raise ValueError(f"playlist[{index}] must be an object") if not item.get("type") or not item.get("id"): raise ValueError(f"playlist[{index}] must include non-empty type and id") pos = item.get("pos") if pos is not None and pos not in {"next", "end"}: raise ValueError(f"playlist[{index}].pos must be one of: next, end") def list_shows(self) -> list: """List channel shows. Requires rank >= MOD.""" return self._request("GET", "/shows") def list_public_shows(self) -> list: """List public show schedule. Read-only.""" return self._request("GET", "/shows/public") def resolve_show_media(self, items: list[dict]) -> dict: """Resolve up to 50 media entries into display-ready titles. Requires rank >= MOD.""" if not isinstance(items, list) or not items: raise ValueError("items is required and must be a non-empty list") if len(items) > 50: raise ValueError("items may contain at most 50 entries") for index, item in enumerate(items): if not isinstance(item, dict): raise ValueError(f"items[{index}] must be an object") if not item.get("type") or not item.get("id"): raise ValueError(f"items[{index}] must include non-empty type and id") return self._request("POST", "/shows/resolve-media", json={"items": items}) def get_show(self, show_id: int | str) -> dict: """Get a single show by id. Requires rank >= MOD.""" return self._request("GET", f"/shows/{show_id}") def create_show(self, payload: dict) -> dict: """Create a show. Requires rank >= MOD.""" self._validate_show_payload(payload) return self._request("POST", "/shows", json=payload) def update_show(self, show_id: int | str, payload: dict) -> dict: """Update a show by id. Requires rank >= MOD.""" self._validate_show_payload(payload) return self._request("PUT", f"/shows/{show_id}", json=payload) def delete_show(self, show_id: int | str) -> None: """Delete a show by id. Requires rank >= ADMIN.""" self._request("DELETE", f"/shows/{show_id}") def show_action(self, show_id: int | str, action: str) -> dict: """ Run a show action. pause/resume/schedule require rank >= MOD. run/cancel require rank >= ADMIN. """ if action not in {"pause", "resume", "schedule", "run", "cancel"}: raise ValueError("action must be one of: pause, resume, schedule, run, cancel") return self._request("POST", f"/shows/{show_id}/action", json={"action": action})