danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

major refactor of archive_file handling

Browse source
This commit is contained in:
arabcoders 2025-08-29 01:34:38 +03:00
parent 29b7748b8a
commit deed79b8d2
14 changed files with 439 additions and 521 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.vscode/settings.json vendored
View file

@ -112,6 +112,7 @@
"timespec",
"tmpfilename",
"ungroup",
"unmark",
"unnegated",
"unpickleable",
"Unraid",

22
API.md
View file

@ -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

1
FAQ.md
View file

@ -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` |

36
app/library/Download.py
View file

@ -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()

37
app/library/DownloadQueue.py
View file

@ -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.

16
app/library/ItemDTO.py
View file

@ -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."""

101
app/library/Tasks.py
View file

@ -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):

248
app/library/Utils.py
View file

@ -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

18
app/library/config.py
View file

@ -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

120
app/library/task_handlers/youtube.py
View file

@ -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/(?P<id>UC[0-9A-Za-z_-]{22})|)/?$")
PLAYLIST_REGEX = re.compile(r"^https?://(?:www\.)?youtube\.com/(?:playlist\?list=(?P<id>[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:
"""

204
app/routes/api/archive.py
View file

@ -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,
)

13
app/routes/api/history.py
View file

@ -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(),

99
app/routes/api/yt_dlp.py
View file

@ -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:
"""

44
app/routes/socket/history.py
View file

@ -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")