Refactor code for improved type hinting; add unarchive functionality to tasks API and UI.

app/library/Download.py

@@ -17,9 +17,8 @@ from .config import Config
 from .Events import EventBus, Events
 from .ffprobe import ffprobe
 from .ItemDTO import ItemDTO
-from .Utils import delete_dir, extract_info, extract_ytdlp_logs, get_archive_id, load_cookies
+from .Utils import delete_dir, extract_info, extract_ytdlp_logs, load_cookies
 from .ytdlp import YTDLP
-from .YTDLPOpts import YTDLPOpts
 
 
 class Terminator:
@@ -27,19 +26,19 @@ class Terminator:
 
 
 class NestedLogger:
-    debug_messages = ["[debug] ", "[download] "]
+    debug_messages: list[str] = ["[debug] ", "[download] "]
 
-    def __init__(self, logger: logging.Logger):
+    def __init__(self, logger: logging.Logger) -> None:
-        self.logger = logger
+        self.logger: logging.Logger = logger
 
-    def debug(self, msg: str):
+    def debug(self, msg: str) -> None:
-        levelno = logging.DEBUG if any(msg.startswith(x) for x in self.debug_messages) else logging.INFO
+        levelno: int = logging.DEBUG if any(msg.startswith(x) for x in self.debug_messages) else logging.INFO
         self.logger.log(level=levelno, msg=re.sub(r"^\[(debug|info)\] ", "", msg, flags=re.IGNORECASE))
 
-    def error(self, msg):
+    def error(self, msg) -> None:
         self.logger.error(msg)
 
-    def warning(self, msg):
+    def warning(self, msg) -> None:
         self.logger.warning(msg)
 
 
@@ -110,7 +109,6 @@ class Download:
         self.template = info.template
         self.template_chapter = info.template_chapter
         self.download_info_expires = int(config.download_info_expires)
-        self.preset = info.preset
         self.info = info
         self.id = info._id
         self.debug = bool(config.debug)
@@ -125,14 +123,14 @@ class Download:
         self.temp_disabled = bool(config.temp_disabled)
         self.is_live = bool(info.is_live) or info.live_in is not None
         self.info_dict = info_dict
-        self.logger = logging.getLogger(f"Download.{info.id if info.id else info._id}")
+        self.logger: logging.Logger = logging.getLogger(f"Download.{info.id if info.id else info._id}")
         self.started_time = 0
-        self.queue_time = datetime.now(tz=UTC)
+        self.queue_time: datetime = datetime.now(tz=UTC)
         self.logs = logs if logs else []
 
     def _progress_hook(self, data: dict):
         if self.debug:
-            d_copy = deepcopy(data)
+            d_copy: dict = deepcopy(data)
             for k in ["formats", "thumbnails", "description", "tags", "_format_sort_fields"]:
                 d_copy["info_dict"].pop(k, None)
 
@@ -148,7 +146,7 @@ class Download:
 
     def _postprocessor_hook(self, data: dict):
         if self.debug:
-            d_copy = deepcopy(data)
+            d_copy: dict = deepcopy(data)
             for k in ["formats", "thumbnails", "description", "tags", "_format_sort_fields"]:
                 d_copy["info_dict"].pop(k, None)
 
@@ -181,10 +179,8 @@ class Download:
 
         try:
             params: dict = (
-                YTDLPOpts.get_instance()
+                self.info.get_ytdlp_opts()
-                .preset(self.preset)
                 .add({"break_on_existing": True})
-                .add_cli(args=self.info.cli, from_user=True)
                 .add(
                     config={
                         "color": "no_color",
@@ -282,10 +278,11 @@ class Download:
                 raise ValueError(msg)  # noqa: TRY301
 
             self.logger.info(
-                f'Task id="{self.info.id}" PID="{os.getpid()}" title="{self.info.title}" preset="{self.preset}" cookies="{bool(params.get("cookiefile"))}" started.'
+                f'Task {self.info.name()}, preset="{self.info.preset}", cookies="{bool(params.get("cookiefile"))}" started.'
             )
 
-            self.logger.debug(f"Params before passing to yt-dlp. {params}")
+            if self.debug:
+                self.logger.debug(f"Params before passing to yt-dlp. {params}")
 
             params["logger"] = NestedLogger(self.logger)
 
@@ -335,7 +332,7 @@ class Download:
                     self.logger.error(f"Failed to delete cookie file: {cookie_file}. {e}")
 
         self.logger.info(
-            f'Task id="{self.info.id}" PID="{os.getpid()}" title="{self.info.title}" preset="{self.preset}" cookies="{bool(params.get("cookiefile"))}" completed.'
+            f'Task {self.info.name()} preset="{self.info.preset}" cookies="{bool(params.get("cookiefile"))}" completed.'
         )
 
     async def start(self):
@@ -373,9 +370,11 @@ class Download:
 
         for i in range(drain_count):
             try:
-                self.logger.debug(f"(50/{i}) Draining the status queue...")
+                if self.debug:
+                    self.logger.debug(f"(50/{i}) Draining the status queue...")
                 if self.final_update:
-                    self.logger.debug("(50/{i}) Draining stopped. Final update received.")
+                    if self.debug:
+                        self.logger.debug("(50/{i}) Draining stopped. Final update received.")
                     break
                 next_status = self.status_queue.get(timeout=0.1)
                 if next_status is None or isinstance(next_status, Terminator):
@@ -499,12 +498,10 @@ class Download:
             return
 
         if str(tmp_dir) == str(self.temp_dir):
-            self.logger.warning(
+            self.logger.warning(f"Refusing to delete video temp folder '{self.temp_path}' as it's temp root.")
-                f"Attempted to delete video temp folder '{self.temp_path}', but it is the same as main temp folder."
-            )
             return
 
-        status = delete_dir(tmp_dir)
+        status: bool = delete_dir(tmp_dir)
         if by_pass:
             tmp_dir.mkdir(parents=True, exist_ok=True)
             self.logger.info(f"Temp folder '{self.temp_path}' emptied.")
@@ -523,7 +520,7 @@ class Download:
             await self._notify.emit(Events.ITEM_UPDATED, data=self.info)
             return
 
-        self.tmpfilename = status.get("tmpfilename")
+        self.tmpfilename: str | None = status.get("tmpfilename")
 
         fl = None
         if "final_name" in status:
@@ -560,7 +557,7 @@ class Download:
 
         if "downloaded_bytes" in status and status.get("downloaded_bytes", 0) > 0:
             self.info.downloaded_bytes = status.get("downloaded_bytes")
-            total = status.get("total_bytes") or status.get("total_bytes_estimate")
+            total: float | None = status.get("total_bytes") or status.get("total_bytes_estimate")
             if total:
                 try:
                     self.info.percent = status["downloaded_bytes"] / total * 100
@@ -617,7 +614,7 @@ class Download:
             return False
 
         if self.started_time < 1:
-            self.logger.debug(f"Download task '{self.info.title}: {self.info.id}' not started yet.")
+            self.logger.debug(f"Download task '{self.info.name()}' not started yet.")
             return False
 
         if int(time.time()) - self.started_time < 300:
@@ -633,7 +630,7 @@ class Download:
             return True
 
         if self.info.status not in ["finished", "error", "cancelled", "downloading", "postprocessing"]:
-            status = self.info.status if self.info.status else "unknown"
+            status: str = self.info.status if self.info.status else "unknown"
             self.logger.warning(
                 f"Download task '{self.info.title}: {self.info.id}' has been stuck in '{status}' state for '{int(time.time()) - self.started_time}' seconds."
             )
@@ -641,34 +638,6 @@ class Download:
 
         return False
 
-    def get_ytdlp_opts(self) -> YTDLPOpts:
-        """
-        Get the yt-dlp options used for this download task.
-
-        Returns:
-            YTDLPOpts: The yt-dlp options instance.
-
-        """
-        params: YTDLPOpts = YTDLPOpts.get_instance().preset(name=self.info.preset)
-        if self.info.cli:
-            params.add_cli(self.info.cli, from_user=True)
-
-        return params
-
-    def get_archive_id(self) -> str | None:
-        """
-        Get the archive ID for the download URL.
-
-        Returns:
-        str | None: The archive ID if available, None otherwise.
-
-        """
-        if not self.info or not self.info.url:
-            return None
-
-        idDict: dict = get_archive_id(self.info.url)
-        return idDict.get("archive_id")
-
     def __getstate__(self):
         state = self.__dict__.copy()
 

app/library/DownloadQueue.py

@@ -85,7 +85,7 @@ class DownloadQueue(metaclass=Singleton):
         self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency)
 
     @staticmethod
-    def get_instance():
+    def get_instance() -> "DownloadQueue":
         """
         Get the instance of the DownloadQueue.
 
@@ -98,7 +98,7 @@ class DownloadQueue(metaclass=Singleton):
 
         return DownloadQueue._instance
 
-    def attach(self, _: web.Application):
+    def attach(self, _: web.Application) -> None:
         """
         Attach the download queue to the application.
 
@@ -134,13 +134,11 @@ class DownloadQueue(metaclass=Singleton):
         await self.done.test()
         return True
 
-    async def initialize(self):
+    async def initialize(self) -> None:
         """
         Initialize the download queue.
         """
-        LOG.info(
+        LOG.info(f"Using '{self.config.max_workers}' workers for downloading.")
-            f"Using '{self.config.max_workers}' worker/s for downloading. Can be configured via `YTP_MAX_WORKERS` environment variable."
-        )
         asyncio.create_task(self._download_pool(), name="download_pool")
 
     async def start_items(self, ids: list[str]) -> dict[str, str]:
@@ -156,11 +154,11 @@ class DownloadQueue(metaclass=Singleton):
         """
         status: dict[str, str] = {"status": "ok"}
         started = False
-        tasks = []
+        tasks: list = []
 
         for item_id in ids:
             try:
-                item = self.queue.get(key=item_id)
+                item: Download = self.queue.get(key=item_id)
             except KeyError as e:
                 status[item_id] = f"not found: {e!s}"
                 status["status"] = "error"
@@ -173,7 +171,7 @@ class DownloadQueue(metaclass=Singleton):
                 continue
 
             item.info.auto_start = True
-            updated = self.queue.put(item)
+            updated: Download = self.queue.put(item)
             tasks.append(self._notify.emit(Events.ITEM_UPDATED, data=updated.info))
             tasks.append(
                 self._notify.emit(
@@ -207,11 +205,11 @@ class DownloadQueue(metaclass=Singleton):
 
         """
         status: dict[str, str] = {"status": "ok"}
-        tasks = []
+        tasks: list = []
 
         for item_id in ids:
             try:
-                item = self.queue.get(key=item_id)
+                item: Download = self.queue.get(key=item_id)
             except KeyError as e:
                 status[item_id] = f"not found: {e!s}"
                 status["status"] = "error"
@@ -229,7 +227,7 @@ class DownloadQueue(metaclass=Singleton):
                 continue
 
             item.info.auto_start = False
-            updated = self.queue.put(item)
+            updated: Download = self.queue.put(item)
             tasks.append(self._notify.emit(Events.ITEM_UPDATED, data=updated.info))
             tasks.append(
                 self._notify.emit(
@@ -414,20 +412,20 @@ class DownloadQueue(metaclass=Singleton):
         LOG.debug(f"Entry id '{entry.get('id')}' url '{entry.get('webpage_url')} - {entry.get('url')}'.")
 
         try:
-            _item = self.done.get(key=entry.get("id"), url=entry.get("webpage_url") or entry.get("url"))
+            _item: Download = self.done.get(key=entry.get("id"), url=entry.get("webpage_url") or entry.get("url"))
-            if _item is not None:
+            err_msg: str = f"Item '{_item.info.id}' - '{_item.info.title}' already exists. Removing from history."
-                err_msg = f"Item '{_item.info.id}' - '{_item.info.title}' already exists. Removing from history."
+            LOG.warning(err_msg)
-                LOG.warning(err_msg)
+            await self.clear([_item.info._id], remove_file=False)
-                await self.clear([_item.info._id], remove_file=False)
         except KeyError:
             pass
 
         try:
-            _item = self.queue.get(key=str(entry.get("id")), url=str(entry.get("webpage_url") or entry.get("url")))
+            _item: Download = self.queue.get(
-            if _item is not None:
+                key=str(entry.get("id")), url=str(entry.get("webpage_url") or entry.get("url"))
-                err_msg = f"Item ID '{_item.info.id}' - '{_item.info.title}' already in download queue."
+            )
-                LOG.info(err_msg)
+            err_msg: str = f"Item ID '{_item.info.id}' - '{_item.info.title}' already in download queue."
-                return {"status": "error", "msg": err_msg}
+            LOG.info(err_msg)
+            return {"status": "error", "msg": err_msg}
         except KeyError:
             pass
 
@@ -600,10 +598,10 @@ class DownloadQueue(metaclass=Singleton):
         if event_type.startswith("url"):
             return await self.add(item=item.new_with(url=entry.get("url")), already=already)
 
-        if not event_type.startswith("video"):
+        if event_type.startswith("video"):
-            return {"status": "error", "msg": f'Unsupported event type "{event_type}".'}
+            return await self._add_video(entry=entry, item=item, logs=logs)
 
-        return await self._add_video(entry=entry, item=item, logs=logs)
+        return {"status": "error", "msg": f'Unsupported event type "{event_type}".'}
 
     async def add(self, item: Item, already: set | None = None):
         """
@@ -683,7 +681,7 @@ class DownloadQueue(metaclass=Singleton):
 
             LOG.info(f"Checking '{item.url}' with {'cookies' if yt_conf.get('cookiefile') else 'no cookies'}.")
 
-            entry = await asyncio.wait_for(
+            entry: dict | None = await asyncio.wait_for(
                 fut=asyncio.get_running_loop().run_in_executor(
                     None,
                     functools.partial(
@@ -1009,7 +1007,7 @@ class DownloadQueue(metaclass=Singleton):
                 nMessage = f"Completed '{entry.info.title}' download."
                 _tasks.append(self._notify.emit(Events.ITEM_COMPLETED, data=entry.info, title=nTitle, message=nMessage))
 
-            await asyncio.sleep(0.2)
+            await asyncio.sleep(0.5)
             self.done.put(entry)
             _tasks.append(
                 self._notify.emit(
@@ -1055,7 +1053,8 @@ class DownloadQueue(metaclass=Singleton):
         if self.is_paused() or self.done.empty():
             return
 
-        LOG.debug("Checking history queue for queued live stream links.")
+        if self.config.debug:
+            LOG.debug("Checking history queue for queued live stream links.")
 
         time_now = datetime.now(tz=UTC)
 

app/library/ItemDTO.py

@@ -193,7 +193,7 @@ class Item:
         from .config import Config
         from .Utils import calc_download_path, strip_newline
 
-        data = {}
+        data: dict = {}
         for k, v in self.serialize().items():
             if not v and k not in ("auto_start"):
                 continue
@@ -209,7 +209,7 @@ class Item:
             else:
                 data[k] = v
 
-        items = "".join(f'{k}="{v}", ' for k, v in data.items() if v)
+        items: str = "".join(f'{k}="{v}", ' for k, v in data.items() if v is not None)
         return f"Item({items.strip(', ')})"
 
 

app/library/Scheduler.py

@@ -126,7 +126,7 @@ class Scheduler(metaclass=Singleton):
 
         self._jobs[job_id] = job
 
-        LOG.debug(f"Added '{job_id}' to the scheduler.")
+        LOG.debug(f"Added '{job_id}' to the scheduler to run on '{timer}'.")
 
         return job_id
 

app/library/Tasks.py

@@ -72,7 +72,7 @@ class Task:
         return (True, f"Task '{self.name}' items marked as downloaded.")
 
     def unmark(self) -> tuple[bool, str]:
-        ret = self._mark_logic()
+        ret: tuple[bool, str] | set[tuple[Path, set[str]]] = self._mark_logic()
         if isinstance(ret, tuple):
             return ret
 
@@ -422,6 +422,7 @@ class Tasks(metaclass=Singleton):
                         "folder": folder,
                         "template": template,
                         "cli": cli,
+                        "auto_start": task.auto_start,
                     }
                 )
             )
@@ -549,7 +550,11 @@ class HandleTask:
             if handler is None:
                 return None
 
-        return await Services.get_instance().handle_async(handler=handler.handle, task=task, **kwargs)
+        try:
+            return await Services.get_instance().handle_async(handler=handler.handle, task=task, **kwargs)
+        except Exception as e:
+            LOG.exception(e)
+            raise
 
     def _discover(self) -> list[type]:
         import importlib

app/library/Utils.py

@@ -1472,14 +1472,20 @@ def archive_delete(file: str | Path, ids: list[str]) -> bool:
 
     changed = False
     kept_lines: list[str] = []
+    removed_ids: list[str] = []
     with path.open("r", encoding="utf-8") as f:
         for line in f:
             s: str = line.strip()
 
-            if not s or len(s.split()) < 2 or s in remove_ids:
+            if not s or len(s.split()) < 2:
                 changed = True
                 continue
 
+            if s in remove_ids:
+                changed = True
+                removed_ids.append(s)
+                continue
+
             kept_lines.append(line)
 
     if not changed:

app/library/task_handlers/youtube.py

@@ -2,15 +2,18 @@ import asyncio
 import logging
 import re
 from typing import TYPE_CHECKING, Any
+from xml.etree.ElementTree import Element
 
 from app.library.config import Config
 from app.library.DownloadQueue import DownloadQueue
 from app.library.Events import EventBus, Events
 from app.library.ItemDTO import Item, ItemDTO
 from app.library.Tasks import Task
-from app.library.Utils import archive_read, get_archive_id
+from app.library.Utils import archive_read
 
 if TYPE_CHECKING:
+    from xml.etree.ElementTree import Element
+
     from app.library.Download import Download
 
 LOG: logging.Logger = logging.getLogger(__name__)
@@ -23,7 +26,7 @@ EventBus.get_instance().subscribe(
 
 
 class YoutubeHandler:
-    queued_ids: set[str] = set()
+    queued: set[str] = set()
     failure_count: dict[str, int] = {}
 
     FEED = "https://www.youtube.com/feeds/videos.xml?{type}={id}"
@@ -35,10 +38,10 @@ class YoutubeHandler:
     @staticmethod
     def can_handle(task: Task) -> bool:
         if not task.get_ytdlp_opts().get_all().get("download_archive"):
-            LOG.debug(f"Task '{task.id}: {task.name}' does not have an archive file configured.")
+            LOG.debug(f"Task '{task.name}' does not have an archive file configured.")
             return False
 
-        LOG.debug(f"Checking if task '{task.id}: {task.name}' can handle YouTube URL: {task.url}")
+        LOG.debug(f"Checking if task '{task.name}' is using parsable YouTube URL: {task.url}")
         return YoutubeHandler.parse(task.url) is not None
 
     @staticmethod
@@ -56,7 +59,7 @@ class YoutubeHandler:
         """
         params: dict = task.get_ytdlp_opts().get_all()
         if not (archive_file := params.get("download_archive")):
-            LOG.error(f"Task '{task.id}: {task.name}' does not have an archive file.")
+            LOG.error(f"Task '{task.name}' does not have an archive file.")
             return
 
         import httpx
@@ -64,12 +67,12 @@ class YoutubeHandler:
 
         parsed: dict[str, str] | None = YoutubeHandler.parse(task.url)
         if not parsed:
-            LOG.error(f"Cannot parse '{task.id}: {task.name}' URL: {task.url}")
+            LOG.error(f"Cannot parse '{task.name}' URL: {task.url}")
             return
 
         feed_url: str = YoutubeHandler.FEED.format(type=parsed["type"], id=parsed["id"])
 
-        LOG.debug(f"Fetching '{task.id}: {task.name}' feed.")
+        LOG.debug(f"Fetching '{task.name}' feed.")
         opts: dict[str, Any] = {
             "proxy": params.get("proxy"),
             "headers": {
@@ -93,71 +96,77 @@ class YoutubeHandler:
             pass
 
         items: list = []
+        has_items = False
 
         async with httpx.AsyncClient(**opts) as client:
             response: httpx.Response = await client.request(method="GET", url=feed_url, timeout=120)
             response.raise_for_status()
 
-            root = fromstring(response.text)
+            root: Element[str] = fromstring(response.text)
-            ns = {"atom": "http://www.w3.org/2005/Atom", "yt": "http://www.youtube.com/xml/schemas/2015"}
+            ns: dict[str, str] = {
+                "atom": "http://www.w3.org/2005/Atom",
+                "yt": "http://www.youtube.com/xml/schemas/2015",
+            }
 
             for entry in root.findall("atom:entry", ns):
-                vid_elem = entry.find("yt:videoId", ns)
+                vid_elem: Element[str] | None = entry.find("yt:videoId", ns)
-                vid = vid_elem.text if vid_elem is not None else ""
+                vid: str | None = vid_elem.text if vid_elem is not None else ""
                 if not vid:
-                    LOG.warning(f"Entry in '{task.id}: {task.name}' feed is missing a video ID. Skipping entry.")
+                    LOG.warning(f"Entry in '{task.name}' feed is missing a video ID. Skipping entry.")
                     continue
 
-                title_elem = entry.find("atom:title", ns)
+                archive_id: str = f"youtube {vid}"
-                pub_elem = entry.find("atom:published", ns)
-                title = title_elem.text if title_elem is not None else ""
-                published = pub_elem.text if pub_elem is not None else ""
                 url: str = f"https://www.youtube.com/watch?v={vid}"
-                idDict = get_archive_id(url=url)
+
-                if not (archive_id := idDict.get("archive_id")):
+                title_elem: Element[str] | None = entry.find("atom:title", ns)
-                    LOG.warning(f"Item '{title}' does not have a valid archive ID. URL: {url}")
+                title: str | None = title_elem.text if title_elem is not None else ""
+
+                pub_elem: Element[str] | None = entry.find("atom:published", ns)
+                published: str | None = pub_elem.text if pub_elem is not None else ""
+                has_items = True
+
+                if archive_id in YoutubeHandler.queued:
+                    LOG.debug(f"Item '{vid}' is already queued for download. Skipping.")
                     continue
 
                 items.append({"id": vid, "url": url, "title": title, "published": published, "archive_id": archive_id})
 
         if len(items) < 1:
-            LOG.warning(f"No entries found in '{task.id}: {task.name}' feed. URL: {feed_url}")
+            if not has_items:
+                LOG.warning(f"No entries found in '{task.name}' feed. URL: {feed_url}")
+            else:
+                LOG.debug(f"No new items found in '{task.name}' feed.")
             return
 
         filtered: list = []
 
-        downloaded: list[str] = archive_read(
+        downloaded: list[str] = archive_read(archive_file, [item["archive_id"] for item in items])
-            file=archive_file,
-            ids=[item["archive_id"] for item in items if item["id"] not in YoutubeHandler.queued_ids],
-        )
 
         for item in items:
+            YoutubeHandler.queued.add(item["archive_id"])
             if item["archive_id"] in downloaded:
-                YoutubeHandler.queued_ids.add(item["id"])
                 continue
 
             if queue.queue.exists(url=item["url"]):
-                LOG.debug(f"Item '{item['id']}' exists in the queue.")
-                YoutubeHandler.queued_ids.add(item["id"])
                 continue
 
             try:
                 done: Download = queue.done.get(url=item["url"])
                 if "error" != done.info.status:
-                    LOG.debug(f"Item '{item['id']}' exists in the history.")
-                    YoutubeHandler.queued_ids.add(item["id"])
                     continue
             except KeyError:
                 pass
 
-            YoutubeHandler.queued_ids.add(item)
+            if item["archive_id"] not in YoutubeHandler.failure_count:
+                YoutubeHandler.failure_count[item["archive_id"]] = 0
+
             filtered.append(item)
 
         if len(filtered) < 1:
-            LOG.debug(f"No new items found in '{task.id}: {task.name}' feed.")
+            LOG.debug(f"No new items found in '{task.name}' feed.")
             return
 
-        LOG.info(f"Found '{len(filtered)}' new items from '{task.id}: {task.name}' feed.")
+        LOG.info(f"Found '{len(filtered)}' new items from '{task.name}' feed.")
 
         rItem: Item = Item.format(
             {
@@ -166,19 +175,18 @@ class YoutubeHandler:
                 "folder": task.folder if task.folder else "",
                 "template": task.template if task.template else "",
                 "cli": task.cli if task.cli else "",
+                "auto_start": task.auto_start,
                 "extras": {"source_task": task.id},
             }
         )
 
         try:
             await asyncio.gather(
-                *[
+                *[notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize()) for item in filtered]
-                    notify.emit(Events.ADD_URL, data=rItem.new_with({"url": item["url"]}).serialize())
-                    for item in filtered
-                ]
             )
         except Exception as e:
-            LOG.error(f"Error while adding items from '{task.id}: {task.name}'. {e!s}")
+            LOG.exception(e)
+            LOG.error(f"Error while adding items from '{task.name}'. {e!s}")
             return
 
     @staticmethod
@@ -216,29 +224,33 @@ class YoutubeHandler:
         if not item or not isinstance(item, ItemDTO):
             return
 
-        cls.queued_ids.add(item.id)
+        if not item.archive_id or not cls.failure_count.get(item.archive_id, None):
-
+            LOG.debug(f"Item '{item.name()}' not queued by the handler.")
-        if item.id not in cls.queued_ids:
             return
 
-        currentFailureCount: int = cls.failure_count.get(item.id, 0)
+        failCount: int = int(cls.failure_count.get(item.archive_id, 0))
 
-        LOG.info(f"Removing '{item.name()}' from queued IDs due to error. Failure count: '{currentFailureCount + 1}'.")
+        LOG.info(f"Removing '{item.name()}' from queued IDs due to error. Failure count: '{failCount + 1}'.")
-        cls.queued_ids.remove(item.id)
+        if item.archive_id in cls.queued:
+            cls.queued.remove(item.archive_id)
 
-        cls.failure_count[item.id] = cls.failure_count.get(item.id, 0) + 1
+        cls.failure_count[item.archive_id] = 1 + failCount
 
     @staticmethod
-    def tests() -> list[str]:
+    def tests() -> list[tuple[str, bool]]:
         """
-        Return a list of test URLs to validate the parsing logic.
+        Test cases for the URL parser.
+
+        Returns:
+            list[tuple[str, bool]]: A list of tuples containing the URL and expected result.
+
         """
         return [
-            "https://www.youtube.com/channel/UCabc123ABCDEFGHIJKLMN",
+            ("https://www.youtube.com/channel/UCabc123ABCDEFGHIJKLMN", True),
-            "https://youtube.com/c/MyCustomName",
+            ("https://youtube.com/c/MyCustomName", False),
-            "https://youtube.com/user/SomeUser123",
+            ("https://youtube.com/user/SomeUser123", False),
-            "https://youtube.com/@SomeHandle",
+            ("https://youtube.com/@SomeHandle", False),
-            "https://youtube.com/playlist?list=PLxyz789ABCDEFGHIJ",
+            ("https://youtube.com/playlist?list=PLxyz789ABCDEFGHIJ", True),
-            "https://youtube.com/watch?v=foo&list=PLxyz789ABCDEFGHIJ",
+            ("https://youtube.com/watch?v=foo&list=PLxyz789ABCDEFGHIJ", True),
-            "https://youtube.com/watch?v=foo",
+            ("https://youtube.com/watch?v=foo", False),
         ]

app/routes/api/tasks.py

@@ -94,7 +94,7 @@ async def tasks_add(request: Request, encoder: Encoder) -> Response:
 
 
 @route("POST", "api/tasks/{id}/mark", "tasks_mark")
-async def mark_task(request: Request, encoder: Encoder) -> Response:
+async def task_mark(request: Request, encoder: Encoder) -> Response:
     """
     Mark all items from task as downloaded.
 
@@ -111,9 +111,9 @@ async def mark_task(request: Request, encoder: Encoder) -> Response:
     if not task_id:
         return web.json_response(data={"error": "No task id."}, status=web.HTTPBadRequest.status_code)
 
-    tasks = Tasks.get_instance()
+    tasks: Tasks = Tasks.get_instance()
     try:
-        task = tasks.get(task_id)
+        task: Task | None = tasks.get(task_id)
         if not task:
             return web.json_response(
                 data={"error": f"Task '{task_id}' does not exist."}, status=web.HTTPNotFound.status_code
@@ -126,3 +126,38 @@ async def mark_task(request: Request, encoder: Encoder) -> Response:
         return web.json_response(data={"message": _message}, status=web.HTTPOk.status_code, dumps=encoder.encode)
     except ValueError as e:
         return web.json_response(data={"error": str(e)}, status=web.HTTPBadRequest.status_code)
+
+
+@route("DELETE", "api/tasks/{id}/mark", "tasks_unmark")
+async def task_unmark(request: Request, encoder: Encoder) -> Response:
+    """
+    Remove All tasks items from download archive.
+
+    Args:
+        request (Request): The request object.
+        encoder (Encoder): The encoder instance.
+
+    Returns:
+        Response: The response object
+
+    """
+    task_id: str = request.match_info.get("id", None)
+
+    if not task_id:
+        return web.json_response(data={"error": "No task id."}, status=web.HTTPBadRequest.status_code)
+
+    tasks: Tasks = Tasks.get_instance()
+    try:
+        task: Task | None = tasks.get(task_id)
+        if not task:
+            return web.json_response(
+                data={"error": f"Task '{task_id}' does not exist."}, status=web.HTTPNotFound.status_code
+            )
+
+        _status, _message = task.unmark()
+        if not _status:
+            return web.json_response(data={"error": _message}, status=web.HTTPBadRequest.status_code)
+
+        return web.json_response(data={"message": _message}, status=web.HTTPOk.status_code, dumps=encoder.encode)
+    except ValueError as e:
+        return web.json_response(data={"error": str(e)}, status=web.HTTPBadRequest.status_code)

ui/app/pages/tasks.vue

@@ -194,9 +194,14 @@
 
                           <hr class="dropdown-divider" />
 
-                          <NuxtLink class="dropdown-item has-text-danger" @click="archiveItems(item)">
+                          <NuxtLink class="dropdown-item" @click="archiveAll(item)">
                             <span class="icon"><i class="fa-solid fa-box-archive" /></span>
-                            <span>Archive all</span>
+                            <span>Archive All</span>
+                          </NuxtLink>
+
+                          <NuxtLink class="dropdown-item" @click="unarchiveAll(item)">
+                            <span class="icon"><i class="fa-solid fa-box-archive" /></span>
+                            <span>Unarchive All</span>
                           </NuxtLink>
 
                           <hr class="dropdown-divider" />
@@ -313,9 +318,14 @@
 
                   <hr class="dropdown-divider" />
 
-                  <NuxtLink class="dropdown-item has-text-danger" @click="archiveItems(item)">
+                  <NuxtLink class="dropdown-item" @click="archiveAll(item)">
                     <span class="icon"><i class="fa-solid fa-box-archive" /></span>
-                    <span>Archive all</span>
+                    <span>Archive All</span>
+                  </NuxtLink>
+
+                  <NuxtLink class="dropdown-item" @click="unarchiveAll(item)">
+                    <span class="icon"><i class="fa-solid fa-box-archive" /></span>
+                    <span>Unarchive All</span>
                   </NuxtLink>
                 </Dropdown>
               </div>
@@ -370,6 +380,7 @@ const box = useConfirm()
 const toast = useNotification()
 const config = useConfigStore()
 const socket = useSocketStore()
+const { confirmDialog: cDialog } = useDialog()
 const display_style = useStorage<string>("tasks_display_style", "cards")
 
 const tasks = ref<Array<task_item>>([])
@@ -428,7 +439,7 @@ watch(() => config.app.basic_mode, async v => {
     return
   }
   await navigateTo('/')
-},{ immediate: true })
+}, { immediate: true })
 
 watch(() => socket.isConnected, async () => {
   if (socket.isConnected && initialLoad.value) {
@@ -764,17 +775,18 @@ const get_tags = (name: string): Array<string> => {
 
 const remove_tags = (name: string): string => name.replace(/\[(.*?)\]/g, '').trim();
 
-const archiveItems = async (item: task_item) => {
-  dialog_confirm.value.visible = true
-  dialog_confirm.value.title = 'Archive All videos'
-  dialog_confirm.value.message = `Archive all items for '${item.name}' task? This will mark all items as downloaded and update the archive file.`
-  dialog_confirm.value.confirm = async () => await archiveAll(item)
-}
-
 const archiveAll = async (item: task_item) => {
   try {
-    dialog_confirm.value.visible = false
+    const { status } = await cDialog({
+      message: `Mark all '${item.name}' items as downloaded in download archive?`
+    })
+
+    if (true !== status) {
+      return;
+    }
+
     item.in_progress = true
+
     const response = await request(`/api/tasks/${item.id}/mark`, { method: 'POST' })
     const data = await response.json()
 
@@ -791,4 +803,33 @@ const archiveAll = async (item: task_item) => {
     item.in_progress = false
   }
 }
+
+const unarchiveAll = async (item: task_item) => {
+  try {
+    const { status } = await cDialog({
+      message: `Remove all '${item.name}' items from download archive?`
+    })
+
+    if (true !== status) {
+      return;
+    }
+
+    item.in_progress = true
+
+    const response = await request(`/api/tasks/${item.id}/mark`, { method: 'DELETE' })
+    const data = await response.json()
+
+    if (data?.error) {
+      toast.error(data.error)
+      return
+    }
+
+    toast.success(data.message)
+  } catch (e: any) {
+    toast.error(`Failed to remove items from archive. ${e.message || 'Unknown error.'}`)
+    return
+  } finally {
+    item.in_progress = false
+  }
+}
 </script>