122 lines
5.2 KiB
Python
122 lines
5.2 KiB
Python
|
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)
|