danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat: add generic browser extractor

Browse source
This commit is contained in:
arabcoders 2026-05-08 01:08:08 +03:00
parent 9b3cc8b310
commit 928ddd5b9a
10 changed files with 659 additions and 184 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

112
FAQ.md
View file

@ -37,6 +37,7 @@ or the `environment:` section in `compose.yaml` file.
| YTP_YTDLP_AUTO_UPDATE | Whether to enable the auto update for yt-dlp | `true` |
| YTP_YTDLP_DEBUG | Whether to turn debug logging for the internal `yt-dlp` package | `false` |
| YTP_YTDLP_VERSION | The version of yt-dlp to use. Defaults to latest version | `(not_set)` |
| YTP_BROWSER_URL | Remote browser endpoint for the bundled `browser` extractor | `(not_set)` |
| YTP_FLARESOLVERR_URL | FlareSolverr endpoint URL. | `(not_set)` |
| YTP_FLARESOLVERR_MAX_TIMEOUT | Max FlareSolverr challenge timeout in seconds | `120` |
| YTP_FLARESOLVERR_CLIENT_TIMEOUT | HTTP client timeout (seconds) when calling FlareSolverr | `120` |
@ -614,3 +615,114 @@ services:
```
For more information please visit [FlareSolverr](https://github.com/FlareSolverr/FlareSolverr) project.
# How to use the browser extractor?
YTPTube ships with a bundled `browser` extractor plugin for `yt-dlp`. It opens the target page in a remote browser,
waits for the page network activity, then tries to extract media urls from the captured requests and media elements.
This is mainly useful for sites where the regular extractor or generic page parsing does not expose the final media url,
but a real browser session does.
## Environment variables
You can define the remote browser endpoint globally with the following environment variable.
### Selenium example
```env
YTP_BROWSER_URL=selenium+http://selenium:4444/wd/hub
```
### Playwright example
For a native Playwright server, point to the Playwright websocket endpoint:
```env
YTP_BROWSER_URL=playwright+ws://playwright:3000/
```
### Playwright CDP example
To connect Playwright over the Chrome DevTools Protocol, use either the shorthand `playwright+cdp://` form or an explicit transport form:
```env
YTP_BROWSER_URL=playwright+cdp://chrome:9222/
# same as playwright+cdp+http://chrome:9222/
```
- `YTP_BROWSER_URL` is the full backend selector and endpoint in one value.
- Supported forms are:
- `selenium+http://...`
- `selenium+https://...`
- `playwright+ws://...`
- `playwright+wss://...`
- `playwright+cdp://...` which is treated as `playwright+cdp+http://...`
- `playwright+cdp+http://...`
- `playwright+cdp+https://...`
- `playwright+cdp+ws://...`
- `playwright+cdp+wss://...`
## yt-dlp usage
To enable the extractor for a preset or task, add the following to the `Command options for yt-dlp` field:
```bash
--use-extractors "browser,default"
```
If you want to set the browser extractor options directly on the yt-dlp side, you can also use `--extractor-args` with `browser:url=...`:
### Selenium example
```bash
--use-extractors "browser,default" --extractor-args "browser:url=selenium+http://selenium:4444/wd/hub"
```
### Playwright example
```bash
--use-extractors "browser,default" --extractor-args "browser:url=playwright+ws://playwright:3000/"
```
### Playwright CDP example
```bash
--use-extractors "browser,default" --extractor-args "browser:url=playwright+cdp://chrome:9222/"
```
The explicit `--extractor-args` value takes priority over `YTP_BROWSER_URL`.
## Example compose setup
```yaml
services:
ytptube:
user: "${UID:-1000}:${UID:-1000}"
image: ghcr.io/arabcoders/ytptube:latest
container_name: ytptube
restart: unless-stopped
environment:
- YTP_BROWSER_URL=selenium+http://selenium:4444/wd/hub # or playwright+ws://playwright:3000/ or playwright+cdp://chrome:9222/
ports:
- "8081:8081"
volumes:
- ./config:/config:rw
- ./downloads:/downloads:rw
selenium:
image: selenium/standalone-chrome:latest
container_name: selenium
restart: unless-stopped
shm_size: 2gb
playwright:
image: mcr.microsoft.com/playwright:v1.59.0-noble
container_name: playwright
restart: unless-stopped
command: /bin/sh -c "npx -y playwright@1.59.0 run-server --port 3000 --host 0.0.0.0"
```
> [!NOTE]
> The browser extractor is slower than the normal extractor flow and should only be used when a site actually needs a real browser session.
> playwright require same version for both the server and the client, so make sure to use the same version in the container and in your local environment if you want to test it locally.

1
README.md
View file

@ -38,6 +38,7 @@ Example of the Simple mode interface.
* Conditions feature to apply custom options based on `yt-dlp` returned info.
* Custom browser extensions, bookmarklets and iOS shortcuts to send links to YTPTube instance.
* A bundled executable version for Windows, macOS and Linux. `MacOS version is untested`.
* Use playwright or selenium for extractors that require a browser. see [related FAQ](FAQ.md#how-to-use-the-browser-extractor).
Please read the [FAQ](FAQ.md) for more information.

126
app/features/ytdlp/extractor.py
View file

@ -18,6 +18,21 @@ from app.library.Singleton import Singleton
LOG: logging.Logger = logging.getLogger("downloads.extractor")
def _ytdlp_logger(target: logging.Logger):
def _log(level: int, msg: str, *args: Any, **kwargs: Any) -> None:
if level <= logging.DEBUG and isinstance(msg, str) and msg.startswith("[debug] "):
target.debug(msg.removeprefix("[debug] "), *args, **kwargs)
return
if level <= logging.DEBUG:
target.info(msg, *args, **kwargs)
return
target.log(level, msg, *args, **kwargs)
return _log
def _get_process_pool_kwargs() -> dict[str, Any]:
"""Use a fork-based pool for frozen Linux builds."""
if sys.platform == "linux" and getattr(sys, "frozen", False):
@ -254,64 +269,69 @@ def extract_info_sync(
log_wrapper = LogWrapper()
id_dict: dict[str, str | None] = get_archive_id(url=url)
archive_id: str | None = f".{id_dict['id']}" if id_dict.get("id") else None
log_wrapper.add_target(
target=logging.getLogger(f"yt-dlp{archive_id or '.extract_info'}"),
level=logging.DEBUG if debug else logging.WARNING,
)
captured_logs: list[str] = kwargs.get("captured_logs", [])
if capture_logs is not None:
log_wrapper.add_target(
target=lambda _, msg: captured_logs.append(msg),
level=capture_logs,
name="log-capture",
)
if log_wrapper.has_targets():
if "logger" in params:
log_wrapper.add_target(target=params["logger"], level=logging.DEBUG)
params["logger"] = log_wrapper
if kwargs.get("no_log", False):
params["logger"] = LogWrapper()
params["quiet"] = True
params["no_warnings"] = True
if no_archive and "download_archive" in params:
del params["download_archive"]
data: dict[str, Any] | None = YTDLP(params=params).extract_info(url, download=False)
if data and follow_redirect and "_type" in data and "url" == data["_type"]:
return extract_info_sync(
config,
data["url"],
debug=debug,
no_archive=no_archive,
follow_redirect=follow_redirect,
sanitize_info=sanitize_info,
capture_logs=capture_logs,
captured_logs=captured_logs,
**kwargs,
)
if not data:
return (data, captured_logs)
logger_name: str = f"yt-dlp{archive_id or '.extract_info'}"
try:
from app.features.ytdlp.mini_filter import match_str
log_wrapper.add_target(
target=_ytdlp_logger(logging.getLogger(logger_name)),
level=logging.DEBUG,
name=logger_name,
)
data["is_premiere"] = match_str("media_type=video & duration & is_live", data)
if not data["is_premiere"]:
data["is_premiere"] = "video" == data.get("media_type") and "is_upcoming" == data.get("live_status")
except ImportError:
pass
captured_logs: list[str] = kwargs.get("captured_logs", [])
if capture_logs is not None:
log_wrapper.add_target(
target=lambda _, msg: captured_logs.append(msg),
level=capture_logs,
name="log-capture",
)
result = YTDLP.sanitize_info(data, remove_private_keys=True) if sanitize_info else data
if log_wrapper.has_targets():
if "logger" in params:
log_wrapper.add_target(target=params["logger"], level=logging.DEBUG)
return (result, captured_logs)
params["logger"] = log_wrapper
if kwargs.get("no_log", False):
params["logger"] = LogWrapper()
params["quiet"] = True
params["no_warnings"] = True
if no_archive and "download_archive" in params:
del params["download_archive"]
data: dict[str, Any] | None = YTDLP(params=params).extract_info(url, download=False)
if data and follow_redirect and "_type" in data and "url" == data["_type"]:
return extract_info_sync(
config,
data["url"],
debug=debug,
no_archive=no_archive,
follow_redirect=follow_redirect,
sanitize_info=sanitize_info,
capture_logs=capture_logs,
captured_logs=captured_logs,
**kwargs,
)
if not data:
return (data, captured_logs)
try:
from app.features.ytdlp.mini_filter import match_str
data["is_premiere"] = match_str("media_type=video & duration & is_live", data)
if not data["is_premiere"]:
data["is_premiere"] = "video" == data.get("media_type") and "is_upcoming" == data.get("live_status")
except ImportError:
pass
result = YTDLP.sanitize_info(data, remove_private_keys=True) if sanitize_info else data
return (result, captured_logs)
finally:
logging.Logger.manager.loggerDict.pop(logger_name, None)
async def fetch_info(

96
app/features/ytdlp/tests/test_ytdlp_extractor.py
View file

@ -1,6 +1,13 @@
import logging
from unittest.mock import MagicMock, patch
from app.features.ytdlp.extractor import ExtractorConfig, ExtractorPool, _get_process_pool_kwargs, extract_info_sync
from app.features.ytdlp.extractor import (
ExtractorConfig,
ExtractorPool,
_get_process_pool_kwargs,
_ytdlp_logger,
extract_info_sync,
)
class TestProcessPoolConfiguration:
@ -76,3 +83,90 @@ class TestExtractInfo:
(result, logs) = extract_info_sync(config, url, debug=True)
assert isinstance(result, dict), "Result should be a dictionary"
assert isinstance(logs, list), "Logs should be a list"
@patch("app.features.ytdlp.extractor.YTDLP")
def test_extract_info_mirrors_debug_to_console(self, mock_ytdlp_class):
seen: list[tuple[int, str]] = []
def fake_extract_info(url, download=False): # noqa: ARG001
logger = mock_ytdlp_class.call_args.kwargs["params"]["logger"]
logger.debug("[generic_browser] Using remote browser for https://example.com/video")
logger.debug("[debug] [generic_browser] Loading page https://example.com/video")
logger.warning("[generic_browser] Browser fallback warning")
return {"title": "Test Video", "id": "test123"}
mock_ytdlp = MagicMock()
mock_ytdlp.extract_info.side_effect = fake_extract_info
mock_ytdlp_class.return_value = mock_ytdlp
logger = logging.getLogger("yt-dlp.extract_info")
with patch.object(
logger, "info", side_effect=lambda msg, *a, **k: seen.append((logging.INFO, msg % a if a else msg))
):
with patch.object(
logger,
"debug",
side_effect=lambda msg, *a, **k: seen.append((logging.DEBUG, msg % a if a else msg)),
):
with patch.object(
logger,
"log",
side_effect=lambda level, msg, *a, **k: seen.append((level, msg % a if a else msg)),
):
(result, logs) = extract_info_sync(
{}, "https://example.com/video", debug=True, capture_logs=logging.WARNING
)
assert result["id"] == "test123"
assert logs == ["[generic_browser] Browser fallback warning"]
assert (logging.INFO, "[generic_browser] Using remote browser for https://example.com/video") in seen
assert (logging.DEBUG, "[generic_browser] Loading page https://example.com/video") in seen
assert (logging.WARNING, "[generic_browser] Browser fallback warning") in seen
@patch("app.features.ytdlp.extractor.YTDLP")
def test_extract_info_mirrors_screen_logs_without_debug(self, mock_ytdlp_class):
seen: list[tuple[int, str]] = []
def fake_extract_info(url, download=False): # noqa: ARG001
logger = mock_ytdlp_class.call_args.kwargs["params"]["logger"]
logger.debug("[generic_browser] Using remote browser for https://example.com/video")
logger.warning("[generic_browser] Browser fallback warning")
return {"title": "Test Video", "id": "test123"}
mock_ytdlp = MagicMock()
mock_ytdlp.extract_info.side_effect = fake_extract_info
mock_ytdlp_class.return_value = mock_ytdlp
logger = logging.getLogger("yt-dlp.extract_info")
with patch.object(
logger, "info", side_effect=lambda msg, *a, **k: seen.append((logging.INFO, msg % a if a else msg))
):
with patch.object(
logger,
"log",
side_effect=lambda level, msg, *a, **k: seen.append((level, msg % a if a else msg)),
):
(result, logs) = extract_info_sync(
{}, "https://example.com/video", debug=False, capture_logs=logging.WARNING
)
assert result["id"] == "test123"
assert logs == ["[generic_browser] Browser fallback warning"]
assert (logging.INFO, "[generic_browser] Using remote browser for https://example.com/video") in seen
assert (logging.WARNING, "[generic_browser] Browser fallback warning") in seen
class TestYtdlpLogger:
def test_debug_prefix_uses_debug(self) -> None:
logger = MagicMock()
_ytdlp_logger(logger)(logging.DEBUG, "[debug] hello")
logger.debug.assert_called_once_with("hello")
def test_screen_style_debug_uses_info(self) -> None:
logger = MagicMock()
_ytdlp_logger(logger)(logging.DEBUG, "screen line")
logger.info.assert_called_once_with("screen line")

2
app/features/ytdlp/tests/test_ytdlp_module.py
View file

@ -1,6 +1,8 @@
import importlib
from unittest.mock import MagicMock, Mock, patch
import pytest
from yt_dlp.globals import extractors as ytdlp_extractors
from app.features.ytdlp.outtmpl import rewrite_outtmpl
from app.features.ytdlp.patches import patch_windows_popen_wait

22
app/features/ytdlp/ytdlp.py
View file

@ -1,8 +1,10 @@
# flake8: noqa: F401, RUF100, W291, I001
import logging
import sys
from typing import Any
import yt_dlp
from yt_dlp.globals import extractors as ytdlp_extractors
from yt_dlp.utils import make_archive_id
from app.features.ytdlp.outtmpl import rewrite_outtmpl
@ -56,6 +58,17 @@ class YTDLP(yt_dlp.YoutubeDL):
def __init__(self, params=None, auto_init=True):
apply_ytdlp_patches()
if not YTDLP._registered:
try:
from app.yt_dlp_plugins.extractor import generic_browser
from app.yt_dlp_plugins.postprocessor.nfo_maker import NFOMakerPP
from yt_dlp.postprocessor import postprocessors
postprocessors.value.update({"NFOMakerPP": NFOMakerPP})
YTDLP._registered = True
except Exception:
logging.getLogger("ytdlp.wrapper").exception("Failed to register yt-dlp plugins")
# Avoid yt-dlp preloading the archive file by stripping the param first
orig_file = None
patched_params: dict[str, Any] | None = None
@ -80,15 +93,6 @@ class YTDLP(yt_dlp.YoutubeDL):
self._ytptube_outtmpl_info: dict[str, Any] | None = None
self._ytptube_outtmpl_cache: dict[str, Any] = {}
self.archive = _ArchiveProxy(orig_file)
if not YTDLP._registered:
try:
from app.yt_dlp_plugins.postprocessor.nfo_maker import NFOMakerPP
from yt_dlp.postprocessor import postprocessors
postprocessors.value.update({"NFOMakerPP": NFOMakerPP})
YTDLP._registered = True
except Exception:
pass
def _delete_downloaded_files(self, *args, **kwargs) -> None:
if self._interrupted:

59
app/tests/test_condition_ignore.py
View file

@ -165,3 +165,62 @@ class TestConditionIgnorePropagation:
assert captured["extras"]["playlist"] == "Playlist"
assert "source_name" not in captured["extras"]
queue.add.assert_awaited_once()
@pytest.mark.asyncio
async def test_playlist_keeps_child_urls(self) -> None:
class FakeItem:
def __init__(self) -> None:
self.extras = {}
def get_ytdlp_opts(self):
return Mock(get_all=Mock(return_value={}))
def new_with(self, **kwargs):
return SimpleNamespace(extras=kwargs["extras"], url=kwargs["url"])
queue = Mock(add=AsyncMock(return_value={"status": "ok"}))
entry = {
"_type": "playlist",
"id": "playlist-1",
"title": "Playlist",
"webpage_url": "https://example.com/page",
"original_url": "https://example.com/page",
"entries": [
{
"_type": "video",
"id": "video-1",
"title": "Video 1",
"url": "https://cdn.example/1.mp3",
"webpage_url": "https://cdn.example/1.mp3",
"original_url": "https://cdn.example/1.mp3",
"formats": [{"url": "https://cdn.example/1.mp3", "ext": "mp3"}],
},
{
"_type": "video",
"id": "video-2",
"title": "Video 2",
"url": "https://cdn.example/2.mp3",
"webpage_url": "https://cdn.example/2.mp3",
"original_url": "https://cdn.example/2.mp3",
"formats": [{"url": "https://cdn.example/2.mp3", "ext": "mp3"}],
},
],
}
with patch("app.library.downloads.playlist_processor.ytdlp_reject", return_value=(True, "")):
result = await process_playlist(queue=queue, entry=entry, item=FakeItem())
assert result == {"status": "ok"}
assert queue.add.await_count == 2
assert [call.kwargs["item"].url for call in queue.add.await_args_list] == [
"https://cdn.example/1.mp3",
"https://cdn.example/2.mp3",
]
assert [call.kwargs["entry"]["webpage_url"] for call in queue.add.await_args_list] == [
"https://cdn.example/1.mp3",
"https://cdn.example/2.mp3",
]
assert [call.kwargs["entry"]["original_url"] for call in queue.add.await_args_list] == [
"https://cdn.example/1.mp3",
"https://cdn.example/2.mp3",
]

88
app/tests/test_download.py
View file

@ -327,94 +327,6 @@ class TestDownloadFlow:
assert queue.items[0]["download_skipped"] is True
assert queue.items[1]["download_skipped"] is True
def test_download_resets_sigint(self, monkeypatch: pytest.MonkeyPatch) -> None:
class Cfg:
debug = False
ytdlp_debug = False
max_workers = 1
temp_keep = False
temp_disabled = True
download_info_expires = 3600
@staticmethod
def get_instance():
return Cfg
monkeypatch.setattr("app.library.downloads.core.Config", Cfg)
item = make_item(id="live-id")
item.is_live = True
download = Download(
info=item,
info_dict={"id": "live-id", "url": "http://u", "formats": [{"format_id": "18"}]},
)
download._process_manager.cancel_event = Mock(wait=Mock())
download.status_queue = cast(Any, DummyQueue())
download._hook_handlers = Mock(
progress_hook=Mock(),
postprocessor_hook=Mock(),
post_hook=Mock(),
)
download.info.get_ytdlp_opts = Mock(
return_value=Mock(add=Mock(return_value=Mock(get_all=Mock(return_value={}))))
)
created_ydl: list[Any] = []
class FakeYTDLP:
def __init__(self, params):
self.params = params
self._download_retcode = 0
self._interrupted = False
self.to_screen = Mock()
created_ydl.append(self)
def process_ie_result(self, ie_result, download):
return ie_result, download
def download(self, url_list):
return 0
signal_mock = Mock()
thread_instances: list[Mock] = []
def build_thread(*_args, **_kwargs):
thread = Mock(start=Mock())
thread_instances.append(thread)
return thread
thread_mock = Mock(side_effect=build_thread)
monkeypatch.setattr("app.library.downloads.core.YTDLP", FakeYTDLP)
monkeypatch.setattr("app.library.downloads.core.signal.signal", signal_mock)
monkeypatch.setattr("app.library.downloads.core.threading.Thread", thread_mock)
download._download()
signal_mock.assert_any_call(signal.SIGINT, signal.default_int_handler)
live_cancel_thread = next(
call for call in thread_mock.call_args_list if call.kwargs.get("name", "").startswith("cancel-watch-")
)
live_cancel_thread_index = thread_mock.call_args_list.index(live_cancel_thread)
thread_instances[live_cancel_thread_index].start.assert_called_once()
target = live_cancel_thread.kwargs["target"]
ydl = created_ydl[0]
if "posix" == os.name:
with (
patch("app.library.downloads.core.os.getpid", return_value=12345),
patch("app.library.downloads.core.os.kill") as mock_kill,
):
target()
mock_kill.assert_called_once_with(12345, signal.SIGINT)
else:
with patch("app.library.downloads.core._thread.interrupt_main") as mock_interrupt_main:
target()
mock_interrupt_main.assert_called_once_with()
assert ydl._interrupted is True
ydl.to_screen.assert_called_once_with("[info] Interrupt received, exiting cleanly...")
def test_playlist_extras(self, monkeypatch: pytest.MonkeyPatch) -> None:
class Cfg:
debug = False

213
app/tests/test_generic_browser.py Normal file
View file

@ -0,0 +1,213 @@
from __future__ import annotations
from unittest.mock import Mock
import pytest
from app.yt_dlp_plugins.extractor import generic_browser
def _make_ie(config: dict[str, str | None] | None = None) -> generic_browser.GenericBrowserIE:
ie = object.__new__(generic_browser.GenericBrowserIE)
values = config or {}
ie._configuration_arg = lambda name, default: [values.get(name)]
ie._generic_id = lambda url: "vid"
ie._generic_title = lambda url, webpage=None: "title"
ie._get_timeout_ms = lambda: None
ie.report_warning = Mock()
ie.report_extraction = Mock()
ie.playlist_result = Mock(side_effect=lambda entries, **kwargs: {"_type": "playlist", "entries": entries, **kwargs})
ie._looks_like_html = lambda webpage: False
ie._merge_requests = lambda network, media: list(network) + list(media)
ie._extract_network_formats = Mock(
return_value={"formats": [{"url": "https://cdn.example/video.mp4", "ext": "mp4"}]}
)
ie.to_screen = Mock()
ie.write_debug = Mock()
ie._downloader = Mock()
ie._downloader.params = {}
ie._failed = False
return ie
def test_cfg_env(monkeypatch: pytest.MonkeyPatch) -> None:
ie = _make_ie()
monkeypatch.setenv("YTP_BROWSER_URL", " selenium+http://browser:4444/wd/hub ")
assert ie._get_config("url", "YTP_BROWSER_URL") == "selenium+http://browser:4444/wd/hub"
def test_cfg_arg_wins(monkeypatch: pytest.MonkeyPatch) -> None:
ie = _make_ie({"url": "selenium+http://arg:4444/wd/hub"})
monkeypatch.setenv("YTP_BROWSER_URL", "selenium+http://env:4444/wd/hub")
assert ie._get_config("url", "YTP_BROWSER_URL") == "selenium+http://arg:4444/wd/hub"
def test_real_extract_env(monkeypatch: pytest.MonkeyPatch) -> None:
ie = _make_ie()
monkeypatch.setenv("YTP_BROWSER_URL", "selenium+http://browser:4444/wd/hub")
monkeypatch.setattr(generic_browser.SeleniumDriver, "is_available", staticmethod(lambda: False))
with pytest.raises(generic_browser.ExtractorError, match="No matching browser driver available"):
ie._real_extract("https://example.com/watch")
def test_real_extract_invalid_url(monkeypatch: pytest.MonkeyPatch) -> None:
ie = _make_ie()
monkeypatch.setenv("YTP_BROWSER_URL", "http://browser:4444/wd/hub")
with pytest.raises(generic_browser.ExtractorError, match="Invalid browser URL"):
ie._real_extract("https://example.com/watch")
def test_real_extract_logs_connect_failure(monkeypatch: pytest.MonkeyPatch) -> None:
ie = _make_ie()
ie.__wrapped__ = Mock()
ie.__wrapped__._real_extract = Mock(return_value={"id": "fallback"})
monkeypatch.setenv("YTP_BROWSER_URL", "selenium+http://browser:4444/wd/hub")
monkeypatch.setattr(generic_browser.SeleniumDriver, "is_available", staticmethod(lambda: True))
class BrokenDriver:
__name__ = "BrokenDriver"
@staticmethod
def connect(ws_url: str, timeout: int | None = None):
raise RuntimeError("remote browser down")
monkeypatch.setattr(generic_browser.GenericBrowserIE, "_select_driver", lambda self, ws_url: BrokenDriver)
result = ie._real_extract("https://example.com/watch")
assert result == {"id": "fallback"}
ie.report_warning.assert_called_once_with(
"Remote browser unavailable: remote browser down, marking as failed.", "vid"
)
ie.to_screen.assert_called_once_with("Using remote browser for https://example.com/watch")
def test_real_extract_logs_session_failure(monkeypatch: pytest.MonkeyPatch) -> None:
ie = _make_ie()
monkeypatch.setenv("YTP_BROWSER_URL", "playwright+ws://browser")
session = Mock()
session.goto.side_effect = RuntimeError("page crashed")
class FakeDriver:
__name__ = "FakeDriver"
@staticmethod
def connect(ws_url: str, timeout: int | None = None):
return session
monkeypatch.setattr(generic_browser.GenericBrowserIE, "_select_driver", lambda self, ws_url: FakeDriver)
with pytest.raises(RuntimeError, match="page crashed"):
ie._real_extract("https://example.com/watch")
assert session.close.call_count == 1
ie.report_warning.assert_called_once_with(
"Browser extractor session failed for url=%r browser_url=%r driver=%s error=%s",
"https://example.com/watch",
"playwright+ws://browser",
"FakeDriver",
session.goto.side_effect,
)
ie.write_debug.assert_any_call("Selected driver FakeDriver for playwright+ws://browser")
ie.write_debug.assert_any_call("Loading page https://example.com/watch")
def test_real_extract_logs_non_html_page(monkeypatch: pytest.MonkeyPatch) -> None:
ie = _make_ie()
monkeypatch.setenv("YTP_BROWSER_URL", "playwright+ws://browser")
session = Mock()
session.content.return_value = "plain text body"
session.get_requests.return_value = [
{"url": "https://cdn.example/video.mp4", "method": "GET", "resourceType": "video"}
]
session.get_media_requests.return_value = []
class FakeDriver:
__name__ = "FakeDriver"
@staticmethod
def connect(ws_url: str, timeout: int | None = None):
return session
monkeypatch.setattr(generic_browser.GenericBrowserIE, "_select_driver", lambda self, ws_url: FakeDriver)
result = ie._real_extract("https://example.com/watch")
assert result["formats"][0]["url"] == "https://cdn.example/video.mp4"
ie.write_debug.assert_any_call("Page content did not look like HTML for https://example.com/watch")
ie.write_debug.assert_any_call("plain text body")
def test_extract_network_formats_logs_no_media() -> None:
ie = _make_ie()
ie.__wrapped__ = Mock()
ie.__wrapped__._real_extract = Mock(return_value={"id": "fallback"})
ie._url = "https://example.com/watch"
ie._extract_network_formats = generic_browser.GenericBrowserIE._extract_network_formats.__get__(
ie, generic_browser.GenericBrowserIE
)
result = ie._extract_network_formats([], "vid", {"title": "title"})
assert result == {"id": "fallback"}
ie.write_debug.assert_called_with("No media formats found in 0 browser request(s)")
ie.report_warning.assert_called_once_with(
"Generic browser extractor found no media formats. falling back to generic extractor.",
"vid",
)
def test_extract_network_formats_playlist_entries_keep_own_urls() -> None:
ie = _make_ie()
ie._extract_network_formats = generic_browser.GenericBrowserIE._extract_network_formats.__get__(
ie, generic_browser.GenericBrowserIE
)
result = ie._extract_network_formats(
[
{"url": "https://cdn.example/1.mp3", "method": "GET", "resourceType": "audio"},
{"url": "https://cdn.example/2.mp3", "method": "GET", "resourceType": "audio"},
],
"vid",
{"title": "Title", "webpage_url": "https://example.com/page", "original_url": "https://example.com/page"},
)
assert result["_type"] == "playlist"
assert result["entries"][0]["url"] == "https://cdn.example/1.mp3"
assert result["entries"][0]["webpage_url"] == "https://cdn.example/1.mp3"
assert result["entries"][0]["original_url"] == "https://cdn.example/1.mp3"
assert result["entries"][0]["_old_archive_ids"] == [generic_browser.make_archive_id("generic", "vid-1")]
assert result["entries"][0]["direct"] is True
assert result["entries"][1]["url"] == "https://cdn.example/2.mp3"
assert result["entries"][1]["webpage_url"] == "https://cdn.example/2.mp3"
assert result["entries"][1]["original_url"] == "https://cdn.example/2.mp3"
assert result["entries"][1]["_old_archive_ids"] == [generic_browser.make_archive_id("generic", "vid-2")]
assert result["entries"][1]["direct"] is True
def test_select_driver_selenium(monkeypatch: pytest.MonkeyPatch) -> None:
ie = _make_ie()
monkeypatch.setattr(generic_browser.SeleniumDriver, "is_available", staticmethod(lambda: True))
assert ie._select_driver("selenium+http://browser:4444/wd/hub") is generic_browser.SeleniumDriver
def test_select_driver_playwright_ws(monkeypatch: pytest.MonkeyPatch) -> None:
ie = _make_ie()
monkeypatch.setattr(generic_browser.PlaywrightDriver, "is_available", staticmethod(lambda: True))
assert ie._select_driver("playwright+ws://playwright:3000/") is generic_browser.PlaywrightDriver
def test_select_driver_playwright_cdp(monkeypatch: pytest.MonkeyPatch) -> None:
ie = _make_ie()
monkeypatch.setattr(generic_browser.PlaywrightDriver, "is_available", staticmethod(lambda: True))
assert ie._select_driver("playwright+cdp://chrome:9222/") is generic_browser.PlaywrightDriver

124
app/yt_dlp_plugins/extractor/generic_browser.py
View file

@ -1,11 +1,12 @@
import base64
import os
import re
import time
import urllib.parse
from typing import Any
from yt_dlp.extractor.generic import GenericIE
from yt_dlp.utils import ExtractorError, determine_ext, mimetype2ext, traverse_obj
from yt_dlp.utils import ExtractorError, determine_ext, make_archive_id, mimetype2ext, traverse_obj
from yt_dlp.utils._utils import _request_dump_filename
MEDIA_EXTENSIONS: set[str] = {
@ -97,6 +98,16 @@ _MEDIA_ELEMENT_JS = """() => {
return mediaUrls;
}"""
_PLAYWRIGHT_PREFIXES: tuple[str, ...] = (
"playwright+ws://",
"playwright+wss://",
"playwright+cdp://",
"playwright+cdp+http://",
"playwright+cdp+https://",
"playwright+cdp+ws://",
"playwright+cdp+wss://",
)
def _has_possible_media(requests_list: list[dict]) -> bool:
for req in requests_list:
@ -186,10 +197,24 @@ class PlaywrightDriver:
p = sync_playwright().start()
if ws_url.startswith("pw+"):
browser = p.chromium.connect(ws_url[3:], timeout=timeout or 30000)
if ws_url.startswith("playwright+ws://"):
browser_url = f"ws://{ws_url.removeprefix('playwright+ws://')}"
browser = p.chromium.connect(browser_url, timeout=timeout or 30000)
elif ws_url.startswith("playwright+wss://"):
browser_url = f"wss://{ws_url.removeprefix('playwright+wss://')}"
browser = p.chromium.connect(browser_url, timeout=timeout or 30000)
elif ws_url.startswith("playwright+cdp://"):
browser_url = f"http://{ws_url.removeprefix('playwright+cdp://')}"
browser = p.chromium.connect_over_cdp(browser_url, timeout=timeout or 30000)
elif ws_url.startswith("playwright+cdp+"):
browser_url = ws_url.removeprefix("playwright+cdp+")
browser = p.chromium.connect_over_cdp(browser_url, timeout=timeout or 30000)
else:
browser = p.chromium.connect_over_cdp(ws_url, timeout=timeout or 30000)
msg = (
"Invalid Playwright browser URL. Use playwright+ws://, playwright+wss://, "
"playwright+cdp://, or playwright+cdp+http(s|ws|wss)://"
)
raise ValueError(msg)
context = browser.new_context()
page = context.new_page()
@ -401,35 +426,43 @@ class SeleniumDriver:
return Session()
DRIVERS: list[type] = [PlaywrightDriver, SeleniumDriver]
class GenericBrowserIE(GenericIE, plugin_name="browser"):
_WORKING = True
_failed: bool = False
_remote_browser_failures: dict[str, str] = {}
_url: str = ""
def _get_config(self, name: str, env_name: str) -> str | None:
value = self._configuration_arg(name, [None])[0]
if value is None:
value = os.environ.get(env_name)
if isinstance(value, str):
value = value.strip() or None
return value
def _real_extract(self, url: str) -> dict[str, Any]:
self._url = url
preferred_driver: str | None = self._configuration_arg("driver", [None])[0]
if not (remote_browser := self._configuration_arg("remote_browser", [None])[0]) or self._failed:
if not (browser_url := self._get_config("url", "YTP_BROWSER_URL")) or self._failed:
return self.__wrapped__._real_extract(self, url)
video_id: str = self._generic_id(url)
timeout: int | None = self._get_timeout_ms()
self.to_screen(f"Using remote browser for {url}")
if not (driver := self._select_driver(preferred_driver, remote_browser)):
available: list[str] = [d.__name__ for d in DRIVERS if d.isAvailable()]
if not (driver := self._select_driver(browser_url)):
msg: str = (
f"No browser driver available. Install playwright or selenium. "
f"Available drivers: {available if available else 'none'}"
"No matching browser driver available for the configured browser URL. "
"Install playwright or selenium as needed."
)
raise ExtractorError(msg)
self.write_debug(f"Selected driver {driver.__name__} for {browser_url}")
try:
session = driver.connect(remote_browser, timeout)
session = driver.connect(browser_url, timeout)
except Exception as e:
message = str(e)
self._failed = True
@ -438,6 +471,7 @@ class GenericBrowserIE(GenericIE, plugin_name="browser"):
try:
self.report_extraction(url)
self.write_debug(f"Loading page {url}")
session.goto(url)
session.wait_for_network_idle(
@ -448,6 +482,7 @@ class GenericBrowserIE(GenericIE, plugin_name="browser"):
webpage = session.content()
requests = self._merge_requests(session.get_requests(), session.get_media_requests())
self.write_debug(f"Captured {len(requests)} network requests")
if self._downloader.params.get("dump_intermediate_pages"):
self.to_screen(f"Browser content dump for: {url}")
@ -483,35 +518,51 @@ class GenericBrowserIE(GenericIE, plugin_name="browser"):
thumbnail = self._og_search_thumbnail(webpage, default=None)
if thumbnail:
info_dict["thumbnail"] = thumbnail
elif webpage:
self.write_debug(f"Page content did not look like HTML for {url}")
self.write_debug(webpage)
network_info = self._extract_network_formats(requests, video_id, info_dict)
if network_info:
self.write_debug(f"Resolved media from browser requests for {url}")
if network_info.get("_type") == "playlist" and network_info.get("entries"):
return self.playlist_result(network_info["entries"], **info_dict)
info_dict.update(network_info)
return info_dict
except Exception as e:
self.report_warning(
"Browser extractor session failed for url=%r browser_url=%r driver=%s error=%s",
url,
browser_url,
driver.__name__,
e,
)
raise
finally:
session.close()
try:
session.close()
except Exception as e:
self.report_warning(
"Browser session close failed for url=%r browser_url=%r driver=%s error=%s",
url,
browser_url,
driver.__name__,
e,
)
def _select_driver(self, preferred: str | None, ws_url: str):
if preferred:
for driver in DRIVERS:
if driver.__name__.lower().startswith(preferred.lower()):
if driver.isAvailable():
return driver
def _select_driver(self, ws_url: str):
if ws_url.startswith(("selenium+http://", "selenium+https://")):
return SeleniumDriver if SeleniumDriver.is_available() else None
self.report_warning(f"Preferred driver {driver.__name__} not available")
return None
if ws_url.startswith(_PLAYWRIGHT_PREFIXES):
return PlaywrightDriver if PlaywrightDriver.is_available() else None
if ws_url.startswith("pw+") and PlaywrightDriver.is_available():
return PlaywrightDriver
for driver in DRIVERS:
if driver.isAvailable():
return driver
return None
msg = (
"Invalid browser URL. Use selenium+http(s)://, playwright+ws(s)://, "
"playwright+cdp://, or playwright+cdp+http(s|ws|wss)://"
)
raise ExtractorError(msg)
def _extract_network_formats(
self, requests: list[dict], video_id: str, base_info: dict[str, Any]
@ -588,6 +639,7 @@ class GenericBrowserIE(GenericIE, plugin_name="browser"):
source_counts["direct"] = source_counts.get("direct", 0) + 1
if not formats:
self.write_debug(f"No media formats found in {len(requests)} browser request(s)")
self.report_warning(
"Generic browser extractor found no media formats. falling back to generic extractor.", video_id
)
@ -597,14 +649,20 @@ class GenericBrowserIE(GenericIE, plugin_name="browser"):
base_title = (base_info.get("title") or "").strip() or video_id
entries = []
for index, fmt in enumerate(direct_formats, start=1):
entry_url = fmt.get("url")
entry_id = f"{video_id}-{index}"
entries.append(
{
"id": f"{video_id}-{index}",
"id": entry_id,
"title": f"{base_title} ({index})",
"_old_archive_ids": [make_archive_id("generic", entry_id)],
"formats": [fmt],
"url": fmt.get("url"),
"url": entry_url,
"webpage_url": entry_url,
"original_url": entry_url,
"ext": fmt.get("ext"),
"protocol": fmt.get("protocol"),
"direct": True,
}
)
return {"_type": "playlist", "entries": entries}