Fix VEO and improve error handling

Author: Commandcracker

src/gucken/__init__.py

@@ -1,4 +1,4 @@
 import warnings
 warnings.filterwarnings('ignore', message='Using slow pure-python SequenceMatcher. Install python-Levenshtein to remove this warning')
 
-__version__ = "0.3.6"
+__version__ = "0.3.7"

src/gucken/gucken.py

@@ -118,28 +118,32 @@ async def get_working_direct_link(hosters: list[Hoster], app: "GuckenApp") -> Un
                 severity="warning",
             )
             continue
-        if direct_link is None:
+        if not direct_link or not direct_link.url or not isinstance(direct_link.url, str) or not isinstance(direct_link, DirectLink):
             logging.warning(
-                "%s: failed to retrieve video URL from: \"%s\"",
+                "%s: Returned empty URL: \"%s\"",
                 name,
-                hoster.url,
-                exc_info=True
+                hoster.url
             )
             app.notify(
-                "Failed to retrieve video URL",
+                "Returned empty URL",
                 title=f"{name} error",
                 severity="warning",
             )
             continue
         is_working = await direct_link.check_is_working()
-        logging.info(
-            'Check: "%s" Working: "%s" URL: "%s"',
-            name,
-            is_working,
-            direct_link,
-        )
         if is_working:
             return direct_link
+        else:
+            logging.warning(
+                "%s: Video URL is not working: \"%s\"",
+                name,
+                direct_link.url or "None"
+            )
+            app.notify(
+                "Video URL is not working",
+                title=f"{name} error",
+                severity="warning",
+            )
     return None
 
 

src/gucken/hoster/common.py

@@ -11,6 +11,8 @@ class DirectLink:
     force_hls: bool = False
 
     async def check_is_working(self) -> bool:
+        if not self.url:
+            return False
         try:
             async with AsyncClient(verify=False) as client:
                 response = await client.head(

src/gucken/hoster/veo.py

@@ -9,7 +9,7 @@
 
 REDIRECT_PATTERN = re_compile("https?://[^'\"<>]+")
 
-# Credit
+# Modified
 # https://github.com/wolfswolke/aniworld_scraper/blob/41bd0f23cbc02352481dd92e6d986d1fe30c76bf/src/logic/search_for_links.py#L23
 
 def deb_func1(input_string):
@@ -56,17 +56,8 @@ def deb_func(input_var):
 
 def find_script_element(raw_html):
     soup = BeautifulSoup(raw_html, features="html.parser")
-    script_object = soup.find_all("script")
-    obfuscated_string = ""
-    for script in script_object:
-        script = str(script)
-        if "KGMAaM=" in script:
-            obfuscated_string = script
-            break
-    if obfuscated_string == "":
-        return None
-    obfuscated_string = obfuscated_string.split('MKGMa="')[1]
-    obfuscated_string = obfuscated_string.split('"')[0]
+    script_object = soup.find("script", type="application/json")
+    obfuscated_string = script_object.text[2:-2]
     output = deb_func(obfuscated_string)
     return output["source"]
 

src/gucken/provider/aniworld.py

@@ -19,38 +19,33 @@
 from ..utils import json_loads
 from ..utils import fully_unescape
 
+HOSTER_MAP = {
+    "VOE": VOEHoster,
+    "Doodstream": DoodstreamHoster,
+    "Vidoza": VidozaHoster,
+    "Streamtape": StreamtapeHoster,
+    "SpeedFiles": SpeedFilesHoster,
+    "Filemoon": FilemoonHoster,
+    "Luluvdo": LuluvdoHoster,
+    "Vidmoly": VidmolyHoster,
+    "LoadX": LoadXHoster
+}
+
+LANGUAGE_MAP = {
+    "english": Language.EN,
+    "english-german": Language.EN_DESUB,
+    "japanese-english": Language.JP_ENSUB,
+    "japanese-german": Language.JP_DESUB,
+    "german": Language.DE
+}
+
 def provider_to_hoster(provider: str, url: str) -> Hoster:
-    if provider == "VOE":
-        return VOEHoster(url)
-    if provider == "Doodstream":
-        return DoodstreamHoster(url)
-    if provider == "Vidoza":
-        return VidozaHoster(url)
-    if provider == "Streamtape":
-        return StreamtapeHoster(url)
-    if provider == "SpeedFiles":
-        return SpeedFilesHoster(url)
-    if provider == "Filemoon":
-        return FilemoonHoster(url)
-    if provider == "Luluvdo":
-        return LuluvdoHoster(url)
-    if provider == "Vidmoly":
-        return VidmolyHoster(url)
-    if provider == "LoadX":
-        return LoadXHoster(url)
+    if provider in HOSTER_MAP:
+        return HOSTER_MAP[provider](url)
 
 
 def lang_img_src_lang_name_to_lang(name: str) -> Language:
-    if name == "english":
-        return Language.EN
-    if name == "english-german":
-        return Language.EN_DESUB
-    if name == "japanese-english":
-        return Language.JP_ENSUB
-    if name == "japanese-german":
-        return Language.JP_DESUB
-    if name == "german":
-        return Language.DE
+    return LANGUAGE_MAP[name]
 
 
 @dataclass
@@ -283,24 +278,8 @@ async def get_episodes_from_soup(
         hoster = set()
         for h in episode.find_all_next("i", class_="icon"):
             t = h.attrs.get("title")
-            if t == "VOE":
-                hoster.add(VOEHoster)
-            if t == "Doodstream":
-                hoster.add(DoodstreamHoster)
-            if t == "Vidoza":
-                hoster.add(VidozaHoster)
-            if t == "Streamtape":
-                hoster.add(StreamtapeHoster)
-            if t == "SpeedFiles":
-                hoster.add(SpeedFilesHoster)
-            if t == "Filemoon":
-                hoster.add(FilemoonHoster)
-            if t == "Luluvdo":
-                hoster.add(LuluvdoHoster)
-            if t == "Vidmoly":
-                hoster.add(VidmolyHoster)
-            if t == "LoadX":
-                hoster.add(LoadXHoster)
+            if t in HOSTER_MAP:
+                hoster.add(HOSTER_MAP[t])
 
         e_count += 1
         title_en = fully_unescape(title.find("span").text.strip())

src/gucken/provider/serienstream.py

@@ -27,38 +27,33 @@
 headers = {"Host": "serienstream.to"}
 extensions = {"sni_hostname": "serienstream.to"}
 
+HOSTER_MAP = {
+    "VOE": VOEHoster,
+    "Doodstream": DoodstreamHoster,
+    "Vidoza": VidozaHoster,
+    "Streamtape": StreamtapeHoster,
+    "SpeedFiles": SpeedFilesHoster,
+    "Filemoon": FilemoonHoster,
+    "Luluvdo": LuluvdoHoster,
+    "Vidmoly": VidmolyHoster,
+    "LoadX": LoadXHoster
+}
+
+LANGUAGE_MAP = {
+    "english": Language.EN,
+    "english-german": Language.EN_DESUB,
+    "japanese-english": Language.JP_ENSUB,
+    "japanese-german": Language.JP_DESUB,
+    "german": Language.DE
+}
 
 def provider_to_hoster(provider: str, url: str) -> Hoster:
-    if provider == "VOE":
-        return VOEHoster(url)
-    if provider == "Doodstream":
-        return DoodstreamHoster(url)
-    if provider == "Vidoza":
-        return VidozaHoster(url)
-    if provider == "Streamtape":
-        return StreamtapeHoster(url)
-    if provider == "SpeedFiles":
-        return SpeedFilesHoster(url)
-    if provider == "Filemoon":
-        return FilemoonHoster(url)
-    if provider == "Luluvdo":
-        return LuluvdoHoster(url)
-    if provider == "Vidmoly":
-        return VidmolyHoster(url)
-    if provider == "LoadX":
-        return LoadXHoster(url)
+    if provider in HOSTER_MAP:
+        return HOSTER_MAP[provider](url)
+
 
 def lang_img_src_lang_name_to_lang(name: str) -> Language:
-    if name == "english":
-        return Language.EN
-    if name == "english-german":
-        return Language.EN_DESUB
-    if name == "japanese-english":
-        return Language.JP_ENSUB
-    if name == "japanese-german":
-        return Language.JP_DESUB
-    if name == "german":
-        return Language.DE
+    return LANGUAGE_MAP[name]
 
 
 @dataclass
@@ -292,24 +287,8 @@ async def get_episodes_from_soup(
         hoster = set()
         for h in episode.find_all_next("i", class_="icon"):
             t = h.attrs.get("title")
-            if t == "VOE":
-                hoster.add(VOEHoster)
-            if t == "Doodstream":
-                hoster.add(DoodstreamHoster)
-            if t == "Vidoza":
-                hoster.add(VidozaHoster)
-            if t == "Streamtape":
-                hoster.add(StreamtapeHoster)
-            if t == "SpeedFiles":
-                hoster.add(SpeedFilesHoster)
-            if t == "Filemoon":
-                hoster.add(FilemoonHoster)
-            if t == "Luluvdo":
-                hoster.add(LuluvdoHoster)
-            if t == "Vidmoly":
-                hoster.add(VidmolyHoster)
-            if t == "LoadX":
-                hoster.add(LoadXHoster)
+            if t in HOSTER_MAP:
+                hoster.add(HOSTER_MAP[t])
 
         e_count += 1
         title_en = fully_unescape(title.find("span").text.strip())

src/gucken/rome.py

@@ -17,8 +17,9 @@ def roman_to_int(roman: str) -> int:
     return result
 
 
-def replace_roman_numerals(text: str) -> str:
-    def repl(match):
-        return str(roman_to_int(match.group(0)))
+def _repl(match):
+    return str(roman_to_int(match.group()))
+
 
-    return ROMAN_PATTERN.sub(repl, text)
+def replace_roman_numerals(text: str) -> str:
+    return ROMAN_PATTERN.sub(_repl, text)

src/gucken/update.py

@@ -1,10 +1,10 @@
 from dataclasses import dataclass
 from typing import Union
 
-from .networking import AsyncClient
 from packaging.version import Version
 
 from . import __version__ as current_version
+from .networking import AsyncClient
 from .utils import json_loads
 
 PACKAGE_NAME = "gucken"
@@ -31,7 +31,6 @@ async def check() -> Union[UpdateResult, None]:
 
 def main():
     from asyncio import run
-
     print(run(check()))
 
 

src/gucken/utils.py

@@ -1,10 +1,8 @@
-#import logging
-import os
 import sys
 from html import unescape
-
-from typing import Union
-from typing import NamedTuple
+from os import getenv
+from os.path import join, dirname
+from typing import Union, NamedTuple
 
 from .player.android import AndroidChoosePlayer
 from .player.common import Player
@@ -15,13 +13,15 @@
 from .player.wmplayer import WMPlayer
 
 is_android = hasattr(sys, "getandroidapilevel")
+is_windows = sys.platform.startswith("win")
+is_linux = sys.platform.startswith("linux")
+is_mac_os = sys.platform.startswith("darwin")
+is_bsd = "freebsd" in sys.platform or sys.platform.startswith("dragonfly")
 
 try:
     from orjson import loads as json_loads
-    # logging.debug("Using orjson")
 except ImportError:
     from json import loads as json_loads
-    # logging.debug("Using default json")
 
 
 def detect_player() -> Union[Player, None]:
@@ -47,51 +47,35 @@ def detect_player() -> Union[Player, None]:
     return None
 
 
-def is_windows():
-    return sys.platform.startswith("win")
-
-
-def is_linux():
-    return sys.platform.startswith("linux")
-
-
-def is_mac_os():
-    return sys.platform.startswith("darwin")
-
-
-def is_bsd():
-    return "freebsd" in sys.platform or sys.platform.startswith("dragonfly")
-
-
 class VLCPaths(NamedTuple):
     vlc_intf_path: str
     vlc_intf_user_path: str
     vlc_module_path: str
 
 
 def get_vlc_intf_user_path(player_path: str) -> VLCPaths:
-    if is_linux():
+    if is_linux:
         if 'snap' in player_path:
             vlc_intf_path = '/snap/vlc/current/usr/lib/vlc/lua/intf/'
-            vlc_intf_user_path = os.path.join(os.getenv('HOME', '.'), "snap/vlc/current/.local/share/vlc/lua/intf/")
+            vlc_intf_user_path = join(getenv('HOME', '.'), "snap/vlc/current/.local/share/vlc/lua/intf/")
         else:
             vlc_intf_path = "/usr/lib/vlc/lua/intf/"
-            vlc_intf_user_path = os.path.join(os.getenv('HOME', '.'), ".local/share/vlc/lua/intf/")
-    elif is_mac_os():
+            vlc_intf_user_path = join(getenv('HOME', '.'), ".local/share/vlc/lua/intf/")
+    elif is_mac_os:
         vlc_intf_path = "/Applications/VLC.app/Contents/MacOS/share/lua/intf/"
-        vlc_intf_user_path = os.path.join(
-            os.getenv('HOME', '.'), "Library/Application Support/org.videolan.vlc/lua/intf/")
-    elif is_bsd():
+        vlc_intf_user_path = join(
+            getenv('HOME', '.'), "Library/Application Support/org.videolan.vlc/lua/intf/")
+    elif is_bsd:
         # *BSD ports/pkgs install to /usr/local by default.
         # This should also work for all the other BSDs, such as OpenBSD or DragonFly.
         vlc_intf_path = "/usr/local/lib/vlc/lua/intf/"
-        vlc_intf_user_path = os.path.join(os.getenv('HOME', '.'), ".local/share/vlc/lua/intf/")
+        vlc_intf_user_path = join(getenv('HOME', '.'), ".local/share/vlc/lua/intf/")
     elif "vlcportable.exe" in player_path.lower():
-        vlc_intf_path = os.path.dirname(player_path).replace("\\", "/") + "/App/vlc/lua/intf/"
+        vlc_intf_path = dirname(player_path).replace("\\", "/") + "/App/vlc/lua/intf/"
         vlc_intf_user_path = vlc_intf_path
     else:
-        vlc_intf_path = os.path.dirname(player_path).replace("\\", "/") + "/lua/intf/"
-        vlc_intf_user_path = os.path.join(os.getenv('APPDATA', '.'), "VLC\\lua\\intf\\")
+        vlc_intf_path = dirname(player_path).replace("\\", "/") + "/lua/intf/"
+        vlc_intf_user_path = join(getenv('APPDATA', '.'), "VLC\\lua\\intf\\")
 
     vlc_module_path = vlc_intf_path + "modules/?.luac"