""" Watch History skill — fetch the user's Jellyfin watch history via JellyStat API. Requires the user to have linked Jellyfin via `/login jellyfin` in Discord. The auth gate (`requires_auth=["jellyfin"]`) is already active — users who haven't linked Jellyfin will be prompted to /login first. Architecture ------------ This skill calls the JellyStat REST API (same FastAPI process, via HTTP) rather than accessing the PostgreSQL database directly. This keeps the bot isolated from database credentials. """ from __future__ import annotations import httpx from agents.skills import Skill, register, ToolResult from src import auth_store from src.config import get_config # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- BASE_URL = (get_config("BASE_URL") or "http://localhost:8000").rstrip("/") # --------------------------------------------------------------------------- # Tool definitions # --------------------------------------------------------------------------- TOOLS = [ { "type": "function", "function": { "name": "watch_history", "description": ( "Get the user's Jellyfin watch history — titles grouped by total " "watch time in a configurable time window. Use this when a user " "asks what they've watched, what they've been watching recently, " "or wants to see their viewing activity." ), "parameters": { "type": "object", "properties": { "limit": { "type": "integer", "description": "How many titles to return (default 10, max 20).", }, "minutes": { "type": "integer", "description": ( "Time window in minutes. Default 10080 (7 days). " "Use a large number like 525600 for 'all time' (1 year)." ), }, }, }, }, }, { "type": "function", "function": { "name": "watch_genres", "description": ( "Get the user's most-watched genres from Jellyfin, ranked by " "total watch time. Use this when a user asks what kinds of " "content they watch most, their favourite genres, or what " "categories dominate their viewing." ), "parameters": { "type": "object", "properties": { "minutes": { "type": "integer", "description": ( "Time window in minutes. Default 10080 (7 days). " "Use a large number like 525600 for 'all time'." ), }, }, }, }, }, { "type": "function", "function": { "name": "watch_summary", "description": ( "Get an all-time Jellyfin watch summary — total watch time, " "most-watched series, most-watched movie, 30-day and 7-day " "activity, and top 3 genres. Use this when a user asks for " "their overall stats, a dashboard, or 'how much have I watched?'." ), "parameters": {"type": "object", "properties": {}}, }, }, ] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _resolve_jellyfin_id(args: dict) -> str | None: """Extract the Jellyfin user ID from auth_store using the injected Discord ID.""" discord_user_id = args.pop("_discord_user_id", None) if discord_user_id is None: return None # not called from Discord — shouldn't happen with auth gate auth = auth_store.get_auth(discord_user_id, "jellyfin") if auth is None or not auth.get("external_user_id"): return None return auth["external_user_id"] async def _fetch_json(url: str) -> dict: """GET *url* and return the parsed JSON body, or {} on failure.""" async with httpx.AsyncClient(timeout=10) as client: resp = await client.get(url) resp.raise_for_status() return resp.json() def _format_seconds(total: float) -> str: """Convert seconds to a human-friendly string.""" total = max(total, 0) hours = int(total // 3600) minutes = int((total % 3600) // 60) if hours and minutes: return f"{hours}h {minutes}m" if hours: return f"{hours}h" if minutes: return f"{minutes}m" return f"{int(total)}s" def _format_history(data: dict, limit: int) -> ToolResult: """Format a watch-history API response for the LLM.""" items = data.get("items", [])[:limit] if not items: return ToolResult.ok("You haven't watched anything in this time window.") lines = [f"**Watch History** (last {data.get('window_minutes', '?')} minutes):"] for i, item in enumerate(items, 1): duration = _format_seconds(item["watch_time_sec"]) icon = "📺" if item["media_type"] == "series" else "🎬" lines.append(f"{i}. {icon} **{item['title']}** — {duration}") return ToolResult.ok("\n".join(lines)) def _format_genres(data: dict) -> ToolResult: """Format a genre-summary API response for the LLM.""" genres = data.get("genres", []) if not genres: return ToolResult.ok("No genre data available for this time window.") lines = [f"**Top Genres** (last {data.get('window_minutes', '?')} minutes):"] for i, g in enumerate(genres, 1): duration = _format_seconds(g["watch_time_sec"]) lines.append(f"{i}. **{g['genre']}** — {duration}") return ToolResult.ok("\n".join(lines)) def _format_summary(data: dict) -> ToolResult: """Format a user-summary API response for the LLM.""" total = _format_seconds(data.get("total_watch_time_sec", 0)) last_30 = _format_seconds(data.get("total_last_30d_sec", 0)) last_7 = _format_seconds(data.get("total_last_7d_sec", 0)) top_series = data.get("most_watched_series") or "—" top_movie = data.get("most_watched_movie") or "—" top_genres = data.get("top_genres", []) genres_str = ", ".join(top_genres) if top_genres else "—" lines = [ "**Your Jellyfin Summary** (all time):", f"⏱️ Total watch time: **{total}**", f"📺 Most-watched series: **{top_series}**", f"🎬 Most-watched movie: **{top_movie}**", f"📅 Last 30 days: **{last_30}**", f"📅 Last 7 days: **{last_7}**", f"🏷️ Top genres: {genres_str}", ] return ToolResult.ok("\n".join(lines)) # --------------------------------------------------------------------------- # Executor # --------------------------------------------------------------------------- async def _execute(tool_name: str, args: dict) -> ToolResult: # 1. Resolve Jellyfin user ID jellyfin_id = _resolve_jellyfin_id(args) if jellyfin_id is None: return ToolResult.fail( "Your Jellyfin account is not linked. Use `/login jellyfin` in a DM to connect." ) # 2. Route to the right JellyStat endpoint try: match tool_name: case "watch_history": limit = args.get("limit", 10) minutes = args.get("minutes", 10080) url = f"{BASE_URL}/jellystat/history/{jellyfin_id}?minutes={minutes}" data = await _fetch_json(url) return _format_history(data, limit) case "watch_genres": minutes = args.get("minutes", 10080) url = f"{BASE_URL}/jellystat/genres/{jellyfin_id}?minutes={minutes}" data = await _fetch_json(url) return _format_genres(data) case "watch_summary": url = f"{BASE_URL}/jellystat/summary/{jellyfin_id}" data = await _fetch_json(url) return _format_summary(data) case _: return ToolResult.fail(f"Unknown tool: {tool_name}") except httpx.HTTPError: return ToolResult.fail( "Could not reach the watch-history service right now. " "Please try again in a moment." ) # --------------------------------------------------------------------------- # Skill registration # --------------------------------------------------------------------------- _PROMPT = ( "## Watch History\n" "\n" "You have THREE tools to answer questions about the user's Jellyfin watch activity:\n" "\n" "1. **`watch_history`** — per-title watch time in a time window (default: 7 days).\n" " Use when a user asks what they've watched, to show their history,\n" " or what they watched this week or yesterday.\n" "\n" "2. **`watch_genres`** — watch time broken down by genre.\n" " Use when a user asks what genres they watch, whether they watch more\n" " comedy than drama, or what their most-watched genre is.\n" "\n" "3. **`watch_summary`** — all-time dashboard: total watch time, most-watched\n" " series and movie, 30-day and 7-day activity, and top 3 genres.\n" " Use when a user asks for their stats, how much they've watched in\n" " total, or what their favourites are.\n" "\n" "Always call the appropriate tool before answering — NEVER guess at watch data.\n" "Format watch times in a human-readable way (hours and minutes), but keep the\n" "raw data visible too." ) watch_history_skill = Skill( name="watch_history", description="User's Jellyfin watch history, genres, and summary stats", requires_auth=["jellyfin"], prompt_fragment=_PROMPT, tools=TOOLS, execute=_execute, ) register(watch_history_skill)