LogalDeveloper/Owlbot
0
Fork 0
Code Issues 13 Actions Packages Releases 3 Wiki Activity

Added pydocstyle (D) rules to Ruff and fixed all violations.
CI / Formatting (push) Successful in 12s
Details
CI / Linting (push) Successful in 13s
Details
CI / Tests (Python 3.12) (push) Successful in 26s
Details
CI / Tests (Python 3.13) (push) Successful in 25s
Details
CI / Tests (Python 3.14) (push) Successful in 25s
Details
CI / Type Checking (push) Successful in 26s
Details

Browse Source
This commit is contained in:
LogalDeveloper
2026-02-19 11:47:47 -05:00
parent 33dd49e20a
commit ca4adbcebf
30 changed files with 309 additions and 591 deletions
Download Patch File Download Diff File Expand all files Collapse all files
owlbot/api/owncast_admin_client.py
+77 -154
View File
@@ -186,8 +186,7 @@ class OwncastAdminClient(OwncastClient):
def __init__(
self, base_url: str, username: str, password: str, http_client: HttpClient
):
"""
Initialize the admin client.
"""Initialize the admin client.
:param base_url: The Owncast server URL (e.g., "https://stream.logal.dev").
:param username: Admin username.
@@ -203,24 +202,21 @@ class OwncastAdminClient(OwncastClient):
)
async def get_status(self) -> dict[str, Any]:
"""
Get the current server status including stream info and viewer count.
"""Get the current server status including stream info and viewer count.
:return: Server status dict.
"""
return dict(await self._get("/api/admin/status"))
async def get_active_viewers(self) -> list[dict[str, Any]]:
"""
Get a list of currently active viewers.
"""Get a list of currently active viewers.
:return: Viewer list as returned by the Owncast API.
"""
return list(await self._get("/api/admin/viewers"))
async def get_viewers_over_time(self, window_start: int) -> list[dict[str, Any]]:
"""
Get viewer count data over time for charting.
"""Get viewer count data over time for charting.
:param window_start: Unix timestamp (seconds since epoch)
for the start of the window.
@@ -233,24 +229,21 @@ class OwncastAdminClient(OwncastClient):
)
async def get_hardware_stats(self) -> dict[str, Any]:
"""
Get server hardware statistics (CPU, memory, disk).
"""Get server hardware statistics (CPU, memory, disk).
:return: Hardware stats dict.
"""
return dict(await self._get("/api/admin/hardwarestats"))
async def get_server_config(self) -> dict[str, Any]:
"""
Get the full server configuration.
"""Get the full server configuration.
:return: Server configuration dict.
"""
return dict(await self._get("/api/admin/serverconfig"))
async def disconnect_stream(self) -> dict[str, Any]:
"""
Disconnect the current inbound stream.
"""Disconnect the current inbound stream.
:return: API response confirming the disconnect.
"""
@@ -258,16 +251,14 @@ class OwncastAdminClient(OwncastClient):
return dict(await self._get("/api/admin/disconnect"))
async def get_chat_messages(self) -> list[dict[str, Any]]:
"""
Get chat messages from the admin perspective.
"""Get chat messages from the admin perspective.
:return: Chat message list as returned by the Owncast API.
"""
return list(await self._get("/api/admin/chat/messages"))
async def get_connected_chat_clients(self) -> list[dict[str, Any]]:
"""
Get currently connected chat clients.
"""Get currently connected chat clients.
:return: Client list as returned by the Owncast API.
"""
@@ -276,8 +267,7 @@ class OwncastAdminClient(OwncastClient):
async def set_message_visibility(
self, message_ids: list[str], visible: bool
) -> str:
"""
Hide or show chat messages.
"""Hide or show chat messages.
:param message_ids: List of message IDs to modify.
:param visible: True to show messages, False to hide them.
@@ -292,8 +282,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_user_enabled(self, user_id: str, enabled: bool) -> str:
"""
Enable or disable a chat user.
"""Enable or disable a chat user.
:param user_id: The user ID to modify.
:param enabled: True to enable, False to disable the user.
@@ -307,16 +296,14 @@ class OwncastAdminClient(OwncastClient):
)
async def get_disabled_users(self) -> list[dict[str, Any]]:
"""
Get a list of disabled chat users.
"""Get a list of disabled chat users.
:return: User list as returned by the Owncast API.
"""
return list(await self._get("/api/admin/chat/users/disabled"))
async def set_user_moderator(self, user_id: str, is_mod: bool) -> str:
"""
Grant or revoke moderator status for a user.
"""Grant or revoke moderator status for a user.
:param user_id: The user ID to modify.
:param is_mod: True to grant moderator, False to revoke.
@@ -330,16 +317,14 @@ class OwncastAdminClient(OwncastClient):
)
async def get_moderators(self) -> list[dict[str, Any]]:
"""
Get a list of moderator users.
"""Get a list of moderator users.
:return: User list as returned by the Owncast API.
"""
return list(await self._get("/api/admin/chat/users/moderators"))
async def ban_ip_address(self, ip: str) -> str:
"""
Ban an IP address from chat.
"""Ban an IP address from chat.
:param ip: The IP address to ban.
:return: Success message from the server.
@@ -348,8 +333,7 @@ class OwncastAdminClient(OwncastClient):
return await self._post("/api/admin/chat/users/ipbans/create", {"value": ip})
async def unban_ip_address(self, ip: str) -> str:
"""
Remove an IP address ban.
"""Remove an IP address ban.
:param ip: The IP address to unban.
:return: Success message from the server.
@@ -358,16 +342,14 @@ class OwncastAdminClient(OwncastClient):
return await self._post("/api/admin/chat/users/ipbans/remove", {"value": ip})
async def get_ip_address_bans(self) -> list[dict[str, Any]]:
"""
Get a list of banned IP addresses.
"""Get a list of banned IP addresses.
:return: IP ban list as returned by the Owncast API.
"""
return list(await self._get("/api/admin/chat/users/ipbans"))
async def set_stream_title(self, title: str) -> str:
"""
Set the stream title.
"""Set the stream title.
:param title: The new stream title.
:return: Success message from the server.
@@ -377,8 +359,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_server_name(self, name: str) -> str:
"""
Set the server name.
"""Set the server name.
:param name: The new server name.
:return: Success message from the server.
@@ -388,8 +369,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_server_summary(self, summary: str) -> str:
"""
Set the server summary.
"""Set the server summary.
:param summary: The new server summary.
:return: Success message from the server.
@@ -399,8 +379,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_welcome_message(self, message: str) -> str:
"""
Set the welcome message shown to new viewers.
"""Set the welcome message shown to new viewers.
:param message: The new welcome message.
:return: Success message from the server.
@@ -410,8 +389,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_offline_message(self, message: str) -> str:
"""
Set the message shown when the stream is offline.
"""Set the message shown when the stream is offline.
:param message: The new offline message.
:return: Success message from the server.
@@ -421,8 +399,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_page_content(self, content: str) -> str:
"""
Set the custom page content (HTML/markdown below the player).
"""Set the custom page content (HTML/markdown below the player).
:param content: The page content.
:return: Success message from the server.
@@ -432,8 +409,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_admin_password(self, password: str) -> str:
"""
Change the admin password.
"""Change the admin password.
:param password: The new admin password.
:return: Success message from the server.
@@ -443,8 +419,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_server_url(self, url: str) -> str:
"""
Set the public server URL.
"""Set the public server URL.
:param url: The new server URL.
:return: Success message from the server.
@@ -454,8 +429,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_custom_styles(self, css: str) -> str:
"""
Set custom CSS styles for the web interface.
"""Set custom CSS styles for the web interface.
:param css: The CSS string.
:return: Success message from the server.
@@ -465,8 +439,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_custom_javascript(self, js: str) -> str:
"""
Set custom JavaScript for the web interface.
"""Set custom JavaScript for the web interface.
:param js: The JavaScript string.
:return: Success message from the server.
@@ -476,8 +449,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_socket_host_override(self, host: str) -> str:
"""
Set the WebSocket host override.
"""Set the WebSocket host override.
:param host: The WebSocket host override value.
:return: Success message from the server.
@@ -487,8 +459,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_ffmpeg_path(self, path: str) -> str:
"""
Set the path to the ffmpeg binary.
"""Set the path to the ffmpeg binary.
:param path: The ffmpeg binary path.
:return: Success message from the server.
@@ -498,8 +469,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_video_serving_endpoint(self, endpoint: str) -> str:
"""
Set the video serving endpoint (e.g., for CDN).
"""Set the video serving endpoint (e.g., for CDN).
:param endpoint: The video serving endpoint URL.
:return: Success message from the server.
@@ -509,8 +479,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_video_codec(self, codec: str) -> str:
"""
Set the video codec.
"""Set the video codec.
:param codec: The video codec name.
:return: Success message from the server.
@@ -520,8 +489,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_chat_disabled(self, disabled: bool) -> str:
"""
Enable or disable the chat.
"""Enable or disable the chat.
:param disabled: True to disable chat, False to enable it.
:return: Success message from the server.
@@ -531,8 +499,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_chat_join_messages_enabled(self, enabled: bool) -> str:
"""
Enable or disable chat join messages.
"""Enable or disable chat join messages.
:param enabled: True to show join messages, False to hide them.
:return: Success message from the server.
@@ -542,8 +509,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_chat_established_mode(self, enabled: bool) -> str:
"""
Enable or disable established user mode for chat.
"""Enable or disable established user mode for chat.
:param enabled: True to enable established mode.
:return: Success message from the server.
@@ -555,8 +521,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_chat_spam_protection(self, enabled: bool) -> str:
"""
Enable or disable chat spam protection.
"""Enable or disable chat spam protection.
:param enabled: True to enable spam protection.
:return: Success message from the server.
@@ -568,8 +533,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_chat_slur_filter(self, enabled: bool) -> str:
"""
Enable or disable the chat slur filter.
"""Enable or disable the chat slur filter.
:param enabled: True to enable the slur filter.
:return: Success message from the server.
@@ -579,8 +543,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_nsfw(self, nsfw: bool) -> str:
"""
Set the NSFW flag for the server.
"""Set the NSFW flag for the server.
:param nsfw: True to mark the server as NSFW.
:return: Success message from the server.
@@ -588,8 +551,7 @@ class OwncastAdminClient(OwncastClient):
return await self._set_config_value("/api/admin/config/nsfw", nsfw, "NSFW flag")
async def set_directory_enabled(self, enabled: bool) -> str:
"""
Enable or disable listing in the Owncast directory.
"""Enable or disable listing in the Owncast directory.
:param enabled: True to enable directory listing.
:return: Success message from the server.
@@ -599,8 +561,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_hide_viewer_count(self, hide: bool) -> str:
"""
Show or hide the viewer count.
"""Show or hide the viewer count.
:param hide: True to hide the viewer count.
:return: Success message from the server.
@@ -610,8 +571,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_disable_search_indexing(self, disabled: bool) -> str:
"""
Enable or disable search engine indexing.
"""Enable or disable search engine indexing.
:param disabled: True to disable search indexing.
:return: Success message from the server.
@@ -623,8 +583,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_forbidden_usernames(self, names: list[str]) -> str:
"""
Set the list of forbidden usernames.
"""Set the list of forbidden usernames.
:param names: List of forbidden username strings.
:return: Success message from the server.
@@ -634,8 +593,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_suggested_usernames(self, names: list[str]) -> str:
"""
Set the list of suggested usernames for new viewers.
"""Set the list of suggested usernames for new viewers.
:param names: List of suggested username strings.
:return: Success message from the server.
@@ -645,8 +603,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_tags(self, tags: list[str]) -> str:
"""
Set the server tags.
"""Set the server tags.
:param tags: List of tag strings.
:return: Success message from the server.
@@ -654,8 +611,7 @@ class OwncastAdminClient(OwncastClient):
return await self._set_config_value("/api/admin/config/tags", tags, "tags")
async def set_federation_blocked_domains(self, domains: list[str]) -> str:
"""
Set the list of blocked federation domains.
"""Set the list of blocked federation domains.
:param domains: List of domain strings to block.
:return: Success message from the server.
@@ -667,8 +623,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_stream_keys(self, keys: list[StreamKey]) -> str:
"""
Set the stream keys.
"""Set the stream keys.
:param keys: List of :class:`StreamKey` objects.
:return: Success message from the server.
@@ -680,8 +635,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_video_variants(self, variants: list[VideoVariant]) -> str:
"""
Set the video output variants (quality levels).
"""Set the video output variants (quality levels).
:param variants: List of :class:`VideoVariant` objects.
:return: Success message from the server.
@@ -693,8 +647,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_social_handles(self, handles: list[SocialHandle]) -> str:
"""
Set the social media handles displayed on the page.
"""Set the social media handles displayed on the page.
:param handles: List of :class:`SocialHandle` objects.
:return: Success message from the server.
@@ -706,8 +659,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_external_actions(self, actions: list[ExternalAction]) -> str:
"""
Set the external actions (buttons/links in the player).
"""Set the external actions (buttons/links in the player).
:param actions: List of :class:`ExternalAction` objects.
:return: Success message from the server.
@@ -727,8 +679,7 @@ class OwncastAdminClient(OwncastClient):
bucket: str,
region: str,
) -> str:
"""
Set the S3 storage configuration.
"""Set the S3 storage configuration.
:param enabled: Whether S3 storage is enabled.
:param endpoint: The S3 endpoint URL.
@@ -757,8 +708,7 @@ class OwncastAdminClient(OwncastClient):
webhook: str,
go_live_message: str,
) -> str:
"""
Set the Discord notification configuration.
"""Set the Discord notification configuration.
:param enabled: Whether Discord notifications are enabled.
:param webhook: The Discord webhook URL.
@@ -780,8 +730,7 @@ class OwncastAdminClient(OwncastClient):
enabled: bool,
go_live_message: str,
) -> str:
"""
Set the browser notification configuration.
"""Set the browser notification configuration.
:param enabled: Whether browser notifications are enabled.
:param go_live_message: The message shown when going live.
@@ -797,8 +746,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_color_variables(self, variables: dict[str, Any]) -> str:
"""
Set the custom color variables for the web interface.
"""Set the custom color variables for the web interface.
:param variables: Color variables dict.
:return: Success message from the server.
@@ -808,8 +756,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_stream_latency(self, level: int) -> str:
"""
Set the stream latency level.
"""Set the stream latency level.
:param level: Latency level value.
:return: Success message from the server.
@@ -819,8 +766,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_rtmp_port(self, port: int) -> str:
"""
Set the RTMP server port.
"""Set the RTMP server port.
:param port: The RTMP port number.
:return: Success message from the server.
@@ -830,8 +776,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_web_server_port(self, port: int) -> str:
"""
Set the web server port.
"""Set the web server port.
:param port: The web server port number.
:return: Success message from the server.
@@ -841,8 +786,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_web_server_ip(self, ip: str) -> str:
"""
Set the web server bind IP address.
"""Set the web server bind IP address.
:param ip: The IP address to bind to.
:return: Success message from the server.
@@ -852,8 +796,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_federation_enabled(self, enabled: bool) -> str:
"""
Enable or disable federation (ActivityPub).
"""Enable or disable federation (ActivityPub).
:param enabled: True to enable federation.
:return: Success message from the server.
@@ -863,8 +806,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_federation_username(self, name: str) -> str:
"""
Set the federation (ActivityPub) username.
"""Set the federation (ActivityPub) username.
:param name: The federation username.
:return: Success message from the server.
@@ -874,8 +816,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_federation_go_live_message(self, message: str) -> str:
"""
Set the message sent to followers when going live.
"""Set the message sent to followers when going live.
:param message: The go-live notification message.
:return: Success message from the server.
@@ -887,8 +828,7 @@ class OwncastAdminClient(OwncastClient):
)
async def set_logo(self, base64_data_url: str) -> str:
"""
Set the server logo from a base64 data URL.
"""Set the server logo from a base64 data URL.
:param base64_data_url: Data URL string
(e.g., ``data:image/png;base64,iVBOR...``).
@@ -899,8 +839,7 @@ class OwncastAdminClient(OwncastClient):
)
async def upload_emoji(self, name: str, data_base64: str) -> str:
"""
Upload a custom emoji.
"""Upload a custom emoji.
:param name: The emoji name.
:param data_base64: Base64-encoded image data for the emoji.
@@ -912,8 +851,7 @@ class OwncastAdminClient(OwncastClient):
)
async def delete_emoji(self, name: str) -> str:
"""
Delete a custom emoji.
"""Delete a custom emoji.
:param name: The emoji name to delete.
:return: Success message from the server.
@@ -922,16 +860,14 @@ class OwncastAdminClient(OwncastClient):
return await self._post("/api/admin/emoji/delete", {"name": name})
async def get_webhooks(self) -> list[dict[str, Any]]:
"""
Get all registered webhooks.
"""Get all registered webhooks.
:return: Webhook list as returned by the Owncast API.
"""
return list(await self._get("/api/admin/webhooks"))
async def create_webhook(self, url: str, events: list[str]) -> str:
"""
Create a new webhook.
"""Create a new webhook.
:param url: The URL to send webhook events to.
:param events: List of event type strings to subscribe to.
@@ -943,8 +879,7 @@ class OwncastAdminClient(OwncastClient):
)
async def delete_webhook(self, webhook_id: int) -> str:
"""
Delete a webhook.
"""Delete a webhook.
:param webhook_id: The ID of the webhook to delete.
:return: Success message from the server.
@@ -953,16 +888,14 @@ class OwncastAdminClient(OwncastClient):
return await self._post("/api/admin/webhooks/delete", {"id": webhook_id})
async def get_access_tokens(self) -> list[dict[str, Any]]:
"""
Get all access tokens.
"""Get all access tokens.
:return: Token list as returned by the Owncast API.
"""
return list(await self._get("/api/admin/accesstokens"))
async def create_access_token(self, name: str, scopes: list[str]) -> str:
"""
Create a new access token.
"""Create a new access token.
:param name: Display name for the token.
:param scopes: List of permission scope strings.
@@ -974,8 +907,7 @@ class OwncastAdminClient(OwncastClient):
)
async def delete_access_token(self, token: str) -> str:
"""
Delete an access token.
"""Delete an access token.
:param token: The token string to delete.
:return: Success message from the server.
@@ -985,8 +917,7 @@ class OwncastAdminClient(OwncastClient):
return await self._post("/api/admin/accesstokens/delete", {"token": token})
async def get_followers(self, offset: int = 0, limit: int = 25) -> dict[str, Any]:
"""
Get a paginated list of followers.
"""Get a paginated list of followers.
:param offset: Number of followers to skip.
:param limit: Maximum number of followers to return.
@@ -999,24 +930,21 @@ class OwncastAdminClient(OwncastClient):
)
async def get_pending_follow_requests(self) -> list[dict[str, Any]]:
"""
Get pending follow requests.
"""Get pending follow requests.
:return: Follower request list as returned by the Owncast API.
"""
return list(await self._get("/api/admin/followers/pending"))
async def get_blocked_followers(self) -> list[dict[str, Any]]:
"""
Get blocked and rejected followers.
"""Get blocked and rejected followers.
:return: Follower list as returned by the Owncast API.
"""
return list(await self._get("/api/admin/followers/blocked"))
async def approve_follower(self, actor_iri: str, approved: bool) -> str:
"""
Approve or reject a follow request.
"""Approve or reject a follow request.
:param actor_iri: The ActivityPub actor IRI of the follower.
:param approved: True to approve, False to reject.
@@ -1030,8 +958,7 @@ class OwncastAdminClient(OwncastClient):
)
async def send_federated_message(self, message: str) -> str:
"""
Send a message to all followers via federation.
"""Send a message to all followers via federation.
:param message: The message text to send.
:return: Success message from the server.
@@ -1040,24 +967,21 @@ class OwncastAdminClient(OwncastClient):
return await self._post("/api/admin/federation/send", {"value": message})
async def get_logs(self) -> list[dict[str, Any]]:
"""
Get server logs.
"""Get server logs.
:return: Log entry list as returned by the Owncast API.
"""
return list(await self._get("/api/admin/logs"))
async def get_warnings(self) -> list[dict[str, Any]]:
"""
Get server warning and error logs.
"""Get server warning and error logs.
:return: Log entry list as returned by the Owncast API.
"""
return list(await self._get("/api/admin/logs/warnings"))
async def get_playback_metrics(self) -> dict[str, Any]:
"""
Get playback quality metrics.
"""Get playback quality metrics.
:return: Playback metrics as returned by the Owncast API.
"""
@@ -1066,8 +990,7 @@ class OwncastAdminClient(OwncastClient):
async def _set_config_value(
self, endpoint: str, value: Any, description: str
) -> str:
"""
Set a configuration value via POST with a ``{"value": ...}`` body.
"""Set a configuration value via POST with a ``{"value": ...}`` body.
:param endpoint: The config API endpoint path.
:param value: The value to set.