diff --git a/Pipfile b/Pipfile index 2a392ea8..a7693427 100644 --- a/Pipfile +++ b/Pipfile @@ -22,6 +22,7 @@ mutagen = "*" brotli = "*" brotlicffi = "*" yt-dlp = "*" +anyio = "*" [dev-packages] diff --git a/Pipfile.lock b/Pipfile.lock index f85c38f2..47831ff2 100644 --- a/Pipfile.lock +++ b/Pipfile.lock @@ -1,7 +1,7 @@ { "_meta": { "hash": { - "sha256": "cd9a56440fe36b901993ced28ee6ef8d301a5125344cda86519fb7c54ede1449" + "sha256": "17668280c8e863033313b83f64240f6837e733957fbc2613767d297fee6c6fc2" }, "pipfile-spec": 6, "requires": { @@ -128,6 +128,7 @@ "sha256:1d9fe889df5212298c0c0723fa20479d1b94883a2df44bd3897aa91083316f7a", "sha256:b5011f270ab5eb0abf13385f851315585cc37ef330dd88e27ec3d34d651fd47a" ], + "index": "pypi", "markers": "python_version >= '3.9'", "version": "==4.8.0" }, diff --git a/app/library/AsyncPool.py b/app/library/AsyncPool.py index e619cc36..f6b58a9a 100644 --- a/app/library/AsyncPool.py +++ b/app/library/AsyncPool.py @@ -14,9 +14,9 @@ class AsyncPool: worker_co, name: str, logger: logging.Logger, - loop: asyncio.AbstractEventLoop|None = None, + loop: asyncio.AbstractEventLoop | None = None, load_factor: int = 1, - max_task_time: int|None = None, + max_task_time: int | None = None, return_futures: bool = False, raise_on_join: bool = False, ): @@ -54,6 +54,12 @@ class AsyncPool: self._status: dict[str, dict | None] = {} async def _worker_loop(self, worker_id: str): + """ + This is the main loop for each worker. It will pull from the queue and call the worker_co + coroutine with the arguments passed to `push`. + :param worker_id: id of the worker + :return: None + """ while True: got_obj = False future = None @@ -84,7 +90,7 @@ class AsyncPool: raise except BaseException as e: if isinstance(e, asyncio.exceptions.CancelledError): - raise e + raise self._exceptions = True @@ -92,7 +98,7 @@ class AsyncPool: # don't log the failure when the client is receiving the future future.set_exception(e) else: - self._logger.exception(f"Worker call failed. {str(e)}") + self._logger.exception(f"Worker call failed. {e!s}") finally: self._status[worker_id] = None @@ -153,7 +159,12 @@ class AsyncPool: self._create_worker(worker_id) async def restart(self, worker_id: str, msg: str = None) -> bool: - """Will restart the worker pool""" + """ + Will restart the worker pool + :param worker_id: worker id to restart + :param msg: message to send to the worker + :return: True if worker was restarted + """ if worker_id not in self._workers: self._logger.warning(f"Worker {worker_id} does not exist.") return False @@ -162,7 +173,7 @@ class AsyncPool: self._workers[worker_id].cancel(msg) await self._workers[worker_id] except asyncio.exceptions.CancelledError as e: - self._logger.warning(f"Worker {worker_id} restarted. {str(e)}") + self._logger.warning(f"Worker {worker_id} restarted. {e!s}") if worker_id in self._status: self._status.pop(worker_id) @@ -173,8 +184,20 @@ class AsyncPool: return True + async def on_shutdown(self, _) -> None: + """Will shutdown the worker pool""" + try: + await asyncio.wait_for(asyncio.gather(*[self.stop(worker_id) for worker_id in self._workers]), timeout=10) + except Exception: + self._logger.error(f"Exception shutting down {self._name}") + async def stop(self, worker_id: str, msg: str = None) -> bool: - """Will stop the worker""" + """ + Will stop the worker + :param worker_id: worker id to stop + :param msg: message to send to the worker + :return: True if worker was stopped + """ if worker_id not in self._workers: self._logger.warning(f"Worker {worker_id} does not exist.") return False @@ -183,7 +206,7 @@ class AsyncPool: try: await self._workers[worker_id] except asyncio.exceptions.CancelledError as e: - self._logger.warning(f"Worker {worker_id} stopped. {str(e)}") + self._logger.warning(f"Worker {worker_id} stopped. {e!s}") if worker_id in self._status: self._status.pop(worker_id) @@ -214,7 +237,8 @@ class AsyncPool: self._logger.info(f"Completed {self._name}") if self._exceptions and self._raise_on_join: - raise Exception(f"Exception occurred in {self._name} pool") + msg = f"Exception occurred in {self._name} pool" + raise Exception(msg) def _create_worker(self, worker_id: str) -> asyncio.Future: if worker_id in self._workers: diff --git a/app/library/DataStore.py b/app/library/DataStore.py index 63ce318a..372e6569 100644 --- a/app/library/DataStore.py +++ b/app/library/DataStore.py @@ -1,7 +1,7 @@ import copy import json from collections import OrderedDict -from datetime import datetime, timezone +from datetime import UTC, datetime from email.utils import formatdate from sqlite3 import Connection @@ -39,13 +39,9 @@ class DataStore: if key and key in self.dict: return True - for i in self.dict: - if (key and self.dict[i].info._id == key) or (url and self.dict[i].info.url == url): - return True + return any((key and self.dict[i].info._id == key) or (url and self.dict[i].info.url == url) for i in self.dict) - return False - - def get(self, key: str, url: str|None = None) -> Download: + def get(self, key: str, url: str | None = None) -> Download: if not key and not url: msg = "key or url must be provided." raise KeyError(msg) @@ -57,8 +53,8 @@ class DataStore: msg = f"{key=} or {url=} not found." raise KeyError(msg) - def getById(self, id: str) -> Download | None: - return self.dict[id] if id in self.dict else None + def get_by_id(self, id: str) -> Download | None: + return self.dict.get(id, None) def items(self) -> list[tuple[str, Download]]: return self.dict.items() @@ -71,25 +67,25 @@ class DataStore: ) for row in cursor: - rowDate = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") + rowDate = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007 data: dict = json.loads(row["data"]) key: str = data.pop("_id") item: ItemDTO = ItemDTO(**data) item._id = key - item.datetime = formatdate(rowDate.replace(tzinfo=timezone.utc).timestamp()) + item.datetime = formatdate(rowDate.replace(tzinfo=UTC).timestamp()) items.append((row["id"], item)) return items def put(self, value: Download) -> Download: self.dict.update({value.info._id: value}) - self._updateStoreItem(self.type, value.info) + self._update_store_item(self.type, value.info) return self.dict[value.info._id] def delete(self, key: str) -> None: self.dict.pop(key, None) - self._deleteStoreItem(key) + self._delete_store_item(key) def next(self) -> tuple[str, Download]: return next(iter(self.dict.items())) @@ -97,17 +93,13 @@ class DataStore: def empty(self): return not bool(self.dict) - def hasDownloads(self): + def has_downloads(self): if 0 == len(self.dict): return False - for key in self.dict: - if self.dict[key].started() is False: - return True + return any(self.dict[key].started() is False for key in self.dict) - return False - - def getNextDownload(self) -> Download: + def get_next_download(self) -> Download: for key in self.dict: if self.dict[key].started() is False and self.dict[key].is_cancelled() is False: return self.dict[key] @@ -118,7 +110,7 @@ class DataStore: self.connection.execute('SELECT "id" FROM "history" LIMIT 1').fetchone() return True - def _updateStoreItem(self, type: str, item: ItemDTO) -> None: + def _update_store_item(self, type: str, item: ItemDTO) -> None: sqlStatement = """ INSERT INTO "history" ("id", "type", "url", "data") VALUES (?, ?, ?, ?) @@ -149,11 +141,11 @@ class DataStore: type, stored.url, stored.json(), - datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"), + datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"), ), ) - def _deleteStoreItem(self, key: str) -> None: + def _delete_store_item(self, key: str) -> None: self.connection.execute( 'DELETE FROM "history" WHERE "type" = ? AND "id" = ?', ( diff --git a/app/library/Download.py b/app/library/Download.py index bd75fcca..c2b041a5 100644 --- a/app/library/Download.py +++ b/app/library/Download.py @@ -15,7 +15,7 @@ from .config import Config from .Emitter import Emitter from .ffprobe import ffprobe from .ItemDTO import ItemDTO -from .Utils import get_opts, jsonCookie, mergeConfig +from .Utils import get_opts, json_cookie, merge_config LOG = logging.getLogger("download") @@ -34,7 +34,7 @@ class Download: info: ItemDTO = None default_ytdl_opts: dict = None debug: bool = False - tempPath: str = None + temp_path: str = None emitter: Emitter = None cancelled: bool = False is_live: bool = False @@ -64,7 +64,7 @@ class Download: ) "Fields to be extracted from yt-dlp progress hook." - tempKeep: bool = False + temp_keep: bool = False "Keep temp folder after download." def __init__(self, info: ItemDTO, info_dict: dict = None, debug: bool = False): @@ -86,7 +86,7 @@ class Download: self.proc = None self.emitter = None self.max_workers = int(config.max_workers) - self.tempKeep = bool(config.temp_keep) + self.temp_keep = bool(config.temp_keep) self.is_live = bool(info.is_live) or info.live_in is not None self.is_manifestless = "is_manifestless" in self.info.options and self.info.options["is_manifestless"] is True self.info_dict = info_dict @@ -94,7 +94,7 @@ class Download: def _progress_hook(self, data: dict): dataDict = {k: v for k, v in data.items() if k in self._ytdlp_fields} - if "finished" == data.get("status", None) and data.get("info_dict", {}).get("filename", None): + if "finished" == data.get("status") and data.get("info_dict", {}).get("filename", None): dataDict["filename"] = data["info_dict"]["filename"] self.status_queue.put({"id": self.id, **dataDict}) @@ -115,11 +115,11 @@ class Download: def _download(self): try: - params: dict = get_opts(self.preset, mergeConfig(self.default_ytdl_opts, self.ytdl_opts)) + params: dict = get_opts(self.preset, merge_config(self.default_ytdl_opts, self.ytdl_opts)) params.update( { "color": "no_color", - "paths": {"home": self.download_dir, "temp": self.tempPath}, + "paths": {"home": self.download_dir, "temp": self.temp_path}, "outtmpl": {"default": self.template, "chapter": self.template_chapter}, "noprogress": True, "break_on_existing": True, @@ -138,17 +138,17 @@ class Download: if self.info.cookies: try: - data = jsonCookie(json.loads(self.info.cookies)) + data = json_cookie(json.loads(self.info.cookies)) if not data: LOG.warning( f"The cookie string that was provided for {self.info.title} is empty or not in expected spec." ) - with open(os.path.join(self.tempPath, f"cookie_{self.info._id}.txt"), "w") as f: + with open(os.path.join(self.temp_path, f"cookie_{self.info._id}.txt"), "w") as f: f.write(data) params["cookiefile"] = f.name except ValueError as e: - LOG.error(f"Invalid cookies: was provided for '{self.info.title}'. '{str(e)}'.") + LOG.error(f"Invalid cookies: was provided for '{self.info.title}'. '{e!s}'.") if self.is_live or self.is_manifestless: hasDeletedOptions = False @@ -191,10 +191,10 @@ class Download: self.status_queue = Config.get_manager().Queue() # Create temp dir for each download. - self.tempPath = os.path.join(self.temp_dir, hashlib.shake_256(f"D-{self.info.id}".encode("utf-8")).hexdigest(5)) + self.temp_path = os.path.join(self.temp_dir, hashlib.shake_256(f"D-{self.info.id}".encode()).hexdigest(5)) - if not os.path.exists(self.tempPath): - os.makedirs(self.tempPath, exist_ok=True) + if not os.path.exists(self.temp_path): + os.makedirs(self.temp_path, exist_ok=True) self.proc = multiprocessing.Process(name=f"download-{self.id}", target=self._download) self.proc.start() @@ -287,26 +287,26 @@ class Download: return False def delete_temp(self): - if self.tempKeep is True or not self.tempPath: + if self.temp_keep is True or not self.temp_path: return if "finished" != self.info.status and self.is_live: LOG.warning( - f"Keeping live temp folder '{self.tempPath}', as the reported status is not finished '{self.info.status}'." + f"Keeping live temp folder '{self.temp_path}', as the reported status is not finished '{self.info.status}'." ) return - if not os.path.exists(self.tempPath): + if not os.path.exists(self.temp_path): return - if self.tempPath == self.temp_dir: + if self.temp_path == self.temp_dir: LOG.warning( - f"Attempted to delete video temp folder: {self.tempPath}, but it is the same as main temp folder." + f"Attempted to delete video temp folder: {self.temp_path}, but it is the same as main temp folder." ) return - LOG.info(f"Deleting Temp folder '{self.tempPath}'.") - shutil.rmtree(self.tempPath, ignore_errors=True) + LOG.info(f"Deleting Temp folder '{self.temp_path}'.") + shutil.rmtree(self.temp_path, ignore_errors=True) async def progress_update(self): """ @@ -342,7 +342,6 @@ class Download: self.info.file_size = os.path.getsize(status.get("filename")) except FileNotFoundError: self.info.file_size = 0 - pass self.info.status = status.get("status", self.info.status) self.info.msg = status.get("msg") diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index e41169c8..a14fd8f9 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -1,13 +1,14 @@ import asyncio -import datetime import json import logging import os import time +from datetime import UTC, datetime from email.utils import formatdate from sqlite3 import Connection import yt_dlp +from aiohttp import web from .AsyncPool import AsyncPool from .config import Config @@ -17,7 +18,7 @@ from .Emitter import Emitter from .EventsSubscriber import Events from .ItemDTO import ItemDTO from .Singleton import Singleton -from .Utils import ExtractInfo, calcDownloadPath, get_opts, isDownloaded, mergeConfig +from .Utils import calc_download_path, extract_info, get_opts, is_downloaded, merge_config LOG = logging.getLogger("DownloadQueue") @@ -42,6 +43,8 @@ class DownloadQueue(metaclass=Singleton): pool: AsyncPool | None = None """Pool of workers to download the files.""" + _active_downloads: dict[str, Download] = {} + _instance = None """Instance of the DownloadQueue.""" @@ -65,12 +68,32 @@ class DownloadQueue(metaclass=Singleton): Returns: DownloadQueue: The instance of the DownloadQueue + """ if not DownloadQueue._instance: DownloadQueue._instance = DownloadQueue() return DownloadQueue._instance + def attach(self, app: web.Application): + """ + Attach the download queue to the application. + + Args: + app (web.Application): The application to attach the download queue to. + + """ + app.on_startup.append(lambda _: self.initialize()) + # app.on_shutdown.append(self.on_shutdown) + + # async def close_pool(_: web.Application): + # try: + # await self.pool.on_shutdown(_) + # except Exception as e: + # LOG.error(f"Failed to cleanup download pool. {e!s}") + + # app.on_cleanup.append(close_pool) + async def test(self) -> bool: """ Test the datastore connection to the database. @@ -89,9 +112,9 @@ class DownloadQueue(metaclass=Singleton): LOG.info( f"Using '{self.config.max_workers}' worker/s for downloading. Can be configured via `YTP_MAX_WORKERS` environment variable." ) - asyncio.create_task(self.__download_pool(), name="download_pool") + asyncio.create_task(self._download_pool(), name="download_pool") - def pause(self) -> bool: + def pause(self, shutdown: bool = False) -> bool: """ Pause the download queue. @@ -101,7 +124,8 @@ class DownloadQueue(metaclass=Singleton): """ if self.paused.is_set(): self.paused.clear() - LOG.warning(f"Download paused at. {datetime.datetime.now().isoformat()}") + if not shutdown: + LOG.warning(f"Download paused at. {datetime.now(tz=UTC).isoformat()}") return True return False @@ -116,12 +140,12 @@ class DownloadQueue(metaclass=Singleton): """ if not self.paused.is_set(): self.paused.set() - LOG.warning(f"Downloading resumed at. {datetime.datetime.now().isoformat()}") + LOG.warning(f"Downloading resumed at. {datetime.now(tz=UTC).isoformat()}") return True return False - def isPaused(self) -> bool: + def is_paused(self) -> bool: """ Check if the download queue is paused. @@ -131,6 +155,15 @@ class DownloadQueue(metaclass=Singleton): """ return not self.paused.is_set() + async def on_shutdown(self, _: web.Application): + LOG.debug("Canceling all active downloads.") + if self._active_downloads: + self.pause() + try: + await self.cancel(list(self._active_downloads.keys())) + except Exception as e: + LOG.error(f"Failed to cancel downloads. {e!s}") + async def __add_entry( self, entry: dict, @@ -169,13 +202,14 @@ class DownloadQueue(metaclass=Singleton): live_in: str | None = None eventType = entry.get("_type") or "video" + if "playlist" == eventType: entries = entry.get("entries", []) playlist_index_digits = len(str(len(entries))) results = [] for i, etr in enumerate(entries, start=1): etr["playlist"] = entry.get("id") - etr["playlist_index"] = "{{0:0{0:d}d}}".format(playlist_index_digits).format(i) + etr["playlist_index"] = f"{{0:0{playlist_index_digits:d}d}}".format(i) for property in ("id", "title", "uploader", "uploader_id"): if property in entry: etr[f"playlist_{property}"] = entry.get(property) @@ -202,7 +236,8 @@ class DownloadQueue(metaclass=Singleton): } return {"status": "ok"} - elif ("video" == eventType or eventType.startswith("url")) and "id" in entry and "title" in entry: + + if ("video" == eventType or eventType.startswith("url")) and "id" in entry and "title" in entry: # check if the video is live stream. if "live_status" in entry and "is_upcoming" == entry.get("live_status"): if "release_timestamp" in entry and entry.get("release_timestamp"): @@ -210,11 +245,9 @@ class DownloadQueue(metaclass=Singleton): else: error = "Live stream not yet started. And no date is set." else: - error = entry.get("msg", None) + error = entry.get("msg") - LOG.debug( - f"Entry id '{entry.get('id', None)}' url '{entry.get('webpage_url', None)} - {entry.get('url', None)}'." - ) + LOG.debug(f"Entry id '{entry.get('id')}' url '{entry.get('webpage_url')} - {entry.get('url')}'.") if self.done.exists(key=entry["id"], url=str(entry.get("webpage_url") or entry.get("url"))): item = self.done.get(key=entry["id"], url=entry.get("webpage_url") or entry["url"]) @@ -234,10 +267,10 @@ class DownloadQueue(metaclass=Singleton): options.update({"is_manifestless": is_manifestless}) live_status: list = ["is_live", "is_upcoming"] - is_live = bool(entry.get("is_live", None) or live_in or entry.get("live_status", None) in live_status) + is_live = bool(entry.get("is_live") or live_in or entry.get("live_status") in live_status) try: - download_dir = calcDownloadPath(basePath=self.config.download_path, folder=folder) + download_dir = calc_download_path(base_path=self.config.download_path, folder=folder) except Exception as e: LOG.exception(e) return {"status": "error", "msg": str(e)} @@ -245,7 +278,7 @@ class DownloadQueue(metaclass=Singleton): extras: dict = {} fields: tuple = ("uploader", "channel", "thumbnail") for field in fields: - if entry.get(field, None): + if entry.get(field): extras[field] = entry.get(field) dl = ItemDTO( @@ -253,7 +286,6 @@ class DownloadQueue(metaclass=Singleton): title=str(entry.get("title")), url=str(entry.get("webpage_url") or entry.get("url")), preset=preset, - thumbnail=entry.get("thumbnail", None), folder=folder, download_dir=download_dir, temp_dir=self.config.temp_path, @@ -275,7 +307,7 @@ class DownloadQueue(metaclass=Singleton): dlInfo: Download = Download(info=dl, info_dict=entry, debug=bool(self.config.ytdl_debug)) - if dlInfo.info.live_in or "is_upcoming" == entry.get("live_status", None): + if dlInfo.info.live_in or "is_upcoming" == entry.get("live_status"): dlInfo.info.status = "not_live" itemDownload = self.done.put(dlInfo) NotifyEvent = Events.COMPLETED @@ -294,7 +326,8 @@ class DownloadQueue(metaclass=Singleton): ) return {"status": "ok"} - elif eventType.startswith("url"): + + if eventType.startswith("url"): return await self.add( url=str(entry.get("url")), preset=preset, @@ -312,7 +345,7 @@ class DownloadQueue(metaclass=Singleton): url: str, preset: str, folder: str, - config: dict|None = None, + config: dict | None = None, cookies: str = "", template: str = "", already=None, @@ -320,7 +353,7 @@ class DownloadQueue(metaclass=Singleton): config = config if config else {} folder = str(folder) if folder else "" - filePath = calcDownloadPath(basePath=self.config.download_path, folder=folder) + filePath = calc_download_path(base_path=self.config.download_path, folder=folder) LOG.info( f"Adding 'URL: {url}' to 'Folder: {filePath}' with 'Preset: {preset}' 'Naming: {template}', 'Cookies: {cookies}' 'YTConfig: {config}'." @@ -342,7 +375,7 @@ class DownloadQueue(metaclass=Singleton): already.add(url) try: - downloaded, id_dict = self.isDownloaded(url) + downloaded, id_dict = self._is_downloaded(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) @@ -354,8 +387,8 @@ class DownloadQueue(metaclass=Singleton): entry = await asyncio.wait_for( fut=asyncio.get_running_loop().run_in_executor( None, - ExtractInfo, - get_opts(preset, mergeConfig(self.config.ytdl_options, config)), + extract_info, + get_opts(preset, merge_config(self.config.ytdl_options, config)), url, bool(self.config.ytdl_debug), ), @@ -392,6 +425,16 @@ class DownloadQueue(metaclass=Singleton): ) async def cancel(self, ids: list[str]) -> dict[str, str]: + """ + Cancel the download. + + Args: + ids (list): The list of ids to cancel. + + Returns: + dict: The status of the operation. + + """ status: dict[str, str] = {"status": "ok"} for id in ids: @@ -403,29 +446,40 @@ class DownloadQueue(metaclass=Singleton): LOG.warning(f"Requested cancel for non-existent download {id=}. {e!s}") continue - itemMessage = f"{id=} {item.info.id=} {item.info.title=}" + item_ref = f"{id=} {item.info.id=} {item.info.title=}" if item.running(): - LOG.debug(f"Canceling {itemMessage}") + LOG.debug(f"Canceling {item_ref}") item.cancel() - LOG.info(f"Cancelled {itemMessage}") + LOG.info(f"Cancelled {item_ref}") await item.close() else: await item.close() - LOG.debug(f"Deleting from queue {itemMessage}") + LOG.debug(f"Deleting from queue {item_ref}") self.queue.delete(id) asyncio.create_task(self.emitter.cancelled(dl=item.info.serialize()), name=f"notifier-c-{id}") item.info.status = "cancelled" item.info.error = "Cancelled by user." self.done.put(item) asyncio.create_task(self.emitter.completed(dl=item.info.serialize()), name=f"notifier-d-{id}") - LOG.info(f"Deleted from queue {itemMessage}") + LOG.info(f"Deleted from queue {item_ref}") status[id] = "ok" return status async def clear(self, ids: list[str], remove_file: bool = False) -> dict[str, str]: + """ + Clear the download history. + + Args: + ids (list): The list of ids to clear. + remove_file (bool): True to remove the file, False otherwise. Default is False. + + Returns: + dict: The status of the operation. + + """ status: dict[str, str] = {"status": "ok"} for id in ids: @@ -447,10 +501,10 @@ class DownloadQueue(metaclass=Singleton): filename = f"{item.info.folder}/{item.info.filename}" try: - realFile: str = calcDownloadPath( - basePath=self.config.download_path, + realFile: str = calc_download_path( + base_path=self.config.download_path, folder=filename, - createPath=False, + create_path=False, ) if realFile and os.path.exists(realFile): os.remove(realFile) @@ -497,11 +551,18 @@ class DownloadQueue(metaclass=Singleton): return items - async def __download_pool(self): + async def _download_pool(self) -> None: + """ + Create a pool of workers to download the files. + + Returns: + None + + """ self.pool = AsyncPool( loop=asyncio.get_running_loop(), num_workers=self.config.max_workers, - worker_co=self.__downloadFile, + worker_co=self._download_file, name="download_pool", logger=logging.getLogger("WorkerPool"), ) @@ -519,19 +580,19 @@ class DownloadQueue(metaclass=Singleton): LOG.info(f"Waiting for worker to be free. {self.pool.get_workers_status()}") await asyncio.sleep(1) - while not self.queue.hasDownloads(): + while not self.queue.has_downloads(): LOG.info(f"Waiting for item to download. '{self.pool.get_available_workers()}' free workers.") if self.event: await self.event.wait() self.event.clear() LOG.debug("Cleared wait event.") - if self.paused and isinstance(self.paused, asyncio.Event) and self.isPaused(): + if self.paused and isinstance(self.paused, asyncio.Event) and self.is_paused(): LOG.info("Download pool is paused.") await self.paused.wait() LOG.info("Download pool resumed downloading.") - entry = self.queue.getNextDownload() + entry = self.queue.get_next_download() await asyncio.sleep(0.2) if entry is None: @@ -544,13 +605,25 @@ class DownloadQueue(metaclass=Singleton): LOG.debug(f"Pushed {entry=} to executor.") await asyncio.sleep(1) - async def __downloadFile(self, id: str, entry: Download): - filePath = calcDownloadPath(basePath=self.config.download_path, folder=entry.info.folder) + async def _download_file(self, id: str, entry: Download) -> None: + """ + Download the file. + + Args: + id (str): The id of the download. + entry (Download): The download entry. + + Returns: + None + + """ + filePath = calc_download_path(base_path=self.config.download_path, folder=entry.info.folder) LOG.info( f"Downloading 'id: {id}', 'Title: {entry.info.title}', 'URL: {entry.info.url}' to 'Folder: {filePath}'." ) try: + self._active_downloads[entry.info._id] = entry await entry.start(self.emitter) if "finished" != entry.info.status: @@ -563,6 +636,9 @@ class DownloadQueue(metaclass=Singleton): entry.info.status = "error" finally: + if entry.info._id in self._active_downloads: + self._active_downloads.pop(entry.info._id, None) + await entry.close() if self.queue.exists(key=id): @@ -582,8 +658,18 @@ class DownloadQueue(metaclass=Singleton): if self.event: self.event.set() - def isDownloaded(self, url: str) -> tuple[bool, dict | None]: + def _is_downloaded(self, url: str) -> tuple[bool, dict | None]: + """ + Check if the url has been downloaded already. + + Args: + url (str): The url to check. + + Returns: + tuple: A tuple with the status of the operation and the id of the downloaded item. + + """ if not url or not self.config.keep_archive: return False, None - return isDownloaded(self.config.ytdl_options.get("download_archive", None), url) + return is_downloaded(self.config.ytdl_options.get("download_archive", None), url) diff --git a/app/library/Emitter.py b/app/library/Emitter.py index b0ef2d89..8348ab70 100644 --- a/app/library/Emitter.py +++ b/app/library/Emitter.py @@ -1,6 +1,6 @@ import asyncio import logging -from typing import Awaitable +from collections.abc import Awaitable from .EventsSubscriber import Events from .Singleton import Singleton @@ -69,19 +69,19 @@ class Emitter(metaclass=Singleton): async def cleared(self, dl: dict | None = None, local: bool = False, **kwargs): await self.emit(Events.CLEARED, data=dl, local=local, **kwargs) - async def error(self, message: str, data: dict = {}, local: bool = False, **kwargs): + async def error(self, message: str, data: dict|None = None, local: bool = False, **kwargs): msg = {"type": "error", "message": message, "data": {}} if data: msg.update({"data": data}) await self.emit(Events.ERROR, data=msg, local=local, **kwargs) - async def info(self, message: str, data: dict = {}, local: bool = False, **kwargs): + async def info(self, message: str, data: dict|None = None, local: bool = False, **kwargs): msg = {"type": "info", "message": message, "data": {}} if data: msg.update({"data": data}) await self.emit(Events.LOG_INFO, data=msg, local=local, **kwargs) - async def success(self, message: str, data: dict = {}, local: bool = False, **kwargs): + async def success(self, message: str, data: dict|None = None, local: bool = False, **kwargs): msg = {"type": "success", "message": message, "data": {}} if data: msg.update({"data": data}) diff --git a/app/library/EventsSubscriber.py b/app/library/EventsSubscriber.py index f97123b7..2fb5a683 100644 --- a/app/library/EventsSubscriber.py +++ b/app/library/EventsSubscriber.py @@ -1,7 +1,7 @@ import asyncio import logging +from collections.abc import Awaitable from dataclasses import dataclass -from typing import Awaitable from .Singleton import Singleton @@ -180,7 +180,7 @@ class EventsSubscriber(metaclass=Singleton): tasks.append(asyncio.create_task(callback(event, data))) except Exception as e: - LOG.error(f"Failed to emit event '{event}' to '{id}'. Error message '{e!s}'.") LOG.exception(e) + LOG.error(f"Failed to emit event '{event}' to '{id}'. Error message '{e!s}'.") return asyncio.gather(*tasks) diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index 76759e2c..f11e087b 100644 --- a/app/library/HttpAPI.py +++ b/app/library/HttpAPI.py @@ -8,9 +8,9 @@ import os import random import time import uuid -from datetime import datetime +from collections.abc import Awaitable +from datetime import UTC, datetime from pathlib import Path -from typing import Awaitable import httpx import magic @@ -31,7 +31,7 @@ from .Playlist import Playlist from .Segments import Segments from .Subtitle import Subtitle from .Tasks import Task, Tasks -from .Utils import StreamingError, arg_converter, calcDownloadPath, getVideoInfo, validate_url, validate_uuid +from .Utils import StreamingError, arg_converter, calc_download_path, get_video_info, validate_url, validate_uuid LOG = logging.getLogger("http_api") MIME = magic.Magic(mime=True) @@ -93,6 +93,9 @@ class HttpAPI(Common): return decorator + async def on_shutdown(self, _: web.Application): + LOG.debug("Shutting down http API server.") + def attach(self, app: web.Application) -> "HttpAPI": """ Attach the routes to the application. @@ -123,9 +126,10 @@ class HttpAPI(Common): except Exception as e: LOG.exception(e) + app.on_shutdown.append(self.on_shutdown) return self - async def staticFile(self, req: Request) -> Response: + async def _static_file(self, req: Request) -> Response: """ Preload static files from the ui/exported folder. @@ -154,7 +158,7 @@ class HttpAPI(Common): status=web.HTTPOk.status_code, ) - def preloadStatic(self, app: web.Application) -> "HttpAPI": + def _preload_static(self, app: web.Application) -> "HttpAPI": """ Preload static files from the ui/exported folder. @@ -187,7 +191,7 @@ class HttpAPI(Common): self._static_holder[urlPath] = {"content": content, "content_type": contentType} LOG.debug(f"Preloading '{urlPath}'.") - app.router.add_get(urlPath, self.staticFile) + app.router.add_get(urlPath, self._static_file) preloaded += 1 if urlPath.endswith("/index.html") and urlPath != "/index.html": @@ -195,8 +199,8 @@ class HttpAPI(Common): parentNoSlash = urlPath.replace("/index.html", "") self._static_holder[parentSlash] = {"content": content, "content_type": contentType} self._static_holder[parentNoSlash] = {"content": content, "content_type": contentType} - app.router.add_get(parentSlash, self.staticFile) - app.router.add_get(parentNoSlash, self.staticFile) + app.router.add_get(parentSlash, self._static_file) + app.router.add_get(parentNoSlash, self._static_file) preloaded += 2 if preloaded < 1: @@ -232,7 +236,7 @@ class HttpAPI(Common): self.routes.route(method._http_method, f"/{http_path}")(method) self.routes.static("/api/download/", self.config.download_path) - self.preloadStatic(app) + self._preload_static(app) try: app.add_routes(self.routes) @@ -389,7 +393,7 @@ class HttpAPI(Common): "proxy": self.config.ytdl_options.get("proxy", None), } - data = getVideoInfo(url=url, ytdlp_opts=opts, no_archive=True) + data = get_video_info(url=url, ytdlp_opts=opts, no_archive=True) self.cache.set(key=self.cache.hash(url), value=data, ttl=300) data["_cached"] = { "key": self.cache.hash(url), @@ -478,7 +482,7 @@ class HttpAPI(Common): """ return web.json_response( - data=Tasks.get_instance().getTasks(), status=web.HTTPOk.status_code, dumps=self.encoder.encode + data=Tasks.get_instance().get_tasks(), status=web.HTTPOk.status_code, dumps=self.encoder.encode ) @route("PUT", "api/tasks") @@ -543,7 +547,7 @@ class HttpAPI(Common): tasks.append(Task(**item)) try: - tasks = ins.save(tasks=tasks).load().getTasks() + tasks = ins.save(tasks=tasks).load().get_tasks() except Exception as e: LOG.exception(e) return web.json_response( @@ -597,7 +601,7 @@ class HttpAPI(Common): if not id: return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code) - item = self.queue.done.getById(id) + item = self.queue.done.get_by_id(id) if not item: return web.json_response(data={"error": "item not found."}, status=web.HTTPNotFound.status_code) @@ -860,7 +864,7 @@ class HttpAPI(Common): if request.if_modified_since: lastMod = time.strftime( - "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple() + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path), tz=UTC).timetuple() ) if os.path.exists(file_path) and request.if_modified_since.timestamp() == os.path.getmtime(file_path): return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod}) @@ -873,7 +877,7 @@ class HttpAPI(Common): segmenter = Segments( index=int(segment), - duration=float("{:.6f}".format(float(sd if sd else M3u8.duration))), + duration=float(f"{float(sd if sd else M3u8.duration):.6f}"), vconvert=vc == 1, aconvert=ac == 1, ) @@ -887,10 +891,10 @@ class HttpAPI(Common): "Pragma": "public", "Cache-Control": f"public, max-age={time.time() + 31536000}", "Last-Modified": time.strftime( - "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple() + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path), tz=UTC).timetuple() ), "Expires": time.strftime( - "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000).timetuple() + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000, tz=UTC).timetuple() ), }, status=web.HTTPOk.status_code, @@ -915,7 +919,7 @@ class HttpAPI(Common): if request.if_modified_since: lastMod = time.strftime( - "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple() + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path), tz=UTC).timetuple() ) if os.path.exists(file_path) and request.if_modified_since.timestamp() == os.path.getmtime(file_path): return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod}) @@ -932,10 +936,10 @@ class HttpAPI(Common): "Pragma": "public", "Cache-Control": f"public, max-age={time.time() + 31536000}", "Last-Modified": time.strftime( - "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple() + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path), tz=UTC).timetuple() ), "Expires": time.strftime( - "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000).timetuple() + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000, tz=UTC).timetuple() ), }, status=web.HTTPOk.status_code, @@ -1008,7 +1012,8 @@ class HttpAPI(Common): "Access-Control-Allow-Origin": "*", "Cache-Control": f"public, max-age={time.time() + 31536000}", "Expires": time.strftime( - "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000).timetuple() + "%a, %d %b %Y %H:%M:%S GMT", + datetime.fromtimestamp(time.time() + 31536000, tz=UTC).timetuple(), ), }, ) @@ -1031,9 +1036,12 @@ class HttpAPI(Common): """ file: str = request.match_info.get("file") + if not file: + return web.json_response(data={"error": "file is required."}, status=web.HTTPBadRequest.status_code) + try: - realFile: str = calcDownloadPath(basePath=self.config.download_path, folder=file, createPath=False) - if not os.path.exists(realFile): + realFile: str = calc_download_path(base_path=self.config.download_path, folder=file, create_path=False) + if not os.path.exists(realFile) or not os.path.isfile(realFile): return web.json_response( data={"error": f"File '{file}' does not exist."}, status=web.HTTPNotFound.status_code ) @@ -1120,8 +1128,8 @@ class HttpAPI(Common): """ return web.json_response( data={ - "notifications": Notification.get_instance().getTargets(), - "allowedTypes": list(NotificationEvents.getEvents().values()), + "notifications": Notification.get_instance().get_targets(), + "allowedTypes": list(NotificationEvents.get_events().values()), }, status=web.HTTPOk.status_code, dumps=self.encoder.encode, @@ -1167,7 +1175,7 @@ class HttpAPI(Common): status=web.HTTPBadRequest.status_code, ) - targets.append(ins.makeTarget(item)) + targets.append(ins.make_target(item)) try: if len(targets) < 1: @@ -1179,7 +1187,7 @@ class HttpAPI(Common): LOG.exception(e) return web.json_response({"error": "Failed to save tasks."}, status=web.HTTPInternalServerError.status_code) - data = {"notifications": targets, "allowedTypes": list(NotificationEvents.getEvents().values())} + data = {"notifications": targets, "allowedTypes": list(NotificationEvents.get_events().values())} return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode) diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index 2f1e4536..435204ca 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -6,9 +6,10 @@ import os import pty import shlex import time -from datetime import datetime +from datetime import UTC, datetime from typing import Any +import anyio import socketio from aiohttp import web @@ -18,7 +19,7 @@ from .DownloadQueue import DownloadQueue from .Emitter import Emitter from .encoder import Encoder from .EventsSubscriber import Event, Events, EventsSubscriber -from .Utils import arg_converter, isDownloaded +from .Utils import arg_converter, is_downloaded LOG = logging.getLogger("socket_api") @@ -67,6 +68,9 @@ class HttpSocket(Common): wrapper._ws_event = func.__name__ # type: ignore return wrapper + async def on_shutdown(self, _: web.Application): + LOG.debug("Shutting down socket server.") + def attach(self, app: web.Application): self.sio.attach(app, socketio_path=self.config.url_socketio) @@ -83,7 +87,10 @@ class HttpSocket(Common): EventsSubscriber.get_instance().subscribe(Events.ADD_URL, "socket_add_url", handle_event) - @ws_event # type: ignore + # register the shutdown event. + app.on_shutdown.append(self.on_shutdown) + + @ws_event async def cli_post(self, sid: str, data): if not data: await self.emitter.emit(Events.CLI_CLOSE, {"exitcode": 0}, to=sid) @@ -172,7 +179,7 @@ class HttpSocket(Common): await self.emitter.emit(Events.CLI_OUTPUT, {"type": "stderr", "line": str(e)}, to=sid) await self.emitter.emit(Events.CLI_CLOSE, {"exitcode": -1}, to=sid) - @ws_event # type: ignore + @ws_event async def add_url(self, sid: str, data: dict): url: str | None = data.get("url") @@ -189,7 +196,7 @@ class HttpSocket(Common): status = await self.add(**item) await self.emitter.emit(event=Events.STATUS, data=status, to=sid) - @ws_event # type: ignore + @ws_event async def item_cancel(self, sid: str, id: str): if not id: await self.emitter.error("Invalid request.", to=sid) @@ -201,7 +208,7 @@ class HttpSocket(Common): await self.emitter.emit(event=Events.ITEM_CANCEL, data=status) - @ws_event # type: ignore + @ws_event async def item_delete(self, sid: str, data: dict): if not data: await self.emitter.error("Invalid request.", to=sid) @@ -218,8 +225,8 @@ class HttpSocket(Common): await self.emitter.emit(event=Events.ITEM_DELETE, data=status) - @ws_event # type: ignore - async def archive_item(self, sid: str, data: dict): + @ws_event + async def archive_item(self, _: str, data: dict): if not isinstance(data, dict) or "url" not in data or not self.config.keep_archive: return @@ -231,36 +238,36 @@ class HttpSocket(Common): if not file: return - exists, idDict = isDownloaded(file, data["url"]) + exists, idDict = is_downloaded(file, data["url"]) if exists or "archive_id" not in idDict or idDict["archive_id"] is None: return - with open(file, "a") as f: - f.write(f"{idDict['archive_id']}\n") + async with await anyio.open_file(file, "a") as f: + await f.write(f"{idDict['archive_id']}\n") manual_archive = self.config.manual_archive if manual_archive: previouslyArchived = False if os.path.exists(manual_archive): - with open(manual_archive, "r") as f: - for line in f.readlines(): + async with await anyio.open_file(manual_archive) as f: + async for line in f: if idDict["archive_id"] in line: previouslyArchived = True break if not previouslyArchived: - with open(manual_archive, "a") as f: - f.write(f"{idDict['archive_id']} - at: {datetime.now().isoformat()}\n") + async with await anyio.open_file(manual_archive, "a") as f: + await f.write(f"{idDict['archive_id']} - at: {datetime.now(UTC).isoformat()}\n") LOG.info(f"Archiving url '{data['url']}' with id '{idDict['archive_id']}'.") - @ws_event # type: ignore + @ws_event async def connect(self, sid: str, _=None): data = { **self.queue.get(), "config": self.config.frontend(), "presets": self.config.presets, - "paused": self.queue.isPaused(), + "paused": self.queue.is_paused(), } # get download folder listing. @@ -269,13 +276,13 @@ class HttpSocket(Common): await self.emitter.emit(event=Events.INITIAL_DATA, data=data, to=sid) - @ws_event # type: ignore - async def pause(self, sid: str, _=None): + @ws_event + async def pause(self, *_, **__): self.queue.pause() await self.emitter.emit(event=Events.PAUSED, data={"paused": True, "at": time.time()}) - @ws_event # type: ignore - async def resume(self, sid: str, _=None): + @ws_event + async def resume(self, *_, **__): self.queue.resume() await self.emitter.emit(event=Events.PAUSED, data={"paused": False, "at": time.time()}) diff --git a/app/library/M3u8.py b/app/library/M3u8.py index ade81f3f..e72a2a5c 100644 --- a/app/library/M3u8.py +++ b/app/library/M3u8.py @@ -3,29 +3,21 @@ import os from urllib.parse import quote from .ffprobe import ffprobe -from .Utils import StreamingError, calcDownloadPath +from .Utils import StreamingError, calc_download_path class M3u8: duration: float = 6.000000 - ok_vcodecs: tuple = ( - "h264", - "x264", - "avc", - ) - ok_acodecs: tuple = ( - "aac", - "m4a", - "mp3", - ) + ok_vcodecs: tuple = ("h264", "x264", "avc") + ok_acodecs: tuple = ("aac", "m4a", "mp3") - def __init__(self, url: str, segment_duration: float|None = None): + def __init__(self, url: str, segment_duration: float | None = None): self.url = url self.duration = float(segment_duration) if segment_duration is not None else self.duration async def make_stream(self, download_path: str, file: str) -> str: - realFile: str = calcDownloadPath(basePath=download_path, folder=file, createPath=False) + realFile: str = calc_download_path(base_path=download_path, folder=file, create_path=False) if not os.path.exists(realFile): error = f"File '{realFile}' does not exist." @@ -79,7 +71,7 @@ class M3u8: return "\n".join(m3u8) async def make_subtitle(self, download_path: str, file: str, duration: float) -> str: - realFile: str = calcDownloadPath(basePath=download_path, folder=file, createPath=False) + realFile: str = calc_download_path(base_path=download_path, folder=file, create_path=False) if not os.path.exists(realFile): error = f"File '{realFile}' does not exist." diff --git a/app/library/Notifications.py b/app/library/Notifications.py index 069b2eaf..d8922c7b 100644 --- a/app/library/Notifications.py +++ b/app/library/Notifications.py @@ -20,7 +20,7 @@ LOG = logging.getLogger("notifications") @dataclass(kw_only=True) -class targetRequestHeader: +class TargetRequestHeader: """Request header details for a notification target.""" key: str @@ -37,13 +37,13 @@ class targetRequestHeader: @dataclass(kw_only=True) -class targetRequest: +class TargetRequest: """Request details for a notification target.""" type: str method: str url: str - headers: list[targetRequestHeader] = field(default_factory=list) + headers: list[TargetRequestHeader] = field(default_factory=list) def serialize(self) -> dict: return { @@ -67,7 +67,7 @@ class Target: id: str name: str on: list[str] - request: targetRequest + request: TargetRequest def serialize(self) -> dict: return { @@ -95,17 +95,12 @@ class NotificationEvents: TEST = Events.TEST @staticmethod - def getEvents() -> dict[str, str]: - data: dict[str, str] = {} - for k, v in vars(NotificationEvents).items(): - if not k.startswith("__") and not callable(v): - data[k] = v - - return data + def get_events() -> dict[str, str]: + return {k: v for k, v in vars(NotificationEvents).items() if not k.startswith("__") and not callable(v)} @staticmethod - def isValid(event: str) -> bool: - return event in NotificationEvents.getEvents().values() + def is_valid(event: str) -> bool: + return event in NotificationEvents.get_events().values() class Notification(metaclass=Singleton): @@ -147,7 +142,7 @@ class Notification(metaclass=Singleton): return Notification._instance - def getTargets(self) -> list[Target]: + def get_targets(self) -> list[Target]: """Get the list of notification targets.""" return self._targets @@ -191,7 +186,7 @@ class Notification(metaclass=Singleton): LOG.info(f"Loading notification targets from '{self._file}'.") try: - with open(self._file, "r") as f: + with open(self._file) as f: targets = json.load(f) except Exception as e: LOG.error(f"Error loading notification targets from '{self._file}'. '{e!s}'") @@ -204,7 +199,7 @@ class Notification(metaclass=Singleton): LOG.error(f"Invalid notification target '{target}'. '{e!s}'") continue - target = self.makeTarget(target) + target = self.make_target(target) self._targets.append(target) @@ -261,7 +256,7 @@ class Notification(metaclass=Singleton): raise ValueError(msg) for e in target["on"]: - if e not in NotificationEvents.getEvents().values(): + if e not in NotificationEvents.get_events().values(): msg = f"Invalid notification target. Invalid event '{e}' found." raise ValueError(msg) @@ -345,7 +340,7 @@ class Notification(metaclass=Singleton): LOG.error(f"Error sending Notification event '{event}' id '{itemId}' to '{target.name}'. '{e}'.") return {"url": target.request.url, "status": 500, "text": str(e)} - def makeTarget(self, target: dict) -> Target: + def make_target(self, target: dict) -> Target: """ Make a notification target from a dictionary. @@ -360,12 +355,12 @@ class Notification(metaclass=Singleton): id=target.get("id"), name=target.get("name"), on=target.get("on", []), - request=targetRequest( + request=TargetRequest( type=target.get("request", {}).get("type", "json"), method=target.get("request", {}).get("method", "POST"), url=target.get("request", {}).get("url"), headers=[ - targetRequestHeader(key=h.get("key"), value=h.get("value")) + TargetRequestHeader(key=h.get("key"), value=h.get("value")) for h in target.get("request", {}).get("headers", []) ], ), @@ -375,7 +370,7 @@ class Notification(metaclass=Singleton): if len(self._targets) < 1: return False - if not NotificationEvents.isValid(event): + if not NotificationEvents.is_valid(event): return False return self.send(event, data) diff --git a/app/library/Playlist.py b/app/library/Playlist.py index 7143a64a..a83074ca 100644 --- a/app/library/Playlist.py +++ b/app/library/Playlist.py @@ -3,11 +3,11 @@ import re from pathlib import Path from urllib.parse import quote -from aiohttp.web import Response +from aiohttp.web import HTTPFound, Response from .ffprobe import ffprobe from .Subtitle import Subtitle -from .Utils import StreamingError, calcDownloadPath, checkId +from .Utils import StreamingError, calc_download_path, check_id class Playlist: @@ -17,16 +17,16 @@ class Playlist: self.url = url async def make(self, download_path: str, file: str) -> str | Response: - rFile = Path(calcDownloadPath(basePath=download_path, folder=file, createPath=False)) + rFile = Path(calc_download_path(base_path=download_path, folder=file, create_path=False)) if not rFile.exists(): - possibleFile = checkId(download_path, rFile) + possibleFile = check_id(file=rFile) if not possibleFile: msg = f"File '{rFile}' does not exist." raise StreamingError(msg) return Response( - status=302, + status=HTTPFound.status_code, headers={ "Location": f"{self.url}api/player/playlist/{quote(str(possibleFile).replace(download_path, '').strip('/'))}.m3u8" }, diff --git a/app/library/Segments.py b/app/library/Segments.py index 6b19af6e..b45d894d 100644 --- a/app/library/Segments.py +++ b/app/library/Segments.py @@ -6,7 +6,7 @@ import tempfile from .config import Config from .ffprobe import ffprobe -from .Utils import StreamingError, calcDownloadPath +from .Utils import StreamingError, calc_download_path LOG = logging.getLogger("player.segments") @@ -25,7 +25,7 @@ class Segments: self.aconvert = True async def stream(self, path: str, file: str) -> bytes: - realFile: str = calcDownloadPath(basePath=path, folder=file, createPath=False) + realFile: str = calc_download_path(base_path=path, folder=file, create_path=False) if not os.path.exists(realFile): msg = f"File {realFile} does not exist." @@ -37,7 +37,7 @@ class Segments: pass tmpDir: str = tempfile.gettempdir() - tmpFile = os.path.join(tmpDir, f'ytptube_stream.{hashlib.md5(realFile.encode("utf-8")).hexdigest()}') # noqa: S324 + tmpFile = os.path.join(tmpDir, f'ytptube_stream.{hashlib.sha256(realFile.encode("utf-8")).hexdigest()}') if not os.path.exists(tmpFile): os.symlink(realFile, tmpFile) @@ -108,8 +108,8 @@ class Segments: data, err = await proc.communicate() if 0 != proc.returncode: - LOG.error(f'Failed to stream {realFile} segment {self.index}. {err.decode("utf-8")}.') - msg = f"Failed to stream {realFile} segment {self.index}." + LOG.error(f"Failed to stream '{realFile}' segment '{self.index}'. {err.decode('utf-8')}.") + msg = f"Failed to stream '{realFile}' segment '{self.index}'." raise StreamingError(msg) return data diff --git a/app/library/Subtitle.py b/app/library/Subtitle.py index 5d964027..31742d45 100644 --- a/app/library/Subtitle.py +++ b/app/library/Subtitle.py @@ -1,11 +1,12 @@ import logging import pathlib +import anyio import pysubs2 from pysubs2.formats.substation import SubstationFormat from pysubs2.time import ms_to_times -from .Utils import calcDownloadPath +from .Utils import calc_download_path LOG = logging.getLogger("player.subtitle") @@ -28,7 +29,7 @@ class Subtitle: ) async def make(self, path: str, file: str) -> str: - realFile: str = calcDownloadPath(basePath=path, folder=file, createPath=False) + realFile: str = calc_download_path(base_path=path, folder=file, create_path=False) rFile = pathlib.Path(realFile) @@ -41,8 +42,8 @@ class Subtitle: raise Exception(msg) if rFile.suffix == ".vtt": - with open(realFile) as f: - return f.read() + async with await anyio.open_file(realFile) as f: + return await f.read() subs = pysubs2.load(path=str(rFile)) diff --git a/app/library/Tasks.py b/app/library/Tasks.py index f9144125..35b52ddc 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -4,7 +4,7 @@ import logging import os import time from dataclasses import dataclass, field -from datetime import datetime +from datetime import UTC, datetime from typing import Any import httpx @@ -106,6 +106,18 @@ class Tasks(metaclass=Singleton): Tasks._instance = Tasks() return Tasks._instance + async def on_shutdown(self, _: web.Application): + """ + Shutdown the socket server. + + Args: + _: The aiohttp application. + + """ + LOG.debug("Shutting down tasks runner.") + self.clear(shutdown=True) + LOG.debug("Tasks runner has been shut down.") + def attach(self, _: web.Application): """ Attach the tasks to the aiohttp application. @@ -119,13 +131,7 @@ class Tasks(metaclass=Singleton): """ self.load() - def shutdown(self): - """ - Shutdown the tasks service. - """ - self.clear() - - def getTasks(self) -> list[Task]: + def get_tasks(self) -> list[Task]: """Return the tasks.""" return [job.task for job in self._jobs] @@ -177,7 +183,7 @@ class Tasks(metaclass=Singleton): return self - def clear(self) -> "Tasks": + def clear(self, shutdown: bool = False) -> "Tasks": """ Clear all tasks. @@ -193,8 +199,9 @@ class Tasks(metaclass=Singleton): LOG.info(f"Stopping job '{task.id}: {task.name}'.") task.job.stop() except Exception as e: - LOG.exception(e) - LOG.error(f"Failed to stop job '{task.id}: {task.name}'. '{e!s}'.") + if not shutdown: + LOG.exception(e) + LOG.error(f"Failed to stop job '{task.id}: {task.name}'. '{e!s}'.") self._jobs.clear() @@ -292,7 +299,7 @@ class Tasks(metaclass=Singleton): """ try: - timeNow = datetime.now().isoformat() + timeNow = datetime.now(UTC).isoformat() started = time.time() if not task.url: LOG.error(f"Failed to dispatch task '{task.id}: {task.name}'. No URL found.") @@ -335,13 +342,13 @@ class Tasks(metaclass=Singleton): await asyncio.wait_for(asyncio.gather(*tasks), timeout=None) - timeNow = datetime.now().isoformat() + timeNow = datetime.now(UTC).isoformat() ended = time.time() LOG.info(f"Task '{task.id}: {task.name}' completed at '{timeNow}' took '{ended - started:.2f}' seconds.") await self._emitter.success(f"Task '{task.name}' completed in '{ended - started:.2f}' seconds.") except Exception as e: - timeNow = datetime.now().isoformat() + timeNow = datetime.now(UTC).isoformat() LOG.error(f"Task '{task.id}: {task.name}' has failed to execute at '{timeNow}'. '{e!s}'.") await self._emitter.error(f"Task '{task.name}' failed to execute at '{timeNow}'. '{e!s}'.") diff --git a/app/library/Utils.py b/app/library/Utils.py index d474e8ed..951650ea 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -8,7 +8,7 @@ import re import shlex import socket import uuid -from datetime import datetime, timezone +from datetime import UTC, datetime from functools import lru_cache from typing import Any @@ -78,7 +78,7 @@ def get_opts(preset: str, ytdl_opts: dict) -> dict: return opts -def getVideoInfo(url: str, ytdlp_opts: dict = None, no_archive: bool = True) -> Any | dict[str, Any] | None: +def get_video_info(url: str, ytdlp_opts: dict = None, no_archive: bool = True) -> Any | dict[str, Any] | None: """ Extracts video information from the given URL. @@ -109,7 +109,7 @@ def getVideoInfo(url: str, ytdlp_opts: dict = None, no_archive: bool = True) -> return yt_dlp.YoutubeDL().extract_info(url, download=False) -def calcDownloadPath(basePath: str, folder: str | None = None, createPath: bool = True) -> str: +def calc_download_path(base_path: str, folder: str | None = None, create_path: bool = True) -> str: """ Calculates download path and prevents folder traversal. @@ -118,25 +118,25 @@ def calcDownloadPath(basePath: str, folder: str | None = None, createPath: bool """ if not folder: - return basePath + return base_path if folder.startswith("/"): folder = folder[1:] - realBasePath = os.path.realpath(basePath) - download_path = os.path.realpath(os.path.join(basePath, folder)) + realBasePath = os.path.realpath(base_path) + download_path = os.path.realpath(os.path.join(base_path, folder)) if not download_path.startswith(realBasePath): msg = f'Folder "{folder}" must resolve inside the base download folder "{realBasePath}".' raise Exception(msg) - if not os.path.isdir(download_path) and createPath: + if not os.path.isdir(download_path) and create_path: os.makedirs(download_path, exist_ok=True) return download_path -def ExtractInfo(config: dict, url: str, debug: bool = False) -> dict: +def extract_info(config: dict, url: str, debug: bool = False) -> dict: params: dict = { "color": "no_color", "extract_flat": True, @@ -168,14 +168,14 @@ def ExtractInfo(config: dict, url: str, debug: bool = False) -> dict: return yt_dlp.YoutubeDL(params=params).extract_info(url, download=False) -def mergeDict(source: dict, destination: dict) -> dict: +def merge_dict(source: dict, destination: dict) -> dict: """Merge data from source into destination""" destination_copy = copy.deepcopy(destination) for key, value in source.items(): destination_key_value = destination_copy.get(key) if isinstance(value, dict) and isinstance(destination_key_value, dict): - destination_copy[key] = mergeDict(source=value, destination=destination_copy.setdefault(key, {})) + destination_copy[key] = merge_dict(source=value, destination=destination_copy.setdefault(key, {})) elif isinstance(value, list) and isinstance(destination_key_value, list): destination_copy[key] = destination_key_value + value else: @@ -184,7 +184,7 @@ def mergeDict(source: dict, destination: dict) -> dict: return destination_copy -def mergeConfig(config: dict, new_config: dict) -> dict: +def merge_config(config: dict, new_config: dict) -> dict: """ Merge user provided config into default config @@ -201,7 +201,7 @@ def mergeConfig(config: dict, new_config: dict) -> dict: LOG.error(f"Key '{key}' is not allowed to be manually set.") del new_config[key] - conf = mergeDict(new_config, config) + conf = merge_dict(new_config, config) if "impersonate" in conf: conf["impersonate"] = ImpersonateTarget.from_str(conf["impersonate"]) @@ -209,8 +209,8 @@ def mergeConfig(config: dict, new_config: dict) -> dict: return conf -def isDownloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, str | None, str | None]]: - global YTDLP_INFO_CLS +def is_downloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, str | None, str | None]]: + global YTDLP_INFO_CLS # noqa: PLW0603 idDict = { "id": None, @@ -263,7 +263,7 @@ def isDownloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, st return (False, idDict) -def jsonCookie(cookies: dict[dict[str, any]]) -> str | None: +def json_cookie(cookies: dict[dict[str, any]]) -> str | None: """ Converts JSON cookies to Netscape cookies @@ -272,7 +272,7 @@ def jsonCookie(cookies: dict[dict[str, any]]) -> str | None: netscapeCookies = "# Netscape HTTP Cookie File\n# https://curl.haxx.se/docs/http-cookies.html\n# This file was generated by libcurl! Edit at your own risk.\n\n" hasCookies: bool = False - for domain in cookies: + for domain in cookies: # noqa: PLC0206 if not isinstance(cookies[domain], dict): continue @@ -287,7 +287,7 @@ def jsonCookie(cookies: dict[dict[str, any]]) -> str | None: cookieDict = cookies[domain][subDomain][cookie] if 0 == int(cookieDict["expirationDate"]): - cookieDict["expirationDate"] = datetime.now(timezone.utc).timestamp() + (86400 * 1000) + cookieDict["expirationDate"] = datetime.now(UTC).timestamp() + (86400 * 1000) hasCookies = True netscapeCookies += ( @@ -365,7 +365,7 @@ def load_file(file: str, check_type=None) -> tuple[dict | list, bool, str]: ) -def checkId(basePath: str, file: pathlib.Path) -> bool | str: +def check_id(file: pathlib.Path) -> bool | str: """ Check if we are able to get an id from the file name. if so check if any video file with the same id exists. @@ -385,16 +385,7 @@ def checkId(basePath: str, file: pathlib.Path) -> bool | str: if id not in f.stem: continue - if f.suffix not in ( - ".mp4", - ".mkv", - ".webm", - ".m4v", - ".m4a", - ".mp3", - ".aac", - ".ogg", - ): + if f.suffix not in (".mp4", ".mkv", ".webm", ".m4v", ".m4a", ".mp3", ".aac", ".ogg"): continue return f.absolute() @@ -510,7 +501,7 @@ def arg_converter(args: str) -> dict: Convert yt-dlp options to a dictionary. Args: - opts (str): yt-dlp options string. + args (str): yt-dlp options string. Returns: dict: yt-dlp options dictionary. @@ -520,7 +511,7 @@ def arg_converter(args: str) -> dict: create_parser = yt_dlp.options.create_parser - def defaultOpts(args: str): + def _default_opts(args: str): patched_parser = create_parser() try: yt_dlp.options.create_parser = lambda: patched_parser @@ -528,7 +519,7 @@ def arg_converter(args: str) -> dict: finally: yt_dlp.options.create_parser = create_parser - default_opts = defaultOpts([]).ydl_opts + default_opts = _default_opts([]).ydl_opts opts = yt_dlp.parse_options(shlex.split(args)).ydl_opts diff --git a/app/library/config.py b/app/library/config.py index 4e32d730..45e8b6e7 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -12,7 +12,7 @@ import coloredlogs from dotenv import load_dotenv from yt_dlp.version import __version__ as YTDLP_VERSION -from .Utils import IGNORED_KEYS, load_file, mergeDict +from .Utils import IGNORED_KEYS, load_file, merge_dict from .version import APP_VERSION @@ -254,7 +254,7 @@ class Config: logging.error(f"Config variable '{k}' had non-existing config reference '{key}'.") sys.exit(1) - v = v.replace(key, getattr(self, localKey)) # noqa: PLW2901 + v = v.replace(key, getattr(self, localKey)) setattr(self, k, v) @@ -304,7 +304,7 @@ class Config: LOG.error(f"Key '{key}' is not allowed to be loaded via 'ytdlp.json' file.") del opts[key] - self.ytdl_options = mergeDict(self.ytdl_options, opts) + self.ytdl_options = merge_dict(self.ytdl_options, opts) else: LOG.error(f"Invalid yt-dlp custom options file '{optsFile}'.") else: diff --git a/app/library/ffprobe.py b/app/library/ffprobe.py index 2653692e..b38da7a7 100644 --- a/app/library/ffprobe.py +++ b/app/library/ffprobe.py @@ -8,6 +8,8 @@ import json import operator import os +import anyio + # parameter-less decorator def async_lru_cache_decorator(async_function): @@ -85,10 +87,7 @@ class FFStream: if self.__dict__.get("codec_type", None) != "video": return False - if self.__dict__.get("codec_name", None) in ["png", "mjpeg", "gif", "bmp", "tiff", "webp"]: - return False - - return True + return self.__dict__.get("codec_name", None) not in ["png", "mjpeg", "gif", "bmp", "tiff", "webp"] def is_subtitle(self): """ @@ -266,7 +265,7 @@ async def ffprobe(file: str) -> FFProbeResult: """ try: - with open(os.devnull, "w") as tempf: + async with await anyio.open_file(os.devnull, "w") as tempf: await asyncio.create_subprocess_exec("ffprobe", "-h", stdout=tempf, stderr=tempf) except FileNotFoundError as e: msg = "ffprobe not found." diff --git a/app/main.py b/app/main.py index 46f39daa..ee857722 100644 --- a/app/main.py +++ b/app/main.py @@ -13,7 +13,6 @@ from library.config import Config from library.DownloadQueue import DownloadQueue from library.Emitter import Emitter from library.EventsSubscriber import EventsSubscriber -from library.HttpAPI import LOG as http_logger # noqa: N811 from library.HttpAPI import HttpAPI from library.HttpSocket import HttpSocket from library.Notifications import Notification @@ -25,110 +24,97 @@ MIME = magic.Magic(mime=True) class Main: - config: Config - app: web.Application - http: HttpAPI - socket: HttpSocket - def __init__(self): - self.config = Config.get_instance() + self._config = Config.get_instance() self.rootPath = str(Path(__file__).parent.absolute()) - self.app = web.Application() + self._app = web.Application() self._check_folders() try: - PackageInstaller(self.config).check() + PackageInstaller(self._config).check() except Exception as e: LOG.exception(e) LOG.error(f"Failed to check for packages. Error message '{e!s}'.") - caribou.upgrade(self.config.db_file, os.path.join(self.rootPath, "migrations")) + caribou.upgrade(self._config.db_file, os.path.join(self.rootPath, "migrations")) - connection = sqlite3.connect(database=self.config.db_file, isolation_level=None) + connection = sqlite3.connect(database=self._config.db_file, isolation_level=None) connection.row_factory = sqlite3.Row connection.execute("PRAGMA journal_mode=wal") + + async def _close_connection(_): + LOG.debug("Closing database connection.") + connection.close() + LOG.debug("Database connection closed.") + try: - if "600" != oct(os.stat(self.config.db_file).st_mode)[-3:]: - os.chmod(self.config.db_file, 0o600) + if "600" != oct(os.stat(self._config.db_file).st_mode)[-3:]: + os.chmod(self._config.db_file, 0o600) except Exception: pass - queue = DownloadQueue(connection=connection) - - self.http = HttpAPI(queue=queue) - self.socket = HttpSocket(queue=queue) + self._queue = DownloadQueue(connection=connection) + self._http = HttpAPI(queue=self._queue) + self._socket = HttpSocket(queue=self._queue) Emitter.get_instance().add_emitter([Notification().emit], local=False).add_emitter( [EventsSubscriber().emit], local=True ) - self.app.on_startup.append(lambda _: queue.initialize()) + self._app.on_cleanup.append(_close_connection) def _check_folders(self): - """ - Check if the required folders exist and create them if they do not. - """ - try: - LOG.debug(f"Checking download folder at '{self.config.download_path}'.") - if not os.path.exists(self.config.download_path): - LOG.info(f"Creating download folder at '{self.config.download_path}'.") - os.makedirs(self.config.download_path, exist_ok=True) - except OSError: - LOG.error(f"Could not create download folder at '{self.config.download_path}'.") - raise + """Check if the required folders exist and create them if they do not.""" + folders = (self._config.download_path, self._config.temp_path, self._config.config_path) + + for folder in folders: + try: + LOG.debug(f"Checking folder at '{folder}'.") + if not os.path.exists(folder): + LOG.info(f"Creating folder at '{folder}'.") + os.makedirs(folder, exist_ok=True) + except OSError: + LOG.error(f"Could not create folder at '{folder}'.") + raise try: - LOG.debug(f"Checking temp folder at '{self.config.temp_path}'.") - if not os.path.exists(self.config.temp_path): - LOG.info(f"Creating temp folder at '{self.config.temp_path}'.") - os.makedirs(self.config.temp_path, exist_ok=True) - except OSError: - LOG.error(f"Could not create temp folder at '{self.config.temp_path}'.") - raise - - try: - LOG.debug(f"Checking config folder at '{self.config.config_path}'.") - if not os.path.exists(self.config.config_path): - LOG.info(f"Creating config folder at '{self.config.config_path}'.") - os.makedirs(self.config.config_path, exist_ok=True) - except OSError: - LOG.error(f"Could not create config folder at '{self.config.config_path}'.") - raise - - try: - LOG.debug(f"Checking database file at '{self.config.db_file}'.") - if not os.path.exists(self.config.db_file): - LOG.info(f"Creating database file at '{self.config.db_file}'.") - with open(self.config.db_file, "w") as _: + LOG.debug(f"Checking database file at '{self._config.db_file}'.") + if not os.path.exists(self._config.db_file): + LOG.info(f"Creating database file at '{self._config.db_file}'.") + with open(self._config.db_file, "w") as _: pass except OSError: - LOG.error(f"Could not create database file at '{self.config.db_file}'.") + LOG.error(f"Could not create database file at '{self._config.db_file}'.") raise def start(self): """ Start the application. """ - self.socket.attach(self.app) - self.http.attach(self.app) - Tasks.get_instance().attach(self.app) + self._socket.attach(self._app) + self._http.attach(self._app) + self._queue.attach(self._app) + Tasks.get_instance().attach(self._app) def started(_): LOG.info("=" * 40) - LOG.info(f"YTPTube v{self.config.version} - started on http://{self.config.host}:{self.config.port}") + LOG.info(f"YTPTube v{self._config.version} - started on http://{self._config.host}:{self._config.port}") LOG.info("=" * 40) - if self.config.access_log: - http_logger.addFilter(lambda record: "GET /api/ping" not in record.getMessage()) + HTTP_LOGGER = None + if self._config.access_log: + from library.HttpAPI import LOG as HTTP_LOGGER + + HTTP_LOGGER.addFilter(lambda record: "GET /api/ping" not in record.getMessage()) web.run_app( - self.app, - host=self.config.host, - port=self.config.port, + self._app, + host=self._config.host, + port=self._config.port, reuse_port=True, loop=asyncio.get_event_loop(), - access_log=http_logger if self.config.access_log else None, + access_log=HTTP_LOGGER, print=started, )