Drop download_archive for generic extractors when using default presets. Fixes #506
This commit is contained in:
parent
a616a77773
commit
c9de6b9f2f
3 changed files with 76 additions and 7 deletions
|
|
@ -20,6 +20,9 @@ from .ItemDTO import ItemDTO
|
||||||
from .Utils import create_cookies_file, delete_dir, extract_info, extract_ytdlp_logs
|
from .Utils import create_cookies_file, delete_dir, extract_info, extract_ytdlp_logs
|
||||||
from .ytdlp import YTDLP
|
from .ytdlp import YTDLP
|
||||||
|
|
||||||
|
GENERIC_EXTRACTORS = ("HTML5MediaEmbed", "generic")
|
||||||
|
"Generic extractors identifiers."
|
||||||
|
|
||||||
|
|
||||||
class Terminator:
|
class Terminator:
|
||||||
pass
|
pass
|
||||||
|
|
@ -246,13 +249,19 @@ class Download:
|
||||||
self.logger.error(err_msg)
|
self.logger.error(err_msg)
|
||||||
raise ValueError(err_msg) from e
|
raise ValueError(err_msg) from e
|
||||||
|
|
||||||
# Safe-guard in-case downloading take too long and the info expires.
|
if self.info_dict and isinstance(self.info_dict, dict):
|
||||||
if self.info_dict and isinstance(self.info_dict, dict) and self.download_info_expires > 0:
|
# If info extractor_key is generic, we need to remove download_archive param from default preset.
|
||||||
_ts: int | None = self.info_dict.get("epoch", self.info_dict.get("timestamp", None))
|
if self.info_dict.get("extractor_key") in GENERIC_EXTRACTORS and self.info.get_preset().default:
|
||||||
_ts = datetime.fromtimestamp(_ts, tz=UTC) if _ts else None
|
self.logger.debug(f"Removing 'download_archive' for generic extractor. {self.info.url=}")
|
||||||
if not _ts or (datetime.now(tz=UTC) - _ts).total_seconds() > self.download_info_expires:
|
params.pop("download_archive", None)
|
||||||
self.info_dict = None
|
|
||||||
self.logger.warning(f"Info for '{self.info.url}' has expired, re-extracting info.")
|
# Safe-guard in-case downloading take too long and the info expires.
|
||||||
|
if self.download_info_expires > 0:
|
||||||
|
_ts: int | None = self.info_dict.get("epoch", self.info_dict.get("timestamp", None))
|
||||||
|
_ts = datetime.fromtimestamp(_ts, tz=UTC) if _ts else None
|
||||||
|
if not _ts or (datetime.now(tz=UTC) - _ts).total_seconds() > self.download_info_expires:
|
||||||
|
self.info_dict = None
|
||||||
|
self.logger.warning(f"Info for '{self.info.url}' has expired, re-extracting info.")
|
||||||
|
|
||||||
if not self.info_dict or not isinstance(self.info_dict, dict):
|
if not self.info_dict or not isinstance(self.info_dict, dict):
|
||||||
self.logger.info(f"Extracting info for '{self.info.url}'.")
|
self.logger.info(f"Extracting info for '{self.info.url}'.")
|
||||||
|
|
|
||||||
|
|
@ -538,6 +538,18 @@ class ItemDTO:
|
||||||
|
|
||||||
return params
|
return params
|
||||||
|
|
||||||
|
def get_preset(self) -> "Preset | None":
|
||||||
|
"""
|
||||||
|
Get the preset for the item.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Preset | None: The preset for the item. If not found, None.
|
||||||
|
|
||||||
|
"""
|
||||||
|
from .Presets import Presets
|
||||||
|
|
||||||
|
return Presets.get_instance().get(self.preset if self.preset else "default")
|
||||||
|
|
||||||
def archive_status(self, force: bool = False) -> None:
|
def archive_status(self, force: bool = False) -> None:
|
||||||
"""
|
"""
|
||||||
Recompute the archive status of the item.
|
Recompute the archive status of the item.
|
||||||
|
|
|
||||||
|
|
@ -279,3 +279,51 @@ class TestItemDTO:
|
||||||
mock_utils_sidecar.assert_not_called()
|
mock_utils_sidecar.assert_not_called()
|
||||||
assert result is existing
|
assert result is existing
|
||||||
assert dto.sidecar is existing
|
assert dto.sidecar is existing
|
||||||
|
|
||||||
|
def test_get_preset_returns_preset_instance(self):
|
||||||
|
"""Test ItemDTO.get_preset returns the Preset instance."""
|
||||||
|
from app.library.Presets import Preset
|
||||||
|
|
||||||
|
mock_preset = Preset(id="test-id", name="test-preset", cli="--test")
|
||||||
|
|
||||||
|
with patch.object(ItemDTO, "__post_init__", lambda _: None):
|
||||||
|
dto = ItemDTO(id="vid", title="t", url="u", folder="f", preset="test-preset")
|
||||||
|
|
||||||
|
with patch("app.library.Presets.Presets.get_instance") as mock_presets:
|
||||||
|
mock_presets.return_value.get.return_value = mock_preset
|
||||||
|
|
||||||
|
result = dto.get_preset()
|
||||||
|
|
||||||
|
mock_presets.return_value.get.assert_called_once_with("test-preset")
|
||||||
|
assert result is mock_preset
|
||||||
|
assert result.name == "test-preset"
|
||||||
|
|
||||||
|
def test_get_preset_uses_default_when_no_preset_set(self):
|
||||||
|
"""Test ItemDTO.get_preset uses 'default' when preset is empty."""
|
||||||
|
from app.library.Presets import Preset
|
||||||
|
|
||||||
|
mock_preset = Preset(id="default-id", name="default", cli="--default")
|
||||||
|
|
||||||
|
with patch.object(ItemDTO, "__post_init__", lambda _: None):
|
||||||
|
dto = ItemDTO(id="vid", title="t", url="u", folder="f", preset="")
|
||||||
|
|
||||||
|
with patch("app.library.Presets.Presets.get_instance") as mock_presets:
|
||||||
|
mock_presets.return_value.get.return_value = mock_preset
|
||||||
|
|
||||||
|
result = dto.get_preset()
|
||||||
|
|
||||||
|
mock_presets.return_value.get.assert_called_once_with("default")
|
||||||
|
assert result is mock_preset
|
||||||
|
|
||||||
|
def test_get_preset_returns_none_when_not_found(self):
|
||||||
|
"""Test ItemDTO.get_preset returns None when preset not found."""
|
||||||
|
with patch.object(ItemDTO, "__post_init__", lambda _: None):
|
||||||
|
dto = ItemDTO(id="vid", title="t", url="u", folder="f", preset="nonexistent")
|
||||||
|
|
||||||
|
with patch("app.library.Presets.Presets.get_instance") as mock_presets:
|
||||||
|
mock_presets.return_value.get.return_value = None
|
||||||
|
|
||||||
|
result = dto.get_preset()
|
||||||
|
|
||||||
|
mock_presets.return_value.get.assert_called_once_with("nonexistent")
|
||||||
|
assert result is None
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue