Plex VineFeeder Service

README.md

@@ -67,11 +67,11 @@ The top level vinefeeder config file at  TwinVine/packages/vinefeeder/src/vinefe
 
 Vinefeeder currently has 8 services for which search, browse and list-select are available  
   
-  ALL4  BBC  ITVX  MY5  STV  TPTV  TVNZ  U 
+  ALL4  BBC  ITVX  MY5 PLEX STV  TPTV  TVNZ  U 
   
 Envied has   
 
-ALL4  AUBC  CBS  DSCP  iP   MAX   MY5   NF   PCOK   ROKU  SPOT  TPTV  TVNZ  YTBE
+ALL4  AUBC  CBS  DSCP  iP   MAX   MY5   NF   PCOK PLEX  ROKU  SPOT  TPTV  TVNZ  YTBE
 ARD   CBC   CTV  DSNP  ITV  MTSP  NBLA  NRK  PLUTO  RTE   STV   TUBI  UKTV  ZDF
 These services have web-origins and not all have been tested by me.  
   

packages/vinefeeder/src/vinefeeder/base_loader.py

@@ -10,6 +10,10 @@ from .pretty import create_clean_panel
 from .pretty import catppuccin_mocha
 import subprocess
 from .config_loader import load_config_with_fallback, get_bool
+import xmltodict
+import json
+import httpx
+import uuid
 
 console = Console()
 
@@ -71,6 +75,21 @@ class BaseLoader:
             raise Exception(f"Request failed with status {response.status_code}")
         return response.text
     
+    def get_json_from_xml_data(self, url, headers=None, params=None):
+        """Fetch data from a given URL."""
+
+        if not headers:
+            headers = self.headers
+        response = self.client.get(
+            url, headers=headers, params=params, follow_redirects=True
+        )
+    
+        if response.status_code != 200:
+            raise Exception(f"Request failed with status {response.status_code}")
+        mydict = xmltodict.parse(response.content)
+        return json.dumps(mydict)
+
+
     def get_options(self, url, headers=None):
         if not headers:
             headers = self.headers
@@ -93,6 +112,40 @@ class BaseLoader:
         """Parse HTML data into JSON format."""
         return parse_json(html)
     
+    def fetch_and_parse(self, url: str, headers):
+        # deals with response.text as xml
+        # first apporach - request json
+        # second approach convert xml to json
+        with httpx.Client(timeout=30) as client:
+            resp = client.get(url, headers=headers)
+            resp.raise_for_status()
+
+            ctype = resp.headers.get("Content-Type", "").lower()
+
+            # Try JSON if the server honored the Accept header
+            if "application/json" in ctype:
+                return resp.json()
+
+            # Fallback: if service still replied with XML (or text), parse it
+            text = resp.text.strip()
+            if text.startswith("<"):
+                # Keep XML attributes and text nodes intact
+                data = xmltodict.parse(
+                    text,
+                    attr_prefix="@",
+                    cdata_key="#text"
+                )
+                return data
+
+            # Last resort: try JSON parse (in case of wrong/missing content-type)
+            try:
+                return resp.json()
+            except json.JSONDecodeError as e:
+                raise RuntimeError(
+                    f"Unexpected response (Content-Type={ctype[:50]}): cannot parse as JSON or XML"
+                ) from e
+
+
     def normalize_episode(self, episode):
         """
         Normalize the episode dictionary for comparison.
@@ -269,8 +322,8 @@ class BaseLoader:
         try:
             self.episode_series_numbers = self.get_episodes_series_numbers(series_name)
 
-        except Exception:
+        except Exception as e:
-            console.print(f"[{catppuccin_mocha['text2']}]No data found")
+            console.print(f"[{catppuccin_mocha['text2']}]No data found: {e}")
             self.clean_terminal()
             sys.exit(0)
 

packages/vinefeeder/src/vinefeeder/services/PLEX/__init__.py

@@ -0,0 +1,266 @@
+## Plex service for twinvine/vinefeeder
+from vinefeeder.base_loader import BaseLoader
+from vinefeeder.parsing_utils import rinse, split_options, extract_script_with_id_json
+from rich.console import Console
+import jmespath
+import re
+import uuid
+
+import json
+
+
+console = Console()
+
+
+class PlexLoader(BaseLoader):
+    options = ""
+
+    def __init__(self):
+        headers = {
+            "Accept": "*/*",
+            "user-agent": "Dalvik/2.9.8 (Linux; U; Android 9.9.2; ALE-L94 Build/NJHGGF)",
+            "Origin": "https://www.itv.com",
+            "Referer": "https://www.itv.com/",
+        }
+        super().__init__(headers)
+
+    def receive(
+        self, inx: None, search_term: None, category=None, hlg_status=False, opts=None
+    ):
+        """
+        First fetch for series titles matching all or part of search_term.
+
+        Uses inx, an int variable to switch:-
+            0 for greedy search using url
+            1 for direct url download
+            2 for category browse
+            3 for search with keyword
+        If search_url_entry is left blank vinefeeder generates a
+        a menu with 3 options:
+            - Greedy Search by URL
+            - Browse by Category
+            - Search by Keyword
+        results are forwarded to receive().
+        If inx == 1, do direct download using url.
+        If inx == 3, do keyword search.
+        If inx == 0, fetch videos using a greedy search or browse-category.
+        If inx == 2, fetch videos from a category url.
+        If an unknown error occurs, exit with code 0.
+        """
+
+        if opts:
+            PlexLoader.options = opts
+        self.options_list = split_options(PlexLoader.options)
+        # direct download
+        if "http" in search_term and inx == 1:
+            self.options_list = split_options(self.options)
+            try:
+                if self.options_list[0] == "":
+                    command = ['uv', 'run', 'envied', "dl", "PLEX", search_term]
+                else:
+                    command = ['uv', 'run', 'envied', "dl", *self.options_list, "PLEX", search_term]
+                self.runsubprocess(command)
+            except Exception as e:
+                print(
+                    "Error downloading video:",
+                    e,
+                    "Is the package for vinefeeder/envied installed correctly ?",
+                )
+            return
+        
+        # keyword search
+        elif inx == 3:
+            print(f"Searching for {search_term}")
+            return self.fetch_videos(search_term)
+
+        # ALTERNATIVES BELOW FROM POP-UP MENU
+        elif inx == 0:
+            # from greedy-search OR selecting Browse-category
+            # example: https://www.channel4.com/programmes/the-great-british-bake-off/on-demand/75228-001
+
+            # need a search keyword(s) from url
+            # split and select series name
+            search_term = search_term.split("/")[4].replace("-", " ")
+            # fetch_videos_by_category search_term may have other params to remove
+            if "?" in search_term:
+                search_term = search_term.split("?")[0].replace("-", " ")
+            return self.fetch_videos(search_term)
+
+        elif "http" in search_term and inx == 2:
+            self.category = category
+            self.fetch_videos_by_category(
+                search_term
+            )  # search_term here holds a URL!!!
+
+        else:
+            print(f"Unknown error when searching for {search_term}")
+
+        return
+    
+    def fetch_videos(self, search_term):
+        URL = ("https://discover.provider.plex.tv/library/search/"
+       "?searchProviders=discover,plexAVOD,plexFAST"
+       "&includeGroups=1"
+       "&searchTypes=all%2Clivetv%2Cmovies%2Ctv%2Cpeople"
+       "&includeMetadata=1"
+       "&filterPeople=1"
+       "&limit=10"
+       f"&query={search_term}")
+
+        # Helpful (and expected!!) Plex headers + ask for JSON
+        HEADERS = {
+            "Accept": "application/json",
+            "X-Plex-Product": "Plex Mediaverse",
+            "X-Plex-Version": "1.0",
+            "X-Plex-Client-Identifier": str(uuid.uuid4()),
+            # If you have a Plex token, add it (some endpoints need it):
+            # "X-Plex-Token": "<YOUR_TOKEN>"
+        }
+        jsn = self.fetch_and_parse(URL, HEADERS)
+        #console.print_json(data=jsn)
+        '''f = open("/home/angela/plex.json", "w")
+        f.write(json.dumps(jsn))
+        f.close()'''
+        
+        res = jmespath.search(
+            """
+            MediaContainer.SearchResults[?id=='plex'] | [0].SearchResult[].{
+            slug:  Metadata.slug,
+            title: Metadata.title,
+            type:  Metadata.type
+            }
+            """,
+            jsn,
+        )
+       
+
+        for item in res:
+            slug = item["slug"]
+            title = item["title"]
+            type = item["type"] 
+            url = f"https://watch.plex.tv/{type}/{slug}"
+
+            episode = {
+                "title": item.get("title", "Unknown Title"),
+                "url": url,
+                "synopsis": item.get("type"),
+            }
+            self.add_episode(title, episode)
+
+        # List series with search_term using beaupy
+        selected_series = self.display_series_list()
+
+        if selected_series:
+            # one series selected
+            return self.second_fetch(selected_series)
+        return None
+    
+    def second_fetch(self, selected):
+        """
+        Given a selected series name, fetch its HTML and extract its episodes.
+        Or if given a movie url, fetch it and process for download.
+        The function will prepare the series data for episode selection and display the final episode list.
+        It will return the URLs for the selected episodes.
+        """
+        episode = self.get_series(selected)
+        url = episode[0]['url']
+        if 'movie' in url:
+            # direct single download
+            self.options_list = split_options(self.options)
+            
+            try:
+                if self.options_list[0] == "":
+                    command = ['uv', 'run', 'envied', "dl", "PLEX", url]
+                else:
+                    command = ['uv', 'run', 'envied', "dl", *self.options_list, "PLEX", url]    
+                self.runsubprocess(command)
+            except Exception as e:
+                print(
+                    "Error downloading video:",
+                    e,
+                    "Is devine/envied installed correctly via 'pip install <program name>?",
+                )
+            return None
+
+        else:
+            try:
+                headers = {
+                    "referer": "https://www.plex.tv",
+                    "user-agent": "Dalvik/2.9.8 (Linux; U; Android 9.9.2; ALE-L94 Build/NJHGGF)",
+                }
+                myhtml = self.get_data(url=url, headers=headers)
+                import re
+                from lxml import html
+
+                doc = html.fromstring(myhtml)
+
+                aria_labels = doc.xpath("//div[@data-id='seasons']//a[contains(@aria-label,'Season ')]/@aria-label")
+                titles = doc.xpath("//div[@data-id='seasons']//*[@title[starts-with(.,'Season ')]]/@title")
+
+                parts = []
+                for lbl in aria_labels:
+                    if "•" in lbl:
+                        parts.append(lbl.split("•", 1)[1].strip())
+                parts.extend(titles)
+
+                def season_num(s: str, default=10**9):
+                    m = re.search(r'\d+', s)
+                    return int(m.group()) if m else default
+
+                seasons = sorted({p.strip().lower() for p in parts}, key=season_num)
+                self.clear_series_data()  #  safely remove any leftovers
+                for item in seasons:
+                    url = f"https://watch.plex.tv/show/{selected.lower().replace(' ','-')}/season/{item.split(' ')[1]}"            
+                    html_bytes = self.get_data(url=url, headers=headers)
+                    doc = html.fromstring(html_bytes)
+                    for fig in doc.xpath("//div[@data-id='episodes']//figure"):
+                        hrefs = fig.xpath(".//a[contains(@href,'/season/') and contains(@href,'/episode/')]/@href")
+                        if not hrefs:
+                            continue
+                        relative_link = hrefs[0].split("?", 1)[0]
+                        m = re.search(r"/season/(\d+)/episode/(\d+)", relative_link)
+                        if not m:
+                            continue
+                        season_no, episode_no = map(int, m.groups())
+                        title = fig.xpath("normalize-space(.//figcaption/span[1]/@title | .//figcaption/span[1]/text())")
+                        synopsis = fig.xpath("normalize-space(.//figcaption/span[2]/@title | .//figcaption/span[2]/text())")
+                        episode = {
+                            "series_no": season_no,
+                            "title": f"{episode_no} {title}",
+                            "url": f"https://watch.plex.tv{relative_link}",
+                            "synopsis": rinse(synopsis) or None,
+                        }
+                        self.add_episode(selected, episode)
+                        #print(selected, episode)
+                
+            except Exception as e:
+                print(f"No valid data at {url} found.\n Exiting  {e}")
+                return
+
+
+        self.prepare_series_for_episode_selection(selected)  # creates list of series; allows user selection of wanted series prepares an episode list over chosen series
+        selected_final_episodes = self.display_final_episode_list(
+            self.final_episode_data
+        )
+        self.options_list = split_options(self.options)
+        for item in selected_final_episodes:
+            url = item.split(",")[2].lstrip()
+            if 'http' not in url:
+                url_pattern = r'https?://[^\s,]+'
+                match = re.search(url_pattern, item)
+                if match:
+                    url = match.group()
+            try:
+                if self.options_list[0] == "":
+                    command = ['uv', 'run', 'envied', "dl", "PLEX", url]
+                else:
+                    command = ['uv','run', 'envied', "dl", *self.options_list, "PLEX", url]
+                self.runsubprocess(command)
+            except Exception as e:
+                print(
+                    "Error downloading video:",
+                    e,
+                    "Is package/envied installed correctly?",
+                )
+
+        return None

packages/vinefeeder/src/vinefeeder/services/PLEX/config.yaml

@@ -0,0 +1,18 @@
+service_name: PLEX
+
+options: -l en
+
+media_dict:
+  action: https://watch.plex.tv/on-demand/category/action
+  animation: https://watch.plex.tv/on-demand/category/animation
+  comedy: https://watch.plex.tv/on-demand/category/comedy
+  crime: https://watch.plex.tv/on-demand/category/crime
+  documantary: https://watch.plex.tv/on-demand/category/documentary
+  drama: https://watch.plex.tv/on-demand/category/drama
+  en epsaniol: https://watch.plex.tv/on-demand/category/en-espanol
+  horror: https://watch.plex.tv/on-demand/category/horror
+  music: https://watch.plex.tv/on-demand/category/music
+  romance: https://watch.plex.tv/on-demand/category/romance
+  sci-fi: https://watch.plex.tv/on-demand/category/sci-fi
+  thriller: https://watch.plex.tv/on-demand/category/thriller
+  western: https://watch.plex.tv/on-demand/category/western