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

fixed linting problems

Browse source
This commit is contained in:
ArabCoders 2025-01-28 23:05:59 +03:00
parent fdec16e895
commit 45878965f4
20 changed files with 402 additions and 313 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Pipfile
View file

@ -22,6 +22,7 @@ mutagen = "*"
brotli = "*" brotli = "*"
brotlicffi = "*" brotlicffi = "*"
yt-dlp = "*" yt-dlp = "*"
anyio = "*"
[dev-packages] [dev-packages]

3
Pipfile.lock generated
View file

@ -1,7 +1,7 @@
{ {
"_meta": { "_meta": {
"hash": { "hash": {
"sha256": "cd9a56440fe36b901993ced28ee6ef8d301a5125344cda86519fb7c54ede1449" "sha256": "17668280c8e863033313b83f64240f6837e733957fbc2613767d297fee6c6fc2"
}, },
"pipfile-spec": 6, "pipfile-spec": 6,
"requires": { "requires": {
@ -128,6 +128,7 @@
"sha256:1d9fe889df5212298c0c0723fa20479d1b94883a2df44bd3897aa91083316f7a", "sha256:1d9fe889df5212298c0c0723fa20479d1b94883a2df44bd3897aa91083316f7a",
"sha256:b5011f270ab5eb0abf13385f851315585cc37ef330dd88e27ec3d34d651fd47a" "sha256:b5011f270ab5eb0abf13385f851315585cc37ef330dd88e27ec3d34d651fd47a"
], ],
"index": "pypi",
"markers": "python_version >= '3.9'", "markers": "python_version >= '3.9'",
"version": "==4.8.0" "version": "==4.8.0"
}, },

42
app/library/AsyncPool.py
View file

@ -14,9 +14,9 @@ class AsyncPool:
worker_co, worker_co,
name: str, name: str,
logger: logging.Logger, logger: logging.Logger,
loop: asyncio.AbstractEventLoop|None = None, loop: asyncio.AbstractEventLoop | None = None,
load_factor: int = 1, load_factor: int = 1,
max_task_time: int|None = None, max_task_time: int | None = None,
return_futures: bool = False, return_futures: bool = False,
raise_on_join: bool = False, raise_on_join: bool = False,
): ):
@ -54,6 +54,12 @@ class AsyncPool:
self._status: dict[str, dict | None] = {} self._status: dict[str, dict | None] = {}
async def _worker_loop(self, worker_id: str): 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: while True:
got_obj = False got_obj = False
future = None future = None
@ -84,7 +90,7 @@ class AsyncPool:
raise raise
except BaseException as e: except BaseException as e:
if isinstance(e, asyncio.exceptions.CancelledError): if isinstance(e, asyncio.exceptions.CancelledError):
raise e raise
self._exceptions = True self._exceptions = True
@ -92,7 +98,7 @@ class AsyncPool:
# don't log the failure when the client is receiving the future # don't log the failure when the client is receiving the future
future.set_exception(e) future.set_exception(e)
else: else:
self._logger.exception(f"Worker call failed. {str(e)}") self._logger.exception(f"Worker call failed. {e!s}")
finally: finally:
self._status[worker_id] = None self._status[worker_id] = None
@ -153,7 +159,12 @@ class AsyncPool:
self._create_worker(worker_id) self._create_worker(worker_id)
async def restart(self, worker_id: str, msg: str = None) -> bool: 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: if worker_id not in self._workers:
self._logger.warning(f"Worker {worker_id} does not exist.") self._logger.warning(f"Worker {worker_id} does not exist.")
return False return False
@ -162,7 +173,7 @@ class AsyncPool:
self._workers[worker_id].cancel(msg) self._workers[worker_id].cancel(msg)
await self._workers[worker_id] await self._workers[worker_id]
except asyncio.exceptions.CancelledError as e: 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: if worker_id in self._status:
self._status.pop(worker_id) self._status.pop(worker_id)
@ -173,8 +184,20 @@ class AsyncPool:
return True 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: 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: if worker_id not in self._workers:
self._logger.warning(f"Worker {worker_id} does not exist.") self._logger.warning(f"Worker {worker_id} does not exist.")
return False return False
@ -183,7 +206,7 @@ class AsyncPool:
try: try:
await self._workers[worker_id] await self._workers[worker_id]
except asyncio.exceptions.CancelledError as e: 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: if worker_id in self._status:
self._status.pop(worker_id) self._status.pop(worker_id)
@ -214,7 +237,8 @@ class AsyncPool:
self._logger.info(f"Completed {self._name}") self._logger.info(f"Completed {self._name}")
if self._exceptions and self._raise_on_join: 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: def _create_worker(self, worker_id: str) -> asyncio.Future:
if worker_id in self._workers: if worker_id in self._workers:

38
app/library/DataStore.py
View file

@ -1,7 +1,7 @@
import copy import copy
import json import json
from collections import OrderedDict from collections import OrderedDict
from datetime import datetime, timezone from datetime import UTC, datetime
from email.utils import formatdate from email.utils import formatdate
from sqlite3 import Connection from sqlite3 import Connection
@ -39,13 +39,9 @@ class DataStore:
if key and key in self.dict: if key and key in self.dict:
return True return True
for i in self.dict: return any((key and self.dict[i].info._id == key) or (url and self.dict[i].info.url == url) 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 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: if not key and not url:
msg = "key or url must be provided." msg = "key or url must be provided."
raise KeyError(msg) raise KeyError(msg)
@ -57,8 +53,8 @@ class DataStore:
msg = f"{key=} or {url=} not found." msg = f"{key=} or {url=} not found."
raise KeyError(msg) raise KeyError(msg)
def getById(self, id: str) -> Download | None: def get_by_id(self, id: str) -> Download | None:
return self.dict[id] if id in self.dict else None return self.dict.get(id, None)
def items(self) -> list[tuple[str, Download]]: def items(self) -> list[tuple[str, Download]]:
return self.dict.items() return self.dict.items()
@ -71,25 +67,25 @@ class DataStore:
) )
for row in cursor: 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"]) data: dict = json.loads(row["data"])
key: str = data.pop("_id") key: str = data.pop("_id")
item: ItemDTO = ItemDTO(**data) item: ItemDTO = ItemDTO(**data)
item._id = key 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)) items.append((row["id"], item))
return items return items
def put(self, value: Download) -> Download: def put(self, value: Download) -> Download:
self.dict.update({value.info._id: value}) 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] return self.dict[value.info._id]
def delete(self, key: str) -> None: def delete(self, key: str) -> None:
self.dict.pop(key, None) self.dict.pop(key, None)
self._deleteStoreItem(key) self._delete_store_item(key)
def next(self) -> tuple[str, Download]: def next(self) -> tuple[str, Download]:
return next(iter(self.dict.items())) return next(iter(self.dict.items()))
@ -97,17 +93,13 @@ class DataStore:
def empty(self): def empty(self):
return not bool(self.dict) return not bool(self.dict)
def hasDownloads(self): def has_downloads(self):
if 0 == len(self.dict): if 0 == len(self.dict):
return False return False
for key in self.dict: return any(self.dict[key].started() is False for key in self.dict)
if self.dict[key].started() is False:
return True
return False def get_next_download(self) -> Download:
def getNextDownload(self) -> Download:
for key in self.dict: for key in self.dict:
if self.dict[key].started() is False and self.dict[key].is_cancelled() is False: if self.dict[key].started() is False and self.dict[key].is_cancelled() is False:
return self.dict[key] return self.dict[key]
@ -118,7 +110,7 @@ class DataStore:
self.connection.execute('SELECT "id" FROM "history" LIMIT 1').fetchone() self.connection.execute('SELECT "id" FROM "history" LIMIT 1').fetchone()
return True return True
def _updateStoreItem(self, type: str, item: ItemDTO) -> None: def _update_store_item(self, type: str, item: ItemDTO) -> None:
sqlStatement = """ sqlStatement = """
INSERT INTO "history" ("id", "type", "url", "data") INSERT INTO "history" ("id", "type", "url", "data")
VALUES (?, ?, ?, ?) VALUES (?, ?, ?, ?)
@ -149,11 +141,11 @@ class DataStore:
type, type,
stored.url, stored.url,
stored.json(), 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( self.connection.execute(
'DELETE FROM "history" WHERE "type" = ? AND "id" = ?', 'DELETE FROM "history" WHERE "type" = ? AND "id" = ?',
( (

41
app/library/Download.py
View file

@ -15,7 +15,7 @@ from .config import Config
from .Emitter import Emitter from .Emitter import Emitter
from .ffprobe import ffprobe from .ffprobe import ffprobe
from .ItemDTO import ItemDTO from .ItemDTO import ItemDTO
from .Utils import get_opts, jsonCookie, mergeConfig from .Utils import get_opts, json_cookie, merge_config
LOG = logging.getLogger("download") LOG = logging.getLogger("download")
@ -34,7 +34,7 @@ class Download:
info: ItemDTO = None info: ItemDTO = None
default_ytdl_opts: dict = None default_ytdl_opts: dict = None
debug: bool = False debug: bool = False
tempPath: str = None temp_path: str = None
emitter: Emitter = None emitter: Emitter = None
cancelled: bool = False cancelled: bool = False
is_live: bool = False is_live: bool = False
@ -64,7 +64,7 @@ class Download:
) )
"Fields to be extracted from yt-dlp progress hook." "Fields to be extracted from yt-dlp progress hook."
tempKeep: bool = False temp_keep: bool = False
"Keep temp folder after download." "Keep temp folder after download."
def __init__(self, info: ItemDTO, info_dict: dict = None, debug: bool = False): def __init__(self, info: ItemDTO, info_dict: dict = None, debug: bool = False):
@ -86,7 +86,7 @@ class Download:
self.proc = None self.proc = None
self.emitter = None self.emitter = None
self.max_workers = int(config.max_workers) 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_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.is_manifestless = "is_manifestless" in self.info.options and self.info.options["is_manifestless"] is True
self.info_dict = info_dict self.info_dict = info_dict
@ -94,7 +94,7 @@ class Download:
def _progress_hook(self, data: dict): def _progress_hook(self, data: dict):
dataDict = {k: v for k, v in data.items() if k in self._ytdlp_fields} 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"] dataDict["filename"] = data["info_dict"]["filename"]
self.status_queue.put({"id": self.id, **dataDict}) self.status_queue.put({"id": self.id, **dataDict})
@ -115,11 +115,11 @@ class Download:
def _download(self): def _download(self):
try: 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( params.update(
{ {
"color": "no_color", "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}, "outtmpl": {"default": self.template, "chapter": self.template_chapter},
"noprogress": True, "noprogress": True,
"break_on_existing": True, "break_on_existing": True,
@ -138,17 +138,17 @@ class Download:
if self.info.cookies: if self.info.cookies:
try: try:
data = jsonCookie(json.loads(self.info.cookies)) data = json_cookie(json.loads(self.info.cookies))
if not data: if not data:
LOG.warning( LOG.warning(
f"The cookie string that was provided for {self.info.title} is empty or not in expected spec." 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) f.write(data)
params["cookiefile"] = f.name params["cookiefile"] = f.name
except ValueError as e: 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: if self.is_live or self.is_manifestless:
hasDeletedOptions = False hasDeletedOptions = False
@ -191,10 +191,10 @@ class Download:
self.status_queue = Config.get_manager().Queue() self.status_queue = Config.get_manager().Queue()
# Create temp dir for each download. # 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): if not os.path.exists(self.temp_path):
os.makedirs(self.tempPath, exist_ok=True) os.makedirs(self.temp_path, exist_ok=True)
self.proc = multiprocessing.Process(name=f"download-{self.id}", target=self._download) self.proc = multiprocessing.Process(name=f"download-{self.id}", target=self._download)
self.proc.start() self.proc.start()
@ -287,26 +287,26 @@ class Download:
return False return False
def delete_temp(self): 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 return
if "finished" != self.info.status and self.is_live: if "finished" != self.info.status and self.is_live:
LOG.warning( 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 return
if not os.path.exists(self.tempPath): if not os.path.exists(self.temp_path):
return return
if self.tempPath == self.temp_dir: if self.temp_path == self.temp_dir:
LOG.warning( 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 return
LOG.info(f"Deleting Temp folder '{self.tempPath}'.") LOG.info(f"Deleting Temp folder '{self.temp_path}'.")
shutil.rmtree(self.tempPath, ignore_errors=True) shutil.rmtree(self.temp_path, ignore_errors=True)
async def progress_update(self): async def progress_update(self):
""" """
@ -342,7 +342,6 @@ class Download:
self.info.file_size = os.path.getsize(status.get("filename")) self.info.file_size = os.path.getsize(status.get("filename"))
except FileNotFoundError: except FileNotFoundError:
self.info.file_size = 0 self.info.file_size = 0
pass
self.info.status = status.get("status", self.info.status) self.info.status = status.get("status", self.info.status)
self.info.msg = status.get("msg") self.info.msg = status.get("msg")

168
app/library/DownloadQueue.py
View file

@ -1,13 +1,14 @@
import asyncio import asyncio
import datetime
import json import json
import logging import logging
import os import os
import time import time
from datetime import UTC, datetime
from email.utils import formatdate from email.utils import formatdate
from sqlite3 import Connection from sqlite3 import Connection
import yt_dlp import yt_dlp
from aiohttp import web
from .AsyncPool import AsyncPool from .AsyncPool import AsyncPool
from .config import Config from .config import Config
@ -17,7 +18,7 @@ from .Emitter import Emitter
from .EventsSubscriber import Events from .EventsSubscriber import Events
from .ItemDTO import ItemDTO from .ItemDTO import ItemDTO
from .Singleton import Singleton 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") LOG = logging.getLogger("DownloadQueue")
@ -42,6 +43,8 @@ class DownloadQueue(metaclass=Singleton):
pool: AsyncPool | None = None pool: AsyncPool | None = None
"""Pool of workers to download the files.""" """Pool of workers to download the files."""
_active_downloads: dict[str, Download] = {}
_instance = None _instance = None
"""Instance of the DownloadQueue.""" """Instance of the DownloadQueue."""
@ -65,12 +68,32 @@ class DownloadQueue(metaclass=Singleton):
Returns: Returns:
DownloadQueue: The instance of the DownloadQueue DownloadQueue: The instance of the DownloadQueue
""" """
if not DownloadQueue._instance: if not DownloadQueue._instance:
DownloadQueue._instance = DownloadQueue() DownloadQueue._instance = DownloadQueue()
return DownloadQueue._instance 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: async def test(self) -> bool:
""" """
Test the datastore connection to the database. Test the datastore connection to the database.
@ -89,9 +112,9 @@ class DownloadQueue(metaclass=Singleton):
LOG.info( LOG.info(
f"Using '{self.config.max_workers}' worker/s for downloading. Can be configured via `YTP_MAX_WORKERS` environment variable." 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. Pause the download queue.
@ -101,7 +124,8 @@ class DownloadQueue(metaclass=Singleton):
""" """
if self.paused.is_set(): if self.paused.is_set():
self.paused.clear() 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 True
return False return False
@ -116,12 +140,12 @@ class DownloadQueue(metaclass=Singleton):
""" """
if not self.paused.is_set(): if not self.paused.is_set():
self.paused.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 True
return False return False
def isPaused(self) -> bool: def is_paused(self) -> bool:
""" """
Check if the download queue is paused. Check if the download queue is paused.
@ -131,6 +155,15 @@ class DownloadQueue(metaclass=Singleton):
""" """
return not self.paused.is_set() 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( async def __add_entry(
self, self,
entry: dict, entry: dict,
@ -169,13 +202,14 @@ class DownloadQueue(metaclass=Singleton):
live_in: str | None = None live_in: str | None = None
eventType = entry.get("_type") or "video" eventType = entry.get("_type") or "video"
if "playlist" == eventType: if "playlist" == eventType:
entries = entry.get("entries", []) entries = entry.get("entries", [])
playlist_index_digits = len(str(len(entries))) playlist_index_digits = len(str(len(entries)))
results = [] results = []
for i, etr in enumerate(entries, start=1): for i, etr in enumerate(entries, start=1):
etr["playlist"] = entry.get("id") 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"): for property in ("id", "title", "uploader", "uploader_id"):
if property in entry: if property in entry:
etr[f"playlist_{property}"] = entry.get(property) etr[f"playlist_{property}"] = entry.get(property)
@ -202,7 +236,8 @@ class DownloadQueue(metaclass=Singleton):
} }
return {"status": "ok"} 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. # check if the video is live stream.
if "live_status" in entry and "is_upcoming" == entry.get("live_status"): if "live_status" in entry and "is_upcoming" == entry.get("live_status"):
if "release_timestamp" in entry and entry.get("release_timestamp"): if "release_timestamp" in entry and entry.get("release_timestamp"):
@ -210,11 +245,9 @@ class DownloadQueue(metaclass=Singleton):
else: else:
error = "Live stream not yet started. And no date is set." error = "Live stream not yet started. And no date is set."
else: else:
error = entry.get("msg", None) error = entry.get("msg")
LOG.debug( LOG.debug(f"Entry id '{entry.get('id')}' url '{entry.get('webpage_url')} - {entry.get('url')}'.")
f"Entry id '{entry.get('id', None)}' url '{entry.get('webpage_url', None)} - {entry.get('url', None)}'."
)
if self.done.exists(key=entry["id"], url=str(entry.get("webpage_url") or 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"]) 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}) options.update({"is_manifestless": is_manifestless})
live_status: list = ["is_live", "is_upcoming"] 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: 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: except Exception as e:
LOG.exception(e) LOG.exception(e)
return {"status": "error", "msg": str(e)} return {"status": "error", "msg": str(e)}
@ -245,7 +278,7 @@ class DownloadQueue(metaclass=Singleton):
extras: dict = {} extras: dict = {}
fields: tuple = ("uploader", "channel", "thumbnail") fields: tuple = ("uploader", "channel", "thumbnail")
for field in fields: for field in fields:
if entry.get(field, None): if entry.get(field):
extras[field] = entry.get(field) extras[field] = entry.get(field)
dl = ItemDTO( dl = ItemDTO(
@ -253,7 +286,6 @@ class DownloadQueue(metaclass=Singleton):
title=str(entry.get("title")), title=str(entry.get("title")),
url=str(entry.get("webpage_url") or entry.get("url")), url=str(entry.get("webpage_url") or entry.get("url")),
preset=preset, preset=preset,
thumbnail=entry.get("thumbnail", None),
folder=folder, folder=folder,
download_dir=download_dir, download_dir=download_dir,
temp_dir=self.config.temp_path, 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)) 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" dlInfo.info.status = "not_live"
itemDownload = self.done.put(dlInfo) itemDownload = self.done.put(dlInfo)
NotifyEvent = Events.COMPLETED NotifyEvent = Events.COMPLETED
@ -294,7 +326,8 @@ class DownloadQueue(metaclass=Singleton):
) )
return {"status": "ok"} return {"status": "ok"}
elif eventType.startswith("url"):
if eventType.startswith("url"):
return await self.add( return await self.add(
url=str(entry.get("url")), url=str(entry.get("url")),
preset=preset, preset=preset,
@ -312,7 +345,7 @@ class DownloadQueue(metaclass=Singleton):
url: str, url: str,
preset: str, preset: str,
folder: str, folder: str,
config: dict|None = None, config: dict | None = None,
cookies: str = "", cookies: str = "",
template: str = "", template: str = "",
already=None, already=None,
@ -320,7 +353,7 @@ class DownloadQueue(metaclass=Singleton):
config = config if config else {} config = config if config else {}
folder = str(folder) if folder 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( LOG.info(
f"Adding 'URL: {url}' to 'Folder: {filePath}' with 'Preset: {preset}' 'Naming: {template}', 'Cookies: {cookies}' 'YTConfig: {config}'." 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) already.add(url)
try: try:
downloaded, id_dict = self.isDownloaded(url) downloaded, id_dict = self._is_downloaded(url)
if downloaded is True and id_dict: 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." message = f"This url with ID '{id_dict.get('id')}' has been downloaded already and recorded in archive."
LOG.info(message) LOG.info(message)
@ -354,8 +387,8 @@ class DownloadQueue(metaclass=Singleton):
entry = await asyncio.wait_for( entry = await asyncio.wait_for(
fut=asyncio.get_running_loop().run_in_executor( fut=asyncio.get_running_loop().run_in_executor(
None, None,
ExtractInfo, extract_info,
get_opts(preset, mergeConfig(self.config.ytdl_options, config)), get_opts(preset, merge_config(self.config.ytdl_options, config)),
url, url,
bool(self.config.ytdl_debug), bool(self.config.ytdl_debug),
), ),
@ -392,6 +425,16 @@ class DownloadQueue(metaclass=Singleton):
) )
async def cancel(self, ids: list[str]) -> dict[str, str]: 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"} status: dict[str, str] = {"status": "ok"}
for id in ids: for id in ids:
@ -403,29 +446,40 @@ class DownloadQueue(metaclass=Singleton):
LOG.warning(f"Requested cancel for non-existent download {id=}. {e!s}") LOG.warning(f"Requested cancel for non-existent download {id=}. {e!s}")
continue continue
itemMessage = f"{id=} {item.info.id=} {item.info.title=}" item_ref = f"{id=} {item.info.id=} {item.info.title=}"
if item.running(): if item.running():
LOG.debug(f"Canceling {itemMessage}") LOG.debug(f"Canceling {item_ref}")
item.cancel() item.cancel()
LOG.info(f"Cancelled {itemMessage}") LOG.info(f"Cancelled {item_ref}")
await item.close() await item.close()
else: else:
await item.close() await item.close()
LOG.debug(f"Deleting from queue {itemMessage}") LOG.debug(f"Deleting from queue {item_ref}")
self.queue.delete(id) self.queue.delete(id)
asyncio.create_task(self.emitter.cancelled(dl=item.info.serialize()), name=f"notifier-c-{id}") asyncio.create_task(self.emitter.cancelled(dl=item.info.serialize()), name=f"notifier-c-{id}")
item.info.status = "cancelled" item.info.status = "cancelled"
item.info.error = "Cancelled by user." item.info.error = "Cancelled by user."
self.done.put(item) self.done.put(item)
asyncio.create_task(self.emitter.completed(dl=item.info.serialize()), name=f"notifier-d-{id}") 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" status[id] = "ok"
return status return status
async def clear(self, ids: list[str], remove_file: bool = False) -> dict[str, str]: 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"} status: dict[str, str] = {"status": "ok"}
for id in ids: for id in ids:
@ -447,10 +501,10 @@ class DownloadQueue(metaclass=Singleton):
filename = f"{item.info.folder}/{item.info.filename}" filename = f"{item.info.folder}/{item.info.filename}"
try: try:
realFile: str = calcDownloadPath( realFile: str = calc_download_path(
basePath=self.config.download_path, base_path=self.config.download_path,
folder=filename, folder=filename,
createPath=False, create_path=False,
) )
if realFile and os.path.exists(realFile): if realFile and os.path.exists(realFile):
os.remove(realFile) os.remove(realFile)
@ -497,11 +551,18 @@ class DownloadQueue(metaclass=Singleton):
return items 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( self.pool = AsyncPool(
loop=asyncio.get_running_loop(), loop=asyncio.get_running_loop(),
num_workers=self.config.max_workers, num_workers=self.config.max_workers,
worker_co=self.__downloadFile, worker_co=self._download_file,
name="download_pool", name="download_pool",
logger=logging.getLogger("WorkerPool"), 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()}") LOG.info(f"Waiting for worker to be free. {self.pool.get_workers_status()}")
await asyncio.sleep(1) 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.") LOG.info(f"Waiting for item to download. '{self.pool.get_available_workers()}' free workers.")
if self.event: if self.event:
await self.event.wait() await self.event.wait()
self.event.clear() self.event.clear()
LOG.debug("Cleared wait event.") 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.") LOG.info("Download pool is paused.")
await self.paused.wait() await self.paused.wait()
LOG.info("Download pool resumed downloading.") LOG.info("Download pool resumed downloading.")
entry = self.queue.getNextDownload() entry = self.queue.get_next_download()
await asyncio.sleep(0.2) await asyncio.sleep(0.2)
if entry is None: if entry is None:
@ -544,13 +605,25 @@ class DownloadQueue(metaclass=Singleton):
LOG.debug(f"Pushed {entry=} to executor.") LOG.debug(f"Pushed {entry=} to executor.")
await asyncio.sleep(1) await asyncio.sleep(1)
async def __downloadFile(self, id: str, entry: Download): async def _download_file(self, id: str, entry: Download) -> None:
filePath = calcDownloadPath(basePath=self.config.download_path, folder=entry.info.folder) """
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( LOG.info(
f"Downloading 'id: {id}', 'Title: {entry.info.title}', 'URL: {entry.info.url}' to 'Folder: {filePath}'." f"Downloading 'id: {id}', 'Title: {entry.info.title}', 'URL: {entry.info.url}' to 'Folder: {filePath}'."
) )
try: try:
self._active_downloads[entry.info._id] = entry
await entry.start(self.emitter) await entry.start(self.emitter)
if "finished" != entry.info.status: if "finished" != entry.info.status:
@ -563,6 +636,9 @@ class DownloadQueue(metaclass=Singleton):
entry.info.status = "error" entry.info.status = "error"
finally: finally:
if entry.info._id in self._active_downloads:
self._active_downloads.pop(entry.info._id, None)
await entry.close() await entry.close()
if self.queue.exists(key=id): if self.queue.exists(key=id):
@ -582,8 +658,18 @@ class DownloadQueue(metaclass=Singleton):
if self.event: if self.event:
self.event.set() 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: if not url or not self.config.keep_archive:
return False, None 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)

8
app/library/Emitter.py
View file

@ -1,6 +1,6 @@
import asyncio import asyncio
import logging import logging
from typing import Awaitable from collections.abc import Awaitable
from .EventsSubscriber import Events from .EventsSubscriber import Events
from .Singleton import Singleton from .Singleton import Singleton
@ -69,19 +69,19 @@ class Emitter(metaclass=Singleton):
async def cleared(self, dl: dict | None = None, local: bool = False, **kwargs): async def cleared(self, dl: dict | None = None, local: bool = False, **kwargs):
await self.emit(Events.CLEARED, data=dl, local=local, **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": {}} msg = {"type": "error", "message": message, "data": {}}
if data: if data:
msg.update({"data": data}) msg.update({"data": data})
await self.emit(Events.ERROR, data=msg, local=local, **kwargs) 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": {}} msg = {"type": "info", "message": message, "data": {}}
if data: if data:
msg.update({"data": data}) msg.update({"data": data})
await self.emit(Events.LOG_INFO, data=msg, local=local, **kwargs) 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": {}} msg = {"type": "success", "message": message, "data": {}}
if data: if data:
msg.update({"data": data}) msg.update({"data": data})

4
app/library/EventsSubscriber.py
View file

@ -1,7 +1,7 @@
import asyncio import asyncio
import logging import logging
from collections.abc import Awaitable
from dataclasses import dataclass from dataclasses import dataclass
from typing import Awaitable
from .Singleton import Singleton from .Singleton import Singleton
@ -180,7 +180,7 @@ class EventsSubscriber(metaclass=Singleton):
tasks.append(asyncio.create_task(callback(event, data))) tasks.append(asyncio.create_task(callback(event, data)))
except Exception as e: except Exception as e:
LOG.error(f"Failed to emit event '{event}' to '{id}'. Error message '{e!s}'.")
LOG.exception(e) LOG.exception(e)
LOG.error(f"Failed to emit event '{event}' to '{id}'. Error message '{e!s}'.")
return asyncio.gather(*tasks) return asyncio.gather(*tasks)

62
app/library/HttpAPI.py
View file

@ -8,9 +8,9 @@ import os
import random import random
import time import time
import uuid import uuid
from datetime import datetime from collections.abc import Awaitable
from datetime import UTC, datetime
from pathlib import Path from pathlib import Path
from typing import Awaitable
import httpx import httpx
import magic import magic
@ -31,7 +31,7 @@ from .Playlist import Playlist
from .Segments import Segments from .Segments import Segments
from .Subtitle import Subtitle from .Subtitle import Subtitle
from .Tasks import Task, Tasks 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") LOG = logging.getLogger("http_api")
MIME = magic.Magic(mime=True) MIME = magic.Magic(mime=True)
@ -93,6 +93,9 @@ class HttpAPI(Common):
return decorator return decorator
async def on_shutdown(self, _: web.Application):
LOG.debug("Shutting down http API server.")
def attach(self, app: web.Application) -> "HttpAPI": def attach(self, app: web.Application) -> "HttpAPI":
""" """
Attach the routes to the application. Attach the routes to the application.
@ -123,9 +126,10 @@ class HttpAPI(Common):
except Exception as e: except Exception as e:
LOG.exception(e) LOG.exception(e)
app.on_shutdown.append(self.on_shutdown)
return self 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. Preload static files from the ui/exported folder.
@ -154,7 +158,7 @@ class HttpAPI(Common):
status=web.HTTPOk.status_code, 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. Preload static files from the ui/exported folder.
@ -187,7 +191,7 @@ class HttpAPI(Common):
self._static_holder[urlPath] = {"content": content, "content_type": contentType} self._static_holder[urlPath] = {"content": content, "content_type": contentType}
LOG.debug(f"Preloading '{urlPath}'.") LOG.debug(f"Preloading '{urlPath}'.")
app.router.add_get(urlPath, self.staticFile) app.router.add_get(urlPath, self._static_file)
preloaded += 1 preloaded += 1
if urlPath.endswith("/index.html") and urlPath != "/index.html": if urlPath.endswith("/index.html") and urlPath != "/index.html":
@ -195,8 +199,8 @@ class HttpAPI(Common):
parentNoSlash = urlPath.replace("/index.html", "") parentNoSlash = urlPath.replace("/index.html", "")
self._static_holder[parentSlash] = {"content": content, "content_type": contentType} self._static_holder[parentSlash] = {"content": content, "content_type": contentType}
self._static_holder[parentNoSlash] = {"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(parentSlash, self._static_file)
app.router.add_get(parentNoSlash, self.staticFile) app.router.add_get(parentNoSlash, self._static_file)
preloaded += 2 preloaded += 2
if preloaded < 1: if preloaded < 1:
@ -232,7 +236,7 @@ class HttpAPI(Common):
self.routes.route(method._http_method, f"/{http_path}")(method) self.routes.route(method._http_method, f"/{http_path}")(method)
self.routes.static("/api/download/", self.config.download_path) self.routes.static("/api/download/", self.config.download_path)
self.preloadStatic(app) self._preload_static(app)
try: try:
app.add_routes(self.routes) app.add_routes(self.routes)
@ -389,7 +393,7 @@ class HttpAPI(Common):
"proxy": self.config.ytdl_options.get("proxy", None), "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) self.cache.set(key=self.cache.hash(url), value=data, ttl=300)
data["_cached"] = { data["_cached"] = {
"key": self.cache.hash(url), "key": self.cache.hash(url),
@ -478,7 +482,7 @@ class HttpAPI(Common):
""" """
return web.json_response( 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") @route("PUT", "api/tasks")
@ -543,7 +547,7 @@ class HttpAPI(Common):
tasks.append(Task(**item)) tasks.append(Task(**item))
try: try:
tasks = ins.save(tasks=tasks).load().getTasks() tasks = ins.save(tasks=tasks).load().get_tasks()
except Exception as e: except Exception as e:
LOG.exception(e) LOG.exception(e)
return web.json_response( return web.json_response(
@ -597,7 +601,7 @@ class HttpAPI(Common):
if not id: if not id:
return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code) 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: if not item:
return web.json_response(data={"error": "item not found."}, status=web.HTTPNotFound.status_code) 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: if request.if_modified_since:
lastMod = time.strftime( 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): 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}) return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod})
@ -873,7 +877,7 @@ class HttpAPI(Common):
segmenter = Segments( segmenter = Segments(
index=int(segment), 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, vconvert=vc == 1,
aconvert=ac == 1, aconvert=ac == 1,
) )
@ -887,10 +891,10 @@ class HttpAPI(Common):
"Pragma": "public", "Pragma": "public",
"Cache-Control": f"public, max-age={time.time() + 31536000}", "Cache-Control": f"public, max-age={time.time() + 31536000}",
"Last-Modified": time.strftime( "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( "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, status=web.HTTPOk.status_code,
@ -915,7 +919,7 @@ class HttpAPI(Common):
if request.if_modified_since: if request.if_modified_since:
lastMod = time.strftime( 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): 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}) return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod})
@ -932,10 +936,10 @@ class HttpAPI(Common):
"Pragma": "public", "Pragma": "public",
"Cache-Control": f"public, max-age={time.time() + 31536000}", "Cache-Control": f"public, max-age={time.time() + 31536000}",
"Last-Modified": time.strftime( "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( "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, status=web.HTTPOk.status_code,
@ -1008,7 +1012,8 @@ class HttpAPI(Common):
"Access-Control-Allow-Origin": "*", "Access-Control-Allow-Origin": "*",
"Cache-Control": f"public, max-age={time.time() + 31536000}", "Cache-Control": f"public, max-age={time.time() + 31536000}",
"Expires": time.strftime( "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") 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: try:
realFile: str = calcDownloadPath(basePath=self.config.download_path, folder=file, createPath=False) realFile: str = calc_download_path(base_path=self.config.download_path, folder=file, create_path=False)
if not os.path.exists(realFile): if not os.path.exists(realFile) or not os.path.isfile(realFile):
return web.json_response( return web.json_response(
data={"error": f"File '{file}' does not exist."}, status=web.HTTPNotFound.status_code data={"error": f"File '{file}' does not exist."}, status=web.HTTPNotFound.status_code
) )
@ -1120,8 +1128,8 @@ class HttpAPI(Common):
""" """
return web.json_response( return web.json_response(
data={ data={
"notifications": Notification.get_instance().getTargets(), "notifications": Notification.get_instance().get_targets(),
"allowedTypes": list(NotificationEvents.getEvents().values()), "allowedTypes": list(NotificationEvents.get_events().values()),
}, },
status=web.HTTPOk.status_code, status=web.HTTPOk.status_code,
dumps=self.encoder.encode, dumps=self.encoder.encode,
@ -1167,7 +1175,7 @@ class HttpAPI(Common):
status=web.HTTPBadRequest.status_code, status=web.HTTPBadRequest.status_code,
) )
targets.append(ins.makeTarget(item)) targets.append(ins.make_target(item))
try: try:
if len(targets) < 1: if len(targets) < 1:
@ -1179,7 +1187,7 @@ class HttpAPI(Common):
LOG.exception(e) LOG.exception(e)
return web.json_response({"error": "Failed to save tasks."}, status=web.HTTPInternalServerError.status_code) 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) return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode)

49
app/library/HttpSocket.py
View file

@ -6,9 +6,10 @@ import os
import pty import pty
import shlex import shlex
import time import time
from datetime import datetime from datetime import UTC, datetime
from typing import Any from typing import Any
import anyio
import socketio import socketio
from aiohttp import web from aiohttp import web
@ -18,7 +19,7 @@ from .DownloadQueue import DownloadQueue
from .Emitter import Emitter from .Emitter import Emitter
from .encoder import Encoder from .encoder import Encoder
from .EventsSubscriber import Event, Events, EventsSubscriber from .EventsSubscriber import Event, Events, EventsSubscriber
from .Utils import arg_converter, isDownloaded from .Utils import arg_converter, is_downloaded
LOG = logging.getLogger("socket_api") LOG = logging.getLogger("socket_api")
@ -67,6 +68,9 @@ class HttpSocket(Common):
wrapper._ws_event = func.__name__ # type: ignore wrapper._ws_event = func.__name__ # type: ignore
return wrapper return wrapper
async def on_shutdown(self, _: web.Application):
LOG.debug("Shutting down socket server.")
def attach(self, app: web.Application): def attach(self, app: web.Application):
self.sio.attach(app, socketio_path=self.config.url_socketio) 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) 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): async def cli_post(self, sid: str, data):
if not data: if not data:
await self.emitter.emit(Events.CLI_CLOSE, {"exitcode": 0}, to=sid) 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_OUTPUT, {"type": "stderr", "line": str(e)}, to=sid)
await self.emitter.emit(Events.CLI_CLOSE, {"exitcode": -1}, 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): async def add_url(self, sid: str, data: dict):
url: str | None = data.get("url") url: str | None = data.get("url")
@ -189,7 +196,7 @@ class HttpSocket(Common):
status = await self.add(**item) status = await self.add(**item)
await self.emitter.emit(event=Events.STATUS, data=status, to=sid) 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): async def item_cancel(self, sid: str, id: str):
if not id: if not id:
await self.emitter.error("Invalid request.", to=sid) 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) 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): async def item_delete(self, sid: str, data: dict):
if not data: if not data:
await self.emitter.error("Invalid request.", to=sid) 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) await self.emitter.emit(event=Events.ITEM_DELETE, data=status)
@ws_event # type: ignore @ws_event
async def archive_item(self, sid: str, data: dict): async def archive_item(self, _: str, data: dict):
if not isinstance(data, dict) or "url" not in data or not self.config.keep_archive: if not isinstance(data, dict) or "url" not in data or not self.config.keep_archive:
return return
@ -231,36 +238,36 @@ class HttpSocket(Common):
if not file: if not file:
return 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: if exists or "archive_id" not in idDict or idDict["archive_id"] is None:
return return
with open(file, "a") as f: async with await anyio.open_file(file, "a") as f:
f.write(f"{idDict['archive_id']}\n") await f.write(f"{idDict['archive_id']}\n")
manual_archive = self.config.manual_archive manual_archive = self.config.manual_archive
if manual_archive: if manual_archive:
previouslyArchived = False previouslyArchived = False
if os.path.exists(manual_archive): if os.path.exists(manual_archive):
with open(manual_archive, "r") as f: async with await anyio.open_file(manual_archive) as f:
for line in f.readlines(): async for line in f:
if idDict["archive_id"] in line: if idDict["archive_id"] in line:
previouslyArchived = True previouslyArchived = True
break break
if not previouslyArchived: if not previouslyArchived:
with open(manual_archive, "a") as f: async with await anyio.open_file(manual_archive, "a") as f:
f.write(f"{idDict['archive_id']} - at: {datetime.now().isoformat()}\n") 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']}'.") 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): async def connect(self, sid: str, _=None):
data = { data = {
**self.queue.get(), **self.queue.get(),
"config": self.config.frontend(), "config": self.config.frontend(),
"presets": self.config.presets, "presets": self.config.presets,
"paused": self.queue.isPaused(), "paused": self.queue.is_paused(),
} }
# get download folder listing. # get download folder listing.
@ -269,13 +276,13 @@ class HttpSocket(Common):
await self.emitter.emit(event=Events.INITIAL_DATA, data=data, to=sid) await self.emitter.emit(event=Events.INITIAL_DATA, data=data, to=sid)
@ws_event # type: ignore @ws_event
async def pause(self, sid: str, _=None): async def pause(self, *_, **__):
self.queue.pause() self.queue.pause()
await self.emitter.emit(event=Events.PAUSED, data={"paused": True, "at": time.time()}) await self.emitter.emit(event=Events.PAUSED, data={"paused": True, "at": time.time()})
@ws_event # type: ignore @ws_event
async def resume(self, sid: str, _=None): async def resume(self, *_, **__):
self.queue.resume() self.queue.resume()
await self.emitter.emit(event=Events.PAUSED, data={"paused": False, "at": time.time()}) await self.emitter.emit(event=Events.PAUSED, data={"paused": False, "at": time.time()})

20
app/library/M3u8.py
View file

@ -3,29 +3,21 @@ import os
from urllib.parse import quote from urllib.parse import quote
from .ffprobe import ffprobe from .ffprobe import ffprobe
from .Utils import StreamingError, calcDownloadPath from .Utils import StreamingError, calc_download_path
class M3u8: class M3u8:
duration: float = 6.000000 duration: float = 6.000000
ok_vcodecs: tuple = ( ok_vcodecs: tuple = ("h264", "x264", "avc")
"h264", ok_acodecs: tuple = ("aac", "m4a", "mp3")
"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.url = url
self.duration = float(segment_duration) if segment_duration is not None else self.duration 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: 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): if not os.path.exists(realFile):
error = f"File '{realFile}' does not exist." error = f"File '{realFile}' does not exist."
@ -79,7 +71,7 @@ class M3u8:
return "\n".join(m3u8) return "\n".join(m3u8)
async def make_subtitle(self, download_path: str, file: str, duration: float) -> str: 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): if not os.path.exists(realFile):
error = f"File '{realFile}' does not exist." error = f"File '{realFile}' does not exist."

37
app/library/Notifications.py
View file

@ -20,7 +20,7 @@ LOG = logging.getLogger("notifications")
@dataclass(kw_only=True) @dataclass(kw_only=True)
class targetRequestHeader: class TargetRequestHeader:
"""Request header details for a notification target.""" """Request header details for a notification target."""
key: str key: str
@ -37,13 +37,13 @@ class targetRequestHeader:
@dataclass(kw_only=True) @dataclass(kw_only=True)
class targetRequest: class TargetRequest:
"""Request details for a notification target.""" """Request details for a notification target."""
type: str type: str
method: str method: str
url: str url: str
headers: list[targetRequestHeader] = field(default_factory=list) headers: list[TargetRequestHeader] = field(default_factory=list)
def serialize(self) -> dict: def serialize(self) -> dict:
return { return {
@ -67,7 +67,7 @@ class Target:
id: str id: str
name: str name: str
on: list[str] on: list[str]
request: targetRequest request: TargetRequest
def serialize(self) -> dict: def serialize(self) -> dict:
return { return {
@ -95,17 +95,12 @@ class NotificationEvents:
TEST = Events.TEST TEST = Events.TEST
@staticmethod @staticmethod
def getEvents() -> dict[str, str]: def get_events() -> dict[str, str]:
data: dict[str, str] = {} return {k: v for k, v in vars(NotificationEvents).items() if not k.startswith("__") and not callable(v)}
for k, v in vars(NotificationEvents).items():
if not k.startswith("__") and not callable(v):
data[k] = v
return data
@staticmethod @staticmethod
def isValid(event: str) -> bool: def is_valid(event: str) -> bool:
return event in NotificationEvents.getEvents().values() return event in NotificationEvents.get_events().values()
class Notification(metaclass=Singleton): class Notification(metaclass=Singleton):
@ -147,7 +142,7 @@ class Notification(metaclass=Singleton):
return Notification._instance return Notification._instance
def getTargets(self) -> list[Target]: def get_targets(self) -> list[Target]:
"""Get the list of notification targets.""" """Get the list of notification targets."""
return self._targets return self._targets
@ -191,7 +186,7 @@ class Notification(metaclass=Singleton):
LOG.info(f"Loading notification targets from '{self._file}'.") LOG.info(f"Loading notification targets from '{self._file}'.")
try: try:
with open(self._file, "r") as f: with open(self._file) as f:
targets = json.load(f) targets = json.load(f)
except Exception as e: except Exception as e:
LOG.error(f"Error loading notification targets from '{self._file}'. '{e!s}'") 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}'") LOG.error(f"Invalid notification target '{target}'. '{e!s}'")
continue continue
target = self.makeTarget(target) target = self.make_target(target)
self._targets.append(target) self._targets.append(target)
@ -261,7 +256,7 @@ class Notification(metaclass=Singleton):
raise ValueError(msg) raise ValueError(msg)
for e in target["on"]: 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." msg = f"Invalid notification target. Invalid event '{e}' found."
raise ValueError(msg) 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}'.") LOG.error(f"Error sending Notification event '{event}' id '{itemId}' to '{target.name}'. '{e}'.")
return {"url": target.request.url, "status": 500, "text": str(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. Make a notification target from a dictionary.
@ -360,12 +355,12 @@ class Notification(metaclass=Singleton):
id=target.get("id"), id=target.get("id"),
name=target.get("name"), name=target.get("name"),
on=target.get("on", []), on=target.get("on", []),
request=targetRequest( request=TargetRequest(
type=target.get("request", {}).get("type", "json"), type=target.get("request", {}).get("type", "json"),
method=target.get("request", {}).get("method", "POST"), method=target.get("request", {}).get("method", "POST"),
url=target.get("request", {}).get("url"), url=target.get("request", {}).get("url"),
headers=[ 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", []) for h in target.get("request", {}).get("headers", [])
], ],
), ),
@ -375,7 +370,7 @@ class Notification(metaclass=Singleton):
if len(self._targets) < 1: if len(self._targets) < 1:
return False return False
if not NotificationEvents.isValid(event): if not NotificationEvents.is_valid(event):
return False return False
return self.send(event, data) return self.send(event, data)

10
app/library/Playlist.py
View file

@ -3,11 +3,11 @@ import re
from pathlib import Path from pathlib import Path
from urllib.parse import quote from urllib.parse import quote
from aiohttp.web import Response from aiohttp.web import HTTPFound, Response
from .ffprobe import ffprobe from .ffprobe import ffprobe
from .Subtitle import Subtitle from .Subtitle import Subtitle
from .Utils import StreamingError, calcDownloadPath, checkId from .Utils import StreamingError, calc_download_path, check_id
class Playlist: class Playlist:
@ -17,16 +17,16 @@ class Playlist:
self.url = url self.url = url
async def make(self, download_path: str, file: str) -> str | Response: 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(): if not rFile.exists():
possibleFile = checkId(download_path, rFile) possibleFile = check_id(file=rFile)
if not possibleFile: if not possibleFile:
msg = f"File '{rFile}' does not exist." msg = f"File '{rFile}' does not exist."
raise StreamingError(msg) raise StreamingError(msg)
return Response( return Response(
status=302, status=HTTPFound.status_code,
headers={ headers={
"Location": f"{self.url}api/player/playlist/{quote(str(possibleFile).replace(download_path, '').strip('/'))}.m3u8" "Location": f"{self.url}api/player/playlist/{quote(str(possibleFile).replace(download_path, '').strip('/'))}.m3u8"
}, },

10
app/library/Segments.py
View file

@ -6,7 +6,7 @@ import tempfile
from .config import Config from .config import Config
from .ffprobe import ffprobe from .ffprobe import ffprobe
from .Utils import StreamingError, calcDownloadPath from .Utils import StreamingError, calc_download_path
LOG = logging.getLogger("player.segments") LOG = logging.getLogger("player.segments")
@ -25,7 +25,7 @@ class Segments:
self.aconvert = True self.aconvert = True
async def stream(self, path: str, file: str) -> bytes: 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): if not os.path.exists(realFile):
msg = f"File {realFile} does not exist." msg = f"File {realFile} does not exist."
@ -37,7 +37,7 @@ class Segments:
pass pass
tmpDir: str = tempfile.gettempdir() 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): if not os.path.exists(tmpFile):
os.symlink(realFile, tmpFile) os.symlink(realFile, tmpFile)
@ -108,8 +108,8 @@ class Segments:
data, err = await proc.communicate() data, err = await proc.communicate()
if 0 != proc.returncode: if 0 != proc.returncode:
LOG.error(f'Failed to stream {realFile} segment {self.index}. {err.decode("utf-8")}.') LOG.error(f"Failed to stream '{realFile}' segment '{self.index}'. {err.decode('utf-8')}.")
msg = f"Failed to stream {realFile} segment {self.index}." msg = f"Failed to stream '{realFile}' segment '{self.index}'."
raise StreamingError(msg) raise StreamingError(msg)
return data return data

9
app/library/Subtitle.py
View file

@ -1,11 +1,12 @@
import logging import logging
import pathlib import pathlib
import anyio
import pysubs2 import pysubs2
from pysubs2.formats.substation import SubstationFormat from pysubs2.formats.substation import SubstationFormat
from pysubs2.time import ms_to_times from pysubs2.time import ms_to_times
from .Utils import calcDownloadPath from .Utils import calc_download_path
LOG = logging.getLogger("player.subtitle") LOG = logging.getLogger("player.subtitle")
@ -28,7 +29,7 @@ class Subtitle:
) )
async def make(self, path: str, file: str) -> str: 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) rFile = pathlib.Path(realFile)
@ -41,8 +42,8 @@ class Subtitle:
raise Exception(msg) raise Exception(msg)
if rFile.suffix == ".vtt": if rFile.suffix == ".vtt":
with open(realFile) as f: async with await anyio.open_file(realFile) as f:
return f.read() return await f.read()
subs = pysubs2.load(path=str(rFile)) subs = pysubs2.load(path=str(rFile))

35
app/library/Tasks.py
View file

@ -4,7 +4,7 @@ import logging
import os import os
import time import time
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime from datetime import UTC, datetime
from typing import Any from typing import Any
import httpx import httpx
@ -106,6 +106,18 @@ class Tasks(metaclass=Singleton):
Tasks._instance = Tasks() Tasks._instance = Tasks()
return Tasks._instance 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): def attach(self, _: web.Application):
""" """
Attach the tasks to the aiohttp application. Attach the tasks to the aiohttp application.
@ -119,13 +131,7 @@ class Tasks(metaclass=Singleton):
""" """
self.load() self.load()
def shutdown(self): def get_tasks(self) -> list[Task]:
"""
Shutdown the tasks service.
"""
self.clear()
def getTasks(self) -> list[Task]:
"""Return the tasks.""" """Return the tasks."""
return [job.task for job in self._jobs] return [job.task for job in self._jobs]
@ -177,7 +183,7 @@ class Tasks(metaclass=Singleton):
return self return self
def clear(self) -> "Tasks": def clear(self, shutdown: bool = False) -> "Tasks":
""" """
Clear all tasks. Clear all tasks.
@ -193,8 +199,9 @@ class Tasks(metaclass=Singleton):
LOG.info(f"Stopping job '{task.id}: {task.name}'.") LOG.info(f"Stopping job '{task.id}: {task.name}'.")
task.job.stop() task.job.stop()
except Exception as e: except Exception as e:
LOG.exception(e) if not shutdown:
LOG.error(f"Failed to stop job '{task.id}: {task.name}'. '{e!s}'.") LOG.exception(e)
LOG.error(f"Failed to stop job '{task.id}: {task.name}'. '{e!s}'.")
self._jobs.clear() self._jobs.clear()
@ -292,7 +299,7 @@ class Tasks(metaclass=Singleton):
""" """
try: try:
timeNow = datetime.now().isoformat() timeNow = datetime.now(UTC).isoformat()
started = time.time() started = time.time()
if not task.url: if not task.url:
LOG.error(f"Failed to dispatch task '{task.id}: {task.name}'. No URL found.") 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) await asyncio.wait_for(asyncio.gather(*tasks), timeout=None)
timeNow = datetime.now().isoformat() timeNow = datetime.now(UTC).isoformat()
ended = time.time() ended = time.time()
LOG.info(f"Task '{task.id}: {task.name}' completed at '{timeNow}' took '{ended - started:.2f}' seconds.") 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.") await self._emitter.success(f"Task '{task.name}' completed in '{ended - started:.2f}' seconds.")
except Exception as e: 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}'.") 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}'.") await self._emitter.error(f"Task '{task.name}' failed to execute at '{timeNow}'. '{e!s}'.")

53
app/library/Utils.py
View file

@ -8,7 +8,7 @@ import re
import shlex import shlex
import socket import socket
import uuid import uuid
from datetime import datetime, timezone from datetime import UTC, datetime
from functools import lru_cache from functools import lru_cache
from typing import Any from typing import Any
@ -78,7 +78,7 @@ def get_opts(preset: str, ytdl_opts: dict) -> dict:
return opts 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. 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) 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. Calculates download path and prevents folder traversal.
@ -118,25 +118,25 @@ def calcDownloadPath(basePath: str, folder: str | None = None, createPath: bool
""" """
if not folder: if not folder:
return basePath return base_path
if folder.startswith("/"): if folder.startswith("/"):
folder = folder[1:] folder = folder[1:]
realBasePath = os.path.realpath(basePath) realBasePath = os.path.realpath(base_path)
download_path = os.path.realpath(os.path.join(basePath, folder)) download_path = os.path.realpath(os.path.join(base_path, folder))
if not download_path.startswith(realBasePath): if not download_path.startswith(realBasePath):
msg = f'Folder "{folder}" must resolve inside the base download folder "{realBasePath}".' msg = f'Folder "{folder}" must resolve inside the base download folder "{realBasePath}".'
raise Exception(msg) 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) os.makedirs(download_path, exist_ok=True)
return download_path 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 = { params: dict = {
"color": "no_color", "color": "no_color",
"extract_flat": True, "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) 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""" """Merge data from source into destination"""
destination_copy = copy.deepcopy(destination) destination_copy = copy.deepcopy(destination)
for key, value in source.items(): for key, value in source.items():
destination_key_value = destination_copy.get(key) destination_key_value = destination_copy.get(key)
if isinstance(value, dict) and isinstance(destination_key_value, dict): 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): elif isinstance(value, list) and isinstance(destination_key_value, list):
destination_copy[key] = destination_key_value + value destination_copy[key] = destination_key_value + value
else: else:
@ -184,7 +184,7 @@ def mergeDict(source: dict, destination: dict) -> dict:
return destination_copy 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 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.") LOG.error(f"Key '{key}' is not allowed to be manually set.")
del new_config[key] del new_config[key]
conf = mergeDict(new_config, config) conf = merge_dict(new_config, config)
if "impersonate" in conf: if "impersonate" in conf:
conf["impersonate"] = ImpersonateTarget.from_str(conf["impersonate"]) conf["impersonate"] = ImpersonateTarget.from_str(conf["impersonate"])
@ -209,8 +209,8 @@ def mergeConfig(config: dict, new_config: dict) -> dict:
return conf return conf
def isDownloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, str | None, str | None]]: def is_downloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, str | None, str | None]]:
global YTDLP_INFO_CLS global YTDLP_INFO_CLS # noqa: PLW0603
idDict = { idDict = {
"id": None, "id": None,
@ -263,7 +263,7 @@ def isDownloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, st
return (False, idDict) 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 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" 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 hasCookies: bool = False
for domain in cookies: for domain in cookies: # noqa: PLC0206
if not isinstance(cookies[domain], dict): if not isinstance(cookies[domain], dict):
continue continue
@ -287,7 +287,7 @@ def jsonCookie(cookies: dict[dict[str, any]]) -> str | None:
cookieDict = cookies[domain][subDomain][cookie] cookieDict = cookies[domain][subDomain][cookie]
if 0 == int(cookieDict["expirationDate"]): if 0 == int(cookieDict["expirationDate"]):
cookieDict["expirationDate"] = datetime.now(timezone.utc).timestamp() + (86400 * 1000) cookieDict["expirationDate"] = datetime.now(UTC).timestamp() + (86400 * 1000)
hasCookies = True hasCookies = True
netscapeCookies += ( 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. 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. 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: if id not in f.stem:
continue continue
if f.suffix not in ( if f.suffix not in (".mp4", ".mkv", ".webm", ".m4v", ".m4a", ".mp3", ".aac", ".ogg"):
".mp4",
".mkv",
".webm",
".m4v",
".m4a",
".mp3",
".aac",
".ogg",
):
continue continue
return f.absolute() return f.absolute()
@ -510,7 +501,7 @@ def arg_converter(args: str) -> dict:
Convert yt-dlp options to a dictionary. Convert yt-dlp options to a dictionary.
Args: Args:
opts (str): yt-dlp options string. args (str): yt-dlp options string.
Returns: Returns:
dict: yt-dlp options dictionary. dict: yt-dlp options dictionary.
@ -520,7 +511,7 @@ def arg_converter(args: str) -> dict:
create_parser = yt_dlp.options.create_parser create_parser = yt_dlp.options.create_parser
def defaultOpts(args: str): def _default_opts(args: str):
patched_parser = create_parser() patched_parser = create_parser()
try: try:
yt_dlp.options.create_parser = lambda: patched_parser yt_dlp.options.create_parser = lambda: patched_parser
@ -528,7 +519,7 @@ def arg_converter(args: str) -> dict:
finally: finally:
yt_dlp.options.create_parser = create_parser 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 opts = yt_dlp.parse_options(shlex.split(args)).ydl_opts

6
app/library/config.py
View file

@ -12,7 +12,7 @@ import coloredlogs
from dotenv import load_dotenv from dotenv import load_dotenv
from yt_dlp.version import __version__ as YTDLP_VERSION 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 from .version import APP_VERSION
@ -254,7 +254,7 @@ class Config:
logging.error(f"Config variable '{k}' had non-existing config reference '{key}'.") logging.error(f"Config variable '{k}' had non-existing config reference '{key}'.")
sys.exit(1) sys.exit(1)
v = v.replace(key, getattr(self, localKey)) # noqa: PLW2901 v = v.replace(key, getattr(self, localKey))
setattr(self, k, v) 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.") LOG.error(f"Key '{key}' is not allowed to be loaded via 'ytdlp.json' file.")
del opts[key] del opts[key]
self.ytdl_options = mergeDict(self.ytdl_options, opts) self.ytdl_options = merge_dict(self.ytdl_options, opts)
else: else:
LOG.error(f"Invalid yt-dlp custom options file '{optsFile}'.") LOG.error(f"Invalid yt-dlp custom options file '{optsFile}'.")
else: else:

9
app/library/ffprobe.py
View file

@ -8,6 +8,8 @@ import json
import operator import operator
import os import os
import anyio
# parameter-less decorator # parameter-less decorator
def async_lru_cache_decorator(async_function): def async_lru_cache_decorator(async_function):
@ -85,10 +87,7 @@ class FFStream:
if self.__dict__.get("codec_type", None) != "video": if self.__dict__.get("codec_type", None) != "video":
return False return False
if self.__dict__.get("codec_name", None) in ["png", "mjpeg", "gif", "bmp", "tiff", "webp"]: return self.__dict__.get("codec_name", None) not in ["png", "mjpeg", "gif", "bmp", "tiff", "webp"]
return False
return True
def is_subtitle(self): def is_subtitle(self):
""" """
@ -266,7 +265,7 @@ async def ffprobe(file: str) -> FFProbeResult:
""" """
try: 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) await asyncio.create_subprocess_exec("ffprobe", "-h", stdout=tempf, stderr=tempf)
except FileNotFoundError as e: except FileNotFoundError as e:
msg = "ffprobe not found." msg = "ffprobe not found."

110
app/main.py
View file

@ -13,7 +13,6 @@ from library.config import Config
from library.DownloadQueue import DownloadQueue from library.DownloadQueue import DownloadQueue
from library.Emitter import Emitter from library.Emitter import Emitter
from library.EventsSubscriber import EventsSubscriber from library.EventsSubscriber import EventsSubscriber
from library.HttpAPI import LOG as http_logger # noqa: N811
from library.HttpAPI import HttpAPI from library.HttpAPI import HttpAPI
from library.HttpSocket import HttpSocket from library.HttpSocket import HttpSocket
from library.Notifications import Notification from library.Notifications import Notification
@ -25,110 +24,97 @@ MIME = magic.Magic(mime=True)
class Main: class Main:
config: Config
app: web.Application
http: HttpAPI
socket: HttpSocket
def __init__(self): def __init__(self):
self.config = Config.get_instance() self._config = Config.get_instance()
self.rootPath = str(Path(__file__).parent.absolute()) self.rootPath = str(Path(__file__).parent.absolute())
self.app = web.Application() self._app = web.Application()
self._check_folders() self._check_folders()
try: try:
PackageInstaller(self.config).check() PackageInstaller(self._config).check()
except Exception as e: except Exception as e:
LOG.exception(e) LOG.exception(e)
LOG.error(f"Failed to check for packages. Error message '{e!s}'.") 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.row_factory = sqlite3.Row
connection.execute("PRAGMA journal_mode=wal") connection.execute("PRAGMA journal_mode=wal")
async def _close_connection(_):
LOG.debug("Closing database connection.")
connection.close()
LOG.debug("Database connection closed.")
try: try:
if "600" != oct(os.stat(self.config.db_file).st_mode)[-3:]: if "600" != oct(os.stat(self._config.db_file).st_mode)[-3:]:
os.chmod(self.config.db_file, 0o600) os.chmod(self._config.db_file, 0o600)
except Exception: except Exception:
pass pass
queue = DownloadQueue(connection=connection) self._queue = DownloadQueue(connection=connection)
self._http = HttpAPI(queue=self._queue)
self.http = HttpAPI(queue=queue) self._socket = HttpSocket(queue=self._queue)
self.socket = HttpSocket(queue=queue)
Emitter.get_instance().add_emitter([Notification().emit], local=False).add_emitter( Emitter.get_instance().add_emitter([Notification().emit], local=False).add_emitter(
[EventsSubscriber().emit], local=True [EventsSubscriber().emit], local=True
) )
self.app.on_startup.append(lambda _: queue.initialize()) self._app.on_cleanup.append(_close_connection)
def _check_folders(self): def _check_folders(self):
""" """Check if the required folders exist and create them if they do not."""
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)
"""
try: for folder in folders:
LOG.debug(f"Checking download folder at '{self.config.download_path}'.") try:
if not os.path.exists(self.config.download_path): LOG.debug(f"Checking folder at '{folder}'.")
LOG.info(f"Creating download folder at '{self.config.download_path}'.") if not os.path.exists(folder):
os.makedirs(self.config.download_path, exist_ok=True) LOG.info(f"Creating folder at '{folder}'.")
except OSError: os.makedirs(folder, exist_ok=True)
LOG.error(f"Could not create download folder at '{self.config.download_path}'.") except OSError:
raise LOG.error(f"Could not create folder at '{folder}'.")
raise
try: try:
LOG.debug(f"Checking temp folder at '{self.config.temp_path}'.") LOG.debug(f"Checking database file at '{self._config.db_file}'.")
if not os.path.exists(self.config.temp_path): if not os.path.exists(self._config.db_file):
LOG.info(f"Creating temp folder at '{self.config.temp_path}'.") LOG.info(f"Creating database file at '{self._config.db_file}'.")
os.makedirs(self.config.temp_path, exist_ok=True) with open(self._config.db_file, "w") as _:
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 _:
pass pass
except OSError: 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 raise
def start(self): def start(self):
""" """
Start the application. Start the application.
""" """
self.socket.attach(self.app) self._socket.attach(self._app)
self.http.attach(self.app) self._http.attach(self._app)
Tasks.get_instance().attach(self.app) self._queue.attach(self._app)
Tasks.get_instance().attach(self._app)
def started(_): def started(_):
LOG.info("=" * 40) 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) LOG.info("=" * 40)
if self.config.access_log: HTTP_LOGGER = None
http_logger.addFilter(lambda record: "GET /api/ping" not in record.getMessage()) 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( web.run_app(
self.app, self._app,
host=self.config.host, host=self._config.host,
port=self.config.port, port=self._config.port,
reuse_port=True, reuse_port=True,
loop=asyncio.get_event_loop(), loop=asyncio.get_event_loop(),
access_log=http_logger if self.config.access_log else None, access_log=HTTP_LOGGER,
print=started, print=started,
) )