minor updates to extract_info

app/library/HttpAPI.py

@@ -40,10 +40,10 @@ from .Utils import (
     arg_converter,
     decrypt_data,
     encrypt_data,
+    extract_info,
     get_file,
     get_mime_type,
     get_sidecar_subtitles,
-    get_video_info,
     validate_url,
     validate_uuid,
 )
@@ -484,11 +484,12 @@ class HttpAPI(Common):
             preset = self.config.default_preset
 
         try:
-            key = self.cache.hash(url+str(preset))
+            key = self.cache.hash(url + str(preset))
 
-            if self.cache.has(key) and not request.query.get("force"):
+            if self.cache.has(key) and not request.query.get("force", False):
                 data = self.cache.get(key)
                 data["_cached"] = {
+                    "status": "hit",
                     "key": key,
                     "ttl": data.get("_cached", {}).get("ttl", 300),
                     "ttl_left": data.get("_cached", {}).get("expires", time.time() + 300) - time.time(),
@@ -501,16 +502,19 @@ class HttpAPI(Common):
             if self.config.ytdl_options.get("proxy", None):
                 opts["proxy"] = self.config.ytdl_options.get("proxy", None)
 
-            data = get_video_info(
+            data = extract_info(
+                config=YTDLPOpts.get_instance().preset(name=preset, with_cookies=True).add(opts).get_all(),
                 url=url,
-                ytdlp_opts=YTDLPOpts.get_instance().preset(name=preset, with_cookies=True).add(opts).get_all(),
+                debug=False,
                 no_archive=True,
+                follow_redirect=True,
             )
 
-            self.cache.set(key=self.cache.hash(url), value=data, ttl=300)
+            self.cache.set(key=key, value=data, ttl=300)
 
             data["_cached"] = {
-                "key": self.cache.hash(url),
+                "status": "miss",
+                "key": key,
                 "ttl": 300,
                 "ttl_left": 300,
                 "expires": time.time() + 300,
@@ -524,6 +528,7 @@ class HttpAPI(Common):
                 data={
                     "error": "failed to get video info.",
                     "message": str(e),
+                    "formats": [],
                 },
                 status=web.HTTPInternalServerError.status_code,
             )
@@ -555,17 +560,17 @@ class HttpAPI(Common):
         tasks = []
         response = []
 
-        def get_video_info_wrapper(id: str, url: str) -> tuple[str, dict]:
+        def info_wrapper(id: str, url: str) -> tuple[str, dict]:
             try:
                 return (
                     id,
-                    get_video_info(
+                    extract_info(
-                        url=url,
+                        config={
-                        ytdlp_opts={
                             "proxy": self.config.ytdl_options.get("proxy", None),
                             "simulate": True,
                             "dump_single_json": True,
                         },
+                        url=url,
                         no_archive=True,
                     ),
                 )
@@ -594,9 +599,7 @@ class HttpAPI(Common):
                     continue
 
                 tasks.append(
-                    asyncio.get_event_loop().run_in_executor(
+                    asyncio.get_event_loop().run_in_executor(None, lambda id=id, url=url: info_wrapper(id=id, url=url))
-                        None, lambda id=id, url=url: get_video_info_wrapper(id=id, url=url)
-                    )
                 )
 
         if len(tasks) > 0:

app/library/Utils.py

@@ -30,42 +30,6 @@ class StreamingError(Exception):
     """Raised when an error occurs during streaming."""
 
 
-def get_video_info(url: str, ytdlp_opts: dict | None = None, no_archive: bool = True) -> Any | dict[str, Any] | None:
-    """
-    Extracts video information from the given URL.
-
-    Args:
-        url (str): URL to extract information from.
-        ytdlp_opts (dict): Additional options to pass to yt-dlp.
-        no_archive (bool): Do not use download archive.
-
-    Returns:
-        dict: Video information.
-
-    """
-    params: dict = {
-        "quiet": True,
-        "color": "no_color",
-        "extract_flat": True,
-        "ignoreerrors": True,
-        "skip_download": True,
-        "ignore_no_formats_error": True,
-    }
-
-    if ytdlp_opts:
-        params = {**params, **ytdlp_opts}
-
-    if no_archive and "download_archive" in params:
-        del params["download_archive"]
-
-    # Remove keys that are not needed for info extraction.
-    keys_to_remove = [key for key in params if str(key).startswith("write") or key in ["postprocessors"]]
-    for key in keys_to_remove:
-        params.pop(key, None)
-
-    return yt_dlp.YoutubeDL(params=params).extract_info(url, download=False)
-
-
 def calc_download_path(base_path: str, folder: str | None = None, create_path: bool = True) -> str:
     """
     Calculates download path and prevents folder traversal.
@@ -98,7 +62,13 @@ def calc_download_path(base_path: str, folder: str | None = None, create_path: b
     return download_path
 
 
-def extract_info(config: dict, url: str, debug: bool = False) -> dict:
+def extract_info(
+    config: dict,
+    url: str,
+    debug: bool = False,
+    no_archive: bool = False,
+    follow_redirect: bool = False,
+) -> dict:
     """
     Extracts video information from the given URL.
 
@@ -149,7 +119,15 @@ def extract_info(config: dict, url: str, debug: bool = False) -> dict:
 
         params["logger"] = log_wrapper
 
-    return yt_dlp.YoutubeDL(params=params).extract_info(url, download=False)
+    if no_archive and "download_archive" in params:
+        del params["download_archive"]
+
+    data = yt_dlp.YoutubeDL(params=params).extract_info(url, download=False)
+
+    if follow_redirect and "_type" in data and "url" == data["_type"]:
+        return extract_info(config, data["url"], debug=debug, no_archive=no_archive, follow_redirect=follow_redirect)
+
+    return data
 
 
 def merge_dict(source: dict, destination: dict) -> dict: