Dedupe duplicate plays on Trakt import

- Skip a Trakt-imported movie/episode watch if it falls within 15 minutes
  of an existing play for the same item, so double-watch entries no
  longer occur when Plex webhook scrobbling and Trakt import are both
  enabled (Plex fires at ~90% progress, Trakt waits for playback stop)
- Dedupe checks against both pre-existing DB rows and other entries in
  the same import run

Fixes #854

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: dannyvfilms

src/integrations/imports/trakt.py

@@ -1,6 +1,7 @@
 import json
 import logging
 from collections import defaultdict
+from datetime import timedelta
 
 import requests
 from django.conf import settings
@@ -24,6 +25,12 @@
 BULK_PAGE_SIZE = 1000
 TRAKT_UNKNOWN_DATE = "1970-01-01T00:00:00.000Z"
 
+# Plex's webhook fires at ~90% progress while Trakt's scrobbler waits for
+# playback to stop, so a webhook-recorded play and its later Trakt-imported
+# counterpart can land several minutes apart. Treat plays for the same item
+# within this window as the same watch instead of double-counting it.
+_DUPLICATE_PLAY_WINDOW = timedelta(minutes=15)
+
 
 def _parse_watched_at(watched_at: str):
     if watched_at == TRAKT_UNKNOWN_DATE:
@@ -383,6 +390,12 @@ def __init__(self, username, user, mode, refresh_token=None):
         # does not create duplicate episode history rows.
         self.existing_episode_watch_keys = self._get_existing_episode_watch_keys()
 
+        # Track existing play timestamps so a play already recorded (e.g. via
+        # Plex webhook) isn't duplicated by a nearby Trakt-imported play for
+        # the same item. See _DUPLICATE_PLAY_WINDOW.
+        self.existing_episode_play_times = self._get_existing_episode_play_times()
+        self.existing_movie_play_times = self._get_existing_movie_play_times()
+
         # Track media IDs to delete in overwrite mode
         self.to_delete = defaultdict(lambda: defaultdict(set))
 
@@ -431,6 +444,43 @@ def _get_existing_episode_watch_keys(self):
             ),
         )
 
+    def _get_existing_episode_play_times(self):
+        """Return existing episode play end_dates keyed by (tmdb_id, season, episode)."""
+        play_times = defaultdict(list)
+        rows = app.models.Episode.objects.filter(
+            related_season__user=self.user,
+            end_date__isnull=False,
+        ).values_list(
+            "item__media_id",
+            "item__season_number",
+            "item__episode_number",
+            "end_date",
+        )
+        for media_id, season_number, episode_number, end_date in rows:
+            play_times[(media_id, season_number, episode_number)].append(end_date)
+        return play_times
+
+    def _get_existing_movie_play_times(self):
+        """Return existing movie play end_dates keyed by tmdb_id."""
+        play_times = defaultdict(list)
+        rows = app.models.Movie.objects.filter(
+            user=self.user,
+            end_date__isnull=False,
+        ).values_list("item__media_id", "end_date")
+        for media_id, end_date in rows:
+            play_times[media_id].append(end_date)
+        return play_times
+
+    @staticmethod
+    def _is_duplicate_play(existing_times, candidate_dt):
+        """Return True if candidate_dt is within _DUPLICATE_PLAY_WINDOW of an existing play."""
+        if candidate_dt is None:
+            return False
+        return any(
+            abs(candidate_dt - existing_dt) <= _DUPLICATE_PLAY_WINDOW
+            for existing_dt in existing_times
+        )
+
     def _raise_for_user_error(self, error):
         """Translate a Trakt HTTP error about the user into a MediaImportError."""
         if error.response.status_code == requests.codes.not_found:
@@ -686,6 +736,17 @@ def process_watched_movie(self, entry):
         watched_at = entry["watched_at"]
         watched_at_dt = _parse_watched_at(watched_at)
 
+        if self._is_duplicate_play(
+            self.existing_movie_play_times[tmdb_id],
+            watched_at_dt,
+        ):
+            logger.debug(
+                "Skipping Trakt movie watch for %s at %s: duplicate of an existing play",
+                movie["title"],
+                watched_at,
+            )
+            return
+
         key = f"{tmdb_id}"
 
         movie_obj = app.models.Movie(
@@ -700,6 +761,8 @@ def process_watched_movie(self, entry):
 
         self.media_instances[MediaTypes.MOVIE.value][key].append(movie_obj)
         self.bulk_media[MediaTypes.MOVIE.value].append(movie_obj)
+        if watched_at_dt is not None:
+            self.existing_movie_play_times[tmdb_id].append(watched_at_dt)
 
     def process_watched_episode(self, entry):
         """Process a single episode watch event."""
@@ -732,6 +795,21 @@ def process_watched_episode(self, entry):
             )
             return
 
+        episode_key = (tmdb_id, season_number, episode_number)
+        if self._is_duplicate_play(
+            self.existing_episode_play_times[episode_key],
+            watched_at_dt,
+        ):
+            logger.debug(
+                "Skipping Trakt episode watch for %s S%sE%s at %s: "
+                "duplicate of an existing play",
+                show["title"],
+                season_number,
+                episode_number,
+                watched_at,
+            )
+            return
+
         tv_exists = (
             tmdb_id in self.existing_media[MediaTypes.TV.value][Sources.TMDB.value]
         )
@@ -875,6 +953,8 @@ def process_watched_episode(self, entry):
         self.media_instances[MediaTypes.EPISODE.value][ep_key].append(episode_obj)
         self.bulk_media[MediaTypes.EPISODE.value].append(episode_obj)
         self.existing_episode_watch_keys.add(episode_watch_key)
+        if watched_at_dt is not None:
+            self.existing_episode_play_times[episode_key].append(watched_at_dt)
 
         # Update status if this is the last episode, but only for rows Floppy
         # just created (or an explicit overwrite re-sync) — never clobber the

src/integrations/tests/imports/test_trakt.py

@@ -62,10 +62,98 @@ def test_process_watched_movie(self, mock_get_metadata):
         movie_obj = trakt_importer.bulk_media[MediaTypes.MOVIE.value][0]
         self.assertEqual(movie_obj.progress, 1)
 
-        # Process the same movie again to test repeat handling
+        # Reprocessing the exact same entry is a duplicate play (issue #854)
+        # and must not create a second row.
         trakt_importer.process_watched_movie(movie_entry)
+        self.assertEqual(len(trakt_importer.bulk_media[MediaTypes.MOVIE.value]), 1)
+
+    @patch("integrations.imports.trakt.TraktImporter._get_metadata")
+    def test_process_watched_movie_dedupes_nearby_play(self, mock_get_metadata):
+        """A movie watch within the dedupe window of an existing play is skipped."""
+        mock_get_metadata.return_value = {
+            "title": "Test Movie",
+            "image": "movie_image.jpg",
+        }
+
+        trakt_importer = TraktImporter("test", self.user, "new")
+        trakt_importer.process_watched_movie(
+            {
+                "type": "movie",
+                "movie": {"title": "Test Movie", "ids": {"tmdb": 67890}},
+                "watched_at": "2023-01-02T00:00:00.000Z",
+            },
+        )
+        self.assertEqual(len(trakt_importer.bulk_media[MediaTypes.MOVIE.value]), 1)
+
+        # 10 minutes later, same movie: within the 15 minute dedupe window.
+        trakt_importer.process_watched_movie(
+            {
+                "type": "movie",
+                "movie": {"title": "Test Movie", "ids": {"tmdb": 67890}},
+                "watched_at": "2023-01-02T00:10:00.000Z",
+            },
+        )
+        self.assertEqual(len(trakt_importer.bulk_media[MediaTypes.MOVIE.value]), 1)
+
+        # 1 day later, same movie: a legitimate rewatch outside the window.
+        trakt_importer.process_watched_movie(
+            {
+                "type": "movie",
+                "movie": {"title": "Test Movie", "ids": {"tmdb": 67890}},
+                "watched_at": "2023-01-03T00:00:00.000Z",
+            },
+        )
         self.assertEqual(len(trakt_importer.bulk_media[MediaTypes.MOVIE.value]), 2)
 
+        # A different movie at a nearby time is not a duplicate.
+        mock_get_metadata.return_value = {
+            "title": "Other Movie",
+            "image": "movie_image.jpg",
+        }
+        trakt_importer.process_watched_movie(
+            {
+                "type": "movie",
+                "movie": {"title": "Other Movie", "ids": {"tmdb": 11111}},
+                "watched_at": "2023-01-03T00:05:00.000Z",
+            },
+        )
+        self.assertEqual(len(trakt_importer.bulk_media[MediaTypes.MOVIE.value]), 3)
+
+    @patch("integrations.imports.trakt.TraktImporter._get_metadata")
+    def test_process_watched_movie_dedupes_against_existing_db_play(
+        self,
+        mock_get_metadata,
+    ):
+        """A Trakt-imported play is skipped if it's near an existing DB play (e.g. webhook)."""
+        item = Item.objects.get_or_create(
+            media_id="67890",
+            source=Sources.TMDB.value,
+            media_type=MediaTypes.MOVIE.value,
+            defaults={"title": "Test Movie"},
+        )[0]
+        Movie.objects.create(
+            item=item,
+            user=self.user,
+            end_date="2023-01-02T00:00:00Z",
+            status=Status.COMPLETED.value,
+            progress=1,
+        )
+
+        mock_get_metadata.return_value = {
+            "title": "Test Movie",
+            "image": "movie_image.jpg",
+        }
+        trakt_importer = TraktImporter("test", self.user, "new")
+        trakt_importer.process_watched_movie(
+            {
+                "type": "movie",
+                "movie": {"title": "Test Movie", "ids": {"tmdb": 67890}},
+                "watched_at": "2023-01-02T00:12:00.000Z",
+            },
+        )
+
+        self.assertEqual(len(trakt_importer.bulk_media[MediaTypes.MOVIE.value]), 0)
+
     @patch("integrations.imports.trakt.TraktImporter._get_metadata")
     def test_process_watched_episode(self, mock_get_metadata):
         """Test processing an episode entry."""
@@ -132,6 +220,152 @@ def mock_metadata_side_effect(media_type, _, __, ___=None):
         )
         self.assertEqual(len(trakt_importer.bulk_media[MediaTypes.EPISODE.value]), 2)
 
+    @patch("integrations.imports.trakt.TraktImporter._get_metadata")
+    def test_process_watched_episode_dedupes_nearby_play(self, mock_get_metadata):
+        """An episode watch within the dedupe window of an existing play is skipped."""
+
+        def mock_metadata_side_effect(media_type, _, __, ___=None):
+            if media_type == MediaTypes.TV.value:
+                return {
+                    "title": "Test Show",
+                    "image": "tv_image.jpg",
+                    "last_episode_season": 1,
+                    "max_progress": 1,
+                }
+            if media_type == MediaTypes.SEASON.value:
+                return {
+                    "title": "Season 1",
+                    "image": "season_image.jpg",
+                    "episodes": [
+                        {
+                            "episode_number": 1,
+                            "still_path": "/still.jpg",
+                            "title": "Pilot Episode Title",
+                        },
+                        {
+                            "episode_number": 2,
+                            "still_path": "/still2.jpg",
+                            "title": "Episode 2 Title",
+                        },
+                    ],
+                    "max_progress": 2,
+                }
+            return None
+
+        mock_get_metadata.side_effect = mock_metadata_side_effect
+
+        episode_entry = {
+            "type": "episode",
+            "episode": {"season": 1, "number": 1, "title": "Pilot"},
+            "show": {"title": "Test Show", "ids": {"tmdb": 12345}},
+            "watched_at": "2023-01-01T00:00:00.000Z",
+        }
+
+        trakt_importer = TraktImporter("testuser", self.user, "new")
+        trakt_importer.process_watched_episode(episode_entry)
+        self.assertEqual(len(trakt_importer.bulk_media[MediaTypes.EPISODE.value]), 1)
+
+        # 10 minutes later, same episode: within the 15 minute dedupe window.
+        trakt_importer.process_watched_episode(
+            {
+                **episode_entry,
+                "watched_at": "2023-01-01T00:10:00.000Z",
+            },
+        )
+        self.assertEqual(len(trakt_importer.bulk_media[MediaTypes.EPISODE.value]), 1)
+
+        # A different episode of the same show at a nearby time is not a duplicate.
+        trakt_importer.process_watched_episode(
+            {
+                "type": "episode",
+                "episode": {"season": 1, "number": 2, "title": "Episode 2"},
+                "show": {"title": "Test Show", "ids": {"tmdb": 12345}},
+                "watched_at": "2023-01-01T00:12:00.000Z",
+            },
+        )
+        self.assertEqual(len(trakt_importer.bulk_media[MediaTypes.EPISODE.value]), 2)
+
+    @patch("integrations.imports.trakt.TraktImporter._get_metadata")
+    def test_process_watched_episode_dedupes_against_existing_db_play(
+        self,
+        mock_get_metadata,
+    ):
+        """A Trakt-imported episode play is skipped if it's near an existing DB play."""
+        tv_item = Item.objects.get_or_create(
+            media_id="12345",
+            source=Sources.TMDB.value,
+            media_type=MediaTypes.TV.value,
+            defaults={"title": "Test Show"},
+        )[0]
+        tv_obj = TV.objects.create(
+            item=tv_item,
+            user=self.user,
+            status=Status.IN_PROGRESS.value,
+        )
+        season_item = Item.objects.get_or_create(
+            media_id="12345",
+            source=Sources.TMDB.value,
+            media_type=MediaTypes.SEASON.value,
+            season_number=1,
+            defaults={"title": "Season 1"},
+        )[0]
+        season_obj = Season.objects.create(
+            item=season_item,
+            related_tv=tv_obj,
+            user=self.user,
+            status=Status.IN_PROGRESS.value,
+        )
+        episode_item = Item.objects.get_or_create(
+            media_id="12345",
+            source=Sources.TMDB.value,
+            media_type=MediaTypes.EPISODE.value,
+            season_number=1,
+            episode_number=1,
+            defaults={"title": "Pilot"},
+        )[0]
+        Episode.objects.create(
+            item=episode_item,
+            related_season=season_obj,
+            end_date="2023-01-01T00:00:00Z",
+        )
+
+        def mock_metadata_side_effect(media_type, _, __, ___=None):
+            if media_type == MediaTypes.TV.value:
+                return {
+                    "title": "Test Show",
+                    "image": "tv_image.jpg",
+                    "last_episode_season": 1,
+                    "max_progress": 1,
+                }
+            if media_type == MediaTypes.SEASON.value:
+                return {
+                    "title": "Season 1",
+                    "image": "season_image.jpg",
+                    "episodes": [
+                        {
+                            "episode_number": 1,
+                            "still_path": "/still.jpg",
+                            "title": "Pilot Episode Title",
+                        },
+                    ],
+                    "max_progress": 1,
+                }
+            return None
+
+        mock_get_metadata.side_effect = mock_metadata_side_effect
+
+        trakt_importer = TraktImporter("testuser", self.user, "new")
+        trakt_importer.process_watched_episode(
+            {
+                "type": "episode",
+                "episode": {"season": 1, "number": 1, "title": "Pilot"},
+                "show": {"title": "Test Show", "ids": {"tmdb": 12345}},
+                "watched_at": "2023-01-01T00:12:00.000Z",
+            },
+        )
+
+        self.assertEqual(len(trakt_importer.bulk_media[MediaTypes.EPISODE.value]), 0)
+
     @patch("integrations.imports.trakt.TraktImporter._get_metadata")
     def test_process_watched_episode_existing_show_imports_new_episode(
         self,