auto-commit for b082c1c8-f6e3-4cfd-82ee-0396c70d1a00

backend/trakt.py

@@ -0,0 +1,101 @@
+"""Trakt.tv API client.
+
+Uses OAuth 2.0 device code flow:
+  1. POST /oauth/device/code with client_id → get device_code + user_code + verification_url
+  2. User visits trakt.tv/activate, enters user_code
+  3. Poll POST /oauth/device/token with device_code until user approves → access_token + refresh_token
+  4. Use access_token as Bearer for all subsequent calls.
+
+Scrobble: POST /scrobble/{start|pause|stop} with movie/episode + progress (0-100).
+"""
+import requests
+from typing import Dict, Optional
+
+BASE = "https://api.trakt.tv"
+
+
+def _headers(client_id: str, access_token: Optional[str] = None) -> dict:
+    h = {
+        "Content-Type": "application/json",
+        "trakt-api-version": "2",
+        "trakt-api-key": client_id,
+    }
+    if access_token:
+        h["Authorization"] = f"Bearer {access_token}"
+    return h
+
+
+def device_code(client_id: str) -> Dict:
+    r = requests.post(f"{BASE}/oauth/device/code", json={"client_id": client_id}, timeout=15)
+    r.raise_for_status()
+    d = r.json()
+    return {
+        "device_code": d["device_code"],
+        "user_code": d["user_code"],
+        "verification_url": d["verification_url"],
+        "expires_in": d["expires_in"],
+        "interval": d["interval"],
+    }
+
+
+def poll_token(client_id: str, client_secret: str, device_code_val: str) -> Optional[Dict]:
+    """Returns access_token dict on success, None if still pending, raises on error."""
+    r = requests.post(f"{BASE}/oauth/device/token", json={
+        "code": device_code_val,
+        "client_id": client_id,
+        "client_secret": client_secret,
+    }, timeout=15)
+    if r.status_code == 200:
+        return r.json()  # {access_token, refresh_token, expires_in, created_at, ...}
+    if r.status_code in (400, 404):  # pending / expired
+        return None
+    r.raise_for_status()
+    return None
+
+
+def refresh(client_id: str, client_secret: str, refresh_token: str) -> Dict:
+    r = requests.post(f"{BASE}/oauth/token", json={
+        "refresh_token": refresh_token,
+        "client_id": client_id,
+        "client_secret": client_secret,
+        "grant_type": "refresh_token",
+    }, timeout=15)
+    r.raise_for_status()
+    return r.json()
+
+
+def user_settings(client_id: str, access_token: str) -> Dict:
+    r = requests.get(f"{BASE}/users/settings", headers=_headers(client_id, access_token), timeout=15)
+    r.raise_for_status()
+    return r.json()
+
+
+def _scrobble(action: str, client_id: str, access_token: str, payload: dict):
+    r = requests.post(f"{BASE}/scrobble/{action}",
+                      headers=_headers(client_id, access_token),
+                      json=payload, timeout=15)
+    if r.status_code in (409, 429):  # duplicate / rate-limit
+        return None
+    r.raise_for_status()
+    return r.json()
+
+
+def scrobble_movie(client_id: str, access_token: str, tmdb_id: int, progress: float, action: str = "stop"):
+    return _scrobble(action, client_id, access_token, {
+        "movie": {"ids": {"tmdb": int(tmdb_id)}},
+        "progress": max(0.0, min(100.0, float(progress))),
+    })
+
+
+def scrobble_episode(client_id: str, access_token: str, tvdb_id: Optional[int], tmdb_id: Optional[int],
+                     season: int, episode: int, progress: float, action: str = "stop"):
+    ids = {}
+    if tvdb_id: ids["tvdb"] = int(tvdb_id)
+    if tmdb_id: ids["tmdb"] = int(tmdb_id)
+    if not ids:
+        return None
+    return _scrobble(action, client_id, access_token, {
+        "show": {"ids": ids},
+        "episode": {"season": int(season), "number": int(episode)},
+        "progress": max(0.0, min(100.0, float(progress))),
+    })