mirror of
https://github.com/myronblair/kino-app
synced 2026-07-28 13:33:49 -05:00
emergent-agent-e1
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
"""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))),
|
|
})
|