nn

Provider/ytdlp.py

@@ -10,7 +10,7 @@ This keeps format selection logic in ytdlp and leaves add-file plug-and-play.
 from __future__ import annotations
 
 import sys
-from typing import Any, Dict, Iterable, List, Optional
+from typing import Any, Dict, Iterable, List, Optional, Tuple
 
 from ProviderCore.base import Provider, SearchResult
 from SYS.provider_helpers import TableProviderMixin
@@ -65,31 +65,31 @@ class ytdlp(TableProviderMixin, Provider):
 
     # Dynamically load URL domains from yt-dlp's extractors
     # This enables provider auto-discovery for format selection routing
-    @property
+    @classmethod
-    def URL(self) -> List[str]:
+    def url_patterns(cls) -> Tuple[str, ...]:
-        """Get list of supported domains from yt-dlp extractors."""
+        """Return supported domains from yt-dlp extractors."""
         try:
             import yt_dlp
             # Build a comprehensive list from known extractors and fallback domains
-            domains = set(self._fallback_domains)
+            domains = set(cls._fallback_domains)
 
             # Try to get extractors and extract domain info
             try:
                 extractors = yt_dlp.gen_extractors()
                 for extractor_class in extractors:
                     # Get extractor name and try to convert to domain
-                    name = getattr(extractor_class, 'IE_NAME', '')
+                    name = getattr(extractor_class, "IE_NAME", "")
-                    if name and name not in ('generic', 'http'):
+                    if name and name not in ("generic", "http"):
                         # Convert extractor name to domain (e.g., 'YouTube' -> 'youtube.com')
-                        name_lower = name.lower().replace('ie', '').strip()
+                        name_lower = name.lower().replace("ie", "").strip()
                         if name_lower and len(name_lower) > 2:
                             domains.add(f"{name_lower}.com")
             except Exception:
                 pass
 
-            return list(domains) if domains else self._fallback_domains
+            return tuple(domains) if domains else tuple(cls._fallback_domains)
         except Exception:
-            return self._fallback_domains
+            return tuple(cls._fallback_domains)
     
     # Fallback common domains in case extraction fails
     _fallback_domains = [

cmdlet/download_file.py

@@ -674,6 +674,35 @@ class Download_File(Cmdlet):
         unsupported = list(set(raw_urls or []) - set(supported or []))
         return supported, unsupported
 
+    @staticmethod
+    def _match_provider_urls(
+        raw_urls: Sequence[str],
+        registry: Dict[str, Any],
+    ) -> Dict[str, str]:
+        matches: Dict[str, str] = {}
+        if not raw_urls:
+            return matches
+
+        match_provider_name_for_url = registry.get("match_provider_name_for_url")
+        if not callable(match_provider_name_for_url):
+            return matches
+
+        for url in raw_urls:
+            try:
+                url_str = str(url or "").strip()
+            except Exception:
+                continue
+            if not url_str:
+                continue
+            try:
+                provider_name = match_provider_name_for_url(url_str)
+            except Exception:
+                provider_name = None
+            if provider_name:
+                matches[url_str] = str(provider_name).strip().lower()
+
+        return matches
+
     def _parse_query_keyed_spec(self, query_spec: Optional[str]) -> Dict[str, List[str]]:
         if not query_spec:
             return {}
@@ -2500,10 +2529,12 @@ class Download_File(Cmdlet):
 
                     original_skip_preflight = None
                     original_timeout = None
+                    original_skip_direct = None
                     try:
                         if isinstance(config, dict):
                             original_skip_preflight = config.get("_skip_url_preflight")
                             original_timeout = config.get("_pipeobject_timeout_seconds")
+                            original_skip_direct = config.get("_skip_direct_on_streaming_failure")
                     except Exception:
                         original_skip_preflight = None
                         original_timeout = None
@@ -2521,6 +2552,7 @@ class Download_File(Cmdlet):
                                 return 0
                             if isinstance(config, dict):
                                 config["_skip_url_preflight"] = True
+                                config["_skip_direct_on_streaming_failure"] = True
 
                         if isinstance(config, dict) and config.get("_pipeobject_timeout_seconds") is None:
                             config["_pipeobject_timeout_seconds"] = 60
@@ -2554,6 +2586,10 @@ class Download_File(Cmdlet):
                                     config.pop("_pipeobject_timeout_seconds", None)
                                 else:
                                     config["_pipeobject_timeout_seconds"] = original_timeout
+                                if original_skip_direct is None:
+                                    config.pop("_skip_direct_on_streaming_failure", None)
+                                else:
+                                    config["_skip_direct_on_streaming_failure"] = original_skip_direct
                         except Exception:
                             pass
 
@@ -2605,12 +2641,18 @@ class Download_File(Cmdlet):
             if picker_result is not None:
                 return int(picker_result)
 
-            streaming_candidates = list(raw_url)
+            provider_url_matches = self._match_provider_urls(raw_url, registry)
+            streaming_candidates = [
+                url for url in raw_url
+                if provider_url_matches.get(str(url).strip()) == "ytdlp"
+            ]
             supported_streaming, unsupported_streaming = self._filter_supported_urls(streaming_candidates)
+            matched_ytdlp = bool(streaming_candidates)
 
             streaming_exit_code: Optional[int] = None
             streaming_downloaded = 0
             if supported_streaming:
+                debug(f"[download-file] Using ytdlp provider for {len(supported_streaming)} URL(s)")
                 streaming_exit_code = self._run_streaming_urls(
                     streaming_urls=supported_streaming,
                     args=args,
@@ -2626,6 +2668,17 @@ class Download_File(Cmdlet):
 
                     if not raw_url and not piped_items:
                         return int(streaming_exit_code or 0)
+                else:
+                    try:
+                        skip_direct = bool(config.get("_skip_direct_on_streaming_failure")) if isinstance(config, dict) else False
+                    except Exception:
+                        skip_direct = False
+                    if matched_ytdlp:
+                        skip_direct = True
+                    if skip_direct:
+                        raw_url = [u for u in raw_url if u not in supported_streaming]
+                        if not raw_url and not piped_items:
+                            return int(streaming_exit_code or 1)
 
             # Re-check picker if partial processing occurred
             picker_result = self._maybe_show_provider_picker(