from __future__ import annotations import asyncio import logging import time from contextlib import asynccontextmanager from datetime import datetime, timezone from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse from app.config import Settings from app.plex_service import PlexService logger = logging.getLogger("plex-age-setter") class RateLimiter: def __init__(self, max_per_minute: int): self.max_per_minute = max_per_minute self._window_start: float = 0 self._count: int = 0 def allow(self) -> bool: now = time.monotonic() if now - self._window_start > 60: self._window_start = now self._count = 0 if self._count >= self.max_per_minute: return False self._count += 1 return True class AppState: def __init__(self, settings: Settings): self.settings = settings self.plex = PlexService( baseurl=settings.plex_baseurl, token=settings.plex_token, username=settings.plex_username, password=settings.plex_password, ) self.trigger_limiter = RateLimiter(settings.trigger_rate_limit) self._health = True self._last_run: str | None = None @property def last_run(self) -> str | None: return self._last_run @last_run.setter def last_run(self, value: str | None): self._last_run = value state: AppState | None = None # ── MCP JSON-RPC helpers ───────────────────────────────────── def mcp_error(id_val: int | str | None, code: int, message: str) -> dict: return {"jsonrpc": "2.0", "id": id_val, "error": {"code": code, "message": message}} def mcp_result(id_val: int | str | None, result: object) -> dict: return {"jsonrpc": "2.0", "id": id_val, "result": result} def handle_mcp_request(body: dict) -> dict: assert state is not None, "App not initialized" req_id = body.get("id") method = body.get("method", "") params = body.get("params", {}) if method == "tools/list": tools = PlexService.tool_registry() return mcp_result( req_id, { "tools": [ { "name": t.name, "description": t.description, "inputSchema": t.inputSchema, } for t in tools ] }, ) if method == "tools/call": tool_name = params.get("name", "") arguments = params.get("arguments", {}) try: result = state.plex.call_tool(tool_name, arguments) return mcp_result( req_id, {"content": [{"type": "text", "text": str(result)}]} ) except ValueError as e: return mcp_error(req_id, -32601, str(e)) except Exception as e: logger.exception("Tool call failed: %s", tool_name) return mcp_error(req_id, -32000, str(e)) if method == "resources/list": return mcp_result(req_id, {"resources": []}) return mcp_error(req_id, -32601, f"Method not found: {method}") # ── FastAPI app ────────────────────────────────────────────── @asynccontextmanager async def lifespan(app: FastAPI): global state settings = Settings() state = AppState(settings) logger.info("Plex Age Limit Setter started") async def scheduled_task(): st = state assert st is not None while True: now = datetime.now(timezone.utc).isoformat() logger.info("Running scheduled task at %s", now) try: result = st.plex.run_scheduled_task( library=settings.target_library, target_rating=settings.target_content_rating, ) st.last_run = now logger.info( "Scheduled task complete: updated=%d skipped=%d errors=%d", result.get("updated", 0), result.get("skipped", 0), result.get("total_errors", 0), ) except Exception as e: logger.exception("Scheduled task failed: %s", e) await asyncio.sleep(settings.schedule_interval_minutes * 60) task = asyncio.create_task(scheduled_task()) yield task.cancel() logger.info("Plex Age Limit Setter stopped") app = FastAPI(lifespan=lifespan, title="Plex Age Limit Setter") @app.get("/health") def health(): st = state return {"status": "healthy", "last_run": st.last_run if st else None} @app.post("/trigger") def trigger(request: Request): if state is None or not state.trigger_limiter.allow(): raise HTTPException(status_code=429, detail="Rate limit exceeded (1 req/min)") try: result = state.plex.run_scheduled_task( library=state.settings.target_library, target_rating=state.settings.target_content_rating, ) state.last_run = datetime.now(timezone.utc).isoformat() return result except Exception as e: logger.exception("Trigger failed") raise HTTPException(status_code=500, detail=str(e)) @app.post("/mcp") async def mcp_endpoint(request: Request): try: body = await request.json() except Exception: return JSONResponse( status_code=400, content={ "jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": "Parse error"}, }, ) if not isinstance(body, dict) or "method" not in body: return JSONResponse( status_code=400, content={ "jsonrpc": "2.0", "id": None, "error": {"code": -32600, "message": "Invalid Request"}, }, ) result = handle_mcp_request(body) return JSONResponse(content=result)