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

simplified the codebase and removed all backwards compat flags and checks.

Browse source
This commit is contained in:
arabcoders 2025-04-01 22:56:28 +03:00
parent 9b4d86041a
commit f7101f6a38
10 changed files with 288 additions and 327 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
app/library/DataStore.py
View file

@ -8,6 +8,7 @@ from sqlite3 import Connection
from .config import Config
from .Download import Download
from .ItemDTO import ItemDTO
from .Utils import clean_item
class DataStore:
@ -68,7 +69,7 @@ class DataStore:
for row in cursor:
rowDate = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007
data: dict = json.loads(row["data"])
data, _ = clean_item(json.loads(row["data"]), keys=ItemDTO.removed_fields())
key: str = data.pop("_id")
item: ItemDTO = ItemDTO(**data)
item._id = key

4
app/library/Download.py
View file

@ -48,7 +48,6 @@ class Download:
temp_dir: str = None
template: str = None
template_chapter: str = None
ytdl_opts: dict = None
info: ItemDTO = None
default_ytdl_opts: dict = None
debug: bool = False
@ -92,7 +91,6 @@ class Download:
self.template = info.template
self.template_chapter = info.template_chapter
self.preset = info.preset
self.ytdl_opts = info.config if info.config else {}
self.info = info
self.id = info._id
self.default_ytdl_opts = config.ytdl_options
@ -147,7 +145,7 @@ class Download:
YTDLPOpts.get_instance()
.preset(self.preset)
.add({"break_on_existing": True})
.add(self.ytdl_opts, from_user=True)
.add_cli(self.info.cli, from_user=True)
.add(
{
"color": "no_color",

179
app/library/DownloadQueue.py
View file

@ -1,5 +1,6 @@
import asyncio
import functools
import glob
import json
import logging
import os
@ -7,6 +8,7 @@ import time
import uuid
from datetime import UTC, datetime
from email.utils import formatdate
from pathlib import Path
from sqlite3 import Connection
import anyio
@ -18,10 +20,10 @@ from .config import Config
from .DataStore import DataStore
from .Download import Download
from .Events import EventBus, Events
from .ItemDTO import ItemDTO
from .ItemDTO import Item, ItemDTO
from .Presets import Presets
from .Singleton import Singleton
from .Utils import arg_converter, calc_download_path, extract_info, is_downloaded, strip_newline
from .Utils import arg_converter, calc_download_path, extract_info, is_downloaded
from .YTDLPOpts import YTDLPOpts
LOG = logging.getLogger("DownloadQueue")
@ -168,45 +170,23 @@ class DownloadQueue(metaclass=Singleton):
except Exception as e:
LOG.error(f"Failed to cancel downloads. {e!s}")
async def __add_entry(
self,
entry: dict,
preset: str,
folder: str,
config: dict | None = None,
cookies: str = "",
template: str = "",
extras: dict | None = None,
cli: str = "",
already=None,
):
async def __add_entry(self, entry: dict, item: Item, already=None):
"""
Add an entry to the download queue.
Args:
entry (dict): The entry to add to the download queue.
preset (str): The preset to use for the download.
folder (str): The folder to save the download to.
config (dict): The yt-dlp configuration to use for the download.
cookies (str): The cookies to use for the download.
template (str): The output template to use for the download.
extras (dict): The extra information to add to the download.
cli (str): The yt-dlp command line options to use for the download.
item (Item): The item to add to the download queue.
already (set): The set of already downloaded items.
Returns:
dict: The status of the operation.
"""
if not config:
config = {}
if not extras:
extras = {}
else:
for key in extras.copy():
if item.has_extras():
for key in item.extras.copy():
if not str(key).startswith("playlist"):
extras.pop(key, None)
item.extras.pop(key, None)
if not entry:
return {"status": "error", "msg": "Invalid/empty data was given."}
@ -239,13 +219,7 @@ class DownloadQueue(metaclass=Singleton):
results.append(
await self.__add_entry(
entry=etr,
preset=preset,
folder=folder,
config=config,
cookies=cookies,
template=template,
extras=extras,
cli=cli,
item=item.new_with(url=etr.get("url") or etr.get("webpage_url")),
already=already,
)
)
@ -291,7 +265,7 @@ class DownloadQueue(metaclass=Singleton):
is_live = bool(entry.get("is_live") or live_in or entry.get("live_status") in live_status)
try:
download_dir = calc_download_path(base_path=self.config.download_path, folder=folder)
download_dir = calc_download_path(base_path=self.config.download_path, folder=item.folder)
except Exception as e:
LOG.exception(e)
return {"status": "error", "msg": str(e)}
@ -299,31 +273,30 @@ class DownloadQueue(metaclass=Singleton):
fields: tuple = ("uploader", "channel", "thumbnail")
for field in fields:
if entry.get(field):
extras[field] = entry.get(field)
item.extras[field] = entry.get(field)
for key in entry:
if isinstance(key, str) and key.startswith("playlist") and entry.get(key):
extras[key] = entry.get(key)
item.extras[key] = entry.get(key)
dl = ItemDTO(
id=str(entry.get("id")),
title=str(entry.get("title")),
url=str(entry.get("webpage_url") or entry.get("url")),
preset=preset,
folder=folder,
preset=item.preset,
folder=item.folder,
download_dir=download_dir,
temp_dir=self.config.temp_path,
cookies=cookies,
config=config,
template=template if template else self.config.output_template,
cookies=item.cookies,
template=item.template if item.template else self.config.output_template,
template_chapter=self.config.output_template_chapter,
datetime=formatdate(time.time()),
error=error,
is_live=is_live,
live_in=live_in,
options=options,
cli=cli,
extras=extras,
cli=item.cli,
extras=item.extras,
)
dlInfo: Download = Download(info=dl, info_dict=entry)
@ -347,44 +320,16 @@ class DownloadQueue(metaclass=Singleton):
return {"status": "ok"}
if eventType.startswith("url"):
return await self.add(
url=str(entry.get("url")),
preset=preset,
folder=folder,
config=config,
cookies=cookies,
template=template,
extras=extras,
cli=cli,
already=already,
)
return await self.add(item=item.new_with(url=entry.get("url")), already=already)
return {"status": "error", "msg": f'Unsupported resource "{eventType}"'}
async def add(
self,
url: str,
preset: str,
folder: str,
config: dict | None = None,
cookies: str = "",
template: str = "",
cli: str = "",
extras: dict | None = None,
already=None,
):
async def add(self, item: Item, already: set | None = None):
"""
Add an item to the download queue.
Args:
url (str): The url to be added to the queue.
preset (str): The preset to be used for the download.
folder (str): The folder to save the download to.
config (dict): The yt-dlp config to be used for the download.
cookies (str): The cookies to be used for the download.
template (str): The template to be used for the download.
cli (str): The yt-dlp cli options to be used for the download.
extras (dict): Extra data to be added to the download
item (Item): The item to be added to the queue.
already (set): Set of already downloaded items.
Returns:
@ -392,37 +337,29 @@ class DownloadQueue(metaclass=Singleton):
{ "status": "text" }
"""
_preset = Presets.get_instance().get(preset)
_preset = Presets.get_instance().get(item.preset)
config = config if config else {}
if cli:
if item.has_cli():
try:
config = arg_converter(args=cli, level=True)
config = arg_converter(args=item.cli, level=True)
except Exception as e:
LOG.error(f"Invalid cli options '{cli}'. {e!s}")
return {"status": "error", "msg": f"Invalid cli options '{cli}'. {e!s}"}
folder = str(folder) if folder else ""
if not extras:
extras = {}
LOG.error(f"Invalid cli options '{item.cli}'. {e!s}")
return {"status": "error", "msg": f"Invalid cli options '{item.cli}'. {e!s}"}
if _preset:
if _preset.folder and not folder:
folder = _preset.folder
if _preset.folder and not item.folder:
item.folder = _preset.folder
if _preset.template and not template:
template = _preset.template
if _preset.template and not item.template:
item.template = _preset.template
if _preset.cookies and not cookies:
cookies = _preset.cookies
if _preset.cookies and not item.cookies:
item.cookies = _preset.cookies
filePath = calc_download_path(base_path=self.config.download_path, folder=folder)
yt_conf = {}
cookie_file = os.path.join(self.config.temp_path, f"c_{uuid.uuid4().hex}.txt")
LOG.info(
f"Adding 'URL: {url}' to 'Folder: {filePath}' with 'Preset: {preset}' 'Naming: {template}', 'Cookies: {len(cookies)}/chars' 'CLI: {strip_newline(cli)}' 'Extras: {extras}'."
)
LOG.info(f"Adding '{item.__repr__()}'.")
if isinstance(config, str):
try:
@ -433,21 +370,21 @@ class DownloadQueue(metaclass=Singleton):
already = set() if already is None else already
if url in already:
LOG.warning(f"Recursion detected with url '{url}' skipping.")
if item.url in already:
LOG.warning(f"Recursion detected with url '{item.url}' skipping.")
return {"status": "ok"}
already.add(url)
already.add(item.url)
try:
downloaded, id_dict = self._is_downloaded(url)
downloaded, id_dict = self._is_downloaded(item.url)
if downloaded is True and id_dict:
message = f"This url with ID '{id_dict.get('id')}' has been downloaded already and recorded in archive."
LOG.info(message)
return {"status": "error", "msg": message}
started = time.perf_counter()
LOG.debug(f"extract_info: checking {url=}")
LOG.debug(f"extract_info: checking '{item.url}'.")
logs = []
@ -456,13 +393,13 @@ class DownloadQueue(metaclass=Singleton):
"func": lambda _, msg: logs.append(msg),
"level": logging.WARNING,
},
**YTDLPOpts.get_instance().preset(name=preset).add(config=config, from_user=True).get_all(),
**YTDLPOpts.get_instance().preset(name=item.preset).add(config=config, from_user=True).get_all(),
}
if cookies:
if item.cookies:
try:
async with await anyio.open_file(cookie_file, "w") as f:
await f.write(cookies)
await f.write(item.cookies)
yt_conf["cookiefile"] = f.name
except ValueError as e:
LOG.error(f"Failed to create cookie file for '{self.info.id}: {self.info.title}'. '{e!s}'.")
@ -473,7 +410,7 @@ class DownloadQueue(metaclass=Singleton):
functools.partial(
extract_info,
config=yt_conf,
url=url,
url=item.url,
debug=bool(self.config.ytdl_debug),
no_archive=False,
follow_redirect=True,
@ -486,7 +423,7 @@ class DownloadQueue(metaclass=Singleton):
return {"status": "error", "msg": "Unable to extract info." + "\n".join(logs)}
LOG.debug(
f"extract_info: for 'URL: {url}' is done in '{time.perf_counter() - started}'. Length: '{len(entry)}'."
f"extract_info: for 'URL: {item.url}' is done in '{time.perf_counter() - started}'. Length: '{len(entry)}/keys'."
)
except yt_dlp.utils.ExistingVideoReached as exc:
LOG.error(f"Video has been downloaded already and recorded in archive.log file. '{exc!s}'.")
@ -508,17 +445,7 @@ class DownloadQueue(metaclass=Singleton):
except Exception as e:
LOG.error(f"Failed to remove cookie file '{yt_conf['cookiefile']}'. {e!s}")
return await self.__add_entry(
entry=entry,
preset=preset,
folder=folder,
config=config,
cookies=cookies,
template=template,
already=already,
extras=extras,
cli=cli,
)
return await self.__add_entry(entry=entry, item=item, already=already)
async def cancel(self, ids: list[str]) -> dict[str, str]:
"""
@ -588,7 +515,7 @@ class DownloadQueue(metaclass=Singleton):
continue
itemRef: str = f"{id=} {item.info.id=} {item.info.title=}"
fileDeleted: bool = False
removed_files = 0
filename: str = ""
if remove_file and self.config.remove_files and "finished" == item.info.status:
@ -602,9 +529,13 @@ class DownloadQueue(metaclass=Singleton):
folder=filename,
create_path=False,
)
if realFile and os.path.exists(realFile):
os.remove(realFile)
fileDeleted = True
rf = Path(realFile)
if rf.is_file() and rf.exists():
for f in rf.parent.glob(f"{glob.escape(rf.stem)}.*"):
if f.is_file() and f.exists() and not f.name.startswith("."):
removed_files += 1
LOG.debug(f"Removing '{itemRef}' local file '{f.name}'.")
os.remove(f)
else:
LOG.warning(f"Failed to remove '{itemRef}' local file '{filename}'. File not found.")
except Exception as e:
@ -614,8 +545,8 @@ class DownloadQueue(metaclass=Singleton):
await self._notify.emit(Events.CLEARED, data=item.info.serialize())
msg = f"Deleted completed download '{itemRef}'."
if fileDeleted and filename:
msg += f" and removed local file '{filename}'."
if removed_files > 0:
msg += f" and removed '{removed_files}' local files."
LOG.info(msg=msg)
status[id] = "ok"

8
app/library/HttpAPI.py
View file

@ -27,6 +27,7 @@ from .DownloadQueue import DownloadQueue
from .encoder import Encoder
from .Events import EventBus, Events, message
from .ffprobe import ffprobe
from .ItemDTO import Item
from .M3u8 import M3u8
from .Notifications import Notification, NotificationEvents
from .Playlist import Playlist
@ -684,7 +685,7 @@ class HttpAPI(Common):
data["preset"] = preset
try:
status = await self.add(**self.format_item(data))
status = await self.add(item=Item.format(data))
except ValueError as e:
return web.json_response(data={"status": False, "message": str(e)}, status=web.HTTPBadRequest.status_code)
@ -711,15 +712,16 @@ class HttpAPI(Common):
if isinstance(data, dict):
data = [data]
items = []
for item in data:
try:
self.format_item(item)
items.append(Item.format(item))
except ValueError as e:
return web.json_response(data={"error": str(e), "data": item}, status=web.HTTPBadRequest.status_code)
return web.json_response(
data=await asyncio.wait_for(
fut=asyncio.gather(*[self.add(**self.format_item(item)) for item in data]),
fut=asyncio.gather(*[self.add(item=item) for item in items]),
timeout=None,
),
status=web.HTTPOk.status_code,

13
app/library/HttpSocket.py
View file

@ -17,6 +17,7 @@ from .config import Config
from .DownloadQueue import DownloadQueue
from .encoder import Encoder
from .Events import Event, EventBus, Events, error
from .ItemDTO import Item
from .Presets import Presets
from .Utils import is_downloaded
@ -79,7 +80,7 @@ class HttpSocket(Common):
self._notify.subscribe(
Events.ADD_URL,
lambda data, _, **kwargs: self.add(**self.format_item(data.data)), # noqa: ARG005
lambda data, _, **kwargs: self.add(item=Item.format(data.data)), # noqa: ARG005
f"{__class__.__name__}.socket_add_url",
)
@ -188,14 +189,16 @@ class HttpSocket(Common):
return
try:
item = self.format_item(data)
await self._notify.emit(
event=Events.STATUS,
data=await self.add(item=Item.format(data)),
to=sid,
)
except ValueError as e:
LOG.exception(e)
await self._notify.emit(Events.ERROR, data=error(str(e)), to=sid)
return
status = await self.add(**item)
await self._notify.emit(Events.STATUS, data=status, to=sid)
@ws_event
async def item_cancel(self, sid: str, id: str):
if not id:

214
app/library/ItemDTO.py
View file

@ -1,10 +1,171 @@
import json
import logging
import time
import uuid
from dataclasses import dataclass, field
from email.utils import formatdate
from typing import Any
LOG = logging.getLogger("ItemDTO")
@dataclass(kw_only=True)
class Item:
"""
This class represents a single download request for data transfer purposes.
"""
url: str
"""The URL of the item to be downloaded."""
preset: str = field(default_factory=lambda: Item._default_preset())
"""The preset to be used for the download."""
folder: str = ""
"""The folder to save the download to."""
cookies: str = ""
"""The cookies to be used for the download."""
template: str = ""
"""The template to be used for the download."""
cli: str = ""
"""The yt-dlp cli options to be used for the download."""
extras: dict = field(default_factory=dict)
"""Extra data to be added to the download."""
def serialize(self) -> dict:
return self.__dict__.copy()
def json(self) -> str:
return json.dumps(self.serialize(), default=lambda o: o.__dict__, sort_keys=True, indent=4)
def get(self, key: str, default: Any = None) -> Any:
return self.__dict__.get(key, default)
def has_extras(self) -> bool:
"""
Check if the item has any extras data associated with it.
Returns:
bool: True if the item has extras, False otherwise.
"""
return self.extras and len(self.extras) > 0
def has_cli(self) -> bool:
"""
Check if the item has any yt-dlp cli options associated with it.
Returns:
bool: True if the item has yt-dlp cli options, False otherwise.
"""
return self.cli and len(self.cli) > 2
def _default_preset() -> str:
from .config import Config
return Config.get_instance().default_preset
def new_with(self, **kwargs) -> "Item":
"""
Create a new instance of Item with the given parameters.
Args:
**kwargs: The parameters to be used for creating the new instance.
Returns:
Item: A new instance of Item with the given parameters.
"""
return Item.format({**self.serialize(), **kwargs})
def __repr__(self):
from .config import Config
from .Utils import calc_download_path, strip_newline
data = {}
for k, v in self.serialize().items():
if not v:
continue
if k == "cli":
data[k] = strip_newline(v)
elif k == "extras":
data[k] = f"{len(v)} items"
elif k == "cookies":
data[k] = f"{len(v)}/chars"
elif k == "folder":
data[k] = calc_download_path(base_path=Config.get_instance().download_path, folder=v)
else:
data[k] = v
items = " ".join(f'{k}="{v}", ' for k, v in data.items() if v)
return f"Item({items.strip(', ')})"
@staticmethod
def format(item: dict) -> "Item":
"""
Format the item to be added to the download queue.
Args:
item (dict): The item to be formatted.
Raises:
ValueError: If the url is not provided.
ValueError: If the yt-cli command line arguments are not valid.
Returns:
Item: The formatted item.
"""
url: str = item.get("url")
if not url or not isinstance(url, str):
msg = "url param is required."
raise ValueError(msg)
data = {
"url": url,
}
preset = item.get("preset")
if preset and isinstance(preset, str) and preset != Item._default_preset():
data["preset"] = preset
if item.get("folder") and isinstance(item.get("folder"), str):
data["folder"] = item.get("folder")
if item.get("cookies") and isinstance(item.get("cookies"), str):
data["cookies"] = item.get("cookies")
if item.get("template") and isinstance(item.get("template"), str):
data["template"] = item.get("template")
extras = item.get("extras")
if extras and isinstance(extras, dict) and len(extras) > 0:
data["extras"] = extras
cli = item.get("cli")
if cli and len(cli) > 2:
from .Utils import arg_converter
try:
removed_options = []
arg_converter(args=cli, level=True, removed_options=removed_options)
if len(removed_options) > 0:
LOG.warning("Removed the following options '%s' for '%s'.", ", ".join(removed_options), url)
data["cli"] = cli
except Exception as e:
msg = f"Failed to parse yt-dlp cli options. {e!s}"
raise ValueError(msg) from e
return Item(**data)
@dataclass(kw_only=True)
class ItemDTO:
@ -49,45 +210,8 @@ class ItemDTO:
speed: str | None = None
eta: str | None = None
# @Deprecated: These fields are deprecated and will be removed in the future.
thumbnail: str | None = None
ytdlp_cookies: str | None = None
ytdlp_config: dict = field(default_factory=dict)
output_template: str | None = None
output_template_chapter: str | None = None
config: dict = field(default_factory=dict)
def serialize(self) -> dict:
deprecated: tuple = (
"thumbnail",
"quality",
"format",
"ytdlp_cookies",
"ytdlp_config",
"output_template",
"output_template_chapter",
"config",
)
if self.thumbnail and "thumbnail" not in self.extras:
self.extras["thumbnail"] = self.thumbnail
mapper: dict = {
"cookies": "ytdlp_cookies",
"config": "ytdlp_config",
"template": "output_template",
"template_chapter": "output_template_chapter",
}
for k, v in mapper.items():
if not self.get(k) and self.get(v):
self.__dict__[k] = self.get(v)
dump = self.__dict__.copy()
for f in deprecated:
dump.pop(f, None)
return dump
return self.__dict__.copy()
def json(self) -> str:
return json.dumps(self.serialize(), default=lambda o: o.__dict__, sort_keys=True, indent=4)
@ -97,3 +221,17 @@ class ItemDTO:
def get_id(self) -> str:
return self._id
@staticmethod
def removed_fields() -> tuple:
"""Fields that once existed but are no longer used."""
return (
"thumbnail",
"quality",
"format",
"ytdlp_cookies",
"ytdlp_config",
"output_template",
"output_template_chapter",
"config",
)

6
app/library/Notifications.py
View file

@ -371,7 +371,9 @@ class Notification(metaclass=Singleton):
if "form" == target.request.type.lower():
reqBody["data"]["data"] = self._encoder.encode(reqBody["data"]["data"])
logging.getLogger("httpx").setLevel(logging.WARNING)
if not self._debug:
logging.getLogger("httpx").setLevel(logging.WARNING)
response = await self._client.request(**reqBody)
respData = {"url": target.request.url, "status": response.status_code, "text": response.text}
@ -404,7 +406,7 @@ class Notification(metaclass=Singleton):
data[k] = [self._deep_unpack(i) for i in v]
if isinstance(v, datetime):
data[k] = v.isoformat()
if isinstance(v, ItemDTO):
if isinstance(v, object) and hasattr(v, "serialize"):
data[k] = v.serialize()
return data

45
app/library/YTDLPOpts.py
View file

@ -36,6 +36,30 @@ class YTDLPOpts(metaclass=Singleton):
return YTDLPOpts._instance
def add_cli(self, args: str, from_user: int | bool = False) -> "YTDLPOpts":
"""
Prase and add yt-dlp cli options to the item options.
Args:
args (str): The cli options to add
from_user (bool): If the options are from the user
Returns:
YTDLPOpts: The instance of the class
"""
if not args or len(args) < 2 or not isinstance(args, str):
return self
removed_options = []
self._item_opts.update(arg_converter(args=args, level=from_user, removed_options=removed_options))
if len(removed_options) > 0:
LOG.warning("Removed the following options: '%s'.", ", ".join(removed_options))
return self
def add(self, config: dict, from_user: bool = False) -> "YTDLPOpts":
"""
Add the options to the item options.
@ -147,25 +171,4 @@ class YTDLPOpts(metaclass=Singleton):
if "format" in data and data["format"] in ["not_set", "default"]:
data["format"] = None
# @Deprecated - To be removed in future versions, All the checks below this line are deprecated.
if "daterange" in data and isinstance(data["daterange"], dict):
from yt_dlp.utils import DateRange
data["daterange"] = DateRange(data["daterange"]["start"], data["daterange"]["end"])
if "impersonate" in data and isinstance(data["impersonate"], str):
from yt_dlp.networking.impersonate import ImpersonateTarget
data["impersonate"] = ImpersonateTarget.from_str(data["impersonate"])
if (
"match_filter" in data
and isinstance(data["match_filter"], dict)
and "filters" in data["match_filter"]
and len(data["match_filter"]["filters"]) > 0
):
from yt_dlp.utils import match_filter_func
data["match_filter"] = match_filter_func(data["match_filter"]["filters"])
return data

98
app/library/common.py
View file

@ -1,10 +1,9 @@
import json
import logging
from .config import Config
from .DownloadQueue import DownloadQueue
from .encoder import Encoder
from .Utils import arg_converter
from .ItemDTO import Item
LOG = logging.getLogger("common")
@ -27,105 +26,16 @@ class Common:
config = config or Config.get_instance()
self.default_preset = config.default_preset
async def add(
self,
url: str,
preset: str,
folder: str,
cookies: str,
# @deprecated: config: dict,
config: dict,
template: str,
extras: dict | None = None,
cli: str = "",
) -> dict[str, str]:
async def add(self, item: Item) -> dict[str, str]:
"""
Add an item to the download queue.
Args:
url (str): The url to be added to the queue.
preset (str): The preset to be used for the download.
folder (str): The folder to save the download to.
cookies (str): The cookies to be used for the download.
config (dict): The yt-dlp config to be used for the download.
template (str): The template to be used for the download.
extras (dict): Extra data to be added to the download
cli (str): The yt-dlp cli options to be used for the download.
item (Item): The item to be added to the queue.
Returns:
dict[str, str]: The status of the download.
{ "status": "text" }
"""
if cookies and isinstance(cookies, dict):
cookies = self.encoder.encode(cookies)
return await self.queue.add(
url=url,
preset=preset if preset else self.config.default_preset,
folder=folder,
cookies=cookies,
config=config if isinstance(config, dict) else {},
template=template,
cli=cli,
extras=extras,
)
def format_item(self, item: dict) -> dict:
"""
Format the item to be added to the download queue.
Args:
item (dict): The item to be formatted.
Raises:
ValueError: If the url is not provided.
ValueError: If the yt-dlp config is not a valid json.
Returns:
dict: The formatted item
"""
url: str = item.get("url")
if not url:
msg = "url param is required."
raise ValueError(msg)
preset: str = str(item.get("preset", self.config.default_preset))
folder: str = str(item.get("folder")) if item.get("folder") else ""
cookies: str = str(item.get("cookies")) if item.get("cookies") else ""
template: str = str(item.get("template")) if item.get("template") else ""
extras = item.get("extras", {})
config = item.get("config")
if isinstance(config, str) and config:
try:
config = json.loads(config)
except Exception as e:
msg = f"Failed to parse json yt-dlp config for '{url}'. {e!s}"
raise ValueError(msg) from e
cli = item.get("cli")
if cli and len(cli) > 2:
try:
removed_options = []
arg_converter(args=cli, level=True, removed_options=removed_options)
if len(removed_options) > 0:
LOG.warning("Removed the following options '%s'.", ", ".join(removed_options))
config = {}
except Exception as e:
msg = f"Failed to parse yt-dlp cli options. {e!s}"
raise ValueError(msg) from e
return {
"url": url,
"preset": preset,
"folder": folder,
"cookies": cookies,
"config": config if isinstance(config, dict) else {},
"template": template,
"cli": cli if isinstance(cli, str) else "",
"extras": extras if isinstance(extras, dict) else {},
}
return await self.queue.add(item=item)

45
app/library/config.py
View file

@ -12,7 +12,7 @@ import coloredlogs
from dotenv import load_dotenv
from yt_dlp.version import __version__ as YTDLP_VERSION
from .Utils import REMOVE_KEYS, arg_converter, load_file, merge_dict
from .Utils import arg_converter
from .version import APP_VERSION
@ -321,11 +321,10 @@ class Config:
except Exception as e:
LOG.error(f"Error starting debugpy server at '0.0.0.0:{self.debugpy_port}'. {e}")
ytdlp_cli: str = os.path.join(self.config_path, "ytdlp.cli")
optsFile: str = os.path.join(self.config_path, "ytdlp.json")
if os.path.exists(ytdlp_cli) and os.path.getsize(ytdlp_cli) > 2:
LOG.info(f"Loading yt-dlp custom options from '{ytdlp_cli}'.")
with open(ytdlp_cli) as f:
opts_file: str = os.path.join(self.config_path, "ytdlp.cli")
if os.path.exists(opts_file) and os.path.getsize(opts_file) > 2:
LOG.info(f"Loading yt-dlp custom options from '{opts_file}'.")
with open(opts_file) as f:
ytdlp_cli_opts = f.read().strip()
if ytdlp_cli_opts:
try:
@ -343,41 +342,15 @@ class Config:
if len(removed_options) > 0:
LOG.warning(
"Removed the following options: '%s' from '%s'", ", ".join(removed_options), ytdlp_cli
"Removed the following options: '%s' from '%s'", ", ".join(removed_options), opts_file
)
except Exception as e:
msg = f"Failed to parse yt-dlp cli options from '{ytdlp_cli}'. '{e!s}'."
msg = f"Failed to parse yt-dlp cli options from '{opts_file}'. '{e!s}'."
raise ValueError(msg) from e
else:
LOG.warning(f"Empty yt-dlp custom options file '{ytdlp_cli}'.")
# @Deprecated - To be removed in future versions.
elif os.path.exists(optsFile) and os.path.getsize(optsFile) > 5:
LOG.warning("The JSON ytdlp.json options file is deprecated, please use 'ytdlp.cli' file instead.")
LOG.info(f"Loading yt-dlp custom options from '{optsFile}'.")
(opts, status, error) = load_file(optsFile, dict)
if not status:
LOG.error(f"Could not load yt-dlp custom options from '{optsFile}'. {error}")
sys.exit(1)
if isinstance(opts, dict):
bad_options = {k: v for d in REMOVE_KEYS for k, v in d.items()}
removed_options = []
for key in opts.copy():
if key not in bad_options:
continue
removed_options.append(bad_options[key])
opts.pop(key, None)
if len(removed_options) > 0:
LOG.warning(
"Removed the following options: '%s' from 'ytdlp.json' file.", ", ".join(removed_options)
)
self.ytdl_options = merge_dict(self.ytdl_options, opts)
else:
LOG.error(f"Invalid yt-dlp custom options file '{optsFile}'.")
LOG.warning(f"Empty yt-dlp custom options file '{opts_file}'.")
else:
LOG.info(f"No yt-dlp custom options found at '{ytdlp_cli}'.")
LOG.info(f"No yt-dlp custom options found at '{opts_file}'.")
self.ytdl_options["socket_timeout"] = self.socket_timeout