Merge pull request #212 from Aerya/fix-arvio-episodes

fix(arvio): improve episode identification, profile resolution, and completion handling

Author: ellite

backend/core/arvio.py

@@ -181,6 +181,43 @@ async def validate_connection(
             raise ArvioAPIError(f"ARVIO profile '{profile_id}' not found in account profiles")
     return session, profiles
 
+def _extract_profile_data(raw: Any, profile_id: str) -> list[Any]:
+    if isinstance(raw, list):
+        return raw
+    if not isinstance(raw, dict) or not raw:
+        return []
+
+    pid_str = str(profile_id)
+    # 1. Exact string match
+    if pid_str in raw and isinstance(raw[pid_str], list):
+        return raw[pid_str]
+
+    # 2. Integer match
+    if pid_str.isdigit():
+        pid_int = int(pid_str)
+        if pid_int in raw and isinstance(raw[pid_int], list):
+            return raw[pid_int]
+
+    # 3. Case-insensitive / suffix match (e.g. "profile_0", "p0", "0")
+    for k, v in raw.items():
+        if isinstance(v, list):
+            k_str = str(k).lower().strip()
+            if k_str == pid_str.lower() or k_str.endswith(f"_{pid_str}") or k_str.endswith(f"-{pid_str}"):
+                return v
+
+    # 4. If single key in dictionary, return that list
+    if len(raw) == 1:
+        val = list(raw.values())[0]
+        if isinstance(val, list):
+            return val
+
+    # 5. Still unmatched - fail safe rather than combine every profile's data
+    # together, which would leak another profile's watch history into this
+    # one on a multi-profile ARVIO account (cases 1-4 above already cover
+    # exact/int/suffix matches and the single-profile shortcut).
+    return []
+
+
 async def pull_sync_data(
     url: str,
     refresh_token: str,
@@ -194,21 +231,31 @@ async def pull_sync_data(
         await on_refresh(session)
     payload = await pull_snapshot(url, session.access_token, api_key=api_key)
 
-    pid_str = str(profile_id)
-    raw_movies = payload.get("localWatchedMoviesByProfile", {})
-    raw_episodes = payload.get("localWatchedEpisodesByProfile", {})
-    raw_cw = payload.get("localContinueWatchingByProfile", {})
-
-    movies_data = raw_movies.get(pid_str, []) if isinstance(raw_movies, dict) else []
-    episodes_data = raw_episodes.get(pid_str, []) if isinstance(raw_episodes, dict) else []
-    cw_data = raw_cw.get(pid_str, []) if isinstance(raw_cw, dict) else []
-
-    if not isinstance(movies_data, list):
-        movies_data = []
-    if not isinstance(episodes_data, list):
-        episodes_data = []
-    if not isinstance(cw_data, list):
-        cw_data = []
+    raw_movies = (
+        payload.get("localWatchedMoviesByProfile")
+        or payload.get("watchedMoviesByProfile")
+        or payload.get("watchedMovies")
+        or payload.get("localWatchedMovies")
+        or {}
+    )
+    raw_episodes = (
+        payload.get("localWatchedEpisodesByProfile")
+        or payload.get("watchedEpisodesByProfile")
+        or payload.get("watchedEpisodes")
+        or payload.get("localWatchedEpisodes")
+        or {}
+    )
+    raw_cw = (
+        payload.get("localContinueWatchingByProfile")
+        or payload.get("continueWatchingByProfile")
+        or payload.get("continueWatching")
+        or payload.get("localContinueWatching")
+        or {}
+    )
+
+    movies_data = _extract_profile_data(raw_movies, profile_id)
+    episodes_data = _extract_profile_data(raw_episodes, profile_id)
+    cw_data = _extract_profile_data(raw_cw, profile_id)
 
     return session, {
         "watched_movies": movies_data,

backend/routers/sync.py

@@ -4903,7 +4903,11 @@ async def _apply_arvio_watched_movie(
     except (TypeError, ValueError):
         return False
 
-    watched_at = _parse_arvio_timestamp(item.get("watchedAt") or item.get("timestamp") or item.get("updatedAtMs"))
+    # updatedAt (in addition to updatedAtMs) matters here now too: a completed
+    # continue-watching movie routed in via _apply_arvio_playback_progress's
+    # high-completion branch may only carry that field, same as the episode
+    # version of this fallback chain below.
+    watched_at = _parse_arvio_timestamp(item.get("watchedAt") or item.get("timestamp") or item.get("updatedAtMs") or item.get("updatedAt"))
 
     result = await db.execute(
         select(Media).where(
@@ -4942,34 +4946,92 @@ async def _apply_arvio_watched_movie(
     return False
 
 
+def _parse_arvio_episode_info(item: dict[str, Any] | str | int) -> tuple[int, int, int] | None:
+    """Extract (show_tmdb_id, season, episode) from various ARVIO item representations."""
+    import re
+    if isinstance(item, (int, str)):
+        item_str = str(item).strip()
+        match = re.search(r"(?:tv:|series:|tmdb:)?(\d+)[:_\-\s]+(?:s|season)?(\d+)[:_\-\s]+(?:e|ep|episode)?(\d+)", item_str, re.IGNORECASE)
+        if match:
+            try:
+                return int(match.group(1)), int(match.group(2)), int(match.group(3))
+            except ValueError:
+                pass
+        try:
+            parsed = json.loads(item_str)
+            if isinstance(parsed, dict):
+                item = parsed
+        except Exception:
+            return None
+
+    if isinstance(item, dict):
+        for field in ("id", "mediaId", "episodeId", "item_id", "itemId"):
+            val = item.get(field)
+            if isinstance(val, str):
+                match = re.search(r"(?:tv:|series:|tmdb:)?(\d+)[:_\-\s]+(?:s|season)?(\d+)[:_\-\s]+(?:e|ep|episode)?(\d+)", val, re.IGNORECASE)
+                if match:
+                    try:
+                        return int(match.group(1)), int(match.group(2)), int(match.group(3))
+                    except ValueError:
+                        pass
+
+        show_tmdb_id_raw = (
+            item.get("showTmdbId")
+            or item.get("show_tmdb_id")
+            or item.get("showId")
+            or item.get("seriesTmdbId")
+            or item.get("series_tmdb_id")
+            or item.get("seriesId")
+            or item.get("series_id")
+            or item.get("tmdbId")
+            or item.get("tmdb_id")
+        )
+        season_raw = (
+            item.get("season")
+            or item.get("seasonNumber")
+            or item.get("season_number")
+            or item.get("seasonIndex")
+            or item.get("s")
+        )
+        episode_raw = (
+            item.get("episode")
+            or item.get("episodeNumber")
+            or item.get("episode_number")
+            or item.get("episodeIndex")
+            or item.get("e")
+        )
+
+        if show_tmdb_id_raw is not None and season_raw is not None and episode_raw is not None:
+            try:
+                return int(show_tmdb_id_raw), int(season_raw), int(episode_raw)
+            except (TypeError, ValueError):
+                pass
+
+    return None
+
+
 async def _apply_arvio_watched_episode(
     db: AsyncSession,
     user_id: int,
     item: dict[str, Any] | int | str,
     tmdb_api_key: str | None,
 ) -> bool:
-    if not isinstance(item, dict):
+    info = _parse_arvio_episode_info(item)
+    if not info:
         return False
 
-    show_tmdb_id_raw = item.get("showTmdbId") or item.get("show_tmdb_id") or item.get("showId") or item.get("id")
-    season_raw = item.get("season") or item.get("seasonNumber")
-    episode_raw = item.get("episode") or item.get("episodeNumber")
-
-    if not show_tmdb_id_raw or season_raw is None or episode_raw is None:
-        return False
-    try:
-        show_tmdb_id = int(show_tmdb_id_raw)
-        season = int(season_raw)
-        episode = int(episode_raw)
-    except (TypeError, ValueError):
-        return False
+    show_tmdb_id, season, episode = info
 
-    watched_at = _parse_arvio_timestamp(item.get("watchedAt") or item.get("timestamp") or item.get("updatedAtMs"))
+    watched_at = None
+    if isinstance(item, dict):
+        watched_at = _parse_arvio_timestamp(item.get("watchedAt") or item.get("timestamp") or item.get("updatedAtMs") or item.get("updatedAt"))
 
     show_res = await db.execute(select(Show).where(Show.tmdb_id == show_tmdb_id))
     show = show_res.scalars().first()
     if not show:
-        show_title = str(item.get("title") or item.get("showTitle") or f"Show {show_tmdb_id}")
+        show_title = f"Show {show_tmdb_id}"
+        if isinstance(item, dict):
+            show_title = str(item.get("title") or item.get("showTitle") or item.get("seriesTitle") or show_title)
         show = Show(tmdb_id=show_tmdb_id, title=show_title)
         db.add(show)
         await db.flush()
@@ -4984,7 +5046,9 @@ async def _apply_arvio_watched_episode(
     )
     media = ep_res.scalars().first()
     if not media:
-        ep_title = str(item.get("episodeTitle") or f"S{season:02d}E{episode:02d}")
+        ep_title = f"S{season:02d}E{episode:02d}"
+        if isinstance(item, dict):
+            ep_title = str(item.get("episodeTitle") or item.get("title") or ep_title)
         media = Media(
             show_id=show.id,
             season_number=season,
@@ -5025,6 +5089,11 @@ async def _apply_arvio_playback_progress(
     item: dict[str, Any] | int | str,
     tmdb_api_key: str | None,
 ) -> bool:
+    if isinstance(item, str):
+        try:
+            item = json.loads(item)
+        except Exception:
+            pass
     if not isinstance(item, dict):
         return False
 
@@ -5037,7 +5106,16 @@ async def _apply_arvio_playback_progress(
     except (TypeError, ValueError):
         progress_pct = 0.0
 
-    if progress_pct < 1.0 or progress_pct >= 90.0:
+    is_completed = item.get("completed") is True or progress_pct >= 85.0
+
+    if is_completed:
+        ep_info = _parse_arvio_episode_info(item)
+        if ep_info:
+            return await _apply_arvio_watched_episode(db, user_id, item, tmdb_api_key)
+        else:
+            return await _apply_arvio_watched_movie(db, user_id, item, tmdb_api_key)
+
+    if progress_pct < 1.0:
         return False
 
     pos_sec = item.get("resumePositionSeconds") or item.get("positionSeconds") or item.get("position")
@@ -5063,19 +5141,15 @@ async def _apply_arvio_playback_progress(
     is_episode = (
         media_type_str in ("TV", "EPISODE", "SERIES")
         or (season_raw is not None and episode_raw is not None)
+        or _parse_arvio_episode_info(item) is not None
     )
 
     media: Media | None = None
     if is_episode:
-        show_tmdb_raw = item.get("showTmdbId") or item.get("show_tmdb_id") or item.get("showId") or item.get("id")
-        if not show_tmdb_raw or season_raw is None or episode_raw is None:
-            return False
-        try:
-            show_tmdb_id = int(show_tmdb_raw)
-            season = int(season_raw)
-            episode = int(episode_raw)
-        except (TypeError, ValueError):
+        ep_info = _parse_arvio_episode_info(item)
+        if not ep_info:
             return False
+        show_tmdb_id, season, episode = ep_info
 
         show_res = await db.execute(select(Show).where(Show.tmdb_id == show_tmdb_id))
         show = show_res.scalars().first()

backend/tests/test_arvio.py

@@ -198,6 +198,29 @@ async def test_apply_arvio_watched_movie(self) -> None:
         self.assertTrue(added)
         self.assertEqual(db.add.call_count, 2)  # Media + WatchEvent
 
+    async def test_apply_arvio_watched_movie_falls_back_to_updated_at(self) -> None:
+        # Regression: a completed continue-watching movie item (routed here by
+        # _apply_arvio_playback_progress's high-completion branch) may only
+        # carry updatedAt, not watchedAt/timestamp/updatedAtMs - without this
+        # fallback the event's timestamp silently became "now" instead of the
+        # real watch time.
+        db = AsyncMock()
+        db.add = MagicMock()
+        db.execute = AsyncMock(side_effect=[
+            _Result(scalars=[]),  # Media search
+            _Result(scalars=[]),  # WatchEvent search
+        ])
+
+        added = await _apply_arvio_watched_movie(
+            db,
+            user_id=1,
+            item={"tmdbId": 550, "title": "Fight Club", "updatedAt": "2026-08-10T12:00:00Z"},
+            tmdb_api_key=None,
+        )
+        self.assertTrue(added)
+        event = next(c.args[0] for c in db.add.call_args_list if isinstance(c.args[0], WatchEvent))
+        self.assertEqual(event.watched_at, datetime(2026, 8, 10, 12, 0, 0))
+
     async def test_apply_arvio_watched_episode(self) -> None:
         db = AsyncMock()
         db.add = MagicMock()
@@ -257,6 +280,37 @@ async def test_apply_arvio_watched_movie_int_item(self) -> None:
         )
         self.assertTrue(added)
 
+    def test_extract_profile_data_fallbacks(self) -> None:
+        self.assertEqual(arvio._extract_profile_data(["item1"], "0"), ["item1"])
+        self.assertEqual(arvio._extract_profile_data({"0": ["item1"]}, "0"), ["item1"])
+        self.assertEqual(arvio._extract_profile_data({"profile_0": ["item1"]}, "0"), ["item1"])
+        self.assertEqual(arvio._extract_profile_data({"uuid-123": ["item1"]}, "0"), ["item1"])
+
+    def test_extract_profile_data_fails_safe_on_multi_profile_mismatch(self) -> None:
+        # A multi-profile dict where profile_id matches none of the keys must
+        # return [] rather than combine every profile's data together - doing
+        # so would leak another profile's watch history into this one on a
+        # multi-profile ARVIO account.
+        self.assertEqual(arvio._extract_profile_data({"p1": ["a"], "p2": ["b"]}, "0"), [])
+
+    async def test_apply_arvio_watched_episode_formats(self) -> None:
+        db = AsyncMock()
+        db.add = MagicMock()
+        db.execute = AsyncMock(side_effect=[
+            _Result(scalars=[]),  # Show search
+            _Result(scalars=[]),  # Media episode search
+            _Result(scalars=[]),  # WatchEvent search
+        ])
+
+        # Test string format "94997:3:1"
+        added = await _apply_arvio_watched_episode(
+            db,
+            user_id=1,
+            item="94997:3:1",
+            tmdb_api_key=None,
+        )
+        self.assertTrue(added)
+
 
 if __name__ == "__main__":
     unittest.main()

frontend/src/pages/connections.astro

@@ -252,15 +252,17 @@ const sonarrConfigured   = !!(settings.sonarr_url && settings.sonarr_token);
                             ))}
                           </select>
                         </div>
-                        <div class="flex items-center justify-between gap-4">
-                          <label class="text-sm text-zinc-300">Push interval</label>
-                          <select class="conn-auto-push bg-zinc-800 border border-zinc-700 rounded-lg px-3 py-2 text-sm text-zinc-100 focus:outline-none focus:ring-2 focus:ring-blue-500 transition-all cursor-pointer" data-conn-id={conn.id}>
-                            <option value="" selected={!conn.auto_push_interval}>Disabled</option>
-                            {AUTO_SYNC_INTERVAL_OPTIONS.map(({ value, label }) => (
-                              <option value={String(value)} selected={conn.auto_push_interval === value}>{label}</option>
-                            ))}
-                          </select>
-                        </div>
+                        {conn.type !== "arvio" && conn.type !== "nuvio" && conn.type !== "stremio" && (
+                          <div class="flex items-center justify-between gap-4">
+                            <label class="text-sm text-zinc-300">Push interval</label>
+                            <select class="conn-auto-push bg-zinc-800 border border-zinc-700 rounded-lg px-3 py-2 text-sm text-zinc-100 focus:outline-none focus:ring-2 focus:ring-blue-500 transition-all cursor-pointer" data-conn-id={conn.id}>
+                              <option value="" selected={!conn.auto_push_interval}>Disabled</option>
+                              {AUTO_SYNC_INTERVAL_OPTIONS.map(({ value, label }) => (
+                                <option value={String(value)} selected={conn.auto_push_interval === value}>{label}</option>
+                              ))}
+                            </select>
+                          </div>
+                        )}
                       </div>
                       {conn.type === "plex" && (
                         <div class="bg-zinc-950 rounded-xl p-4 space-y-3">