Merge pull request #1855 from glensc/trakt-id

Add trakt url support for inspect: Movies

plextraktsync/plex/PlexApi.py

@@ -99,6 +99,38 @@ def download(self, m: SubtitleStream | MediaPart, **kwargs):
 
         return plexapi.utils.download(url, token, **kwargs)
 
+    def search_by_guid(self, guids: dict, libtype: str):
+        r"""
+                    fetchItem(ekey, guid__regex=r"com\.plexapp\.agents\.(imdb|themoviedb)://|tt\d+")
+                    fetchItem(ekey, guid__id__regex=r"(imdb|tmdb|tvdb)://")
+        """
+
+        from plextraktsync.plex.PlexGuid import PlexGuid
+
+        # Build regex of possible matches
+        keys = "|".join(guids.keys())
+        values = "|".join(map(str, guids.values()))
+        regex = f"({keys})://({values})"
+
+        plexguids = [PlexGuid(f"{k}://{v}", type=libtype) for k, v in guids.items()]
+        sections = self.movie_sections() if libtype == "movie" else None
+        if not sections:
+            raise RuntimeError(f"Not supported search type: {libtype}")
+        results = []
+        for section in sections:
+            result = section.search(libtype=libtype, guid__id__regex=regex)
+            for m in result:
+                pm = PlexLibraryItem(m, self)
+                # Do proper matching with provider type and provider id
+                matched = len([[True for g1 in pm.guids if g1 == g2] for g2 in plexguids])
+                if matched:
+                    results.append(pm)
+
+        if not len(results):
+            return None
+
+        return results
+
     @property
     def version(self):
         return self.server.version

plextraktsync/plex/PlexIdFactory.py

@@ -10,6 +10,8 @@ class PlexIdFactory:
     def create(cls, key: str | int):
         if isinstance(key, int) or key.isnumeric():
             return PlexId(int(key))
+        elif key.startswith("https://trakt.tv/"):
+            return cls.from_trakt_url(key)
         elif key.startswith("https:") or key.startswith("http:"):
             return cls.from_url(key)
         elif key.startswith("plex://"):
@@ -61,3 +63,27 @@ def from_url(url: str):
                 return PlexId(int(filters["metadataItemID"][0]), server=server)
 
         raise RuntimeError(f"Unable to parse: {url}")
+
+    @classmethod
+    def from_trakt_url(cls, url: str):
+        path = urlparse(url).path
+
+        if path.startswith("/movies/"):
+            media_type = "movie"
+            slug = path[len("/movies/"):]
+        else:
+            from click import ClickException
+            raise ClickException(f"Unable to create PlexId: {path}")
+
+        from plextraktsync.factory import factory
+        from plextraktsync.trakt.TraktItem import TraktItem
+
+        tm = TraktItem(factory.trakt_api.find_by_slug(slug, media_type))
+        results = factory.plex_api.search_by_guid(tm.guids, libtype=tm.type)
+        if results is None:
+            raise RuntimeError(f"Unable to find Plex Match: {url}")
+        if len(results) > 1:
+            raise RuntimeError(f"Failed to find unique match: {url}")
+        pm = results[0]
+
+        return PlexId(pm.key, server=pm.plex.server.machineIdentifier)

plextraktsync/trakt/TraktApi.py

@@ -9,8 +9,8 @@
 import trakt.sync
 import trakt.users
 from click import ClickException
-from trakt.errors import (ForbiddenException, OAuthException,
-                          OAuthRefreshException)
+from trakt.errors import (ForbiddenException, NotFoundException,
+                          OAuthException, OAuthRefreshException)
 
 from plextraktsync import pytrakt_extensions
 from plextraktsync.decorators.flatten import flatten_list
@@ -257,6 +257,19 @@ def find_by_guid(self, guid: PlexGuid):
 
             return tm
 
+    @staticmethod
+    def find_by_slug(slug: str, media_type: str):
+        if media_type == "movie":
+            from trakt.movies import Movie
+            factory_method = Movie
+        else:
+            raise RuntimeError(f"Unsupported media_type={media_type}")
+
+        try:
+            return factory_method(title=slug, slug=slug)
+        except NotFoundException:
+            raise RuntimeError(f"Unable to find by slug: {slug}")
+
     @rate_limit()
     @retry()
     def search_by_id(self, media_id: str, id_type: str, media_type: str) -> TVShow | Movie | None:

plextraktsync/trakt/TraktItem.py

@@ -18,3 +18,7 @@ def type(self):
         """
         # NB: TVSeason does not have "media_type" property
         return self.item.media_type[:-1]
+
+    @property
+    def guids(self):
+        return {k: v for k, v in self.item.ids["ids"].items() if k in ["imdb", "tmdb", "tvdb"]}