diff --git a/.vscode/settings.json b/.vscode/settings.json index e7fec592..96a1dd3c 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -112,6 +112,7 @@ "timespec", "tmpfilename", "ungroup", + "unmark", "unnegated", "unpickleable", "Unraid", diff --git a/API.md b/API.md index 046a4141..68717ba8 100644 --- a/API.md +++ b/API.md @@ -41,7 +41,6 @@ This document describes the available endpoints and their usage. All endpoints r - [POST /api/file/action/{path:.\*}](#post-apifileactionpath) - [POST /api/file/download](#post-apifiledownload) - [GET /api/file/download/{token}](#get-apifiledownloadtoken) - - [GET /api/yt-dlp/archive/recheck](#get-apiyt-dlparchiverecheck) - [GET /api/random/background](#get-apirandombackground) - [GET /api/presets](#get-apipresets) - [GET /api/dl\_fields](#get-apidl_fields) @@ -714,25 +713,6 @@ or an error: --- -### GET /api/yt-dlp/archive/recheck -**Purpose**: Recheck manual archive entries to see if become available or not. - -**Response**: -```json -[ - { - "id": "youtube_video_id", - "url": "https://youtube.com/watch?v=...", - "status": "available|unavailable|error", - "info": { ... } // video info if available - }, - ... -] -``` -- Returns `404 Not Found` if manual archive is not enabled or file doesn't exist. - ---- - ### GET /api/random/background **Purpose**: Get a random background image from configured backends. @@ -1105,7 +1085,7 @@ or an error: --- ### GET /api/dev/loop -**Purpose**: Development-only. Show asyncio loop details and running tasks. +**Purpose**: Development-only. Show event loop details and running tasks. **Response**: ```json diff --git a/FAQ.md b/FAQ.md index b3cfd6bb..27e03ffc 100644 --- a/FAQ.md +++ b/FAQ.md @@ -30,7 +30,6 @@ or the `environment:` section in `compose.yaml` file. | YTP_SOCKET_TIMEOUT | The timeout for the yt-dlp socket connection variable | `30` | | YTP_EXTRACT_INFO_TIMEOUT | The timeout for extracting video information | `70` | | YTP_DB_FILE | The path to the SQLite database file | `{config_path}/ytptube.db` | -| YTP_MANUAL_ARCHIVE | The path to the manual archive file | `{config_path}/manual_archive.log` | | YTP_UI_UPDATE_TITLE | Whether to update the title of the page with the current stats | `true` | | YTP_PIP_PACKAGES | A space separated list of pip packages to install | `empty string` | | YTP_PIP_IGNORE_UPDATES | Do not update the custom pip packages | `false` | diff --git a/app/library/Download.py b/app/library/Download.py index 79fac733..80a007f0 100644 --- a/app/library/Download.py +++ b/app/library/Download.py @@ -17,7 +17,7 @@ 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, load_cookies +from .Utils import delete_dir, extract_info, extract_ytdlp_logs, get_archive_id, load_cookies from .ytdlp import YTDLP from .YTDLPOpts import YTDLPOpts @@ -263,21 +263,19 @@ class Download: self.info_dict = info if self.is_live: - hasDeletedOptions = False deletedOpts: list = [] for opt in self.bad_live_options: if opt in params: params.pop(opt, None) - hasDeletedOptions = True deletedOpts.append(opt) - if hasDeletedOptions: + if len(deletedOpts) > 0: self.logger.warning( f"Live stream detected for '{self.info.title}', The following opts '{deletedOpts=}' have been deleted." ) if isinstance(self.info_dict, dict) and len(self.info_dict.get("formats", [])) < 1: - msg = f"Failed to extract any formats for '{self.info.url}'." + msg: str = f"Failed to extract any formats for '{self.info.url}'." if filtered_logs := extract_ytdlp_logs(self.logs): msg += " " + ", ".join(filtered_logs) @@ -643,6 +641,34 @@ 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() diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index 2ca39cd2..f25fee31 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -24,12 +24,13 @@ from .Presets import Presets from .Scheduler import Scheduler from .Singleton import Singleton from .Utils import ( + archive_read, arg_converter, calc_download_path, dt_delta, extract_info, extract_ytdlp_logs, - is_downloaded, + get_archive_id, load_cookies, str_to_dt, ytdlp_reject, @@ -39,7 +40,7 @@ from .YTDLPOpts import YTDLPOpts if TYPE_CHECKING: from app.library.Presets import Preset -LOG = logging.getLogger("DownloadQueue") +LOG: logging.Logger = logging.getLogger("DownloadQueue") class DownloadQueue(metaclass=Singleton): @@ -665,12 +666,15 @@ class DownloadQueue(metaclass=Singleton): LOG.warning(f"Using external downloader '{yt_conf.get('external_downloader')}' for '{item.url}'.") item.extras.update({"external_downloader": True}) - downloaded, id_dict = self._is_downloaded(file=yt_conf.get("download_archive"), url=item.url) - if downloaded is True and id_dict: - message = f"'{id_dict.get('id')}': The URL '{item.url}' is already downloaded and recorded in archive." - LOG.error(message) - await self._notify.emit(Events.LOG_INFO, title="Already Downloaded", message=message) - return {"status": "error", "msg": message} + if archive_file := yt_conf.get("download_archive"): + idDict: dict = get_archive_id(item.url) + if (archive_id := idDict.get("archive_id")) and len(archive_read(archive_file, [archive_id])) > 0: + message: str = ( + f"'{idDict.get('id')}': The URL '{item.url}' is already downloaded and recorded in archive." + ) + LOG.error(message) + await self._notify.emit(Events.LOG_INFO, title="Already Downloaded", message=message) + return {"status": "error", "msg": message} started: float = time.perf_counter() @@ -1029,23 +1033,6 @@ class DownloadQueue(metaclass=Singleton): if self.event: self.event.set() - def _is_downloaded(self, url: str, file: str | None = None) -> tuple[bool, dict | None]: - """ - Check if the url has been downloaded already. - - Args: - url (str): The url to check. - file (str | None): The archive file to check. - - Returns: - tuple: A tuple with the status of the operation and the id of the downloaded item. - - """ - if not url or not file: - return False, None - - return is_downloaded(file, url) - async def _check_for_stale(self): """ Monitor pool for stale downloads and cancel them if needed. diff --git a/app/library/ItemDTO.py b/app/library/ItemDTO.py index 4a264e67..749b661c 100644 --- a/app/library/ItemDTO.py +++ b/app/library/ItemDTO.py @@ -6,7 +6,7 @@ from dataclasses import dataclass, field from email.utils import formatdate from typing import Any -from app.library.Utils import clean_item +from app.library.Utils import clean_item, get_archive_id LOG = logging.getLogger("ItemDTO") @@ -248,6 +248,20 @@ class ItemDTO: def name(self) -> str: return f'id="{self.id}", title="{self.title}"' + 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: + return None + + idDict: dict = get_archive_id(self.url) + return idDict.get("archive_id") + @staticmethod def removed_fields() -> tuple: """Fields that once existed but are no longer used.""" diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 00bdd1e8..638a3183 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -19,7 +19,7 @@ from .ItemDTO import Item from .Scheduler import Scheduler from .Services import Services from .Singleton import Singleton -from .Utils import extract_info, init_class, validate_url +from .Utils import archive_add, archive_delete, extract_info, init_class, validate_url from .YTDLPOpts import YTDLPOpts LOG: logging.Logger = logging.getLogger("tasks") @@ -47,51 +47,69 @@ class Task: def get(self, key: str, default: Any = None) -> Any: return self.serialize().get(key, default) - def mark(self) -> tuple[bool, str]: - if not self.url: - return False, "No URL found in task parameters." + def get_ytdlp_opts(self) -> YTDLPOpts: + params: YTDLPOpts = YTDLPOpts.get_instance() + + if self.preset: + params = params.preset(name=self.preset) - params: YTDLPOpts = YTDLPOpts.get_instance().preset(name=self.preset) if self.cli: - params.add_cli(self.cli, from_user=True) + params = params.add_cli(self.cli, from_user=True) - params = params.get_all() - if not (_archive := params.get("download_archive", None)): - return False, "No archive file found in task parameters." + return params - _archive: Path = Path(_archive) - if not _archive.parent.exists(): - _archive.parent.mkdir(parents=True, exist_ok=True) + def mark(self) -> tuple[bool, str]: + ret = self._mark_logic() + if isinstance(ret, tuple): + return ret - if not _archive.exists(): - _archive.touch() + archive_file: Path = ret.get("file") + items: set[str] = ret.get("items", set()) - _info = extract_info(params, self.url, no_archive=True, follow_redirect=True) - if not _info or not isinstance(_info, dict): - return False, "Failed to extract information from URL." + if len(items) < 1 or not archive_add(archive_file, list(items)): + return (True, "No new items to mark as downloaded.") - if "playlist" != _info.get("_type"): - return False, "Expected a playlist type from extract_info." + return (True, f"Task '{self.name}' items marked as downloaded.") - archived_items: list[str] = [] - with _archive.open() as f: - for line in f: - line = line.strip() - if not line or not isinstance(line, str) or line.startswith("#"): - continue + def unmark(self) -> tuple[bool, str]: + ret = self._mark_logic() + if isinstance(ret, tuple): + return ret - archived_items.append(line) + archive_file: Path = ret.get("file") + items: set[str] = ret.get("items", set()) - def _process(item: dict) -> bool: - status = False + if len(items) < 1 or not archive_delete(archive_file, list(items)): + return (True, "No items to remove from archive file.") + + return (True, f"Removed '{self.name}' items from archive file.") + + def _mark_logic(self) -> tuple[bool, str] | set[tuple[Path, set[str]]]: + if not self.url: + return (False, "No URL found in task parameters.") + + params: dict = self.get_ytdlp_opts().get_all() + if not (archive_file := params.get("download_archive")): + return (False, "No archive file found.") + + archive_file: Path = Path(archive_file) + + ie_info: dict | None = extract_info(params, self.url, no_archive=True, follow_redirect=True) + if not ie_info or not isinstance(ie_info, dict): + return (False, "Failed to extract information from URL.") + + if "playlist" != ie_info.get("_type"): + return (False, "Expected a playlist type from extract_info.") + + items: set[str] = set() + + def _process(item: dict): for entry in item.get("entries", []): if not isinstance(entry, dict): continue if "playlist" == entry.get("_type"): - _status = _process(entry) - if status is False: - status = _status + _process(entry) continue if entry.get("_type") not in ("video", "url"): @@ -100,26 +118,13 @@ class Task: if not entry.get("id") or not entry.get("ie_key"): continue - archive_id = f'{entry.get("ie_key","").lower()} {entry.get("id")}' + archive_id: str = f"{entry.get('ie_key', '').lower()} {entry.get('id')}" - if archive_id in archived_items: - continue + items.add(archive_id) - archived_items.append(archive_id) - status = True + _process(ie_info) - return status - - updated = _process(_info) - - if not updated: - return True, "No new items to mark as downloaded." - - with _archive.open("a") as f: - for item in archived_items: - f.write(f"{item}\n") - - return True, f"Task '{self.name}' marked as downloaded. Updated archive file '{_archive}'." + return {"file": archive_file, "items": items} class Tasks(metaclass=Singleton): diff --git a/app/library/Utils.py b/app/library/Utils.py index 4a54145f..74c3d613 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -63,7 +63,7 @@ REMOVE_KEYS: list = [ YTDLP_INFO_CLS: YTDLP = None -ALLOWED_SUBS_EXTENSIONS: tuple[str] = (".srt", ".vtt", ".ass") +ALLOWED_SUBS_EXTENSIONS: tuple[str, str, str] = (".srt", ".vtt", ".ass") FILES_TYPE: list = [ {"rx": re.compile(r"\.(avi|ts|mkv|mp4|mp3|mpv|ogm|m4v|webm|m4b)$", re.IGNORECASE), "type": "video"}, @@ -262,113 +262,6 @@ def merge_dict(source: dict, destination: dict) -> dict: return destination_copy - -def is_downloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, str | None, str | None]]: - """ - Check if the video is already downloaded. - - Args: - archive_file (str): Archive file path. - url (str): URL to check. - - Returns: - bool: True if the video is already downloaded. - dict: Video information. - - """ - idDict = {"id": None, "ie_key": None, "archive_id": None} - - if not url or not archive_file: - return (False, idDict) - - if not Path(archive_file).exists(): - return (False, idDict) - - idDict = get_archive_id(url=url) - - if idDict.get("archive_id"): - with open(archive_file) as f: - for line in f: - if idDict["archive_id"] in line: - return (True, idDict) - - return (False, idDict) - - -def remove_from_archive(archive_file: str | Path, url: str) -> bool: - """ - Remove the downloaded video record from the archive file. - - Args: - archive_file (str): Archive file path. - url (str): URL to check and remove. - - Returns: - bool: True if the record removed, False otherwise. - - """ - if not url or not archive_file: - return False - - archive_path: Path = Path(archive_file) if not isinstance(archive_file, Path) else archive_file - if not archive_path.exists(): - return False - - idDict = get_archive_id(url=url) - archive_id: str | None = idDict.get("archive_id") - - if not archive_id: - return False - - lines: list[str] = archive_path.read_text(encoding="utf-8").splitlines() - new_lines: list[str] = [line for line in lines if archive_id not in line] - - if len(lines) == len(new_lines): - return False - - archive_path.write_text("\n".join(new_lines) + "\n", encoding="utf-8") - return True - - -def load_file(file: str, check_type=None) -> tuple[dict | list, bool, str]: - """ - Load a JSON or JSON5 file and return the contents as a dictionary - - Args: - file (str): File path - check_type (type): Type to check the loaded file against. - - Returns tuple: - dict|list: Dictionary or list of the file contents. Empty dict if the file could not be loaded. - bool: True if the file was loaded successfully. - str: Error message if the file could not be loaded. - - """ - try: - with open(file) as json_data: - opts = json.load(json_data) - - if check_type: - assert isinstance(opts, check_type) - - return (opts, True, "") - except Exception: - with open(file) as json_data: - from pyjson5 import load as json5_load - - try: - opts = json5_load(json_data) - - if check_type: - assert isinstance(opts, check_type) - - return (opts, True, "") - except AssertionError: - return ({}, False, f"Failed to assert that the contents '{type(opts)}' are of type '{check_type}'.") - except Exception as e: - return ({}, False, f"{e}") - - def check_id(file: Path) -> bool | str: """ Check if we are able to get an id from the file name. @@ -1107,7 +1000,7 @@ def load_cookies(file: str | Path) -> tuple[bool, MozillaCookieJar]: raise ValueError(msg) from e -def get_archive_id(url: str) -> tuple[bool, dict[str | None, str | None, str | None]]: +def get_archive_id(url: str) -> dict[str, str | None]: """ Get the archive ID for a given URL. @@ -1115,8 +1008,11 @@ def get_archive_id(url: str) -> tuple[bool, dict[str | None, str | None, str | N url (str): URL to check. Returns: - bool: True if the video is already downloaded. - dict: Video information. + dict: { + "id": str | None, + "ie_key": str | None, + "archive_id": str | None + } """ global YTDLP_INFO_CLS # noqa: PLW0603 @@ -1455,3 +1351,133 @@ def list_folders(path: Path, base: Path, depth_limit: int) -> list[str]: folders.extend(list_folders(entry, base, depth_limit)) return folders + + +def archive_add(file: str | Path, ids: list[str], skip_check: bool = False) -> bool: + """ + Add IDs to an archive file. + + Args: + file (str|Path): The archive file path. + ids (list[str]): List of IDs to add. + skip_check (bool): If True, skip checking for existing IDs. + + """ + if not ids: + return False + + path: Path = Path(file) if not isinstance(file, Path) else file + existing_ids: set[str] = set() + + if not skip_check and path.exists(): + with path.open("r", encoding="utf-8") as f: + for line in f: + s = line.strip() + + if not s or len(s.split()) < 2: + continue + + existing_ids.add(s) + + new_ids: list[str] = [] + for raw in ids: + s: str = str(raw).strip() + + if not s or len(s.split()) < 2 or s in existing_ids or s in new_ids: + continue + + new_ids.append(s) + + if not new_ids: + return False + + try: + if not path.parent.exists(): + path.parent.mkdir(parents=True, exist_ok=True) + + with path.open("a", encoding="utf-8") as f: + f.write("".join(f"{x}\n" for x in new_ids)) + + return True + except OSError as e: + LOG.error(f"Failed to write to archive file '{path!s}'. {e!s}") + return False + + +def archive_read(file: str | Path, ids: list[str] | None = None) -> list[str]: + """ + Read IDs from an archive file with optional filtering. + + - If `ids` is empty, return all IDs present in the archive file. + - If `ids` is provided, return only the ids that exist in the archive file, + + Args: + file (str|Path): The archive file path. + ids (list[str]): Optional list of IDs to query; empty list returns all. + + Returns: + list[str]: List of ids found in the archive file filtered by `ids` if provided. + + """ + path: Path = Path(file) if not isinstance(file, Path) else file + if not path.exists(): + return [] + + ids_set: set[str] | None = ( + {s.strip() for s in ids if str(s).strip() and len(str(s).strip().split()) < 2} if ids else None + ) + + found: 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: + continue + + if ids_set is None or s in ids_set: + found.append(s) + + return found + + +def archive_delete(file: str | Path, ids: list[str]) -> bool: + """ + Delete the given IDs from an archive file. + + Args: + file (str|Path): The archive file path. + ids (list[str]): List of IDs to remove. + + Returns: + bool: True if deletion succeeded (or nothing to do), False on error. + + """ + path: Path = Path(file) if not isinstance(file, Path) else file + + if not path.exists() or not ids: + return False + + remove_ids: set[str] = {x.strip() for x in ids if str(x).strip() and len(str(x).strip().split()) < 2} + if not remove_ids: + return True + + changed = False + kept_lines: 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: + changed = True + continue + + kept_lines.append(line) + + if not changed: + return True + + with path.open("w", encoding="utf-8") as f: + f.writelines(kept_lines) + + return True diff --git a/app/library/config.py b/app/library/config.py index a94d2e1c..fcbf77bb 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -103,9 +103,6 @@ class Config: archive_file: str = "{config_path}/archive.log" """The path to the download archive file.""" - manual_archive: str = "{config_path}/archive.manual.log" - """The path to the manual archive file.""" - apprise_config: str = "{config_path}/apprise.yml" """The path to the Apprise configuration file.""" @@ -424,7 +421,7 @@ class Config: LOG.info(f"Creating archive file '{archive_file}'.") archive_file.touch(exist_ok=True) - LOG.info(f"keep archive option is enabled. Using archive file '{archive_file}'.") + LOG.info(f"keep archive option is enabled. Using archive file '{archive_file}' by default.") self._ytdlp_cli_mutable += f"\n--download-archive {archive_file.as_posix()!s}" if self.temp_keep: @@ -519,6 +516,16 @@ class Config: return "production" == self.app_env def get_ytdlp_args(self) -> dict: + """ + Get the yt-dlp command line options as a dictionary. + + Returns: + dict: The yt-dlp command line options. + + Todo: + Rename to get_ytdlp_opts() to match the system. + + """ try: return arg_converter(args=self._ytdlp_cli_mutable, level=True) except Exception as e: @@ -537,6 +544,7 @@ class Config: ytdlp_args = self.get_ytdlp_args() + # TODO: this doesn't make sense, as each item might have it's own archive file or none at all. if not data.get("keep_archive", False) and ytdlp_args.get("download_archive", None): data["keep_archive"] = True @@ -544,7 +552,7 @@ class Config: return data @staticmethod - def _ytdlp_version(): + def _ytdlp_version() -> str: try: from yt_dlp.version import __version__ as YTDLP_VERSION diff --git a/app/library/task_handlers/youtube.py b/app/library/task_handlers/youtube.py index 97552e2d..38ba01ed 100644 --- a/app/library/task_handlers/youtube.py +++ b/app/library/task_handlers/youtube.py @@ -1,16 +1,14 @@ import asyncio import logging import re -from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any 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 is_downloaded -from app.library.YTDLPOpts import YTDLPOpts +from app.library.Utils import archive_read, get_archive_id if TYPE_CHECKING: from app.library.Download import Download @@ -29,16 +27,14 @@ class YoutubeHandler: failure_count: dict[str, int] = {} FEED = "https://www.youtube.com/feeds/videos.xml?{type}={id}" - FEED_PLAYLIST = "https://www.youtube.com/feeds/videos.xml?playlist_id={id}" CHANNEL_REGEX = re.compile(r"^https?://(?:www\.)?youtube\.com/(?:channel/(?PUC[0-9A-Za-z_-]{22})|)/?$") PLAYLIST_REGEX = re.compile(r"^https?://(?:www\.)?youtube\.com/(?:playlist\?list=(?P[A-Za-z0-9_-]+)|).*$") @staticmethod - def can_handle(task: Task, config: Config) -> bool: - has, _ = YoutubeHandler.has_archive(task, config) - if not has: + 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.") return False @@ -58,24 +54,24 @@ class YoutubeHandler: queue (DownloadQueue): The download queue instance. """ - archive_file, params = YoutubeHandler.has_archive(task, config) - if not archive_file: - LOG.error(f"Task '{task.id}: {task.name}' does not have an archive file configured.") + 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.") return import httpx from defusedxml.ElementTree import fromstring - parsed = YoutubeHandler.parse(task.url) + parsed: dict[str, str] | None = YoutubeHandler.parse(task.url) if not parsed: LOG.error(f"Cannot parse '{task.id}: {task.name}' URL: {task.url}") return - feed_url = YoutubeHandler.FEED.format(type=parsed["type"], id=parsed["id"]) + feed_url: str = YoutubeHandler.FEED.format(type=parsed["type"], id=parsed["id"]) LOG.debug(f"Fetching '{task.id}: {task.name}' feed.") - opts = { - "proxy": params.get("proxy", None), + opts: dict[str, Any] = { + "proxy": params.get("proxy"), "headers": { "User-Agent": params.get( "user_agent", @@ -84,10 +80,22 @@ class YoutubeHandler: }, } + try: + from httpx_curl_cffi import AsyncCurlTransport, CurlOpt + + opts["transport"] = AsyncCurlTransport( + impersonate="chrome", + default_headers=True, + curl_options={CurlOpt.FRESH_CONNECT: True}, + ) + opts.pop("headers", None) + except Exception: + pass + items: list = [] async with httpx.AsyncClient(**opts) as client: - response = await client.request(method="GET", url=feed_url, timeout=120) + response: httpx.Response = await client.request(method="GET", url=feed_url, timeout=120) response.raise_for_status() root = fromstring(response.text) @@ -95,28 +103,37 @@ class YoutubeHandler: for entry in root.findall("atom:entry", ns): vid_elem = entry.find("yt:videoId", ns) + vid = 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.") + continue + title_elem = entry.find("atom:title", ns) pub_elem = entry.find("atom:published", ns) - vid = vid_elem.text if vid_elem is not None else "" title = title_elem.text if title_elem is not None else "" published = pub_elem.text if pub_elem is not None else "" - items.append( - { - "id": vid, - "url": f"https://www.youtube.com/watch?v={vid}", - "title": title, - "published": published, - } - ) + url: str = f"https://www.youtube.com/watch?v={vid}" + idDict = get_archive_id(url=url) + if not (archive_id := idDict.get("archive_id")): + LOG.warning(f"Item '{title}' does not have a valid archive ID. URL: {url}") + 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}") return filtered: list = [] + + downloaded: list[str] = archive_read( + file=archive_file, + ids=[item["archive_id"] for item in items if item["id"] not in YoutubeHandler.queued_ids], + ) + for item in items: - status, _ = is_downloaded(archive_file, url=item["url"]) - if status is True or item["id"] in YoutubeHandler.queued_ids: + if item["archive_id"] in downloaded: + YoutubeHandler.queued_ids.add(item["id"]) continue if queue.queue.exists(url=item["url"]): @@ -127,13 +144,13 @@ class YoutubeHandler: try: done: Download = queue.done.get(url=item["url"]) if "error" != done.info.status: - LOG.debug(f"Item '{item['id']}' exists in the download history.") + 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["id"]) + YoutubeHandler.queued_ids.add(item) filtered.append(item) if len(filtered) < 1: @@ -142,20 +159,21 @@ class YoutubeHandler: LOG.info(f"Found '{len(filtered)}' new items from '{task.id}: {task.name}' feed.") - preset: str = str(task.preset or config.default_preset) - folder: str = task.folder if task.folder else "" - template: str = task.template if task.template else "" - cli: str = task.cli if task.cli else "" + rItem: Item = Item.format( + { + "url": feed_url, + "preset": str(task.preset or config.default_preset), + "folder": task.folder if task.folder else "", + "template": task.template if task.template else "", + "cli": task.cli if task.cli else "", + "extras": {"source_task": task.id}, + } + ) try: await asyncio.gather( *[ - notify.emit( - Events.ADD_URL, - data=Item.format( - {"url": item["url"], "preset": preset, "folder": folder, "template": template, "cli": cli} - ).serialize(), - ) + notify.emit(Events.ADD_URL, data=rItem.new_with({"url": item["url"]}).serialize()) for item in filtered ] ) @@ -163,30 +181,6 @@ class YoutubeHandler: LOG.error(f"Error while adding items from '{task.id}: {task.name}'. {e!s}") return - @staticmethod - def has_archive(task: Task, config: Config) -> tuple[Path | None, dict]: - archive_file: Path | None = Path(config.archive_file) if config.keep_archive else None - params: YTDLPOpts = YTDLPOpts.get_instance() - - if task.preset: - params.preset(name=task.preset) - - if task.cli: - params.add_cli(task.cli, from_user=True) - - params = params.get_all() - if user_archive_file := params.get("download_archive", None): - archive_file = Path(user_archive_file) - - if not archive_file: - return (None, params) - - if not archive_file.exists(): - archive_file.parent.mkdir(parents=True, exist_ok=True) - archive_file.touch(exist_ok=True) - - return (archive_file, params) - @staticmethod def parse(url: str) -> dict[str, str] | None: """ diff --git a/app/routes/api/archive.py b/app/routes/api/archive.py index 2d425e74..dabb2044 100644 --- a/app/routes/api/archive.py +++ b/app/routes/api/archive.py @@ -1,17 +1,14 @@ import logging -from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING -import anyio from aiohttp import web from aiohttp.web import Request, Response -from app.library.config import Config from app.library.Download import Download from app.library.DownloadQueue import DownloadQueue from app.library.router import route -from app.library.Utils import is_downloaded, remove_from_archive +from app.library.Utils import archive_add, archive_delete, archive_read, get_archive_id from app.library.YTDLPOpts import YTDLPOpts if TYPE_CHECKING: @@ -20,86 +17,8 @@ if TYPE_CHECKING: LOG: logging.Logger = logging.getLogger(__name__) -@route("DELETE", r"api/archive/{id}", "archive.remove") -async def archive_remove(request: Request, queue: DownloadQueue, config: Config) -> Response: - """ - Remove an item from the archive. - - Args: - request (Request): The request object. - queue (DownloadQueue): The download queue instance. - config (Config): The configuration instance. - - Returns: - Response: The response object. - - """ - item = None - try: - data: dict | None = await request.json() - except Exception: - data = {} - - title: str = "" - - url: str | None = data.get("url") - - if not url: - id: str = request.match_info.get("id") - if not id: - return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code) - - try: - item: Download | None = queue.done.get_by_id(id) - if not item: - return web.json_response(data={"error": f"item '{id}' not found."}, status=web.HTTPNotFound.status_code) - - url = item.info.url - title = f" '{item.info.title}'" - except KeyError: - return web.json_response(data={"error": f"item '{id}' not found."}, status=web.HTTPNotFound.status_code) - - if config.manual_archive: - remove_from_archive(archive_file=Path(config.manual_archive), url=url) - - archive_file: Path | None = Path(config.archive_file) if config.keep_archive else None - if item: - params: YTDLPOpts = YTDLPOpts.get_instance().preset(name=item.info.preset) - if item.info.cli: - params.add_cli(item.info.cli, from_user=True) - - params = params.get_all() - if user_file := params.get("download_archive", None): - archive_file = Path(user_file) - - if not archive_file: - return web.json_response( - data={ - "error": "Archive file is not configured." if not config.keep_archive else "Archive file is not set." - }, - status=web.HTTPBadRequest.status_code, - ) - - if not archive_file.exists(): - return web.json_response( - data={"error": f"Archive file '{archive_file}' does not exist."}, - status=web.HTTPNotFound.status_code, - ) - - if not remove_from_archive(archive_file=archive_file, url=url): - return web.json_response( - data={"error": f"item{title} not found in '{archive_file}' archive."}, - status=web.HTTPNotFound.status_code, - ) - - return web.json_response( - data={"message": f"item{title} removed from '{archive_file}' archive."}, - status=web.HTTPOk.status_code, - ) - - @route("POST", r"api/archive/{id}", "archive.item") -async def archive_item(request: Request, queue: DownloadQueue, config: Config): +async def archive_item(request: Request, queue: DownloadQueue): """ Manually mark an item as archived. @@ -112,57 +31,118 @@ async def archive_item(request: Request, queue: DownloadQueue, config: Config): Response: The response object. """ - id: str = request.match_info.get("id") - - if not id: + if not (id := request.match_info.get("id")): return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code) try: item: Download | None = queue.done.get_by_id(id) if not item: - return web.json_response(data={"error": "item not found."}, status=web.HTTPNotFound.status_code) + return web.json_response(data={"error": f"item '{id}' not found."}, status=web.HTTPNotFound.status_code) except KeyError: - return web.json_response(data={"error": "item not found."}, status=web.HTTPNotFound.status_code) + return web.json_response(data={"error": f"item '{id}' not found."}, status=web.HTTPNotFound.status_code) - if config.manual_archive: - manual_archive = Path(config.manual_archive) - if manual_archive.exists(): - exists, idDict = is_downloaded(manual_archive, item.info.url) - if exists is False and idDict.get("archive_id"): - async with await anyio.open_file(manual_archive, "a") as f: - await f.write(f"{idDict['archive_id']} - at: {datetime.now(UTC).isoformat()}\n") + params: dict = item.get_ytdlp_opts().get_all() - params: YTDLPOpts = YTDLPOpts.get_instance().preset(name=item.info.preset) - if item.info.cli: - params.add_cli(item.info.cli, from_user=True) + if not (archive_file := params.get("download_archive")): + return web.json_response( + data={"error": f"item '{item.info.title}' does not have an archive file."}, + status=web.HTTPBadRequest.status_code, + ) - params = params.get_all() + idDict = get_archive_id(url=item.info.url) + if not (archive_id := idDict.get("archive_id")): + return web.json_response( + data={"error": f"item '{item.info.title}' does not have an archive ID."}, + status=web.HTTPBadRequest.status_code, + ) + + if len(archive_read(archive_file, [archive_id])) > 0: + return web.json_response( + data={"error": f"item '{item.info.title}' already archived."}, + status=web.HTTPConflict.status_code, + ) + + archive_add(archive_file, [archive_id]) + + return web.json_response( + data={"message": f"item '{item.info.title}' archived."}, + status=web.HTTPOk.status_code, + ) + + +@route("DELETE", r"api/archive/{id}", "archive.remove") +async def archive_remove(request: Request, queue: DownloadQueue) -> Response: + """ + Remove an item from the archive. + + Args: + request (Request): The request object. + queue (DownloadQueue): The download queue instance. + + Returns: + Response: The response object. + + """ + item = None + try: + data: dict | None = await request.json() + except Exception: + data = {} + + title: str = "" + url: str | None = data.get("url") + id: str | None = request.match_info.get("id") + preset: str | None = data.get("preset",) + + if not id and not url: + return web.json_response(data={"error": "id or url is required."}, status=web.HTTPBadRequest.status_code) + + if id: + try: + item: Download | None = queue.done.get_by_id(id) + if not item: + return web.json_response(data={"error": f"item '{id}' not found."}, status=web.HTTPNotFound.status_code) + + url = item.info.url + title = f" '{item.info.title}'" + params = item.get_ytdlp_opts().get_all() + except KeyError: + return web.json_response(data={"error": f"item '{id}' not found."}, status=web.HTTPNotFound.status_code) + else: + params: YTDLPOpts = YTDLPOpts.get_instance() + if preset := data.get("preset"): + params = params.preset(name=preset) + + params = params.get_all() + + if not (archive_file := params.get("download_archive", None)): + return web.json_response( + data={"error": "Archive file is not configured."}, + status=web.HTTPBadRequest.status_code, + ) + + archive_file = Path(archive_file) - user_file: str | None = params.get("download_archive", None) - archive_file: Path = Path(user_file) if user_file else Path(config.archive_file) if not archive_file.exists(): return web.json_response( data={"error": f"Archive file '{archive_file}' does not exist."}, status=web.HTTPNotFound.status_code, ) - exists, idDict = is_downloaded(archive_file, item.info.url) - item_id: str | None = idDict.get("archive_id") - if not item_id: + idDict = get_archive_id(url=url) + if not (archive_id := idDict.get("archive_id")): return web.json_response( - data={"error": "item does not have an archive ID."}, status=web.HTTPBadRequest.status_code + data={"error": "item does not have an archive ID."}, + status=web.HTTPBadRequest.status_code, ) - if exists is True: + if not archive_delete(archive_file, [archive_id]): return web.json_response( - data={"error": f"item '{item_id}' already archived in file '{archive_file}'."}, - status=web.HTTPConflict.status_code, + data={"error": f"item{title} not found in '{archive_file}' archive."}, + status=web.HTTPNotFound.status_code, ) - async with await anyio.open_file(archive_file, "a") as f: - await f.write(f"{item_id}\n") - return web.json_response( - data={"message": f"item '{item_id}' archived in file '{archive_file}'."}, + data={"message": f"item{title} removed from '{archive_file}' archive."}, status=web.HTTPOk.status_code, ) diff --git a/app/routes/api/history.py b/app/routes/api/history.py index b9dcc4f6..aba91bad 100644 --- a/app/routes/api/history.py +++ b/app/routes/api/history.py @@ -13,8 +13,7 @@ from app.library.Events import EventBus, Events from app.library.ItemDTO import Item from app.library.Presets import Preset, Presets from app.library.router import route -from app.library.Utils import is_downloaded -from app.library.YTDLPOpts import YTDLPOpts +from app.library.Utils import archive_read if TYPE_CHECKING: from library.Download import Download @@ -101,13 +100,9 @@ async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder, co return web.json_response(data={"error": "item has no info."}, status=web.HTTPNotFound.status_code) is_archived = False - params: YTDLPOpts = YTDLPOpts.get_instance().preset(name=item.info.preset) - if item.info.cli: - params.add_cli(item.info.cli, from_user=True) - - params = params.get_all() - if archive_file := params.get("download_archive"): - is_archived, _ = is_downloaded(archive_file, item.info.url) + params: dict = item.get_ytdlp_opts().get_all() + if (archive_file := params.get("download_archive")) and (archive_id := item.get_archive_id()): + is_archived: bool = len(archive_read(archive_file, [archive_id])) > 0 info = { **item.info.serialize(), diff --git a/app/routes/api/yt_dlp.py b/app/routes/api/yt_dlp.py index 2643b829..1b56a0db 100644 --- a/app/routes/api/yt_dlp.py +++ b/app/routes/api/yt_dlp.py @@ -1,12 +1,9 @@ -import asyncio import json import logging import time from collections import OrderedDict -from pathlib import Path from typing import Any -import anyio from aiohttp import web from aiohttp.web import Request, Response @@ -14,7 +11,14 @@ from app.library.cache import Cache from app.library.config import Config from app.library.Presets import Presets from app.library.router import route -from app.library.Utils import REMOVE_KEYS, arg_converter, extract_info, is_downloaded, validate_url +from app.library.Utils import ( + REMOVE_KEYS, + archive_read, + arg_converter, + extract_info, + get_archive_id, + validate_url, +) from app.library.YTDLPOpts import YTDLPOpts LOG: logging.Logger = logging.getLogger(__name__) @@ -200,8 +204,10 @@ async def get_info(request: Request, cache: Cache, config: Config) -> Response: } is_archived = False - if user_file := ytdlp_opts.get("download_archive"): - is_archived, _ = is_downloaded(user_file, url) + if (archive_file := ytdlp_opts.get("download_archive")) and ( + archive_id := get_archive_id(url=url).get("archive_id") + ): + is_archived: bool = len(archive_read(archive_file, [archive_id])) > 0 data["is_archived"] = is_archived @@ -223,87 +229,6 @@ async def get_info(request: Request, cache: Cache, config: Config) -> Response: ) -@route("GET", "api/yt-dlp/archive/recheck/", "archive_recheck") -async def archive_recheck(cache: Cache) -> Response: - """ - Recheck the manual archive entries. - - Args: - cache (Cache): The cache instance. - - Returns: - Response: The response object - - """ - config: Config = Config.get_instance() - if not config.manual_archive: - return web.json_response(data={"error": "Manual archive is not enabled."}, status=web.HTTPNotFound.status_code) - - manual_archive = Path(config.manual_archive) - if not manual_archive.exists(): - return web.json_response( - data={"error": "Manual archive file not found.", "file": manual_archive}, - status=web.HTTPNotFound.status_code, - ) - - tasks: list = [] - response: list = [] - - def info_wrapper(id: str, url: str) -> tuple[str, dict]: - try: - return ( - id, - extract_info( - config={ - "proxy": config.get_ytdlp_args().get("proxy", None), - "simulate": True, - "dump_single_json": True, - }, - url=url, - no_archive=True, - ), - ) - except Exception as e: - return (id, {"error": str(e)}) - - async with await anyio.open_file(manual_archive) as f: - # line format is "youtube ID - at: ISO8601" - async for line in f: - line = line.strip() - - if not line or not line.startswith("youtube"): - continue - - id = line.split(" ")[1].strip() - - if not id: - continue - - url = f"https://www.youtube.com/watch?v={id}" - key = cache.hash(id) - - if cache.has(key): - data = cache.get(key) - response.append({id: bool(data.get("id", None)) if isinstance(data, dict) else False}) - continue - - tasks.append( - asyncio.get_event_loop().run_in_executor(None, lambda i=id, url=url: info_wrapper(id=i, url=url)) - ) - - if len(tasks) > 0: - results = await asyncio.gather(*tasks) - for data in results: - if not data: - continue - - id, info = data - cache.set(key=cache.hash(id), value=info, ttl=3600 * 6) - response.append({id: bool(data.get("id", None)) if isinstance(data, dict) else False}) - - return web.json_response(data=response, status=web.HTTPOk.status_code) - - @route("GET", "api/yt-dlp/options/", "get_options") async def get_options() -> Response: """ diff --git a/app/routes/socket/history.py b/app/routes/socket/history.py index 1202deb6..3904d218 100644 --- a/app/routes/socket/history.py +++ b/app/routes/socket/history.py @@ -1,16 +1,11 @@ import logging import time -from datetime import UTC, datetime -from pathlib import Path -import anyio - -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 from app.library.router import RouteType, route -from app.library.Utils import is_downloaded +from app.library.Utils import archive_add, archive_read, get_archive_id from app.library.YTDLPOpts import YTDLPOpts LOG: logging.Logger = logging.getLogger(__name__) @@ -80,7 +75,7 @@ async def item_delete(queue: DownloadQueue, notify: EventBus, sid: str, data: di @route(RouteType.SOCKET, "archive_item", "archive_item") -async def archive_item(config: Config, data: dict): +async def archive_item(data: dict): if not isinstance(data, dict) or "url" not in data: return @@ -94,40 +89,23 @@ async def archive_item(config: Config, data: dict): params = params.get_all() - file: str = params.get("download_archive", None) - - if not file: + if not (file := params.get("download_archive")): return - exists, idDict = is_downloaded(file, data["url"]) - if exists or "archive_id" not in idDict or idDict["archive_id"] is None: + idDict = get_archive_id(url=data["url"]) + if not idDict.get("archive_id"): + LOG.warning(f"URL '{data['url']}' does not have an archive ID.") return - async with await anyio.open_file(file, "a") as f: - await f.write(f"{idDict['archive_id']}\n") + archive_id: str = idDict.get("archive_id") - manual_archive: str = config.manual_archive - if not manual_archive: + if len(archive_read(file, [archive_id])) > 0: + LOG.info(f"URL '{data['url']}' already archived.") return - manual_archive = Path(manual_archive) + archive_add(file, [archive_id]) - if not manual_archive.exists(): - manual_archive.touch(exist_ok=True) - - previouslyArchived = False - async with await anyio.open_file(manual_archive) as f: - async for line in f: - if idDict["archive_id"] in line: - previouslyArchived = True - break - - if not previouslyArchived: - async with await anyio.open_file(manual_archive, "a") as f: - await f.write(f"{idDict['archive_id']} - at: {datetime.now(UTC).isoformat()}\n") - LOG.info(f"Archiving url '{data['url']}' with id '{idDict['archive_id']}'.") - else: - LOG.info(f"URL '{data['url']}' with id '{idDict['archive_id']}' already archived.") + LOG.info(f"Archiving url '{data['url']}' with id '{archive_id}'.") @route(RouteType.SOCKET, "item_start", "item_start")