Merge pull request #1565 from glensc/plexid

Author: glensc

plextraktsync/commands/download.py

@@ -5,7 +5,7 @@
 from typing import TYPE_CHECKING
 
 from plextraktsync.factory import factory
-from plextraktsync.util.expand_id import expand_id
+from plextraktsync.util.expand_id import expand_plexid
 
 if TYPE_CHECKING:
     from plextraktsync.plex.PlexApi import PlexApi
@@ -60,10 +60,12 @@ def download(input: list[str], only_subs: bool, target: str):
 
     # Expand ~ as HOME
     savepath = Path(target).expanduser()
-    for id in expand_id(input):
-        pm = plex.fetch_item(id)
+    for plex_id in expand_plexid(input):
+        if plex_id.server:
+            plex = factory.get_plex_by_id(plex_id.server)
+        pm = plex.fetch_item(plex_id.key)
         if not pm:
-            print(f"Not found: {id}. Skipping")
+            print(f"Not found: {plex_id} from {plex}. Skipping")
             continue
 
         if not only_subs:

plextraktsync/commands/plex_login.py

@@ -220,6 +220,7 @@ def login(username: str, password: str):
 
     sc = ServerConfig()
     sc.add_server(
+        id=plex.machineIdentifier,
         name=server.name,
         token=token,
         urls=server_urls(server),

plextraktsync/config/PlexServerConfig.py

@@ -12,6 +12,8 @@ class PlexServerConfig:
     name: str
     token: str
     urls: list[str]
+    # The machineIdentifier value of this server
+    id: str = None
 
     def asdict(self):
         data = asdict(self)

plextraktsync/config/ServerConfig.py

@@ -20,6 +20,11 @@ def get_server(self, name: str):
         except KeyError:
             raise RuntimeError(f"Server with name {name} is not defined")
 
+    def server_by_id(self, id: str):
+        for name, server in self.servers.items():
+            if "id" in server and id == server["id"]:
+                return self.get_server(name)
+
     def load(self):
         if self.loaded:
             return self

plextraktsync/plex/PlexApi.py

@@ -32,6 +32,9 @@ class PlexApi:
     def __init__(self, plex: PlexServer):
         self.plex = plex
 
+    def __str__(self):
+        return str(self.plex)
+
     def plex_base_url(self, section="server"):
         return f"https://app.plex.tv/desktop/#!/{section}/{self.plex.machineIdentifier}"
 

plextraktsync/plex/PlexId.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(unsafe_hash=True)
+class PlexId:
+    key: int | str
+    media_type: str = None
+    provider: str = None
+    server: str = None
+
+    METADATA = "metadata.provider.plex.tv"
+    METADATA_URL = "https://metadata.provider.plex.tv/library/metadata"
+
+    @property
+    def metadata_url(self):
+        return f"{self.METADATA_URL}/{self.key}"
+
+    @property
+    def is_discover(self):
+        return self.provider == self.METADATA
+
+    def __repr__(self):
+        keys = [self.__class__.__name__, self.server, self.provider, self.key]
+        fields = map(str, filter(None, keys))
+
+        return f'<{":".join(fields)}>'

plextraktsync/plex/PlexIdFactory.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from urllib.parse import parse_qs, urlparse
+
+from .PlexId import PlexId
+
+
+class PlexIdFactory:
+    @classmethod
+    def create(cls, key: str | int):
+        if isinstance(key, int) or key.isnumeric():
+            return PlexId(int(key))
+        elif key.startswith("https:") or key.startswith("http:"):
+            return cls.from_url(key)
+        elif key.startswith("plex://"):
+            return cls.from_plex_guid(key)
+
+        raise RuntimeError(f"Unable to create PlexId: {key}")
+
+    @classmethod
+    def from_plex_guid(cls, id):
+        key = id.rsplit('/', 1)[-1]
+        return PlexId(key, provider=PlexId.METADATA)
+
+    @staticmethod
+    def from_url(url: str):
+        """
+        Extracts id from urls like:
+          https://app.plex.tv/desktop/#!/server/abcdefg/details?key=%2Flibrary%2Fmetadata%2F13202
+          https://app.plex.tv/desktop/#!/server/abcdefg/playHistory?filters=metadataItemID%3D6041&filterTitle=&isParentType=false
+          https://app.plex.tv/desktop/#!/provider/tv.plex.provider.discover/details?key=%2Flibrary%2Fmetadata%2F5d7768532e80df001ebe18e7
+          https://app.plex.tv/desktop/#!/provider/tv.plex.provider.discover/details?key=/library/metadata/5d776a8e51dd69001fe24eb8'
+        """
+
+        result = urlparse(url)
+        if result.fragment[0] != "!":
+            raise RuntimeError(f"Unable to parse: {url}")
+
+        fragment = urlparse(result.fragment)
+        parsed = parse_qs(fragment.query)
+
+        if fragment.path.startswith('!/server/'):
+            server = fragment.path.split('/')[2]
+        else:
+            server = None
+
+        if "key" in parsed:
+            key = ",".join(parsed["key"])
+            if key.startswith("/library/metadata/"):
+                id = key[len("/library/metadata/"):]
+                if fragment.path == '!/provider/tv.plex.provider.discover/details':
+                    return PlexId(id, provider=PlexId.METADATA)
+                return PlexId(int(id), server=server)
+
+        if "filters" in parsed:
+            filters = parse_qs(parsed["filters"][0])
+            if "metadataItemID" in filters:
+                return PlexId(int(filters["metadataItemID"][0]), server=server)
+
+        raise RuntimeError(f"Unable to parse: {url}")

plextraktsync/util/Factory.py

@@ -58,6 +58,14 @@ def media_factory(self):
 
         return mf
 
+    def get_plex_by_id(self, server_id: str):
+        server_config = self.server_config_factory.server_by_id(server_id)
+        if server_config is not None and server_config is not self.server_config:
+            self.invalidate(["plex_api", "plex_server", "server_config"])
+            self.run_config.server = server_config.name
+
+        return self.plex_api
+
     @cached_property
     def plex_server(self):
         from plextraktsync.factory import factory
@@ -79,12 +87,17 @@ def has_plex_token(self):
             return False
 
     @cached_property
-    def server_config(self):
+    def server_config_factory(self):
         from plextraktsync.config.ServerConfig import ServerConfig
 
+        return ServerConfig()
+
+    @cached_property
+    def server_config(self):
+
         config = self.config
         run_config = self.run_config
-        server_config = ServerConfig()
+        server_config = self.server_config_factory
         server_name = run_config.server
 
         if server_name is None:

plextraktsync/util/expand_id.py

@@ -32,6 +32,13 @@ def plex_id(id):
     return f"https://metadata.provider.plex.tv/library/metadata/{key}"
 
 
+def expand_plexid(input):
+    from plextraktsync.plex.PlexIdFactory import PlexIdFactory
+
+    for id in input:
+        yield PlexIdFactory.create(id)
+
+
 def expand_id(input):
     """
     Takes list of id or urls and resolves to Plex media id.

tests/test_plex_id.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3 -m pytest
+from plextraktsync.plex.PlexId import PlexId
+from plextraktsync.plex.PlexIdFactory import PlexIdFactory
+
+# sha1 of "PlexTraktSync"
+SERVER_ID = "2ac365831928c7ad926484eae3c65e3a88112c54"
+
+
+def test_plex_id_numeric():
+    pid = PlexIdFactory.create(10)
+    assert pid.key == 10
+    assert pid.media_type is None
+    assert pid.provider is None
+    assert pid.server is None
+    assert pid.is_discover is False
+
+    pid = PlexIdFactory.create("10")
+    assert pid.key == 10
+    assert pid.media_type is None
+    assert pid.provider is None
+    assert pid.server is None
+    assert pid.is_discover is False
+
+
+def test_plex_id_urls():
+    pid = PlexIdFactory.create(
+        f"https://app.plex.tv/desktop/#!/server/{SERVER_ID}/details?key=%2Flibrary%2Fmetadata%2F13202"
+    )
+    assert pid.key == 13202
+    assert pid.media_type is None
+    assert pid.provider is None
+    assert pid.server == SERVER_ID
+    assert pid.is_discover is False
+
+    pid = PlexIdFactory.create(
+        f"https://app.plex.tv/desktop/#!/server/{SERVER_ID}/playHistory?filters=metadataItemID%3D6041&filterTitle=&isParentType=false"
+    )
+    assert pid.key == 6041
+    assert pid.media_type is None
+    assert pid.provider is None
+    assert pid.server == SERVER_ID
+    assert pid.is_discover is False
+
+
+def test_plex_id_discover_url():
+    pid = PlexIdFactory.create(
+        "https://app.plex.tv/desktop/#!/provider/tv.plex.provider.discover/details?key=%2Flibrary%2Fmetadata%2F5d7768532e80df001ebe18e7"
+    )
+    assert pid.key == "5d7768532e80df001ebe18e7"
+    assert pid.media_type is None
+    assert pid.provider == PlexId.METADATA
+    assert pid.server is None
+    assert pid.is_discover is True
+
+
+def test_plex_id_discover_url2():
+    pid = PlexIdFactory.create(
+        "https://app.plex.tv/desktop/#!/provider/tv.plex.provider.discover/details?key=/library/metadata/5d776a8e51dd69001fe24eb8"
+    )
+    assert pid.key == "5d776a8e51dd69001fe24eb8"
+    assert pid.media_type is None
+    assert pid.provider == PlexId.METADATA
+    assert pid.server is None
+    assert pid.is_discover is True