Fix viewing-history artwork and add transcode playback metrics

backend/app/routers/insights.py

@@ -6,10 +6,19 @@ from pydantic import BaseModel, ConfigDict, field_validator
 from ..auth import get_current_user
 from ..clients.jellystat import HistoryLimitError, JellystatError
 from ..services.insights import get_insights
+from ..services.insights_artwork import get_artwork
+from ..runtime import get_runtime_settings
 
 router = APIRouter(prefix="/insights", tags=["insights"])
 
 
+@router.get("/artwork/{item_id}")
+async def artwork(item_id: str, token: Annotated[str, Query(max_length=100)], user: dict = Depends(get_current_user)):
+    content, media_type = await get_artwork(user, get_runtime_settings(), item_id, token)
+    return Response(content=content, media_type=media_type,
+                    headers={"Cache-Control": "private, max-age=600", "Vary": "Cookie, Authorization", "X-Content-Type-Options": "nosniff"})
+
+
 class InsightsQuery(BaseModel):
     model_config = ConfigDict(extra="forbid")
     days: int = 30

backend/app/services/insights.py

@@ -1,5 +1,6 @@
 import asyncio
 import hashlib
+import json
 import math
 import sqlite3
 import time
@@ -12,10 +13,50 @@ from ..clients.jellyfin import JellyfinClient
 from ..clients.jellystat import JellystatClient, JellystatError
 from ..runtime import get_runtime_settings
 from .jellyfin_identity import link_user, linked_user_id
+from .insights_artwork import item_id as artwork_item_id, with_artwork
 
 _cache: dict[tuple, tuple[float, dict]] = {}
 CACHE_SECONDS = 60
 
+HARDWARE = {"amf": "AMD AMF", "qsv": "Intel Quick Sync", "nvenc": "NVIDIA NVENC",
+            "v4l2m2m": "V4L2", "vaapi": "VAAPI", "videotoolbox": "Apple VideoToolbox", "rkmpp": "Rockchip MPP"}
+HARDWARE_ENUM = {0: "none", 1: "amf", 2: "qsv", 3: "nvenc", 4: "v4l2m2m", 5: "vaapi", 6: "videotoolbox", 7: "rkmpp"}
+
+
+def add_transcoding(row, duration, media_type, totals, hardware, audio_codecs):
+    # Jellystat can retain stale transcoding metadata after a switch to DirectPlay.
+    method = row.get("PlayMethod")
+    if method not in {"Transcode", "DirectStream"}:
+        return
+    info = row.get("TranscodingInfo")
+    if isinstance(info, str):
+        try:
+            info = json.loads(info)
+        except ValueError:
+            info = None
+    info = info if isinstance(info, dict) else {}
+    video_present = media_type in {"movie", "episode"} or bool(info.get("VideoCodec"))
+    if method == "Transcode" and video_present:
+        if info.get("IsVideoDirect") is False:
+            totals["video_minutes"] += duration
+            value = info.get("HardwareAccelerationType")
+            value = HARDWARE_ENUM.get(value) if type(value) is int else str(value or "").strip().lower()
+            if value in HARDWARE:
+                totals["hardware_video_minutes"] += duration
+                hardware[HARDWARE[value]] += duration
+            elif value == "none":
+                totals["software_video_minutes"] += duration
+            else:
+                totals["unknown_hardware_minutes"] += duration
+        elif info.get("IsVideoDirect") is not True:
+            totals["unknown_video_minutes"] += duration
+    if info.get("IsAudioDirect") is False:
+        totals["audio_minutes"] += duration
+        codec = str(info.get("AudioCodec") or "Unknown").upper()[:30]
+        audio_codecs[codec] += duration
+    elif info.get("IsAudioDirect") is not True:
+        totals["unknown_audio_minutes"] += duration
+
 
 def _date(value) -> datetime:
     try:
@@ -85,6 +126,9 @@ def summarize(history: list, libraries: list, start: datetime, end: datetime) ->
     daily_seconds = defaultdict(float)
     clients = defaultdict(float)
     methods = defaultdict(float)
+    transcoding = dict.fromkeys(("video_minutes", "audio_minutes", "hardware_video_minutes", "software_video_minutes",
+                                "unknown_hardware_minutes", "unknown_video_minutes", "unknown_audio_minutes"), 0.0)
+    hardware, audio_codecs = defaultdict(float), defaultdict(float)
     titles = {}
     movie_ids, episode_ids, seen = set(), set(), set()
     recent = []
@@ -107,6 +151,7 @@ def summarize(history: list, libraries: list, start: datetime, end: datetime) ->
         episode_id = row.get("EpisodeId")
         library_type = library_types.get(str(row.get("ParentId")), "")
         media_type = "episode" if episode_id else "movie" if library_type == "movies" else "other"
+        add_transcoding(row, duration / 60, media_type, transcoding, hardware, audio_codecs)
         if media_type == "episode":
             episode_ids.add(str(episode_id))
         elif media_type == "movie":
@@ -126,7 +171,7 @@ def summarize(history: list, libraries: list, start: datetime, end: datetime) ->
         recent.append({"id": row_id, "title": name, "series": series, "type": media_type,
                        "episode": f"S{row.get('SeasonNumber', '?')} · E{row.get('EpisodeNumber', '?')}" if episode_id else None,
                        "minutes": round(duration / 60, 1), "played_at": date.isoformat(), "client": client,
-                       "method": method})
+                       "method": method, "artwork_item_id": artwork_item_id(row.get("NowPlayingItemId"))})
     count = (end.date() - start.date()).days + 1
     daily = [{"date": (start.date() + timedelta(days=i)).isoformat(),
               "minutes": round(daily_seconds.get((start.date() + timedelta(days=i)).isoformat(), 0) / 60, 2)} for i in range(count)]
@@ -149,6 +194,10 @@ def summarize(history: list, libraries: list, start: datetime, end: datetime) ->
             "daily": daily, "top_titles": top,
             "clients": [{"name": name, "minutes": round(value / 60, 1)} for name, value in sorted(clients.items(), key=lambda pair: -pair[1])[:6]],
             "methods": [{"name": name, "minutes": round(value / 60, 1)} for name, value in sorted(methods.items(), key=lambda pair: -pair[1])],
+            "transcoding": {**{name: round(value, 1) for name, value in transcoding.items()},
+                            "hardware": [{"name": name, "minutes": round(value, 1)} for name, value in sorted(hardware.items(), key=lambda pair: -pair[1])],
+                            "audio_codecs": [{"name": name, "minutes": round(value, 1)} for name, value in sorted(audio_codecs.items(), key=lambda pair: -pair[1])],
+                            "gpu_busy_minutes": None},
             "recent": sorted(recent, key=lambda row: row["played_at"], reverse=True)[:20]}
 
 
@@ -169,7 +218,7 @@ async def get_insights(user: dict, days: int) -> dict:
            runtime.jellyfin_base_url, identity, days)
     cached = _cache.get(key)
     if cached and cached[0] > time.monotonic():
-        return {**base, **cached[1]}
+        return {**base, **with_artwork(cached[1], user, runtime)}
     history, libraries = await client.get_user_history(identity, start, end)
     data = {**summarize(history, libraries, start, end), "state": "ready", "updated_at": end.isoformat(),
             "period_start": start.isoformat(), "period_end": end.isoformat()}
@@ -178,4 +227,4 @@ async def get_insights(user: dict, days: int) -> dict:
     if len(_cache) >= 128:
         _cache.pop(next(iter(_cache)))
     _cache[key] = (time.monotonic() + CACHE_SECONDS, data)
-    return {**base, **data}
+    return {**base, **with_artwork(data, user, runtime)}

backend/app/services/insights_artwork.py

@@ -0,0 +1,98 @@
+"""Private Jellyfin thumbnails for items returned in a user's own viewing history."""
+
+import asyncio
+import hashlib
+import hmac
+import re
+import time
+from collections import OrderedDict
+
+import httpx
+from fastapi import HTTPException
+
+from ..config import settings
+
+TOKEN_SECONDS = 3600
+MAX_IMAGE_BYTES = 1024 * 1024
+MAX_CACHE_BYTES = 16 * 1024 * 1024
+_cache = OrderedDict()
+_downloads = asyncio.Semaphore(6)
+
+
+def item_id(value):
+    value = str(value or "").replace("-", "").lower()
+    return value if re.fullmatch(r"[a-f0-9]{32}", value) else None
+
+
+def source(runtime):
+    return hashlib.sha256(f"{runtime.jellyfin_base_url}|{runtime.jellyfin_api_key}".encode()).hexdigest()
+
+
+def signature(user, runtime, media_id, expires):
+    message = f"insights-artwork\n{user['username']}\n{source(runtime)}\n{media_id}\n{expires}"
+    return hmac.new(settings.jwt_secret.encode(), message.encode(), hashlib.sha256).hexdigest()
+
+
+def with_artwork(data, user, runtime):
+    expires = int(time.time()) + TOKEN_SECONDS
+    recent = []
+    for play in data.get("recent", []):
+        row = {**play}
+        media_id = row.pop("artwork_item_id", None)
+        row["artwork_url"] = None
+        if item_id(media_id) and settings.jwt_secret and runtime.jellyfin_base_url and runtime.jellyfin_api_key:
+            token = f"{expires}.{signature(user, runtime, media_id, expires)}"
+            row["artwork_url"] = f"/insights/artwork/{media_id}?token={token}"
+        recent.append(row)
+    return {**data, "recent": recent}
+
+
+def verify_artwork_token(user, runtime, media_id, token):
+    if not settings.jwt_secret or not re.fullmatch(r"[a-f0-9]{32}", media_id):
+        raise HTTPException(404, "Artwork unavailable")
+    if not re.fullmatch(r"[0-9]{1,12}\.[a-f0-9]{64}", token):
+        raise HTTPException(403, "Artwork link is invalid or expired")
+    try:
+        expires_text, supplied = token.split(".", 1)
+        expires = int(expires_text)
+    except (ValueError, TypeError):
+        raise HTTPException(403, "Artwork link is invalid or expired") from None
+    now = int(time.time())
+    if expires < now or expires > now + TOKEN_SECONDS or not hmac.compare_digest(supplied, signature(user, runtime, media_id, expires)):
+        raise HTTPException(403, "Artwork link is invalid or expired")
+
+
+async def get_artwork(user, runtime, media_id, token):
+    verify_artwork_token(user, runtime, media_id, token)
+    if not runtime.jellyfin_base_url or not runtime.jellyfin_api_key:
+        raise HTTPException(404, "Artwork unavailable")
+    key = (source(runtime), media_id)
+    async with _downloads:
+        cached = _cache.get(key)
+        if cached and cached[0] > time.monotonic():
+            _cache.move_to_end(key)
+            return cached[1], cached[2]
+        try:
+            async with httpx.AsyncClient(timeout=8.0) as client:
+                async with client.stream("GET", f"{runtime.jellyfin_base_url.rstrip('/')}/Items/{media_id}/Images/Primary",
+                    headers={"X-Emby-Token": runtime.jellyfin_api_key},
+                    params={"maxWidth": 120, "maxHeight": 180, "quality": 85, "format": "Webp"}) as response:
+                    response.raise_for_status()
+                    content_type = response.headers.get("content-type", "").split(";", 1)[0].strip().lower()
+                    if content_type not in {"image/jpeg", "image/png", "image/webp"}:
+                        raise ValueError()
+                    content = bytearray()
+                    async for chunk in response.aiter_bytes():
+                        content.extend(chunk)
+                        if len(content) > MAX_IMAGE_BYTES:
+                            raise ValueError()
+                    if not content:
+                        raise ValueError()
+        except (httpx.HTTPError, ValueError) as exc:
+            raise HTTPException(404, "Artwork unavailable") from exc
+        for expired in [entry for entry, value in _cache.items() if value[0] <= time.monotonic()]:
+            _cache.pop(expired, None)
+        while _cache and (len(_cache) >= 128 or sum(len(value[1]) for value in _cache.values()) + len(content) > MAX_CACHE_BYTES):
+            _cache.popitem(last=False)
+        _cache[key] = (time.monotonic() + 600, bytes(content), content_type)
+        return bytes(content), content_type

backend/tests/test_insights_media.py

@@ -0,0 +1,165 @@
+import asyncio
+import base64
+import copy
+import unittest
+from datetime import timedelta
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+from urllib.parse import parse_qs, urlsplit
+
+import httpx
+from fastapi import FastAPI, HTTPException
+from fastapi.testclient import TestClient
+
+from backend.app.routers import insights as router
+from backend.app.services import insights, insights_artwork as artwork
+from backend.tests.test_insights import NOW, LIBRARIES, USER, play
+
+ITEM = "a" * 32
+OTHER = "b" * 32
+PNG = base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+jK1sAAAAASUVORK5CYII=")
+
+
+def transcode(activity, minutes, video_direct=False, audio_direct=False, hardware="nvenc", **kwargs):
+    return play(activity, NowPlayingItemId=ITEM, PlaybackDuration=minutes * 60, PlayMethod="Transcode",
+                TranscodingInfo={"IsVideoDirect": video_direct, "IsAudioDirect": audio_direct,
+                    "HardwareAccelerationType": hardware, "VideoCodec": "h264", "AudioCodec": "aac"}, **kwargs)
+
+
+class TranscodingSummaryTests(unittest.TestCase):
+    def test_gpu_audio_software_and_unknown_time_are_separate_and_deduplicated(self):
+        gpu = transcode("gpu", 10)
+        rows = [gpu, dict(gpu), transcode("audio-only", 5, video_direct=True),
+                transcode("software", 3, audio_direct=True, hardware="none"),
+                transcode("remux", 2, video_direct=True, audio_direct=True),
+                {**transcode("stale", 1), "PlayMethod": "DirectPlay"},
+                transcode("unknown-hardware", 2, audio_direct=True, hardware=None),
+                {**transcode("unknown-streams", 1), "TranscodingInfo": None},
+                transcode("old", 100, ActivityDateInserted=(NOW - timedelta(days=31)).isoformat())]
+        result = insights.summarize(rows, LIBRARIES, NOW - timedelta(days=30), NOW)
+        stats = result["transcoding"]
+        self.assertEqual(stats["hardware_video_minutes"], 10)
+        self.assertEqual(stats["audio_minutes"], 15)
+        self.assertEqual(stats["video_minutes"], 15)
+        self.assertEqual(stats["software_video_minutes"], 3)
+        self.assertEqual(stats["unknown_hardware_minutes"], 2)
+        self.assertEqual(stats["unknown_video_minutes"], 1)
+        self.assertEqual(stats["unknown_audio_minutes"], 1)
+        self.assertEqual(stats["hardware"], [{"name": "NVIDIA NVENC", "minutes": 10}])
+        self.assertIsNone(stats["gpu_busy_minutes"])
+
+    def test_audio_media_cannot_accumulate_video_gpu_time(self):
+        row = transcode("music", 4, ParentId="music")
+        row["TranscodingInfo"]["VideoCodec"] = None
+        result = insights.summarize([row], LIBRARIES, NOW - timedelta(days=7), NOW)["transcoding"]
+        self.assertEqual(result["audio_minutes"], 4)
+        self.assertEqual(result["video_minutes"], 0)
+        self.assertEqual(result["hardware_video_minutes"], 0)
+
+    def test_direct_stream_counts_audio_but_does_not_claim_video_encoding(self):
+        row = {**transcode("stream", 5), "PlayMethod": "DirectStream"}
+        result = insights.summarize([row], LIBRARIES, NOW - timedelta(days=7), NOW)["transcoding"]
+        self.assertEqual(result["audio_minutes"], 5)
+        self.assertEqual(result["hardware_video_minutes"], 0)
+
+    def test_numeric_hardware_enum_and_legacy_json_are_supported(self):
+        import json
+        row = transcode("enum", 5, hardware=3)
+        row["TranscodingInfo"] = json.dumps(row["TranscodingInfo"])
+        result = insights.summarize([row], LIBRARIES, NOW - timedelta(days=7), NOW)["transcoding"]
+        self.assertEqual(result["hardware_video_minutes"], 5)
+        row["TranscodingInfo"] = "invalid JSON"
+        result = insights.summarize([row], LIBRARIES, NOW - timedelta(days=7), NOW)["transcoding"]
+        self.assertEqual(result["unknown_video_minutes"], 5)
+        self.assertEqual(result["hardware_video_minutes"], 0)
+
+    def test_episode_artwork_uses_series_id_and_invalid_ids_are_ignored(self):
+        rows = [play("episode", EpisodeId=OTHER, NowPlayingItemId=ITEM),
+                play("invalid", NowPlayingItemId="../../secret")]
+        result = insights.summarize(rows, LIBRARIES, NOW - timedelta(days=7), NOW)
+        indexed = {row["id"]: row for row in result["recent"]}
+        self.assertEqual(indexed["episode"]["artwork_item_id"], ITEM)
+        self.assertIsNone(indexed["invalid"]["artwork_item_id"])
+
+
+class ArtworkTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.runtime = SimpleNamespace(jellyfin_base_url="http://jellyfin", jellyfin_api_key="PRIVATE-API-KEY")
+        self.secret = patch.object(artwork.settings, "jwt_secret", "test-artwork-signing-secret")
+        self.secret.start()
+        self.addCleanup(self.secret.stop)
+        artwork._cache.clear()
+        artwork._downloads = asyncio.Semaphore(6)
+
+    def url_and_token(self, user=USER):
+        data = {"recent": [{"id": "play-1", "artwork_item_id": ITEM}]}
+        before = copy.deepcopy(data)
+        result = artwork.with_artwork(data, user, self.runtime)
+        self.assertEqual(data, before)
+        row = result["recent"][0]
+        self.assertNotIn("artwork_item_id", row)
+        self.assertNotIn("PRIVATE-API-KEY", row["artwork_url"])
+        return row["artwork_url"], parse_qs(urlsplit(row["artwork_url"]).query)["token"][0]
+
+    async def test_ticket_is_bound_to_user_item_server_credentials_and_time(self):
+        with patch.object(artwork.time, "time", return_value=1000):
+            _, token = self.url_and_token()
+            artwork.verify_artwork_token(USER, self.runtime, ITEM, token)
+            for user, runtime, media_id in [({**USER, "username": "different"}, self.runtime, ITEM),
+                    (USER, self.runtime, OTHER), (USER, SimpleNamespace(jellyfin_base_url="http://other", jellyfin_api_key="PRIVATE-API-KEY"), ITEM),
+                    (USER, SimpleNamespace(jellyfin_base_url="http://jellyfin", jellyfin_api_key="changed"), ITEM)]:
+                with self.assertRaises(HTTPException) as raised:
+                    artwork.verify_artwork_token(user, runtime, media_id, token)
+                self.assertEqual(raised.exception.status_code, 403)
+        with patch.object(artwork.time, "time", return_value=5000), self.assertRaises(HTTPException):
+            artwork.verify_artwork_token(USER, self.runtime, ITEM, token)
+        for invalid in ["invalid", "1.", "1." + "\u2603" * 64]:
+            with self.assertRaises(HTTPException):
+                artwork.verify_artwork_token(USER, self.runtime, ITEM, invalid)
+
+    async def test_private_proxy_returns_image_and_validates_before_cache_access(self):
+        calls = []
+        def handler(request):
+            calls.append(request)
+            self.assertEqual(request.url.path, f"/Items/{ITEM}/Images/Primary")
+            self.assertNotIn("PRIVATE", str(request.url))
+            self.assertEqual(request.headers["X-Emby-Token"], "PRIVATE-API-KEY")
+            return httpx.Response(200, content=PNG, headers={"Content-Type": "image/png"})
+        real = httpx.AsyncClient
+        _, token = self.url_and_token()
+        with patch.object(artwork.httpx, "AsyncClient", side_effect=lambda **kwargs: real(transport=httpx.MockTransport(handler), **kwargs)):
+            self.assertEqual(await artwork.get_artwork(USER, self.runtime, ITEM, token), (PNG, "image/png"))
+            self.assertEqual(await artwork.get_artwork(USER, self.runtime, ITEM, token), (PNG, "image/png"))
+            with self.assertRaises(HTTPException):
+                await artwork.get_artwork({**USER, "username": "someone-else"}, self.runtime, ITEM, token)
+        self.assertEqual(len(calls), 1)
+
+    async def test_non_images_missing_images_and_large_images_fail_closed(self):
+        _, token = self.url_and_token()
+        real = httpx.AsyncClient
+        for status, body, mime in [(404, b"PRIVATE", "text/plain"), (200, b"<svg>PRIVATE</svg>", "image/svg+xml"),
+                                   (200, b"x" * (artwork.MAX_IMAGE_BYTES + 1), "image/png")]:
+            transport = httpx.MockTransport(lambda request: httpx.Response(status, content=body, headers={"Content-Type": mime}))
+            with patch.object(artwork.httpx, "AsyncClient", side_effect=lambda **kwargs: real(transport=transport, **kwargs)):
+                with self.assertRaises(HTTPException) as raised:
+                    await artwork.get_artwork(USER, self.runtime, ITEM, token)
+                self.assertEqual(raised.exception.status_code, 404)
+                self.assertNotIn("PRIVATE", raised.exception.detail)
+        self.assertEqual(len(artwork._cache), 0)
+
+
+class ArtworkRouteTests(unittest.TestCase):
+    def test_authentication_and_private_response_headers(self):
+        app = FastAPI()
+        app.include_router(router.router)
+        client = TestClient(app)
+        self.assertEqual(client.get(f"/insights/artwork/{ITEM}?token=invalid").status_code, 401)
+        app.dependency_overrides[router.get_current_user] = lambda: USER
+        with patch.object(router, "get_runtime_settings", return_value=None), \
+             patch.object(router, "get_artwork", new_callable=AsyncMock, return_value=(PNG, "image/png")):
+            result = client.get(f"/insights/artwork/{ITEM}?token=fixture")
+        self.assertEqual(result.status_code, 200)
+        self.assertEqual(result.content, PNG)
+        self.assertEqual(result.headers["cache-control"], "private, max-age=600")
+        self.assertEqual(result.headers["vary"], "Cookie, Authorization")
+        self.assertEqual(result.headers["x-content-type-options"], "nosniff")