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

fixes

Browse source
This commit is contained in:
arabcoders 2025-09-12 17:47:21 +03:00
parent 5d1eed876b
commit 669f70d9b8
34 changed files with 878 additions and 446 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
.vscode/settings.json vendored
View file

@ -133,6 +133,7 @@
"SUPPRESSHELP",
"tebibytes",
"testpaths",
"threadsafe",
"tiktok",
"timespec",
"tmpfilename",
@ -170,5 +171,10 @@
"**/.venv": true,
},
"eslint.format.enable": true,
"eslint.ignoreUntitled": true
"eslint.ignoreUntitled": true,
"python.testing.pytestArgs": [
"app/tests"
],
"python.testing.unittestEnabled": true,
"python.testing.pytestEnabled": true
}

10
app/library/Archiver.py
View file

@ -40,19 +40,15 @@ class Archiver(metaclass=ThreadSafe):
"""
def __init__(self) -> None:
if getattr(self, "_initialized", False):
return
self._cache: dict[str, _Entry] = {}
self._locks: dict[str, threading.RLock] = {}
self._global_lock = threading.RLock()
self._stats_check: bool = True
self._stats_ttl: float = 0.2
self._initialized = True
@classmethod
def get_instance(cls) -> "Archiver":
return cls()
@staticmethod
def get_instance() -> "Archiver":
return Archiver()
def _key(self, file: str | Path) -> str:
"""

23
app/library/BackgroundWorker.py
View file

@ -22,15 +22,17 @@ class BackgroundWorker(metaclass=Singleton):
It is designed to run in a separate thread and uses asyncio to handle asynchronous tasks.
"""
_instance = None
"""The instance of the Notification class."""
thread: threading.Thread
"""The thread that runs the background worker."""
def __init__(self):
self.queue = Queue()
self.queue: Queue = Queue()
"The queue to hold the tasks."
self.running = True
"Whether the background worker is running or not."
self.thread: threading.Thread = None
"The thread that runs the background worker."
@staticmethod
def get_instance() -> "BackgroundWorker":
return BackgroundWorker()
def attach(self, app: web.Application):
app.on_shutdown.append(self.on_shutdown)
@ -49,13 +51,6 @@ class BackgroundWorker(metaclass=Singleton):
except Exception as e:
LOG.error(f"Failed to shut down background worker: {e}")
@staticmethod
def get_instance() -> "BackgroundWorker":
if BackgroundWorker._instance is None:
BackgroundWorker._instance = BackgroundWorker()
return BackgroundWorker._instance
def _run(self):
asyncio.set_event_loop(asyncio.new_event_loop())
loop = asyncio.get_event_loop()

26
app/library/DataStore.py
View file

@ -11,12 +11,12 @@ from .Download import Download
from .ItemDTO import ItemDTO
from .Utils import init_class
LOG = logging.getLogger("datastore")
LOG: logging.Logger = logging.getLogger("datastore")
class StoreType(str, Enum):
HISTORY = "done"
QUEUE = "queue"
HISTORY: str = "done"
QUEUE: str = "queue"
@classmethod
def all(cls) -> list[str]:
@ -53,19 +53,13 @@ class DataStore:
Persistent queue.
"""
_type: StoreType = None
"""Type of the store, e.g., DONE, QUEUE, PENDING."""
_dict: OrderedDict[str, Download] = None
"""Ordered dictionary to store Download objects."""
_connection: Connection
"""SQLite connection to the database."""
def __init__(self, type: StoreType, connection: Connection):
self._dict = OrderedDict()
self._type = type
self._connection = connection
self._dict: OrderedDict[str, Download] = OrderedDict()
"The dictionary of items."
self._type: StoreType = type
"The type of the datastore."
self._connection: Connection = connection
"The database connection."
def load(self) -> None:
for id, item in self.saved_items():
@ -136,7 +130,7 @@ class DataStore:
)
for row in cursor:
rowDate = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007
rowDate: datetime = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007
data: dict = json.loads(row["data"])
data.pop("_id", None)
item: ItemDTO = init_class(ItemDTO, data)

61
app/library/Download.py
View file

@ -47,22 +47,8 @@ class Download:
Download task.
"""
id: str = None
download_dir: str = None
temp_dir: str = None
temp_disabled: bool = False
"Disable the temporary files feature."
template: str = None
template_chapter: str = None
info: ItemDTO = None
debug: bool = False
temp_path: str = None
cancelled: bool = False
is_live: bool = False
info_dict: dict = None
"yt-dlp metadata dict."
update_task = None
cancel_in_progress: bool = False
final_update = False
@ -86,12 +72,6 @@ class Download:
)
"Fields to be extracted from yt-dlp progress hook."
temp_keep: bool = False
"Keep temp folder after download."
logs: list = []
"Logs from yt-dlp."
def __init__(self, info: ItemDTO, info_dict: dict = None, logs: list | None = None):
"""
Initialize download task.
@ -104,29 +84,52 @@ class Download:
"""
config: Config = Config.get_instance()
self.download_dir = info.download_dir
self.temp_dir = info.temp_dir
self.template = info.template
self.template_chapter = info.template_chapter
self.download_dir: str = info.download_dir
"Download directory."
self.temp_dir: str | None = info.temp_dir
"Temporary directory."
self.template: str | None = info.template
"Filename template."
self.template_chapter: str | None = info.template_chapter
"Chapter filename template."
self.download_info_expires = int(config.download_info_expires)
self.info = info
self.id = info._id
"Time in seconds before the download info is considered expired."
self.info: ItemDTO = info
"ItemDTO object."
self.id: str = info._id
"Download ID."
self.debug = bool(config.debug)
"Debug mode."
self.debug_ytdl = bool(config.ytdlp_debug)
"Debug mode for yt-dlp."
self.cancelled = False
"Download cancelled."
self.tmpfilename = None
"Temporary filename."
self.status_queue = None
"Status queue."
self.proc = None
self._notify = EventBus.get_instance()
"yt-dlp process."
self._notify: EventBus = EventBus.get_instance()
"Event bus instance."
self.max_workers = int(config.max_workers)
"Maximum number of concurrent downloads."
self.temp_keep = bool(config.temp_keep)
"Keep temp folder after download."
self.temp_disabled = bool(config.temp_disabled)
self.is_live = bool(info.is_live) or info.live_in is not None
self.info_dict = info_dict
"Disable the temporary files feature."
self.is_live: bool = bool(info.is_live) or info.live_in is not None
"Is the download a live stream."
self.info_dict: dict = info_dict
"yt-dlp metadata dict."
self.logger: logging.Logger = logging.getLogger(f"Download.{info.id if info.id else info._id}")
"Logger for the download task."
self.started_time = 0
"Time when the download started."
self.queue_time: datetime = datetime.now(tz=UTC)
"Time when the download was queued."
self.logs = logs if logs else []
"Logs from yt-dlp."
def _progress_hook(self, data: dict):
if self.debug:

55
app/library/DownloadQueue.py
View file

@ -47,44 +47,29 @@ class DownloadQueue(metaclass=Singleton):
DownloadQueue class is a singleton class that manages the download queue and the download history.
"""
paused: asyncio.Event
"""Event to pause the download queue."""
event: asyncio.Event
"""Event to signal the download queue to start downloading."""
_active: dict[str, Download] = {}
"""Dictionary of active downloads."""
_instance = None
"""Instance of the DownloadQueue."""
queue: DataStore
"""DataStore for the download queue."""
done: DataStore
"""DataStore for the completed downloads."""
workers: asyncio.Semaphore
"""Semaphore to limit the number of concurrent downloads."""
processors: asyncio.Semaphore
"""Semaphore to limit the number of concurrent processors."""
def __init__(self, connection: Connection, config: Config | None = None):
DownloadQueue._instance = self
self.config = config or Config.get_instance()
self._notify = EventBus.get_instance()
self.config: Config = config or Config.get_instance()
"Configuration instance."
self._notify: EventBus = EventBus.get_instance()
"Event bus instance."
self.done = DataStore(type=StoreType.HISTORY, connection=connection)
"DataStore for the completed downloads."
self.queue = DataStore(type=StoreType.QUEUE, connection=connection)
"DataStore for the download queue."
self.workers = asyncio.Semaphore(self.config.max_workers)
"Semaphore to limit the number of concurrent downloads."
self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency)
"Semaphore to limit the number of concurrent processors."
self.paused = asyncio.Event()
"Event to pause the download queue."
self.event = asyncio.Event()
"Event to signal the download queue to start downloading."
self._active: dict[str, Download] = {}
"""Dictionary of active downloads."""
self.done.load()
self.queue.load()
self.paused = asyncio.Event()
self.paused.set()
self.event = asyncio.Event()
self.workers = asyncio.Semaphore(self.config.max_workers)
self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency)
@staticmethod
def get_instance() -> "DownloadQueue":
@ -95,10 +80,7 @@ class DownloadQueue(metaclass=Singleton):
DownloadQueue: The instance of the DownloadQueue
"""
if not DownloadQueue._instance:
DownloadQueue._instance = DownloadQueue()
return DownloadQueue._instance
return DownloadQueue()
def attach(self, _: web.Application) -> None:
"""
@ -1205,6 +1187,5 @@ class DownloadQueue(metaclass=Singleton):
return
if exc := task.exception():
task_name: str = task.get_name() if task.get_name() else "unknown_task"
LOG.error(f"Unhandled exception in background task '{task_name}': {exc!s}. {traceback.format_exc()}")

39
app/library/Events.py
View file

@ -194,14 +194,13 @@ class Event:
class EventListener:
name: str
is_coroutine: bool = False
call_back: callable
def __init__(self, name: str, callback: callable):
self.name = name
self.call_back = callback
self.is_coroutine = asyncio.iscoroutinefunction(callback)
self.name: str = name
"The name of the listener."
self.call_back: callable = callback
"The callback function to call when the event is emitted."
self.is_coroutine: bool = asyncio.iscoroutinefunction(callback)
"Whether the callback is a coroutine function or not."
async def handle(self, event: Event, **kwargs):
if self.is_coroutine:
@ -215,20 +214,15 @@ class EventBus(metaclass=Singleton):
This class is used to subscribe to and emit events to the registered listeners.
"""
_instance = None
"""the instance of the EventsSubscriber"""
_listeners: dict[str, list[str, EventListener]] = {}
"""The listeners for the events."""
debug: bool = False
"""Whether to log debug messages or not."""
_offload: BackgroundWorker = None
"""The background worker to offload tasks to."""
def __init__(self):
EventBus._instance = self
self._listeners: dict[str, list[str, EventListener]] = {}
"The listeners for the events."
self.debug: bool = False
"Whether to log debug messages or not."
self._offload: BackgroundWorker = None
"The background worker to offload tasks to."
@staticmethod
def get_instance() -> "EventBus":
@ -239,10 +233,7 @@ class EventBus(metaclass=Singleton):
EventsSubscriber: The instance of the EventsSubscriber
"""
if not EventBus._instance:
EventBus._instance = EventBus()
return EventBus._instance
return EventBus()
def subscribe(self, event: str | list | tuple, callback: Awaitable, name: str | None = None) -> "EventBus":
"""

2
app/library/HttpSocket.py
View file

@ -57,7 +57,7 @@ class HttpSocket:
async def event_handler(e: Event, _, **kwargs):
await self.sio.emit(event=e.event, data=encoder.encode(e), **kwargs)
services = Services.get_instance()
services: Services = Services.get_instance()
services.add_all(
{
k: v

4
app/library/LogWrapper.py
View file

@ -59,11 +59,9 @@ class LogWrapper:
"""
targets: list[LogTarget] = []
"""A list of dictionaries where each dictionary represents a logging target with its level and type."""
def __init__(self):
self.targets: list[LogTarget] = []
"""A list of dictionaries where each dictionary represents a logging target with its level and type."""
def add_target(self, target: logging.Logger | Callable, level: int = logging.DEBUG, name: str | None = None):
"""

34
app/library/Notifications.py
View file

@ -130,12 +130,6 @@ class NotificationEvents:
class Notification(metaclass=Singleton):
_targets: list[Target] = []
"""Notification targets to send events to."""
_instance = None
"""The instance of the Notification class."""
def __init__(
self,
file: str | None = None,
@ -144,15 +138,23 @@ class Notification(metaclass=Singleton):
config: Config | None = None,
background_worker: BackgroundWorker | None = None,
):
Notification._instance = self
self._targets: list[Target] = []
"Notification targets to send events to."
config: Config = config or Config.get_instance()
self._debug: bool = config.debug
"Debug mode."
self._file: Path = Path(file) if file else Path(config.config_path).joinpath("notifications.json")
"File to store notification targets."
self._client: httpx.AsyncClient = client or httpx.AsyncClient()
"HTTP client to send requests."
self._encoder: Encoder = encoder or Encoder()
"Encoder to encode data."
self._version: str = config.app_version
"Application version."
self._offload: BackgroundWorker = background_worker or BackgroundWorker.get_instance()
"Background worker to offload tasks to."
if self._file.exists() and "600" != self._file.stat().st_mode:
try:
@ -161,11 +163,19 @@ class Notification(metaclass=Singleton):
pass
@staticmethod
def get_instance() -> "Notification":
if Notification._instance is None:
Notification._instance = Notification()
return Notification._instance
def get_instance(
file: str | None = None,
client: httpx.AsyncClient | None = None,
encoder: Encoder | None = None,
config: Config | None = None,
background_worker: BackgroundWorker | None = None,
) -> "Notification":
"""
Get the instance of the class.
"""
return Notification(
file=file, client=client, encoder=encoder, config=config, background_worker=background_worker
)
def attach(self, _: web.Application):
"""

5
app/library/PackageInstaller.py
View file

@ -37,9 +37,10 @@ class Packages:
class PackageInstaller:
user_site: Path | None = None
def __init__(self, pkg_path: Path | None = None):
self.user_site: Path | None = None
"Where to install user packages."
if pkg_path:
self.user_site = pkg_path

4
app/library/Playlist.py
View file

@ -6,11 +6,11 @@ from .Utils import StreamingError, get_file_sidecar
class Playlist:
_url: str = None
def __init__(self, download_path: Path, url: str):
self.url: str = url
"The base URL for the playlist."
self.download_path: Path = download_path
"The path where files are downloaded."
async def make(self, file: Path) -> str:
ref: str = Path(str(file.relative_to(self.download_path)).strip("/"))

33
app/library/Presets.py
View file

@ -105,24 +105,20 @@ class Presets(metaclass=Singleton):
This class is used to manage the presets.
"""
_items: list[Preset] = []
"""The list of presets."""
_instance = None
"""The instance of the class."""
_config: Config = None
"""The config instance."""
_default: list[Preset] = []
"""The list of default presets."""
def __init__(self, file: str | Path | None = None, config: Config | None = None):
Presets._instance = self
self._items: list[Preset] = []
"The list of presets."
self._config: Config = None
"The config instance."
self._default: list[Preset] = []
"The list of default presets."
self._config = config or Config.get_instance()
self._file: Path = Path(file) if file else Path(self._config.config_path).joinpath("presets.json")
"The path to the presets file."
if self._file.exists() and "600" != self._file.stat().st_mode:
try:
@ -139,18 +135,19 @@ class Presets(metaclass=Singleton):
continue
@staticmethod
def get_instance() -> "Presets":
def get_instance(file: str | Path | None = None, config: Config | None = None) -> "Presets":
"""
Get the instance of the class.
Args:
file (str|Path|None): The path to the presets file.
config (Config|None): The config instance.
Returns:
Presets: The instance of the class
"""
if not Presets._instance:
Presets._instance = Presets()
return Presets._instance
return Presets(file=file, config=config)
async def on_shutdown(self, _: web.Application):
pass

28
app/library/Scheduler.py
View file

@ -15,25 +15,15 @@ class Scheduler(metaclass=Singleton):
This class is used to manage the schedule.
"""
_jobs: dict[str, Cron] = {}
"""The scheduled jobs."""
_instance = None
"""The instance of the class."""
def __init__(self, loop: asyncio.AbstractEventLoop | None = None):
Scheduler._instance = self
self._jobs: dict[str, Cron] = {}
"The scheduled jobs."
self._loop = loop or asyncio.get_event_loop()
async def event_handler(data, _):
if data and data.data:
self.add(**data.data)
EventBus.get_instance().subscribe(Events.SCHEDULE_ADD, event_handler, f"{__class__.__name__}.add")
"The event loop to use."
@staticmethod
def get_instance() -> "Scheduler":
def get_instance(loop: asyncio.AbstractEventLoop | None = None) -> "Scheduler":
"""
Get the instance of the class.
@ -41,9 +31,7 @@ class Scheduler(metaclass=Singleton):
Scheduler: The instance of the class
"""
if not Scheduler._instance:
Scheduler._instance = Scheduler()
return Scheduler._instance
return Scheduler(loop=loop)
async def on_shutdown(self, _: web.Application):
"""
@ -70,6 +58,12 @@ class Scheduler(metaclass=Singleton):
def attach(self, app: web.Application):
app.on_shutdown.append(self.on_shutdown)
async def event_handler(data, _):
if data and data.data:
self.add(**data.data)
EventBus.get_instance().subscribe(Events.SCHEDULE_ADD, event_handler, f"{__class__.__name__}.add")
def get_all(self) -> dict[str, Cron]:
"""Return the jobs."""
return self._jobs

20
app/library/Segments.py
View file

@ -11,22 +11,32 @@ from aiohttp import web
from .config import Config
from .ffprobe import ffprobe
LOG = logging.getLogger("player.segments")
LOG: logging.Logger = logging.getLogger("player.segments")
class Segments:
def __init__(self, download_path: str, index: int, duration: float, vconvert: bool, aconvert: bool):
config = Config.get_instance()
self.download_path = download_path
config: Config = Config.get_instance()
self.download_path: str = download_path
"The path where files are downloaded."
self.index = int(index)
"The index of the segment."
self.duration = float(duration)
"The duration of the segment."
self.vconvert = bool(vconvert)
"Whether to convert video."
self.aconvert = bool(aconvert)
self.vcodec = config.streamer_vcodec
self.acodec = config.streamer_acodec
"Whether to convert audio."
self.vcodec: str = config.streamer_vcodec
"The video codec to use."
self.acodec: str = config.streamer_acodec
"The audio codec to use."
# sadly due to unforeseen circumstances, we have to convert the video for now.
self.vconvert = True
"Whether to convert video."
self.aconvert = True
"Whether to convert audio."
async def build_ffmpeg_args(self, file: Path) -> list[str]:
try:

25
app/library/Services.py
View file

@ -9,42 +9,37 @@ LOG: logging.Logger = logging.getLogger(__name__)
class Services(metaclass=Singleton):
_dct: dict[str, T] = {}
_instance = None
"""The instance of the class."""
def __init__(self):
self._services: dict[str, T] = {}
@staticmethod
def get_instance() -> "Services":
if Services._instance is None:
Services._instance = Services()
return Services._instance
return Services()
def add(self, name: str, service: T):
self._dct[name] = service
self._services[name] = service
def add_all(self, services: dict[str, T]):
for name, service in services.items():
self.add(name, service)
def get(self, name: str) -> T | None:
return self._dct.get(name)
return self._services.get(name)
def has(self, name: str) -> bool:
return name in self._dct
return name in self._services
def remove(self, name: str):
if name not in self._dct:
if name not in self._services:
return
self._dct.pop(name, None)
self._services.pop(name, None)
def clear(self):
self._dct.clear()
self._services.clear()
def get_all(self) -> dict[str, T]:
return self._dct.copy()
return self._services.copy()
async def handle_async(self, handler: callable, **kwargs) -> Any:
context = {**self.get_all(), **kwargs}

4
app/library/Singleton.py
View file

@ -8,6 +8,7 @@ class Singleton(type):
"""
_instances: dict[type, Any] = {}
"The singleton instances."
def __call__(cls, *args: Any, **kwargs: Any) -> Any:
if cls not in cls._instances:
@ -30,7 +31,10 @@ class ThreadSafe(type):
"""
_instances: dict[type, Any] = {}
"The singleton instances."
_lock = threading.Lock()
"A lock to ensure thread-safe singleton creation."
def __call__(cls, *args: Any, **kwargs: Any) -> Any:
with cls._lock:

62
app/library/Tasks.py
View file

@ -132,12 +132,6 @@ class Tasks(metaclass=Singleton):
This class is used to manage the tasks.
"""
_tasks: list[Task] = []
"""The tasks."""
_instance = None
"""The instance of the Tasks."""
def __init__(
self,
file: str | None = None,
@ -146,19 +140,28 @@ class Tasks(metaclass=Singleton):
encoder: Encoder | None = None,
scheduler: Scheduler | None = None,
):
Tasks._instance = self
self._tasks: list[Task] = []
"The tasks."
config = config or Config.get_instance()
self._debug: bool = config.debug
"Debug mode."
self._default_preset: str = config.default_preset
"The default preset."
self._file: Path = Path(file) if file else Path(config.config_path).joinpath("tasks.json")
"The tasks file."
self._encoder: Encoder = encoder or Encoder()
"The JSON encoder."
self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_event_loop()
"The event loop."
self._scheduler: Scheduler = scheduler or Scheduler.get_instance()
"The scheduler."
self._notify: EventBus = EventBus.get_instance()
"The event bus."
self._task_handler = HandleTask(self._scheduler, self, config)
"The task handler."
self._downloadQueue = DownloadQueue.get_instance()
"The download queue."
if self._file.exists() and "600" != self._file.stat().st_mode:
try:
@ -167,7 +170,13 @@ class Tasks(metaclass=Singleton):
pass
@staticmethod
def get_instance() -> "Tasks":
def get_instance(
file: str | None = None,
loop: asyncio.AbstractEventLoop | None = None,
config: Config | None = None,
encoder: Encoder | None = None,
scheduler: Scheduler | None = None,
) -> "Tasks":
"""
Get the instance of the class.
@ -175,10 +184,7 @@ class Tasks(metaclass=Singleton):
Tasks: The instance of the class.
"""
if not Tasks._instance:
Tasks._instance = Tasks()
return Tasks._instance
return Tasks(file=file, loop=loop, config=config, encoder=encoder, scheduler=scheduler)
async def on_shutdown(self, _: web.Application):
self.clear(shutdown=True)
@ -454,17 +460,17 @@ class Tasks(metaclass=Singleton):
class HandleTask:
_tasks: Tasks
_handlers: list[type]
_scheduler: Scheduler
_config: Config
_task_name: str
def __init__(self, scheduler: Scheduler, tasks: Tasks, config: Config) -> None:
self._tasks = tasks
self._scheduler = scheduler
self._config = config
self._handlers: list[type] = []
"The available handlers."
self._tasks: Tasks = tasks
"The tasks manager."
self._scheduler: Scheduler = scheduler
"The scheduler."
self._config: Config = config
"The configuration."
self._task_name: str = f"{__class__.__name__}._dispatcher"
"The task name for the scheduler."
def load(self) -> None:
self._handlers: list[type] = self._discover()
@ -496,24 +502,36 @@ class HandleTask:
self._scheduler.remove(self._task_name)
def _dispatcher(self):
s: dict[list[str]] = {"h": [], "d": [], "u": [], "f": []}
for task in self._tasks.get_all():
if not task.handler_enabled:
s["d"].append(task.name)
continue
if not task.get_ytdlp_opts().get_all().get("download_archive"):
LOG.debug(f"Task '{task.name}' does not have an archive file configured.")
s["f"].append(task.name)
continue
try:
handler = self._find_handler(task)
if handler is None:
s["u"].append(task.name)
continue
coro = self.dispatch(task, handler=handler)
t = asyncio.create_task(coro, name=f"task-{task.id}")
t.add_done_callback(lambda fut, t=task: self._handle_exception(fut, t))
s["h"].append(task.name)
except Exception as e:
LOG.error(f"Failed to handle task '{task.name}'. '{e!s}'.")
s["f"].append(task.name)
if len(self._tasks.get_all()) > 0:
LOG.info(
f"Task Handler summary: Handled: {len(s['h'])}, Unhandled: {len(s['u'])}, Disabled: {len(s['d'])}, Failed: {len(s['f'])}."
)
def _handle_exception(self, fut: asyncio.Task, task: Task) -> None:
if fut.cancelled():

139
app/library/Utils.py
View file

@ -61,10 +61,13 @@ REMOVE_KEYS: list = [
"opt_list_impersonate_targets": "--list-impersonate-targets",
},
]
"Keys to remove from yt-dlp options at various levels."
YTDLP_INFO_CLS: YTDLP = None
"Cached YTDLP info class."
ALLOWED_SUBS_EXTENSIONS: set[str] = {".srt", ".vtt", ".ass"}
"Allowed subtitle file extensions."
FILES_TYPE: list = [
{"rx": re.compile(r"\.(avi|ts|mkv|mp4|mp3|mpv|ogm|m4v|webm|m4b)$", re.IGNORECASE), "type": "video"},
@ -74,11 +77,15 @@ FILES_TYPE: list = [
{"rx": re.compile(r"\.(txt|nfo|md|json|yml|yaml|plexmatch)$", re.IGNORECASE), "type": "text"},
{"rx": re.compile(r"\.(nfo|json|jpg|torrent|\.info\.json)$", re.IGNORECASE), "type": "metadata"},
]
"File type patterns for sidecar files."
TAG_REGEX: re.Pattern[str] = re.compile(r"%{([^:}]+)(?::([^}]*))?}c")
"Regex to find tags in templates."
DT_PATTERN: re.Pattern[str] = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[+-]\d{2}:\d{2}))\s?")
"Regex to match ISO 8601 datetime strings."
T = TypeVar("T")
"Generic type variable."
class StreamingError(Exception):
@ -111,11 +118,11 @@ def calc_download_path(base_path: str | Path, folder: str | None = None, create_
folder = folder.removeprefix("/")
realBasePath = base_path.resolve()
download_path = Path(realBasePath).joinpath(folder).resolve(strict=False)
realBasePath: Path = base_path.resolve()
download_path: Path = Path(realBasePath).joinpath(folder).resolve(strict=False)
if not str(download_path).startswith(str(realBasePath)):
msg = f'Folder "{folder}" must resolve inside the base download folder "{realBasePath}".'
msg: str = f'Folder "{folder}" must resolve inside the base download folder "{realBasePath}".'
raise Exception(msg)
try:
@ -165,7 +172,7 @@ def extract_info(
}
# Remove keys that are not needed for info extraction.
keys_to_remove = [key for key in params if str(key).startswith("write") or key in ["postprocessors"]]
keys_to_remove: list = [key for key in params if str(key).startswith("write") or key in ["postprocessors"]]
for key in keys_to_remove:
params.pop(key, None)
@ -175,8 +182,8 @@ def extract_info(
params["quiet"] = True
log_wrapper = LogWrapper()
idDict = get_archive_id(url=url)
archive_id = f".{idDict['id']}" if idDict.get("id") else None
idDict: dict[str, str | None] = get_archive_id(url=url)
archive_id: str | None = f".{idDict['id']}" if idDict.get("id") else None
log_wrapper.add_target(
target=logging.getLogger(f"yt-dlp{archive_id if archive_id else '.extract_info'}"),
@ -244,7 +251,7 @@ def _is_safe_key(key: any) -> bool:
return False
# Block only truly dangerous dunder patterns
key_stripped = key.strip()
key_stripped: str = key.strip()
# Block dunder attributes (starts AND ends with __)
return not (key_stripped.startswith("__") and key_stripped.endswith("__"))
@ -279,7 +286,7 @@ def merge_dict(
# Prevent deep recursion DoS
if _depth > max_depth:
msg = f"Recursion depth limit exceeded ({max_depth})"
msg: str = f"Recursion depth limit exceeded ({max_depth})"
raise RecursionError(msg)
# Initialize circular reference tracking
@ -294,10 +301,10 @@ def merge_dict(
raise ValueError(msg)
# Track current objects
current_seen = _seen | {source_id, dest_id}
current_seen: set[Any | int] = _seen | {source_id, dest_id}
# Create a clean copy of destination with only safe keys
destination_copy = {}
destination_copy: dict = {}
for k, v in destination.items():
if _is_safe_key(k):
# Prevent memory DoS from large lists
@ -311,7 +318,7 @@ def merge_dict(
if not _is_safe_key(key):
continue
destination_value = destination_copy.get(key)
destination_value: Any | None = destination_copy.get(key)
# Recursively merge dictionaries with safety checks
if isinstance(value, dict) and isinstance(destination_value, dict):
@ -325,8 +332,8 @@ def merge_dict(
combined_size = len(value) + len(destination_value)
if combined_size > max_list_size:
# Truncate to stay within limits
available_space = max_list_size - len(destination_value)
truncated_value = value[: max(0, available_space)]
available_space: int = max_list_size - len(destination_value)
truncated_value: list = value[: max(0, available_space)]
destination_copy[key] = copy.deepcopy(destination_value) + copy.deepcopy(truncated_value)
else:
destination_copy[key] = copy.deepcopy(destination_value) + copy.deepcopy(value)
@ -354,11 +361,13 @@ def check_id(file: Path) -> bool | str:
bool|str: False if no file found, else the file path.
"""
match = re.search(r"(?<=\[)(?:youtube-)?(?P<id>[a-zA-Z0-9\-_]{11})(?=\])", file.stem, re.IGNORECASE)
match: re.Match[str] | None = re.search(
r"(?<=\[)(?:youtube-)?(?P<id>[a-zA-Z0-9\-_]{11})(?=\])", file.stem, re.IGNORECASE
)
if not match:
return False
id = match.groupdict().get("id")
id: str | None = match.groupdict().get("id")
try:
for f in file.parent.iterdir():
@ -378,8 +387,8 @@ def check_id(file: Path) -> bool | str:
@lru_cache(maxsize=512)
def is_private_address(hostname: str) -> bool:
ip = socket.gethostbyname(hostname)
ip_obj = ipaddress.ip_address(ip)
ip: str = socket.gethostbyname(hostname)
ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address = ipaddress.ip_address(ip)
return ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_reserved or ip_obj.is_link_local
@ -545,7 +554,7 @@ def get_file_sidecar(file: Path) -> list[dict]:
list: List of sidecar files.
"""
files = {}
files: dict = {}
for i, f in enumerate(file.parent.glob(f"{glob.escape(file.stem)}.*")):
if f == file or f.is_file() is False or f.stem.startswith("."):
@ -557,15 +566,15 @@ def get_file_sidecar(file: Path) -> list[dict]:
content_type = "Unknown"
for pattern in FILES_TYPE:
if pattern["rx"].search(f.name):
content_type = pattern["type"]
content_type: str = pattern["type"]
break
if content_type == "subtitle":
if f.suffix not in ALLOWED_SUBS_EXTENSIONS:
continue
lg = re.search(r"\.(?P<lang>\w{2,3})\.\w{3}$", f.name)
lang = lg.groupdict().get("lang") if lg else "und"
content = {"file": f, "lang": lang, "name": f"{f.suffix[1:].upper()} ({i}) - {lang}"}
lg: re.Match[str] | None = re.search(r"\.(?P<lang>\w{2,3})\.\w{3}$", f.name)
lang: str | None = lg.groupdict().get("lang") if lg else "und"
content: dict[str, Any] = {"file": f, "lang": lang, "name": f"{f.suffix[1:].upper()} ({i}) - {lang}"}
else:
content = {"file": f}
@ -574,7 +583,7 @@ def get_file_sidecar(file: Path) -> list[dict]:
files[content_type].append(content)
images = get_possible_images(str(file.parent))
images: list[dict] = get_possible_images(str(file.parent))
if len(images) > 0:
if "image" not in files:
files["image"] = []
@ -586,7 +595,7 @@ def get_file_sidecar(file: Path) -> list[dict]:
@lru_cache(maxsize=512)
def get_possible_images(dir: str) -> list[dict]:
images = []
images: list = []
path_loc = Path(dir, "test.jpg")
@ -615,7 +624,7 @@ def get_mime_type(metadata: dict, file_path: Path) -> str:
format_name = metadata.get("format_name", "")
# Define mappings for HTML5-compatible video types
format_to_mime = {
format_to_mime: dict[str, str] = {
"matroska": "video/x-matroska", # Default for MKV
"webm": "video/webm", # MKV can also be WebM
"mp4": "video/mp4",
@ -628,7 +637,7 @@ def get_mime_type(metadata: dict, file_path: Path) -> str:
for fmt in format_name.split(","):
fmt = fmt.strip().lower()
if fmt in format_to_mime:
selected = format_to_mime[fmt]
selected: str = format_to_mime[fmt]
if selected:
return selected
@ -668,7 +677,7 @@ def get_file(download_path: str | Path, file: str | Path) -> tuple[Path, int]:
LOG.error(f"Error calculating download path. {e!s}")
return (Path(file), 404)
possibleFile = check_id(file=realFile)
possibleFile: bool | str = check_id(file=realFile)
if not possibleFile:
return (realFile, 404)
@ -775,7 +784,7 @@ def get(
return default() if callable(default) else default
# Split the path by the separator and traverse the data structure.
segments = path.split(separator)
segments: list[str] = path.split(separator)
for segment in segments:
if isinstance(data, dict):
if segment in data:
@ -873,7 +882,7 @@ def get_files(base_path: Path | str, dir: str | None = None):
if not content_type:
content_type = "download"
stat = file.stat()
stat: os.stat_result = file.stat()
contents.append(
{
@ -897,9 +906,6 @@ def get_files(base_path: Path | str, dir: str | None = None):
def clean_item(item: dict, keys: list | tuple) -> tuple[dict, bool]:
"""
Remove given keys from a dictionary.
This function modifies the dictionary in place and returns a tuple
containing the modified dictionary and a boolean indicating
whether any keys were removed.
Args:
item (dict): The item to clean.
@ -913,6 +919,7 @@ def clean_item(item: dict, keys: list | tuple) -> tuple[dict, bool]:
"""
status = False
item = copy.deepcopy(item)
if not isinstance(item, dict):
msg = "Item must be a dictionary."
@ -946,7 +953,7 @@ def strip_newline(string: str) -> str:
if not string:
return ""
res = re.sub(r"(\r\n|\r|\n)", " ", string)
res: str = re.sub(r"(\r\n|\r|\n)", " ", string)
return res.strip() if res else ""
@ -978,34 +985,34 @@ async def read_logfile(file: Path, offset: int = 0, limit: int = 50) -> dict:
try:
async with await open_file(file, "rb") as f:
await f.seek(0, os.SEEK_END)
file_size = await f.tell()
file_size: int = await f.tell()
block_size = 1024
block_end = file_size
buffer = b""
lines = []
block_end: int = file_size
buffer: bytes = b""
lines: list = []
required_count = offset + limit + 1
required_count: int = offset + limit + 1
while len(lines) < required_count and block_end > 0:
block_start = max(0, block_end - block_size)
block_start: int = max(0, block_end - block_size)
await f.seek(block_start)
chunk = await f.read(block_end - block_start)
buffer = chunk + buffer # prepend the chunk
chunk: bytes = await f.read(block_end - block_start)
buffer: bytes = chunk + buffer # prepend the chunk
lines = buffer.splitlines()
block_end = block_start
if len(lines) > offset + limit:
next_offset = offset + limit
next_offset: int = offset + limit
end_is_reached = False
else:
next_offset = None
end_is_reached = True
for line in lines[-(offset + limit) : -offset] if offset else lines[-limit:]:
line_bytes = line if isinstance(line, bytes) else line.encode()
msg = line.decode(errors="replace")
dt_match = DT_PATTERN.match(msg)
line_bytes: bytes | str = line if isinstance(line, bytes) else line.encode()
msg: str = line.decode(errors="replace")
dt_match: re.Match[str] | None = DT_PATTERN.match(msg)
result.append(
{
"id": sha256(line_bytes).hexdigest(),
@ -1041,13 +1048,13 @@ async def tail_log(file: Path, emitter: callable, sleep_time: float = 0.5):
async with await open_file(file, "rb") as f:
await f.seek(0, os.SEEK_END)
while True:
line = await f.readline()
line: bytes = await f.readline()
if not line:
await asyncio_sleep(sleep_time)
continue
msg = line.decode(errors="replace")
dt_match = DT_PATTERN.match(msg)
msg: str = line.decode(errors="replace")
dt_match: re.Match[str] | None = DT_PATTERN.match(msg)
await emitter(
{
@ -1101,7 +1108,7 @@ def get_archive_id(url: str) -> dict[str, str | None]:
"""
global YTDLP_INFO_CLS # noqa: PLW0603
idDict = {
idDict: dict[str, None] = {
"id": None,
"ie_key": None,
"archive_id": None,
@ -1158,7 +1165,7 @@ def dt_delta(delta: timedelta) -> str:
hours, rem = divmod(rem, 3600)
minutes, secs = divmod(rem, 60)
parts = []
parts: list[str] = []
if days:
parts.append(f"{days}d")
if hours:
@ -1269,6 +1276,7 @@ def load_modules(root_path: Path, directory: Path):
full_name: str = f"{package_name}.{name}"
if name.startswith("_"):
continue
try:
importlib.import_module(full_name)
except ImportError as e:
@ -1377,37 +1385,6 @@ def ytdlp_reject(entry: dict, yt_params: dict) -> tuple[bool, str]:
return (True, "")
def find_unpickleable(obj, name="root", seen=None):
import pickle
if seen is None:
seen = set()
if id(obj) in seen:
return
seen.add(id(obj))
try:
pickle.dumps(obj)
except Exception as e:
LOG.error(f"[UNPICKLEABLE] {name}: {e}")
if isinstance(obj, dict):
for k, v in obj.items():
find_unpickleable(v, f"{name}[{repr(k)!s}]", seen)
elif hasattr(obj, "__dict__"):
for attr in vars(obj):
try:
value = getattr(obj, attr)
find_unpickleable(value, f"{name}.{attr}", seen)
except Exception as ie:
LOG.error(f"[ERROR] Accessing {name}.{attr}: {ie}")
elif isinstance(obj, list | tuple | set):
for idx, item in enumerate(obj):
find_unpickleable(item, f"{name}[{idx}]", seen)
def list_folders(path: Path, base: Path, depth_limit: int) -> list[str]:
"""
List all folders relative to a base path, up to a specified depth limit.

21
app/library/YTDLPOpts.py
View file

@ -9,20 +9,17 @@ LOG: logging.Logger = logging.getLogger("YTDLPOpts")
class YTDLPOpts:
_item_opts: dict = {}
"""The item options."""
_preset_opts: dict = {}
"""The preset options."""
_item_cli: list = []
"""The command options for yt-dlp from item."""
_preset_cli: str = ""
"""The command options for yt-dlp from preset."""
def __init__(self):
self._config: Config = Config.get_instance()
"The config instance."
self._item_opts: dict = {}
"The item options."
self._preset_opts: dict = {}
"The preset options."
self._item_cli: list = []
"The command options for yt-dlp from item."
self._preset_cli: str = ""
"The command options for yt-dlp from preset."
@staticmethod
def get_instance() -> "YTDLPOpts":

4
app/library/cache.py
View file

@ -11,12 +11,8 @@ class Cache(metaclass=ThreadSafe):
"""
Initialize the Cache.
"""
# Prevent reinitialization in singleton context.
if hasattr(self, "_initialized") and self._initialized:
return
self._cache: dict[str, tuple[Any, float | None]] = {}
self._lock = threading.Lock()
self._initialized = True
def set(self, key: str, value: Any, ttl: float | None = None) -> None:
"""

31
app/library/conditions.py
View file

@ -14,7 +14,7 @@ from .mini_filter import match_str
from .Singleton import Singleton
from .Utils import arg_converter, init_class
LOG = logging.getLogger("conditions")
LOG: logging.Logger = logging.getLogger("conditions")
@dataclass(kw_only=True)
@ -49,18 +49,14 @@ class Conditions(metaclass=Singleton):
This class is used to manage the download conditions.
"""
_items: list[Condition] = []
"""The list of items."""
_instance = None
"""The instance of the class."""
def __init__(self, file: Path | str | None = None, config: Config | None = None):
Conditions._instance = self
self._items: list[Condition] = []
"The list of items."
config = config or Config.get_instance()
self._file: Path = Path(file) if file else Path(config.config_path) / "conditions.json"
"The path to the file where the items are stored."
if self._file.exists() and "600" != self._file.stat().st_mode:
try:
@ -68,14 +64,8 @@ class Conditions(metaclass=Singleton):
except Exception:
pass
async def event_handler(_, __):
msg = "Not implemented"
raise Exception(msg)
EventBus.get_instance().subscribe(Events.CONDITIONS_ADD, event_handler, f"{__class__.__name__}.save")
@staticmethod
def get_instance() -> "Conditions":
def get_instance(file: Path | str | None = None, config: Config | None = None) -> "Conditions":
"""
Get the instance of the class.
@ -83,10 +73,7 @@ class Conditions(metaclass=Singleton):
Conditions: The instance of the class
"""
if not Conditions._instance:
Conditions._instance = Conditions()
return Conditions._instance
return Conditions(file=file, config=config)
async def on_shutdown(self, _: web.Application):
pass
@ -104,6 +91,12 @@ class Conditions(metaclass=Singleton):
"""
self.load()
async def event_handler(_, __):
msg = "Not implemented"
raise Exception(msg)
EventBus.get_instance().subscribe(Events.CONDITIONS_ADD, event_handler, f"{__class__.__name__}.save")
def get_all(self) -> list[Condition]:
"""Return the items."""
return self._items

17
app/library/config.py
View file

@ -12,6 +12,7 @@ from typing import TYPE_CHECKING
import coloredlogs
from dotenv import load_dotenv
from .Singleton import Singleton
from .Utils import FileLogFormatter, arg_converter
from .version import APP_BRANCH, APP_BUILD_DATE, APP_COMMIT_SHA, APP_VERSION
@ -19,7 +20,7 @@ if TYPE_CHECKING:
from subprocess import CompletedProcess
class Config:
class Config(metaclass=Singleton):
app_env: str = "production"
"""The application environment, can be 'production' or 'development'."""
@ -131,9 +132,6 @@ class Config:
app_branch: str = APP_BRANCH
"The branch of the application."
__instance = None
"The instance of the class."
started: int = 0
"The time the application was started."
@ -206,7 +204,6 @@ class Config:
"The variables that are set manually."
_immutable: tuple = (
"__instance",
"ytdl_options",
"started",
"ytdlp_cli",
@ -284,8 +281,7 @@ class Config:
@staticmethod
def get_instance(is_native: bool = False) -> "Config":
"""Static access method."""
cls: Config = Config(is_native) if not Config.__instance else Config.__instance
cls = Config(is_native)
cls.is_native = is_native or cls.is_native
return cls
@ -297,13 +293,6 @@ class Config:
return Config._manager
def __init__(self, is_native: bool = False):
"""Virtually private constructor."""
if Config.__instance is not None:
msg = "This class is a singleton. Use Config.get_instance() instead."
raise Exception(msg)
Config.__instance = self
baseDefaultPath: str = str(Path(__file__).parent.parent.parent.absolute())
self.is_native = is_native

39
app/library/dl_fields.py
View file

@ -100,21 +100,13 @@ class DLFields(metaclass=Singleton):
This class is used to manage the DLFields.
"""
_items: list[DLField] = []
"""The list of items."""
_instance = None
"""The instance of the class."""
_config: Config = None
"""The config instance."""
def __init__(self, file: str | Path | None = None, config: Config | None = None):
DLFields._instance = self
self._config = config or Config.get_instance()
self._file: Path = Path(file) if file else Path(self._config.config_path).joinpath("dl_fields.json")
self._items: list[DLField] = []
"The list of items."
config: Config = config or Config.get_instance()
"The configuration instance."
self._file: Path = Path(file) if file else Path(config.config_path).joinpath("dl_fields.json")
"The path to the file where the items are stored."
if self._file.exists() and "600" != self._file.stat().st_mode:
try:
@ -122,14 +114,8 @@ class DLFields(metaclass=Singleton):
except Exception:
pass
async def event_handler(_, __):
msg = "Not implemented"
raise Exception(msg)
EventBus.get_instance().subscribe(Events.DLFIELDS_ADD, event_handler, f"{__class__.__name__}.add")
@staticmethod
def get_instance() -> "DLFields":
def get_instance(file: str | Path | None = None, config: Config | None = None) -> "DLFields":
"""
Get the instance of the class.
@ -137,10 +123,7 @@ class DLFields(metaclass=Singleton):
DLFields: The instance of the class
"""
if not DLFields._instance:
DLFields._instance = DLFields()
return DLFields._instance
return DLFields(file=file, config=config)
async def on_shutdown(self, _: web.Application):
pass
@ -158,6 +141,12 @@ class DLFields(metaclass=Singleton):
"""
self.load()
async def event_handler(_, __):
msg = "Not implemented"
raise Exception(msg)
EventBus.get_instance().subscribe(Events.DLFIELDS_ADD, event_handler, f"{__class__.__name__}.add")
def get_all(self) -> list[DLField]:
"""Return the items."""
return self._items

6
app/library/ffprobe.py
View file

@ -277,7 +277,7 @@ async def ffprobe(file: str) -> FFProbeResult:
msg = "ffprobe not found."
raise OSError(msg) from e
args = ["-v", "quiet", "-of", "json", "-show_streams", "-show_format", str(f)]
args: list[str] = ["-v", "quiet", "-of", "json", "-show_streams", "-show_format", str(f)]
p = await asyncio.create_subprocess_exec(
"ffprobe",
@ -287,13 +287,13 @@ async def ffprobe(file: str) -> FFProbeResult:
creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0,
)
exitCode = await p.wait()
exitCode: int = await p.wait()
data, err = await p.communicate()
if 0 == exitCode:
parsed: dict = json.loads(data.decode("utf-8"))
else:
msg = f"ffprobe returned with non-0 exit code. '{err.decode('utf-8')}'"
msg: str = f"ffprobe returned with non-0 exit code. '{err.decode('utf-8')}'"
raise FFProbeError(msg)
result = FFProbeResult()

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

@ -61,7 +61,7 @@ class YoutubeHandler(BaseHandler):
params: dict = task.get_ytdlp_opts().get_all()
feed_url: str = YoutubeHandler.FEED.format(type=parsed["type"], id=parsed["id"])
LOG.info(f"'{task.name}': Fetching feed.")
LOG.debug(f"'{task.name}': Fetching feed.")
items: list = []
@ -106,7 +106,7 @@ class YoutubeHandler(BaseHandler):
if real_count < 1:
LOG.warning(f"'{task.name}': No entries found the RSS feed. URL: {feed_url}")
else:
LOG.info(f"'{task.name}': Feed has '{real_count}' entries, all already downloaded/queued.")
LOG.debug(f"'{task.name}': Feed has '{real_count}' entries, all already downloaded/queued.")
return
filtered: list = []
@ -134,7 +134,7 @@ class YoutubeHandler(BaseHandler):
filtered.append(item)
if len(filtered) < 1:
LOG.info(f"'{task.name}': Feed has '{real_count}' entries, all already downloaded/queued.")
LOG.debug(f"'{task.name}': Feed has '{real_count}' entries, all already downloaded/queued.")
return
LOG.info(f"'{task.name}': Found '{len(filtered)}/{real_count}' new items from feed.")

1
app/library/ytdlp.py
View file

@ -14,6 +14,7 @@ class _ArchiveProxy:
def __init__(self, file: str | None):
self._file: str | None = file
"The archive file path."
def __contains__(self, item: str) -> bool:
if not self._file or not item:

4
app/tests/test_cache.py
View file

@ -29,9 +29,7 @@ class TestCache:
def setup_method(self):
"""Set up test fixtures."""
# Clear any existing cache instance
if hasattr(Cache, "_instances"):
Cache._instances.clear()
Cache._reset_singleton()
self.cache = Cache()
def test_singleton_behavior(self):

7
app/tests/test_conditions.py
View file

@ -21,7 +21,6 @@ from unittest.mock import MagicMock, patch
import pytest
from app.library.conditions import Condition, Conditions
from app.library.Singleton import Singleton
class TestCondition:
@ -112,11 +111,7 @@ class TestConditions:
def setup_method(self):
"""Set up test fixtures by clearing singleton instances."""
# Clear singleton instances before each test
Singleton._instances.clear()
# Reset class variable
Conditions._items = []
Conditions._instance = None
Conditions._reset_singleton()
def test_conditions_singleton(self):
"""Test that Conditions follows singleton pattern."""

14
app/tests/test_dl_fields.py
View file

@ -22,7 +22,6 @@ from unittest.mock import MagicMock, patch
import pytest
from app.library.dl_fields import DLField, DLFields, FieldType
from app.library.Singleton import Singleton
class TestFieldType:
@ -156,12 +155,7 @@ class TestDLFields:
def setup_method(self):
"""Set up test fixtures."""
# Clear singleton instances before each test
Singleton._instances.clear()
if hasattr(DLFields, "_instances"):
DLFields._instances.clear()
# Also clear the class-level _items list
DLFields._items = []
DLFields._reset_singleton()
@pytest.fixture
def temp_file(self):
@ -219,11 +213,9 @@ class TestDLFields:
assert fields1 is fields2
@patch("app.library.dl_fields.Config")
@patch("app.library.dl_fields.EventBus")
def test_dl_fields_initialization(self, mock_event_bus, mock_config):
def test_dl_fields_initialization(self, mock_config):
"""Test DLFields initialization."""
mock_config.get_instance.return_value.config_path = "/tmp"
mock_event_bus.get_instance.return_value.subscribe = MagicMock()
with tempfile.TemporaryDirectory() as temp_dir:
temp_file = Path(temp_dir) / "test_fields.json"
@ -231,8 +223,6 @@ class TestDLFields:
fields = DLFields(file=str(temp_file))
assert fields._file == temp_file
assert fields._config is not None
mock_event_bus.get_instance.return_value.subscribe.assert_called_once()
@patch("app.library.dl_fields.Config")
def test_dl_fields_load_empty_file(self, mock_config, temp_file):

8
app/tests/test_events.py
View file

@ -627,10 +627,7 @@ class TestEventBus:
def setup_method(self):
"""Clear EventBus listeners and reset singleton before each test."""
# Reset singleton instance to allow mocking to work
EventBus._instance = None
bus = EventBus.get_instance()
bus.clear()
EventBus._reset_singleton()
@patch("app.library.config.Config")
@patch("app.library.BackgroundWorker.BackgroundWorker")
@ -656,9 +653,6 @@ class TestEventBus:
def test_event_bus_initialization(self):
"""Test EventBus initialization with new clean design."""
# Reset singleton to ensure fresh instance
EventBus._instance = None
# Create EventBus with clean initialization
bus = EventBus()

7
app/tests/test_presets.py
View file

@ -99,13 +99,6 @@ class TestPresets:
def setup_method(self):
"""Set up test environment before each test."""
# Reset singleton completely
Presets._instance = None
def teardown_method(self):
"""Clean up after each test."""
# Reset singleton completely
Presets.get_instance()._items.clear()
Presets._instance = None
Presets._reset_singleton()
@patch("app.library.Presets.Config")

552
app/tests/test_services.py Normal file
View file

@ -0,0 +1,552 @@
import logging
from unittest.mock import MagicMock, patch
import pytest
from app.library.Services import Services
class TestServices:
"""Test the Services singleton class."""
def setup_method(self):
"""Clear services before each test."""
Services._reset_singleton()
def test_singleton_behavior(self):
"""Test that Services follows singleton pattern."""
service1 = Services()
service2 = Services()
service3 = Services.get_instance()
assert service1 is service2, "Multiple Services() calls should return same instance"
assert service1 is service3, "get_instance() should return same instance"
assert id(service1) == id(service2) == id(service3), "All references should point to same object"
def test_add_and_get_service(self):
"""Test adding and retrieving services."""
services = Services()
test_service = "test_value"
services.add("test_service", test_service)
retrieved = services.get("test_service")
assert retrieved == test_service, "Should retrieve the same service that was added"
def test_get_nonexistent_service(self):
"""Test retrieving a service that doesn't exist."""
services = Services()
result = services.get("nonexistent")
assert result is None, "Should return None for nonexistent service"
def test_has_service(self):
"""Test checking if service exists."""
services = Services()
services.add("existing", "value")
assert services.has("existing") is True, "Should return True for existing service"
assert services.has("nonexistent") is False, "Should return False for nonexistent service"
def test_remove_service(self):
"""Test removing a service."""
services = Services()
services.add("to_remove", "value")
assert services.has("to_remove") is True, "Service should exist before removal"
services.remove("to_remove")
assert services.has("to_remove") is False, "Service should not exist after removal"
def test_remove_nonexistent_service(self):
"""Test removing a service that doesn't exist."""
services = Services()
# Should not raise an exception
services.remove("nonexistent")
assert services.has("nonexistent") is False
def test_clear_services(self):
"""Test clearing all services."""
services = Services()
services.add("service1", "value1")
services.add("service2", "value2")
assert len(services.get_all()) == 2, "Should have 2 services before clear"
services.clear()
assert len(services.get_all()) == 0, "Should have 0 services after clear"
def test_add_all_services(self):
"""Test adding multiple services at once."""
services = Services()
services_dict = {
"service1": "value1",
"service2": "value2",
"service3": "value3"
}
services.add_all(services_dict)
assert services.get("service1") == "value1"
assert services.get("service2") == "value2"
assert services.get("service3") == "value3"
assert len(services.get_all()) == 3
def test_get_all_returns_copy(self):
"""Test that get_all returns a copy, not the original dict."""
services = Services()
services.add("test", "value")
all_services = services.get_all()
all_services["injected"] = "malicious"
assert "injected" not in services.get_all(), "Modifying returned dict should not affect internal state"
def test_handle_sync_with_matching_args(self):
"""Test synchronous handler with matching arguments."""
services = Services()
services.add("db", "database_connection")
services.add("logger", "logger_instance")
def test_handler(db, logger):
return f"Handler called with {db} and {logger}"
result = services.handle_sync(test_handler)
expected = "Handler called with database_connection and logger_instance"
assert result == expected
def test_handle_sync_with_extra_kwargs(self):
"""Test synchronous handler with additional kwargs."""
services = Services()
services.add("db", "database_connection")
def test_handler(db, user_id):
return f"Handler called with {db} and {user_id}"
result = services.handle_sync(test_handler, user_id=123)
expected = "Handler called with database_connection and 123"
assert result == expected
def test_handle_sync_with_missing_args(self):
"""Test synchronous handler with missing arguments."""
services = Services()
services.add("db", "database_connection")
def test_handler(db_param, missing_service_param): # noqa: ARG001
return "Should not reach here"
with patch("app.library.Services.LOG") as mock_logger:
# The current implementation still calls the handler even with missing args
# This causes a TypeError, which is the actual current behavior
with pytest.raises(TypeError, match=r"missing .* required positional argument"):
services.handle_sync(test_handler)
# Should still log error about missing arguments
mock_logger.error.assert_called_once()
error_call = mock_logger.error.call_args[0][0]
assert "Missing arguments" in error_call
assert "missing_service_param" in error_call
def test_handle_sync_no_args_handler(self):
"""Test synchronous handler that takes no arguments."""
services = Services()
services.add("unused", "value")
def test_handler():
return "No args handler"
result = services.handle_sync(test_handler)
assert result == "No args handler"
@pytest.mark.asyncio
async def test_handle_async_with_matching_args(self):
"""Test asynchronous handler with matching arguments."""
services = Services()
services.add("db", "database_connection")
services.add("logger", "logger_instance")
async def test_handler(db, logger):
return f"Async handler called with {db} and {logger}"
result = await services.handle_async(test_handler)
expected = "Async handler called with database_connection and logger_instance"
assert result == expected
@pytest.mark.asyncio
async def test_handle_async_with_extra_kwargs(self):
"""Test asynchronous handler with additional kwargs."""
services = Services()
services.add("db", "database_connection")
async def test_handler(db, user_id):
return f"Async handler called with {db} and {user_id}"
result = await services.handle_async(test_handler, user_id=456)
expected = "Async handler called with database_connection and 456"
assert result == expected
@pytest.mark.asyncio
async def test_handle_async_with_missing_args(self):
"""Test asynchronous handler with missing arguments."""
services = Services()
services.add("db", "database_connection")
async def test_handler(db_param, missing_service_param): # noqa: ARG001
return "Should not reach here"
with patch("app.library.Services.LOG") as mock_logger:
# The current implementation still calls the handler even with missing args
# This causes a TypeError, which is the actual current behavior
with pytest.raises(TypeError, match=r"missing .* required positional argument"):
await services.handle_async(test_handler)
# Should still log error about missing arguments
mock_logger.error.assert_called_once()
error_call = mock_logger.error.call_args[0][0]
assert "Missing arguments" in error_call
assert "missing_service_param" in error_call
@pytest.mark.asyncio
async def test_handle_async_no_args_handler(self):
"""Test asynchronous handler that takes no arguments."""
services = Services()
services.add("unused", "value")
async def test_handler():
return "No args async handler"
result = await services.handle_async(test_handler)
assert result == "No args async handler"
def test_handle_sync_kwargs_override_services(self):
"""Test that kwargs override services with same name."""
services = Services()
services.add("param", "service_value")
def test_handler(param):
return f"Received: {param}"
result = services.handle_sync(test_handler, param="override_value")
assert result == "Received: override_value"
@pytest.mark.asyncio
async def test_handle_async_kwargs_override_services(self):
"""Test that kwargs override services with same name in async handler."""
services = Services()
services.add("param", "service_value")
async def test_handler(param):
return f"Received: {param}"
result = await services.handle_async(test_handler, param="override_value")
assert result == "Received: override_value"
def test_handle_sync_with_complex_signature(self):
"""Test synchronous handler with complex function signature."""
services = Services()
services.add("db", "database")
services.add("cache", "redis")
def complex_handler(db, cache, *args, **kwargs):
return f"db:{db}, cache:{cache}, args:{args}, kwargs:{kwargs}"
result = services.handle_sync(complex_handler, extra="value")
expected = "db:database, cache:redis, args:(), kwargs:{}"
assert result == expected
@pytest.mark.asyncio
async def test_handle_async_with_complex_signature(self):
"""Test asynchronous handler with complex function signature."""
services = Services()
services.add("db", "database")
services.add("cache", "redis")
async def complex_handler(db, cache, *args, **kwargs):
return f"db:{db}, cache:{cache}, args:{args}, kwargs:{kwargs}"
result = await services.handle_async(complex_handler, extra="value")
expected = "db:database, cache:redis, args:(), kwargs:{}"
assert result == expected
def test_service_types_preserved(self):
"""Test that different service types are preserved correctly."""
services = Services()
# Test various types
string_service = "string_value"
int_service = 42
list_service = [1, 2, 3]
dict_service = {"key": "value"}
custom_object = MagicMock()
services.add("string", string_service)
services.add("int", int_service)
services.add("list", list_service)
services.add("dict", dict_service)
services.add("object", custom_object)
assert services.get("string") == string_service
assert services.get("int") == int_service
assert services.get("list") == list_service
assert services.get("dict") == dict_service
assert services.get("object") is custom_object
def test_add_none_service(self):
"""Test adding None as a service value."""
services = Services()
services.add("none_service", None)
assert services.has("none_service") is True
assert services.get("none_service") is None
def test_service_name_edge_cases(self):
"""Test edge cases for service names."""
services = Services()
# Empty string name
services.add("", "empty_name_value")
assert services.get("") == "empty_name_value"
# Numeric string name
services.add("123", "numeric_name")
assert services.get("123") == "numeric_name"
# Special characters in name
services.add("special-chars_123!@#", "special_value")
assert services.get("special-chars_123!@#") == "special_value"
def test_overwrite_existing_service(self):
"""Test overwriting an existing service."""
services = Services()
services.add("service", "original_value")
services.add("service", "new_value")
assert services.get("service") == "new_value"
def test_singleton_persistence_across_operations(self):
"""Test that singleton behavior persists across various operations."""
# Get instance and add a service
services1 = Services()
services1.add("persistent", "value")
# Get another instance and verify service exists
services2 = Services.get_instance()
assert services2.get("persistent") == "value"
# Clear from one instance
services1.clear()
# Verify cleared in other instance
assert services2.get("persistent") is None
def test_handler_exception_propagation(self):
"""Test that exceptions in handlers are properly propagated."""
services = Services()
def failing_handler():
msg = "Handler failed"
raise ValueError(msg)
with pytest.raises(ValueError, match="Handler failed"):
services.handle_sync(failing_handler)
@pytest.mark.asyncio
async def test_async_handler_exception_propagation(self):
"""Test that exceptions in async handlers are properly propagated."""
services = Services()
async def failing_async_handler():
msg = "Async handler failed"
raise RuntimeError(msg)
with pytest.raises(RuntimeError, match="Async handler failed"):
await services.handle_async(failing_async_handler)
def test_handle_sync_with_callable_object(self):
"""Test handle_sync with callable object instead of function."""
services = Services()
services.add("data", "test_data")
class CallableHandler:
def __call__(self, data):
return f"Callable received: {data}"
handler = CallableHandler()
result = services.handle_sync(handler)
assert result == "Callable received: test_data"
@pytest.mark.asyncio
async def test_handle_async_with_callable_object(self):
"""Test handle_async with callable object instead of function."""
services = Services()
services.add("data", "test_data")
class AsyncCallableHandler:
async def __call__(self, data):
return f"Async callable received: {data}"
handler = AsyncCallableHandler()
result = await services.handle_async(handler)
assert result == "Async callable received: test_data"
def test_inspect_signature_edge_cases(self):
"""Test that inspect.signature works correctly with edge cases."""
services = Services()
# Lambda function replacement
def lambda_handler(x):
return f"Lambda: {x}"
services.add("x", "lambda_value")
result = services.handle_sync(lambda_handler)
assert result == "Lambda: lambda_value"
def test_logging_configuration(self):
"""Test that logging is properly configured."""
# This test verifies the module-level logger setup
from app.library.Services import LOG
assert isinstance(LOG, logging.Logger)
assert LOG.name == "app.library.Services"
def test_service_container_isolation(self):
"""Test that services don't interfere with each other."""
services = Services()
# Add services with potentially conflicting names
services.add("data", {"type": "database"})
services.add("data_backup", {"type": "backup"})
assert services.get("data")["type"] == "database"
assert services.get("data_backup")["type"] == "backup"
# Remove one, other should remain
services.remove("data")
assert services.get("data") is None
assert services.get("data_backup")["type"] == "backup"
def test_large_number_of_services(self):
"""Test handling a large number of services."""
services = Services()
# Add many services
num_services = 1000
for i in range(num_services):
services.add(f"service_{i}", f"value_{i}")
# Verify all exist
assert len(services.get_all()) == num_services
# Verify specific services
assert services.get("service_0") == "value_0"
assert services.get("service_500") == "value_500"
assert services.get("service_999") == "value_999"
# Clear should work efficiently
services.clear()
assert len(services.get_all()) == 0
def test_add_all_overwrites_existing(self):
"""Test that add_all overwrites existing services."""
services = Services()
services.add("existing", "original")
new_services = {
"existing": "overwritten",
"new": "value"
}
services.add_all(new_services)
assert services.get("existing") == "overwritten"
assert services.get("new") == "value"
def test_add_all_empty_dict(self):
"""Test add_all with empty dictionary."""
services = Services()
services.add("existing", "value")
services.add_all({})
# Should not affect existing services
assert services.get("existing") == "value"
assert len(services.get_all()) == 1
def test_type_var_generic_behavior(self):
"""Test that TypeVar T is handled correctly."""
services = Services()
# Add different types and ensure they're returned correctly
services.add("string", "text")
services.add("number", 42)
services.add("boolean", True) # noqa: FBT003
# Type should be preserved (runtime check)
assert isinstance(services.get("string"), str)
assert isinstance(services.get("number"), int)
assert isinstance(services.get("boolean"), bool)
def test_concurrent_access_safety(self):
"""Test basic thread safety aspects of singleton."""
import threading
import time
results = []
def get_instance():
time.sleep(0.01) # Small delay to increase chance of race condition
instance = Services()
results.append(id(instance))
# Create multiple threads
threads = []
for _ in range(10):
thread = threading.Thread(target=get_instance)
threads.append(thread)
thread.start()
# Wait for all threads
for thread in threads:
thread.join()
# All should be the same instance
assert len(set(results)) == 1, "All threads should get the same singleton instance"
def test_method_chaining_possibility(self):
"""Test that methods can be potentially chained."""
services = Services()
# While current implementation doesn't return self, test the pattern works
services.add("test1", "value1")
services.add("test2", "value2")
services.remove("test1")
assert services.get("test1") is None
assert services.get("test2") == "value2"
def test_edge_case_empty_handler_name(self):
"""Test handlers with minimal or no names."""
services = Services()
services.add("param", "value")
# Anonymous lambda
result = services.handle_sync(lambda param: f"anon: {param}")
assert result == "anon: value"
# Function with minimal signature info
def minimal(param): return param
result = services.handle_sync(minimal)
assert result == "value"
def test_services_state_isolation(self):
"""Test that different Services instances share state properly."""
# This test verifies the singleton behavior more thoroughly
s1 = Services()
s1.add("shared", "data")
s2 = Services.get_instance()
assert s2.get("shared") == "data"
s2.add("another", "value")
assert s1.get("another") == "value"
# Clear from s1 affects s2
s1.clear()
assert len(s2.get_all()) == 0

25
app/tests/test_utils.py
View file

@ -26,7 +26,6 @@ from app.library.Utils import (
encrypt_data,
extract_info,
extract_ytdlp_logs,
find_unpickleable,
get,
get_archive_id,
get_file,
@ -1424,30 +1423,6 @@ class TestYtdlpReject:
assert isinstance(message, str)
class TestFindUnpickleable:
"""Test the find_unpickleable function."""
def test_find_unpickleable_simple(self):
"""Test with simple pickleable object."""
obj = {"key": "value", "number": 42}
try:
find_unpickleable(obj)
# Should not find any unpickleable items
except Exception:
# Function might raise exceptions for complex objects
pass
def test_find_unpickleable_complex(self):
"""Test with complex object."""
obj = {"func": lambda x: x} # Lambda is not pickleable
try:
find_unpickleable(obj)
except Exception:
pass
class TestInitClass:
"""Test the init_class function."""