veretube_bot_lib/veretube_bot/_api.py

import urllib.parse

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 ─────────────────────────────────────────────────────────────────

    def _validate_show_payload(self, payload: dict) -> None:
        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")

        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", "completed", "failed", "canceled"}:
            raise ValueError("status must be one of: draft, scheduled, paused, completed, failed, canceled")

        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 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})