Locally check the id instead fo requesting extract_info twice.

.vscode/launch.json

@@ -27,6 +27,22 @@
       "request": "launch",
       "program": "app/main.py",
       "console": "internalConsole",
+      "justMyCode": true,
+      "env": {
+        "YTP_CONFIG_PATH": "${workspaceFolder}/var/config",
+        "YTP_DOWNLOAD_PATH": "${workspaceFolder}/var/downloads",
+        "YTP_TEMP_PATH": "${workspaceFolder}/var/tmp",
+        "YTP_URL_HOST": "http://localhost:8081",
+        "YTP_LOG_LEVEL": "DEBUG"
+      }
+    },
+    {
+      "name": "Python: main.py (Deep)",
+      "type": "debugpy",
+      "request": "launch",
+      "program": "app/main.py",
+      "console": "internalConsole",
+      "justMyCode": false,
       "env": {
         "YTP_CONFIG_PATH": "${workspaceFolder}/var/config",
         "YTP_DOWNLOAD_PATH": "${workspaceFolder}/var/downloads",

app/DownloadQueue.py

@@ -219,7 +219,15 @@ class DownloadQueue:
         already.add(url)
 
         try:
-            LOG.debug(f'extracting info from {url=}')
+            downloaded, id_dict = self.isDownloaded(url)
+            if downloaded is True:
+                message = f"[ { id_dict.get('id') } ]: has been downloaded already."
+                LOG.info(message)
+                return {'status': 'error', 'msg': message}
+
+            started = time.perf_counter()
+            LOG.debug(f'extract_info: checking {url=}')
+
             entry = await asyncio.wait_for(
                 fut=asyncio.get_running_loop().run_in_executor(
                     None,
@@ -231,40 +239,9 @@ class DownloadQueue:
                 timeout=self.config.extract_info_timeout)
 
             if not entry:
-                if not self.config.keep_archive:
-                    return {
-                        'status': 'error',
-                        'msg': 'No metadata, most likely video has been downloaded before.' if self.config.keep_archive else 'Unable to extract info check logs.'
-                    }
+                return {'status': 'error', 'msg': 'Unable to extract info check logs.'}
 
-                LOG.debug(f'No metadata, Rechecking with archive disabled. {url=}')
-
-                entry = await asyncio.wait_for(
-                    fut=asyncio.get_running_loop().run_in_executor(
-                        None,
-                        ExtractInfo,
-                        mergeConfig(self.config.ytdl_options, ytdlp_config),
-                        url,
-                        bool(self.config.ytdl_debug),
-                        True
-                    ),
-                    timeout=self.config.extract_info_timeout
-                )
-
-                if not entry:
-                    return {'status': 'error', 'msg': 'Unable to extract info check logs.'}
-
-                if self.isDownloaded(entry):
-                    LOG.info(f'[{entry.get("id")}: {entry.get("title")}]: has been downloaded already.')
-                    return {
-                        'status': 'error',
-                        'msg': f'[{entry.get("id")}: {entry.get("title")}]: has been downloaded already.'
-                    }
-
-            if self.isDownloaded(entry):
-                raise yt_dlp.utils.ExistingVideoReached()
-
-            LOG.debug(f'extract_info for [{url=}] length: {len(entry)}')
+            LOG.debug(f'extract_info: for [{url=}] is done in {time.perf_counter() - started}. Length: {len(entry)}')
         except yt_dlp.utils.ExistingVideoReached:
             return {'status': 'error', 'msg': 'Video has been downloaded already and recorded in archive.log file.'}
         except yt_dlp.utils.YoutubeDLError as exc:
@@ -422,5 +399,8 @@ class DownloadQueue:
 
         self.event.set()
 
-    def isDownloaded(self, info: dict) -> bool:
-        return self.config.keep_archive and isDownloaded(self.config.ytdl_options.get('download_archive', None), info)
+    def isDownloaded(self, url: str) -> tuple[bool, dict[str | None, str | None, str | None]]:
+        if not url or not self.config.keep_archive:
+            return False, None
+
+        return isDownloaded(self.config.ytdl_options.get('download_archive', None), url)

app/Utils.py

@@ -14,6 +14,8 @@ LOG = logging.getLogger('Utils')
 AUDIO_FORMATS: tuple = ('m4a', 'mp3', 'opus', 'wav')
 IGNORED_KEYS: tuple = ('cookiefile', 'paths', 'outtmpl', 'progress_hooks', 'postprocessor_hooks',)
 
+YTDLP_INFO_CLS = None
+
 
 def get_format(format: str, quality: str) -> str:
     """
@@ -197,24 +199,53 @@ def mergeConfig(config: dict, new_config: dict) -> dict:
     return mergeDict(new_config, config)
 
 
-def isDownloaded(archive_file: str, info: dict) -> bool:
-    if not info or not archive_file or not os.path.exists(archive_file):
-        return False
+def isDownloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, str | None, str | None]]:
+    global YTDLP_INFO_CLS
 
-    try:
-        id = yt_dlp.YoutubeDL()._make_archive_id(info)
-        if not id:
-            return False
-    except Exception as e:
-        LOG.error(f'Error generating archive id: {e}')
-        return False
+    idDict = {
+        'id': None,
+        'ie_key': None,
+        'archive_id': None,
+    }
+
+    if not url or not archive_file or not os.path.exists(archive_file):
+        return False, idDict,
+
+    if not YTDLP_INFO_CLS:
+        YTDLP_INFO_CLS = yt_dlp.YoutubeDL(params={
+            'color': 'no_color',
+            'extract_flat': True,
+            'skip_download': True,
+            'ignoreerrors': True,
+            'ignore_no_formats_error': True,
+            'quiet': True,
+        })
+
+    for key, ie in YTDLP_INFO_CLS._ies.items():
+        if not ie.suitable(url):
+            continue
+
+        if not ie.working():
+            break
+
+        temp_id = ie.get_temp_id(url)
+        if not temp_id:
+            break
+
+        idDict['id'] = temp_id
+        idDict['ie_key'] = key
+        idDict['archive_id'] = YTDLP_INFO_CLS._make_archive_id(idDict)
+        break
+
+    if not idDict['archive_id']:
+        return False, idDict,
 
     with open(archive_file, 'r') as f:
         for line in f.readlines():
-            if id in line:
-                return True
+            if idDict['archive_id'] in line:
+                return True, idDict,
 
-    return False
+    return False, idDict,
 
 
 def jsonCookie(cookies: dict[dict[str, any]]) -> str | None: