From ad54d80077e538a9e4c2729711cf76179866edb1 Mon Sep 17 00:00:00 2001 From: arabcoders Date: Wed, 14 Jan 2026 19:50:35 +0300 Subject: [PATCH] Refactor: better final filename status tracking --- app/library/ItemDTO.py | 1 + app/library/downloads/core.py | 3 +- app/library/downloads/hooks.py | 28 +++----- app/library/downloads/status_tracker.py | 59 +++++++++-------- app/library/downloads/utils.py | 1 + app/tests/test_download.py | 24 +------ .../test_item_adder_extract_concurrency.py | 66 ------------------- ui/app/components/Queue.vue | 19 +++++- ui/app/types/store.d.ts | 4 +- 9 files changed, 63 insertions(+), 142 deletions(-) delete mode 100644 app/tests/test_item_adder_extract_concurrency.py diff --git a/app/library/ItemDTO.py b/app/library/ItemDTO.py index 05b5000c..d3faaf2f 100644 --- a/app/library/ItemDTO.py +++ b/app/library/ItemDTO.py @@ -399,6 +399,7 @@ class ItemDTO: percent: int | None = None speed: str | None = None eta: str | None = None + postprocessor: str | None = None _recomputed: bool = False _archive_file: str | None = None diff --git a/app/library/downloads/core.py b/app/library/downloads/core.py index 242b9bcd..416e80fd 100644 --- a/app/library/downloads/core.py +++ b/app/library/downloads/core.py @@ -271,8 +271,7 @@ class Download: """ self.status_queue = Config.get_manager().Queue() - temp_path = self._temp_manager.create_temp_path() - if temp_path: + if temp_path := self._temp_manager.create_temp_path(): self.info.temp_path = str(temp_path) self._status_tracker = StatusTracker( diff --git a/app/library/downloads/hooks.py b/app/library/downloads/hooks.py index 44b9dfeb..830056ec 100644 --- a/app/library/downloads/hooks.py +++ b/app/library/downloads/hooks.py @@ -2,7 +2,6 @@ import logging import re -from pathlib import Path from typing import TYPE_CHECKING, Any from .utils import DEBUG_MESSAGE_PREFIXES, YTDLP_PROGRESS_FIELDS, create_debug_safe_dict @@ -55,24 +54,17 @@ class HookHandlers: except Exception as e: self.logger.debug(f"PP Hook: Error creating debug info: {e}") - if "MoveFiles" == data.get("postprocessor") and "finished" == data.get("status"): - if "__finaldir" in data.get("info_dict", {}) and "filepath" in data.get("info_dict", {}): - filename = str(Path(data["info_dict"]["__finaldir"]) / Path(data["info_dict"]["filepath"]).name) - else: - filename: str | None = data.get("info_dict", {}).get("filepath", data.get("filename")) + self.status_queue.put( + { + "id": self.id, + "action": "postprocessing", + **{k: v for k, v in data.items() if k in YTDLP_PROGRESS_FIELDS}, + "status": "postprocessing", + } + ) - self.logger.debug(f"Final filename: '{filename}'.") - self.status_queue.put({"id": self.id, "action": "moved", "status": "finished", "final_name": filename}) - return - - dataDict = {k: v for k, v in data.items() if k in YTDLP_PROGRESS_FIELDS} - self.status_queue.put({"id": self.id, "action": "postprocessing", **dataDict, "status": "postprocessing"}) - - def post_hook(self, filename: str | None = None) -> None: - if not filename: - return - - self.status_queue.put({"id": self.id, "filename": filename}) + def post_hook(self, filename: str) -> None: + self.status_queue.put({"id": self.id, "status": "finished", "final_name": filename}) class NestedLogger: diff --git a/app/library/downloads/status_tracker.py b/app/library/downloads/status_tracker.py index da4e8c47..02671da3 100644 --- a/app/library/downloads/status_tracker.py +++ b/app/library/downloads/status_tracker.py @@ -66,6 +66,9 @@ class StatusTracker: status: Status dictionary from yt-dlp hooks """ + if self.final_update: + return + if status.get("id") != self.id or len(status) < 2: self.logger.warning(f"Received invalid status update. {status}") return @@ -79,22 +82,9 @@ class StatusTracker: self.tmpfilename = status.get("tmpfilename") - fl = None - if "final_name" in status: - fl = Path(status.get("final_name")) - self.info.filename = safe_relative_path(fl, Path(self.download_dir), self.temp_path) - - if fl.is_file() and fl.exists(): - self.final_update = True - self.logger.debug(f"Final file name: '{fl}'.") - - try: - self.info.file_size = fl.stat().st_size - except FileNotFoundError: - self.info.file_size = 0 - self.info.status = status.get("status", self.info.status) self.info.msg = status.get("msg") + self.info.postprocessor = status.get("postprocessor", None) if "error" == self.info.status and "error" in status: self.info.error = status.get("error") @@ -118,24 +108,35 @@ class StatusTracker: self.info.speed = status.get("speed") self.info.eta = status.get("eta") - if "finished" == self.info.status and fl and fl.is_file() and fl.exists(): - self.info.file_size = fl.stat().st_size + if final_name := status.get("final_name", None): + final_name = Path(final_name) self.info.datetime = str(formatdate(time.time())) + self.info.status = "finished" + self.final_update = True + self.info.filename = safe_relative_path(final_name, Path(self.download_dir), self.temp_path) + self.logger.debug(f"Final file name: '{final_name}'.") - try: - ff = await ffprobe(str(fl)) - self.info.extras["is_video"] = ff.has_video() - self.info.extras["is_audio"] = ff.has_audio() - if (ff.has_video() or ff.has_audio()) and not self.info.extras.get("duration"): - self.info.extras["duration"] = int(float(ff.metadata.get("duration", 0.0))) - except Exception as e: - self.info.extras["is_video"] = True - self.info.extras["is_audio"] = True - self.logger.exception(e) - self.logger.error(f"Failed to run ffprobe. {e}") + if final_name.is_file() and final_name.exists(): + try: + self.info.file_size = final_name.stat().st_size + except FileNotFoundError: + self.info.file_size = 0 - if not self.final_update or fl: - self._notify.emit(Events.ITEM_UPDATED, data=self.info) + try: + ff = await ffprobe(final_name) + self.info.extras["is_video"] = ff.has_video() + self.info.extras["is_audio"] = ff.has_audio() + if ff.has_video() or ff.has_audio(): + self.info.extras["duration"] = int( + float(ff.metadata.get("duration", self.info.extras.get("duration", 0.0))) + ) + except Exception as e: + self.info.extras["is_video"] = True + self.info.extras["is_audio"] = True + self.logger.exception(e) + self.logger.error(f"Failed to run ffprobe. {e}") + + self._notify.emit(Events.ITEM_UPDATED, data=self.info) async def progress_update(self) -> None: """ diff --git a/app/library/downloads/utils.py b/app/library/downloads/utils.py index a420335f..898c0946 100644 --- a/app/library/downloads/utils.py +++ b/app/library/downloads/utils.py @@ -23,6 +23,7 @@ YTDLP_PROGRESS_FIELDS = ( "downloaded_bytes", "speed", "eta", + "postprocessor", ) # Live Stream Options (known to cause issues) diff --git a/app/tests/test_download.py b/app/tests/test_download.py index 7fb0e544..e6bd8d21 100644 --- a/app/tests/test_download.py +++ b/app/tests/test_download.py @@ -128,35 +128,13 @@ class TestDownloadHooks: ): assert k in ev, f"Key '{k}' should be in event" - def test_postprocessor_hook_movefiles_sets_final_name(self, tmp_path: Path) -> None: - d = Download(make_item()) - q = DummyQueue() - hooks = HookHandlers(d.id, q, d.logger, d.debug) - - finaldir = tmp_path / "out" - finaldir.mkdir() - path = finaldir / "file.mp4" - payload = { - "postprocessor": "MoveFiles", - "status": "finished", - "info_dict": {"__finaldir": str(finaldir), "filepath": str(path)}, - } - hooks.postprocessor_hook(payload) - assert 1 == len(q.items), "Should have 1 item in queue" - ev = q.items[0] - assert ev["action"] == "moved", "Action should be 'moved'" - assert ev["status"] == "finished", "Status should be 'finished'" - assert ev["final_name"].endswith("file.mp4"), "Final name should end with 'file.mp4'" - def test_post_hooks_pushes_filename(self) -> None: d = Download(make_item()) q = DummyQueue() hooks = HookHandlers(d.id, q, d.logger, d.debug) - hooks.post_hook(None) - assert 0 == len(q.items), "Should have no items when filename is None" hooks.post_hook("name.ext") assert 1 == len(q.items), "Should have 1 item when filename is provided" - assert q.items[0]["filename"] == "name.ext", "Filename should match" + assert q.items[0]["final_name"] == "name.ext", "Filename should match" class TestDownloadStale: diff --git a/app/tests/test_item_adder_extract_concurrency.py b/app/tests/test_item_adder_extract_concurrency.py deleted file mode 100644 index d1211e55..00000000 --- a/app/tests/test_item_adder_extract_concurrency.py +++ /dev/null @@ -1,66 +0,0 @@ -import time -import asyncio -import threading -import pytest - -from app.library.config import Config -from app.library.downloads.queue_manager import DownloadQueue -from app.library.ItemDTO import Item - - -@pytest.mark.asyncio -async def test_extract_concurrency_limited(monkeypatch): - """Ensure that concurrent extract_info calls are limited by the configured semaphore.""" - # Configure a low concurrency to make the timing assertions stable - cfg = Config.get_instance() - cfg.extract_info_concurrency = 2 - - # Reset singleton so new DownloadQueue picks up updated config - DownloadQueue._reset_singleton() - queue = DownloadQueue.get_instance() - - sleep_time = 0.18 - - # Thread-safe counters to observe concurrency usage in the executor threads - lock = threading.Lock() - current = {"val": 0} - max_seen = {"max": 0} - - def fake_extract_info(config, url, **kwargs): - # Track concurrent starts - with lock: - current["val"] += 1 - if current["val"] > max_seen["max"]: - max_seen["max"] = current["val"] - - # Blocking call to simulate expensive IO/work executed in executor - time.sleep(sleep_time) - - with lock: - current["val"] -= 1 - - return {"_type": "video", "id": url.split("/")[-1], "title": "t", "url": url} - - monkeypatch.setattr("app.library.Utils.extract_info", fake_extract_info) - - items = [Item(url=f"http://example.com/{i}") for i in range(6)] - - start = time.perf_counter() - tasks = [asyncio.create_task(queue.add(item=item)) for item in items] - - await asyncio.gather(*tasks) - elapsed = time.perf_counter() - start - - # Assert we never exceeded the configured concurrency - assert max_seen["max"] <= cfg.extract_info_concurrency, ( - f"Max concurrent extractions {max_seen['max']} exceeded limit {cfg.extract_info_concurrency}" - ) - - # Sanity timing check (relaxed to avoid flakiness) - rounds = -(-len(items) // cfg.extract_info_concurrency) # ceil division - assert elapsed >= rounds * sleep_time * 0.8, ( - f"Elapsed {elapsed:.2f}s too low; expected at least {rounds * sleep_time * 0.8:.2f}s" - ) - - # Cleanup - DownloadQueue._reset_singleton() diff --git a/ui/app/components/Queue.vue b/ui/app/components/Queue.vue index c11321f4..d6cbf309 100644 --- a/ui/app/components/Queue.vue +++ b/ui/app/components/Queue.vue @@ -320,8 +320,7 @@
  • source_name:task_name - items added by the specified task.
    • - +

    Download queue is empty.

     @@ -448,6 +447,11 @@ const setStatus = (item: StoreItem): string => { if ('downloading' === item.status && item.is_live) { return 'Streaming' } + + if ('started' === item.status) { + return 'Starting' + } + if ('preparing' === item.status) { return ag(item, 'extras.external_downloader') ? 'External-DL' : 'Preparing..' } @@ -458,7 +462,7 @@ const setStatus = (item: StoreItem): string => { } const setIconColor = (item: StoreItem): string => { - if ('downloading' === item.status) { + if (['downloading', 'started'].includes(item.status as string)) { return 'has-text-success' } if ('postprocessing' === item.status) { @@ -511,9 +515,18 @@ const updateProgress = (item: StoreItem): string => { if (null === item.status && true === config.paused) { return 'Global Pause' } + + if ('started' === item.status) { + return 'Starting' + } + if ('postprocessing' === item.status) { + if (item.postprocessor) { + return `PP Running: ${item.postprocessor}` + } return 'Post-processors are running.' } + if ('preparing' === item.status) { return ag(item, 'extras.external_downloader') ? 'External downloader.' : 'Preparing' } diff --git a/ui/app/types/store.d.ts b/ui/app/types/store.d.ts index 9192b04b..ba71a123 100644 --- a/ui/app/types/store.d.ts +++ b/ui/app/types/store.d.ts @@ -1,4 +1,4 @@ -type ItemStatus = 'finished' | 'preparing' | 'error' | 'cancelled' | 'downloading' | 'postprocessing' | 'not_live' | 'skip' | null; +type ItemStatus = 'started' | 'finished' | 'preparing' | 'error' | 'cancelled' | 'downloading' | 'postprocessing' | 'not_live' | 'skip' | null; type SideCar = { file: string @@ -106,6 +106,8 @@ type StoreItem = { is_archived?: boolean /** Item archive ID */ archive_id?: string | null + /** Postprocessor running for the item if available */ + postprocessor?: string | null } export type { ItemStatus, StoreItem }