From c70d5340f13473f2b5ea986d14b87c764498af96 Mon Sep 17 00:00:00 2001 From: arabcoders Date: Sat, 11 Apr 2026 23:31:48 +0300 Subject: [PATCH] feat: implement patch to make Windows Popen support stopping live stream --- app/features/ytdlp/patches.py | 98 +++++++++++++++++++ app/features/ytdlp/tests/test_ytdlp_module.py | 56 +++++++++++ app/features/ytdlp/utils.py | 52 +--------- app/features/ytdlp/ytdlp.py | 7 +- app/library/downloads/core.py | 3 +- app/tests/test_download.py | 26 +++-- pyproject.toml | 1 + 7 files changed, 183 insertions(+), 60 deletions(-) create mode 100644 app/features/ytdlp/patches.py diff --git a/app/features/ytdlp/patches.py b/app/features/ytdlp/patches.py new file mode 100644 index 00000000..d38b2a64 --- /dev/null +++ b/app/features/ytdlp/patches.py @@ -0,0 +1,98 @@ +import logging +import subprocess +import sys +from typing import Any + +LOG: logging.Logger = logging.getLogger("ytdlp.utils") + + +def patch_metadataparser() -> None: + """ + Patches yt_dlp MetadataParserPP action to handle subprocess pickling issues. + """ + try: + from yt_dlp.postprocessor.metadataparser import MetadataParserPP + from yt_dlp.utils import Namespace + except Exception as exc: + LOG.warning(f"Unable to import yt_dlp metadata parser for patching: {exc!s}") + return + + if getattr(MetadataParserPP.Actions, "_ytptube_patched", False): + return + + class _ActionNS(Namespace): + _ACTIONS_STR: list[str] = [] + + @staticmethod + def _get_name(func) -> str | None: + if not callable(func): + return None + + target = getattr(func, "__func__", func) + module_name = getattr(target, "__module__", None) + qual_name = getattr(target, "__qualname__", getattr(target, "__name__", None)) + + return f"{module_name}.{qual_name}" if module_name and qual_name else None + + def __contains__(self, candidate: object) -> bool: + if candidate in self.__dict__.values(): + return True + + if func_name := _ActionNS._get_name(candidate): + if len(_ActionNS._ACTIONS_STR) < 1: + _ActionNS._ACTIONS_STR.extend( + [value for value in (_ActionNS._get_name(value) for value in self.__dict__.values()) if value] + ) + + return func_name in _ActionNS._ACTIONS_STR + + return False + + actions_dict: dict[str, Any] = dict(MetadataParserPP.Actions.items_) + MetadataParserPP.Actions = _ActionNS(**actions_dict) + MetadataParserPP.Actions._ytptube_patched = True + LOG.debug("MetadataParserPP action namespace patch applied successfully.") + + +def patch_windows_popen_wait() -> None: + if sys.platform != "win32": + return + + try: + from yt_dlp.utils import Popen + except Exception as exc: + LOG.warning(f"Unable to import yt_dlp Popen for patching: {exc!s}") + return + + if getattr(Popen, "_ytptube_wait_patched", False): + return + + original_wait = Popen.wait + + # Windows subprocess waits can swallow the synthetic interrupt we use to + # stop live downloads, especially while yt-dlp is blocked on ffmpeg. + def interruptible_wait(self, timeout=None): + if timeout is not None: + return original_wait(self, timeout=timeout) + + while True: + try: + return original_wait(self, timeout=0.1) + except subprocess.TimeoutExpired: + continue + + Popen.wait = interruptible_wait + Popen._ytptube_wait_patched = True + LOG.debug("yt_dlp Popen.wait Windows patch applied successfully.") + + +def apply_ytdlp_patches() -> None: + try: + patch_metadataparser() + except Exception as exc: + LOG.debug("Metadata parser patch failed to apply: %s", exc) + + try: + patch_windows_popen_wait() + except Exception as exc: + LOG.debug("Windows Popen wait patch failed to apply: %s", exc) diff --git a/app/features/ytdlp/tests/test_ytdlp_module.py b/app/features/ytdlp/tests/test_ytdlp_module.py index 1adf9112..66215795 100644 --- a/app/features/ytdlp/tests/test_ytdlp_module.py +++ b/app/features/ytdlp/tests/test_ytdlp_module.py @@ -1,5 +1,6 @@ from unittest.mock import MagicMock, Mock, patch +from app.features.ytdlp.patches import patch_windows_popen_wait from app.features.ytdlp.utils import _DATA from app.features.ytdlp.ytdlp import YTDLP, _ArchiveProxy, ytdlp_options @@ -145,6 +146,61 @@ class TestYTDLP: assert isinstance(ytdlp.archive, _ArchiveProxy) assert not ytdlp.archive + @patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL.__init__") + def test_init_patches_windows_popen_wait_once(self, mock_super_init) -> None: + mock_super_init.return_value = None + + class FakePopen: + def wait(self, timeout=None): + return timeout + + with patch("app.features.ytdlp.patches.sys.platform", "win32"): + with patch("yt_dlp.utils.Popen", FakePopen): + YTDLP(params={}) + + assert getattr(FakePopen, "_ytptube_wait_patched", False) is True + + def test_windows_wait_patch_uses_polling_for_blocking_wait(self) -> None: + calls: list[float | None] = [] + + class FakePopen: + _ytptube_wait_patched = False + + def wait(self, timeout=None): + calls.append(timeout) + if len(calls) < 3: + raise TimeoutError + return 0 + + with patch("app.features.ytdlp.patches.sys.platform", "win32"): + with ( + patch("yt_dlp.utils.Popen", FakePopen), + patch("app.features.ytdlp.patches.subprocess.TimeoutExpired", TimeoutError), + ): + patch_windows_popen_wait() + result = FakePopen().wait() + + assert result == 0 + assert calls == [0.1, 0.1, 0.1] + + def test_windows_wait_patch_preserves_explicit_timeout(self) -> None: + calls: list[float | None] = [] + + class FakePopen: + _ytptube_wait_patched = False + + def wait(self, timeout=None): + calls.append(timeout) + return 0 + + with patch("app.features.ytdlp.patches.sys.platform", "win32"): + with patch("yt_dlp.utils.Popen", FakePopen): + patch_windows_popen_wait() + result = FakePopen().wait(timeout=5) + + assert result == 0 + assert calls == [5] + @patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL._delete_downloaded_files") def test_delete_downloaded_files_skips_when_interrupted(self, mock_super_delete) -> None: """Test _delete_downloaded_files skips cleanup when _interrupted is True.""" diff --git a/app/features/ytdlp/utils.py b/app/features/ytdlp/utils.py index 50e1ab2f..93034d89 100644 --- a/app/features/ytdlp/utils.py +++ b/app/features/ytdlp/utils.py @@ -9,6 +9,7 @@ from email.utils import formatdate from pathlib import Path from typing import Any +from app.features.ytdlp.patches import apply_ytdlp_patches from app.features.ytdlp.ytdlp import YTDLP from app.library.Utils import merge_dict, timed_lru_cache @@ -143,52 +144,6 @@ class LogWrapper: self._log(logging.CRITICAL, msg, *args, **kwargs) -def patch_metadataparser() -> None: - """ - Patches yt_dlp MetadataParserPP action to handle subprocess pickling issues. - """ - try: - from yt_dlp.postprocessor.metadataparser import MetadataParserPP - from yt_dlp.utils import Namespace - except Exception as exc: - LOG.warning(f"Unable to import yt_dlp metadata parser for patching: {exc!s}") - return - - if getattr(MetadataParserPP.Actions, "_ytptube_patched", False): - return - - class _ActionNS(Namespace): - _ACTIONS_STR: list[str] = [] - - @staticmethod - def _get_name(func) -> str | None: - if not callable(func): - return None - - target = getattr(func, "__func__", func) - module_name = getattr(target, "__module__", None) - qual_name = getattr(target, "__qualname__", getattr(target, "__name__", None)) - - return f"{module_name}.{qual_name}" if module_name and qual_name else None - - def __contains__(self, candidate: object) -> bool: - if candidate in self.__dict__.values(): - return True - - if func_name := _ActionNS._get_name(candidate): - if len(_ActionNS._ACTIONS_STR) < 1: - _ActionNS._ACTIONS_STR.extend([_ActionNS._get_name(value) for value in self.__dict__.values()]) - - return func_name in _ActionNS._ACTIONS_STR - - return False - - actions_dict: dict[str, Any] = dict(MetadataParserPP.Actions.items_) - MetadataParserPP.Actions = _ActionNS(**actions_dict) - MetadataParserPP.Actions._ytptube_patched = True - LOG.debug("MetadataParserPP action namespace patch applied successfully.") - - def arg_converter( args: str, level: int | bool | None = None, @@ -222,10 +177,7 @@ def arg_converter( finally: yt_dlp.options.create_parser = create_parser - try: - patch_metadataparser() - except Exception as exc: - LOG.debug("Metadata parser patch failed to apply: %s", exc) + apply_ytdlp_patches() default_opts = _default_opts([]).ydl_opts diff --git a/app/features/ytdlp/ytdlp.py b/app/features/ytdlp/ytdlp.py index 1a1f0a84..cccba7e5 100644 --- a/app/features/ytdlp/ytdlp.py +++ b/app/features/ytdlp/ytdlp.py @@ -5,6 +5,7 @@ from typing import Any import yt_dlp from yt_dlp.utils import make_archive_id +from app.features.ytdlp.patches import apply_ytdlp_patches from app.library.cf_solver_handler import set_cf_handler @@ -52,13 +53,15 @@ class YTDLP(yt_dlp.YoutubeDL): _registered = False def __init__(self, params=None, auto_init=True): + apply_ytdlp_patches() + # Avoid yt-dlp preloading the archive file by stripping the param first orig_file = None - patched_params = None + patched_params: dict[str, Any] | None = None if params is not None: try: orig_file: str | None = params.get("download_archive") - patched_params: dict = dict(params) + patched_params = dict(params) if "download_archive" in patched_params: patched_params.pop("download_archive", None) except Exception: diff --git a/app/library/downloads/core.py b/app/library/downloads/core.py index 52911f05..778e1906 100644 --- a/app/library/downloads/core.py +++ b/app/library/downloads/core.py @@ -2,6 +2,7 @@ from __future__ import annotations +import _thread import asyncio import logging import os @@ -217,7 +218,7 @@ class Download: if "posix" == os.name: os.kill(os.getpid(), signal.SIGINT) else: - signal.raise_signal(signal.SIGINT) + _thread.interrupt_main() threading.Thread(target=trigger_live_cancel, name=f"cancel-watch-{self.id}", daemon=True).start() diff --git a/app/tests/test_download.py b/app/tests/test_download.py index 96c03f73..c1601a7f 100644 --- a/app/tests/test_download.py +++ b/app/tests/test_download.py @@ -371,20 +371,32 @@ class TestDownloadFlow: def process_ie_result(self, ie_result, download): return ie_result, download + def download(self, url_list): + return 0 + signal_mock = Mock() - thread_instance = Mock(start=Mock()) - thread_mock = Mock(return_value=thread_instance) + 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() - thread_mock.assert_called_once() - thread_instance.start.assert_called_once() signal_mock.assert_any_call(signal.SIGINT, signal.default_int_handler) - target = thread_mock.call_args.kwargs["target"] + 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: @@ -395,9 +407,9 @@ class TestDownloadFlow: target() mock_kill.assert_called_once_with(12345, signal.SIGINT) else: - with patch("app.library.downloads.core.signal.raise_signal") as mock_raise_signal: + with patch("app.library.downloads.core._thread.interrupt_main") as mock_interrupt_main: target() - mock_raise_signal.assert_called_once_with(signal.SIGINT) + mock_interrupt_main.assert_called_once_with() assert ydl._interrupted is True ydl.to_screen.assert_called_once_with("[info] Interrupt received, exiting cleanly...") diff --git a/pyproject.toml b/pyproject.toml index 456d7f3d..2b96d829 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -194,6 +194,7 @@ testpaths = ["app/tests", "app/features"] addopts = "-v --tb=short" filterwarnings = [ "ignore:Parsing dates involving a day of month without a year:DeprecationWarning", + "ignore:This process \\(pid\\=.*\\) is multi-threaded, use of fork\\(\\) may lead to deadlocks in the child\\.:DeprecationWarning", ] [dependency-groups]