Maddox
8c62c9fd31
Maubot plugin: Matrix companion for the homelab media stack. Services wrapped: - Seerr: search, request, requests, trending - Emby: nowplaying, recent, watched - Sonarr/Radarr: queue, upcoming, missing - NZBGet/qBittorrent: activity Each Matrix sender is mapped to per-service user IDs via plugin config; unmapped senders are rejected. Replies are MXID-prefixed for shared rooms.
52 lines
2 KiB
Python
52 lines
2 KiB
Python
"""Emby HTTP client.
|
|
|
|
API key passed as ?api_key=... (also accepts X-Emby-Token header).
|
|
Base URL must include the /emby path prefix.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import aiohttp
|
|
|
|
|
|
class EmbyError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class EmbyClient:
|
|
def __init__(self, session: aiohttp.ClientSession, base_url: str, api_key: str) -> None:
|
|
self.session = session
|
|
self.base = base_url.rstrip("/")
|
|
self.api_key = api_key
|
|
|
|
async def _get(self, path: str, params: dict | None = None) -> dict | list:
|
|
merged = {"api_key": self.api_key, **(params or {})}
|
|
async with self.session.get(f"{self.base}{path}", params=merged) as r:
|
|
if r.status >= 400:
|
|
raise EmbyError(f"GET {path} → {r.status}: {(await r.text())[:200]}")
|
|
return await r.json()
|
|
|
|
async def sessions(self) -> list[dict]:
|
|
data = await self._get("/Sessions")
|
|
return data if isinstance(data, list) else (data.get("Items") or [])
|
|
|
|
async def recently_added(self, user_id: str, limit: int = 10,
|
|
item_types: str | None = None) -> list[dict]:
|
|
params: dict = {"Limit": limit, "Fields": "PremiereDate,ProductionYear"}
|
|
if item_types:
|
|
params["IncludeItemTypes"] = item_types
|
|
data = await self._get(f"/Users/{user_id}/Items/Latest", params=params)
|
|
return data if isinstance(data, list) else (data.get("Items") or [])
|
|
|
|
async def user_played(self, user_id: str, limit: int = 10) -> list[dict]:
|
|
params = {
|
|
"IncludeItemTypes": "Movie,Episode",
|
|
"Recursive": "true",
|
|
"Filters": "IsPlayed",
|
|
"SortBy": "DatePlayed",
|
|
"SortOrder": "Descending",
|
|
"Limit": limit,
|
|
"Fields": "ProductionYear,SeriesName,IndexNumber,ParentIndexNumber",
|
|
}
|
|
data = await self._get(f"/Users/{user_id}/Items", params=params)
|
|
return (data or {}).get("Items", []) if isinstance(data, dict) else (data or [])
|