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

Merge pull request #148 from arabcoders/dev

Browse source 
Dev
This commit is contained in:
Abdulmohsen 2024-12-20 22:16:28 +03:00 committed by GitHub
commit 7d2828734b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
27 changed files with 1304 additions and 1016 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.editorconfig
View file

@ -15,3 +15,4 @@ trim_trailing_whitespace = false
[*.py]
indent_size = 4
max_line_length = 120

1
.vscode/launch.json vendored
View file

@ -37,6 +37,7 @@
"YTP_LOG_LEVEL": "DEBUG",
"YTP_DEBUG": "false",
"YTP_YTDL_DEBUG": "false",
"YTP_MAX_WORKERS": "2",
}
},
{

54
app/library/AsyncPool.py
View file

@ -1,6 +1,6 @@
import asyncio
from datetime import datetime, timezone
import logging
from datetime import datetime, timezone
class Terminator:
@ -8,13 +8,18 @@ class Terminator:
class AsyncPool:
def __init__(self,
num_workers: int, worker_co,
name: str, logger: logging.Logger,
loop: asyncio.AbstractEventLoop = None,
load_factor: int = 1, max_task_time: int = None,
return_futures: bool = False, raise_on_join: bool = False
):
def __init__(
self,
num_workers: int,
worker_co,
name: str,
logger: logging.Logger,
loop: asyncio.AbstractEventLoop = None,
load_factor: int = 1,
max_task_time: int = None,
return_futures: bool = False,
raise_on_join: bool = False,
):
"""
This class will create `num_workers` asyncio tasks to work against a queue of
`num_workers * load_factor` items of back-pressure (IOW we will block after such
@ -62,8 +67,8 @@ class AsyncPool:
future, args, kwargs = item
self._status[worker_id] = {
'started': self._time().isoformat(),
'data': kwargs.get('entry', {'info': {}}).info.__dict__,
"started": self._time().isoformat(),
"data": kwargs.get("entry", {"info": {}}).info.__dict__,
}
# the wait_for will cancel the task (task sees CancelledError) and raises a TimeoutError from here
@ -87,7 +92,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. {str(e)}")
finally:
self._status[worker_id] = None
@ -140,7 +145,7 @@ class AsyncPool:
return future
def start(self):
""" Will start up worker pool and reset exception state """
"""Will start up worker pool and reset exception state"""
self._exceptions = False
for worker_number in range(self._num_workers):
@ -148,16 +153,16 @@ class AsyncPool:
self._createWorker(worker_id)
async def restart(self, worker_id: str, msg: str = None) -> bool:
""" Will restart the worker pool """
"""Will restart the worker pool"""
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
try:
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. {str(e)}")
if worker_id in self._status:
self._status.pop(worker_id)
@ -169,16 +174,16 @@ class AsyncPool:
return True
async def stop(self, worker_id: str, msg: str = None) -> bool:
""" Will stop the worker """
"""Will stop the worker"""
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
if self._workers[worker_id].cancel(msg):
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. {str(e)}")
if worker_id in self._status:
self._status.pop(worker_id)
@ -192,7 +197,7 @@ class AsyncPool:
if len(self._workers) < 1:
return
self._logger.info(f'Joining {self._name}')
self._logger.info(f"Joining {self._name}")
# The Terminators will kick each worker from being blocked against the _queue.get() and allow
# each one to exit.
@ -203,26 +208,25 @@ class AsyncPool:
await asyncio.gather(*list(self._workers))
self._workers = {}
except:
self._logger.exception(f'Exception joining {self._name}')
self._logger.exception(f"Exception joining {self._name}")
raise
finally:
self._logger.info(f'Completed {self._name}')
self._logger.info(f"Completed {self._name}")
if self._exceptions and self._raise_on_join:
raise Exception(f"Exception occurred in {self._name} pool")
def _createWorker(self, worker_id: str) -> asyncio.Future:
if worker_id in self._workers:
self._logger.debug(f'Worker {worker_id} already exists.')
self._logger.debug(f"Worker {worker_id} already exists.")
return self._workers[worker_id]
self._status[worker_id] = None
self._workers[worker_id] = asyncio.ensure_future(
coro_or_future=self._worker_loop(worker_id=worker_id),
loop=self._loop
coro_or_future=self._worker_loop(worker_id=worker_id), loop=self._loop
)
self._logger.debug(f'Created {worker_id}')
self._logger.debug(f"Created {worker_id}")
return self._workers[worker_id]

56
app/library/DataStore.py
View file

@ -1,9 +1,10 @@
from collections import OrderedDict
import copy
import json
from collections import OrderedDict
from datetime import datetime, timezone
from email.utils import formatdate
import json
from sqlite3 import Connection
from .config import Config
from .Download import Download
from .ItemDTO import ItemDTO
@ -13,6 +14,7 @@ class DataStore:
"""
Persistent queue.
"""
type: str = None
dict: OrderedDict[str, Download] = None
config: Config = None
@ -31,7 +33,7 @@ class DataStore:
def exists(self, key: str = None, url: str = None) -> bool:
if not key and not url:
raise KeyError('key or url must be provided.')
raise KeyError("key or url must be provided.")
if key and key in self.dict:
return True
@ -44,13 +46,13 @@ class DataStore:
def get(self, key: str, url: str = None) -> Download:
if not key and not url:
raise KeyError('key or url must be provided.')
raise KeyError("key or url must be provided.")
for i in self.dict:
if (key and self.dict[i].info._id == key) or (url and self.dict[i].info.url == url):
return self.dict[i]
raise KeyError(f'{key=} or {url=} not found.')
raise KeyError(f"{key=} or {url=} not found.")
def getById(self, id: str) -> Download | None:
return self.dict[id] if id in self.dict else None
@ -62,18 +64,17 @@ class DataStore:
items: list[tuple[str, ItemDTO]] = []
cursor = self.connection.execute(
f'SELECT "id", "data", "created_at" FROM "history" WHERE "type" = ? ORDER BY "created_at" ASC',
(self.type,)
'SELECT "id", "data", "created_at" FROM "history" WHERE "type" = ? ORDER BY "created_at" ASC', (self.type,)
)
for row in cursor:
rowDate = datetime.strptime(row['created_at'], '%Y-%m-%d %H:%M:%S')
data: dict = json.loads(row['data'])
key: str = data.pop('_id')
rowDate = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S")
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())
items.append((row['id'], item))
items.append((row["id"], item))
return items
@ -123,22 +124,37 @@ class DataStore:
stored = copy.deepcopy(item)
if hasattr(stored, 'datetime'):
if hasattr(stored, "datetime"):
try:
delattr(stored, 'datetime')
delattr(stored, "datetime")
except AttributeError:
pass
if hasattr(stored, 'live_in') and stored.status == 'finished':
if hasattr(stored, "live_in") and stored.status == "finished":
try:
delattr(stored, 'live_in')
delattr(stored, "live_in")
except AttributeError:
pass
self.connection.execute(sqlStatement.strip(), (
stored._id, type, stored.url, stored.json(), type, stored.url,
stored.json(), datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')
))
self.connection.execute(
sqlStatement.strip(),
(
stored._id,
type,
stored.url,
stored.json(),
type,
stored.url,
stored.json(),
datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"),
),
)
def _deleteStoreItem(self, key: str) -> None:
self.connection.execute('DELETE FROM "history" WHERE "type" = ? AND "id" = ?', (self.type, key,))
self.connection.execute(
'DELETE FROM "history" WHERE "type" = ? AND "id" = ?',
(
self.type,
key,
),
)

174
app/library/Download.py
View file

@ -1,25 +1,28 @@
import asyncio
import hashlib
import json
import logging
import multiprocessing
import os
import re
import shutil
from multiprocessing.managers import SyncManager
import yt_dlp
import hashlib
from .AsyncPool import Terminator
from .Utils import get_opts, jsonCookie, mergeConfig
from .ItemDTO import ItemDTO
from .config import Config
from .Emitter import Emitter
from multiprocessing.managers import SyncManager
LOG = logging.getLogger('download')
from .ItemDTO import ItemDTO
from .Utils import get_opts, jsonCookie, mergeConfig
LOG = logging.getLogger("download")
class Download:
"""
Download task.
"""
id: str = None
manager = None
download_dir: str = None
@ -47,15 +50,15 @@ class Download:
"Bad yt-dlp options which are known to cause issues with live stream and post manifestless mode."
_ytdlp_fields: tuple = (
'tmpfilename',
'filename',
'status',
'msg',
'total_bytes',
'total_bytes_estimate',
'downloaded_bytes',
'speed',
'eta',
"tmpfilename",
"filename",
"status",
"msg",
"total_bytes",
"total_bytes_estimate",
"downloaded_bytes",
"speed",
"eta",
)
"Fields to be extracted from yt-dlp progress hook."
@ -83,69 +86,62 @@ class Download:
self.max_workers = int(config.max_workers)
self.tempKeep = 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.is_manifestless = "is_manifestless" in self.info.options and self.info.options["is_manifestless"] is True
self.info_dict = info_dict
def _progress_hook(self, data: dict):
dataDict = {k: v for k, v in data.items() if k in self._ytdlp_fields}
self.status_queue.put({'id': self.id, **dataDict})
self.status_queue.put({"id": self.id, **dataDict})
def _postprocessor_hook(self, data: dict):
if data.get('postprocessor') != 'MoveFiles' or data.get('status') != 'finished':
if data.get("postprocessor") != "MoveFiles" or data.get("status") != "finished":
return
if '__finaldir' in data['info_dict']:
filename = os.path.join(data['info_dict']['__finaldir'], os.path.basename(data['info_dict']['filepath']))
if "__finaldir" in data["info_dict"]:
filename = os.path.join(data["info_dict"]["__finaldir"], os.path.basename(data["info_dict"]["filepath"]))
else:
filename = data['info_dict']['filepath']
filename = data["info_dict"]["filepath"]
self.status_queue.put({
'id': self.id,
'status': 'finished',
'filename': filename
})
self.status_queue.put({"id": self.id, "status": "finished", "filename": filename})
def _download(self):
try:
params: dict = {
'color': 'no_color',
'paths': {
'home': self.download_dir,
'temp': self.tempPath,
"color": "no_color",
"paths": {
"home": self.download_dir,
"temp": self.tempPath,
},
'outtmpl': {
'default': self.output_template,
'chapter': self.output_template_chapter
},
'noprogress': True,
'break_on_existing': True,
'progress_hooks': [self._progress_hook],
'postprocessor_hooks': [self._postprocessor_hook],
"outtmpl": {"default": self.output_template, "chapter": self.output_template_chapter},
"noprogress": True,
"break_on_existing": True,
"progress_hooks": [self._progress_hook],
"postprocessor_hooks": [self._postprocessor_hook],
**get_opts(self.preset, mergeConfig(self.default_ytdl_opts, self.ytdl_opts)),
}
if 'format' not in params and self.default_ytdl_opts.get('format', None):
params['format'] = self.default_ytdl_opts['format']
if "format" not in params and self.default_ytdl_opts.get("format", None):
params["format"] = self.default_ytdl_opts["format"]
params['ignoreerrors'] = False
params["ignoreerrors"] = False
if self.debug:
params['verbose'] = True
params['noprogress'] = False
params["verbose"] = True
params["noprogress"] = False
if self.info.ytdlp_cookies:
try:
data = jsonCookie(json.loads(self.info.ytdlp_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:
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:
f.write(data)
params['cookiefile'] = f.name
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} - {str(e)}")
if self.is_live or self.is_manifestless:
hasDeletedOptions = False
@ -158,29 +154,26 @@ class Download:
if hasDeletedOptions:
LOG.warning(
f"Live stream detected for '{self.info.title}', The following opts '{deletedOpts=}' have been deleted which are known to cause issues with live stream and post stream manifestless mode.")
f"Live stream detected for '{self.info.title}', The following opts '{deletedOpts=}' have been deleted which are known to cause issues with live stream and post stream manifestless mode."
)
LOG.info(
f'Downloading pid: {os.getpid()} id="{self.info.id}" title="{self.info.title}" preset="{self.preset}".')
f'Downloading pid: {os.getpid()} id="{self.info.id}" title="{self.info.title}" preset="{self.preset}".'
)
cls = yt_dlp.YoutubeDL(params=params)
if isinstance(self.info_dict, dict) and len(self.info_dict) > 1:
LOG.debug(f'Downloading using pre-info.')
LOG.debug("Downloading using pre-info.")
cls.process_ie_result(self.info_dict, download=True)
ret = cls._download_retcode
else:
LOG.debug(f'Downloading using url: {self.info.url}')
LOG.debug(f"Downloading using url: {self.info.url}")
ret = cls.download([self.info.url])
self.status_queue.put({'id': self.id, 'status': 'finished' if ret == 0 else 'error'})
self.status_queue.put({"id": self.id, "status": "finished" if ret == 0 else "error"})
except Exception as exc:
self.status_queue.put({
'id': self.id,
'status': 'error',
'msg': str(exc),
'error': str(exc)
})
self.status_queue.put({"id": self.id, "status": "error", "msg": str(exc), "error": str(exc)})
LOG.info(f'Finished {os.getpid()=} id="{self.info.id}" title="{self.info.title}".')
@ -190,17 +183,14 @@ class Download:
self.status_queue = self.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.tempPath = os.path.join(self.temp_dir, hashlib.shake_256(f"D-{self.info.id}".encode("utf-8")).hexdigest(5))
if not os.path.exists(self.tempPath):
os.makedirs(self.tempPath, exist_ok=True)
self.proc = multiprocessing.Process(name=f"download-{self.id}", target=self._download)
self.proc.start()
self.info.status = 'preparing'
self.info.status = "preparing"
asyncio.create_task(self.emitter.updated(dl=self.info), name=f"emitter-{self.id}")
asyncio.create_task(self.progress_update(), name=f"update-{self.id}")
@ -230,7 +220,7 @@ class Download:
try:
LOG.info(f"Closing download process: '{procId}'.")
try:
if 'update_task' in self.__dict__ and self.update_task.done() is False:
if "update_task" in self.__dict__ and self.update_task.done() is False:
self.update_task.cancel()
except Exception as e:
LOG.error(f"Failed to close status queue: '{procId}'. {e}")
@ -290,9 +280,10 @@ class Download:
if self.tempKeep is True or not self.tempPath:
return
if self.info.status != 'finished' and self.is_live:
if self.info.status != "finished" 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.tempPath}], as the reported status is not finished [{self.info.status}]."
)
return
if not os.path.exists(self.tempPath):
@ -300,10 +291,11 @@ class Download:
if self.tempPath == 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.tempPath}, but it is the same as main temp folder."
)
return
LOG.info(f'Deleting Temp folder: {self.tempPath}')
LOG.info(f"Deleting Temp folder: {self.tempPath}")
shutil.rmtree(self.tempPath, ignore_errors=True)
async def progress_update(self):
@ -314,54 +306,54 @@ class Download:
try:
self.update_task = asyncio.get_running_loop().run_in_executor(None, self.status_queue.get)
except asyncio.CancelledError:
LOG.debug(f'Closing progress update for: {self.info._id=}.')
LOG.debug(f"Closing progress update for: {self.info._id=}.")
return
status = await self.update_task
if status is None or status.__class__ is Terminator:
LOG.debug(f'Closing progress update for: {self.info._id=}.')
LOG.debug(f"Closing progress update for: {self.info._id=}.")
return
if status.get('id') != self.id or len(status) < 2:
if status.get("id") != self.id or len(status) < 2:
continue
if self.debug:
LOG.debug(f'Status Update: {self.info._id=} {status=}')
LOG.debug(f"Status Update: {self.info._id=} {status=}")
if isinstance(status, str):
asyncio.create_task(self.emitter.updated(dl=self.info), name=f"emitter-u-{self.id}")
continue
self.tmpfilename = status.get('tmpfilename')
self.tmpfilename = status.get("tmpfilename")
if 'filename' in status:
self.info.filename = os.path.relpath(status.get('filename'), self.download_dir)
if "filename" in status:
self.info.filename = os.path.relpath(status.get("filename"), self.download_dir)
if os.path.exists(status.get('filename')):
self.info.file_size = os.path.getsize(status.get('filename'))
if os.path.exists(status.get("filename")):
self.info.file_size = os.path.getsize(status.get("filename"))
self.info.status = status.get('status', self.info.status)
self.info.msg = status.get('msg')
self.info.status = status.get("status", self.info.status)
self.info.msg = status.get("msg")
if self.info.status == 'error' and 'error' in status:
self.info.error = status.get('error')
if self.info.status == "error" and "error" in status:
self.info.error = status.get("error")
asyncio.create_task(
self.emitter.error(message=self.info.error, data=self.info),
name=f"emitter-e-{self.id}")
self.emitter.error(message=self.info.error, data=self.info), name=f"emitter-e-{self.id}"
)
if 'downloaded_bytes' in status:
total = status.get('total_bytes') or status.get('total_bytes_estimate')
if "downloaded_bytes" in status:
total = status.get("total_bytes") or status.get("total_bytes_estimate")
if total:
self.info.percent = status['downloaded_bytes'] / total * 100
self.info.percent = status["downloaded_bytes"] / total * 100
self.info.total_bytes = total
self.info.speed = status.get('speed')
self.info.eta = status.get('eta')
self.info.speed = status.get("speed")
self.info.eta = status.get("eta")
if self.info.status == 'finished' and 'filename' in status:
if self.info.status == "finished" and "filename" in status:
try:
self.info.file_size = os.path.getsize(status.get('filename'))
self.info.file_size = os.path.getsize(status.get("filename"))
except FileNotFoundError:
LOG.warning(f'File not found: {status.get("filename")}')
self.info.file_size = None

259
app/library/DownloadQueue.py
View file

@ -1,31 +1,29 @@
import asyncio
from email.utils import formatdate
import json
import logging
import os
import time
import yt_dlp
from email.utils import formatdate
from sqlite3 import Connection
from .config import Config
from .Download import Download
from .ItemDTO import ItemDTO
from .DataStore import DataStore
from .Utils import ag, calcDownloadPath, ExtractInfo, isDownloaded, mergeConfig
from .AsyncPool import AsyncPool
from .Emitter import Emitter
LOG = logging.getLogger('DownloadQueue')
TYPE_DONE: str = 'done'
TYPE_QUEUE: str = 'queue'
import yt_dlp
from .AsyncPool import AsyncPool
from .config import Config
from .DataStore import DataStore
from .Download import Download
from .Emitter import Emitter
from .ItemDTO import ItemDTO
from .Utils import ExtractInfo, calcDownloadPath, isDownloaded, mergeConfig
LOG = logging.getLogger("DownloadQueue")
TYPE_DONE: str = "done"
TYPE_QUEUE: str = "queue"
class DownloadQueue:
config: Config = None
emitter: Emitter = None
queue: DataStore = None
done: DataStore = None
event: asyncio.Event = None
pool: AsyncPool = None
event: asyncio.Event | None = None
pool: AsyncPool | None = None
def __init__(self, emitter: Emitter, connection: Connection):
self.config = Config.get_instance()
@ -41,11 +39,11 @@ class DownloadQueue:
async def initialize(self):
self.event = asyncio.Event()
LOG.info(f'Using {self.config.max_workers} workers for downloading.')
LOG.info(f"Using {self.config.max_workers} workers for downloading.")
asyncio.create_task(
self.__download_pool() if self.config.max_workers > 1 else self.__download(),
name='download_pool' if self.config.max_workers > 1 else 'download_worker'
name="download_pool" if self.config.max_workers > 1 else "download_worker",
)
async def __add_entry(
@ -54,29 +52,29 @@ class DownloadQueue:
preset: str,
folder: str,
ytdlp_config: dict = {},
ytdlp_cookies: str = '',
output_template: str = '',
already=None
ytdlp_cookies: str = "",
output_template: str = "",
already=None,
):
if not entry:
return {'status': 'error', 'msg': 'Invalid/empty data was given.'}
return {"status": "error", "msg": "Invalid/empty data was given."}
options: dict = {}
error: str = None
live_in: str = None
eventType = entry.get('_type') or 'video'
if eventType == 'playlist':
entries = entry['entries']
eventType = entry.get("_type") or "video"
if eventType == "playlist":
entries = entry["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)
for property in ('id', 'title', 'uploader', 'uploader_id'):
etr["playlist"] = entry.get("id")
etr["playlist_index"] = "{{0:0{0:d}d}}".format(playlist_index_digits).format(i)
for property in ("id", "title", "uploader", "uploader_id"):
if property in entry:
etr[f'playlist_{property}'] = entry.get(property)
etr[f"playlist_{property}"] = entry.get(property)
results.append(
await self.__add_entry(
@ -86,60 +84,63 @@ class DownloadQueue:
ytdlp_config=ytdlp_config,
ytdlp_cookies=ytdlp_cookies,
output_template=output_template,
already=already
already=already,
)
)
if any(res['status'] == 'error' for res in results):
return {'status': 'error', 'msg': ', '.join(res['msg'] for res in results if res['status'] == 'error' and 'msg' in res)}
if any(res["status"] == "error" for res in results):
return {
"status": "error",
"msg": ", ".join(res["msg"] for res in results if res["status"] == "error" and "msg" in res),
}
return {'status': 'ok'}
elif (eventType == 'video' or eventType.startswith('url')) and 'id' in entry and 'title' in entry:
return {"status": "ok"}
elif (eventType == "video" 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 entry.get('live_status') == 'is_upcoming':
if 'release_timestamp' in entry and entry.get('release_timestamp'):
live_in = formatdate(entry.get('release_timestamp'))
if "live_status" in entry and entry.get("live_status") == "is_upcoming":
if "release_timestamp" in entry and entry.get("release_timestamp"):
live_in = formatdate(entry.get("release_timestamp"))
else:
error = 'Live stream not yet started. And no date is set.'
error = "Live stream not yet started. And no date is set."
else:
error = entry.get('msg', None)
error = entry.get("msg", None)
LOG.debug(
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=entry.get('webpage_url') or entry.get('url')):
item = self.done.get(key=entry['id'], url=entry.get('webpage_url') or entry['url'])
if self.done.exists(key=entry["id"], url=entry.get("webpage_url") or entry.get("url")):
item = self.done.get(key=entry["id"], url=entry.get("webpage_url") or entry["url"])
LOG.warning(f"Item '{item.info.id}' - '{item.info.title}' already downloaded. Removing from history.")
await self.clear([item.info._id])
try:
item = self.queue.get(key=entry.get('id'), url=entry.get('webpage_url') or entry.get('url'))
item = self.queue.get(key=entry.get("id"), url=entry.get("webpage_url") or entry.get("url"))
if item is not None:
err_message = f"Item ID '{item.info.id}' - '{item.info.title}' already in download queue."
LOG.info(err_message)
return {'status': 'error', 'msg': err_message}
return {"status": "error", "msg": err_message}
except KeyError:
pass
is_manifestless = entry.get('is_manifestless', False)
options.update({'is_manifestless': is_manifestless})
is_manifestless = entry.get("is_manifestless", False)
options.update({"is_manifestless": is_manifestless})
live_status: list = ['is_live', 'is_upcoming']
is_live = entry.get('is_live', None) or live_in or entry.get('live_status', None) in live_status
live_status: list = ["is_live", "is_upcoming"]
is_live = entry.get("is_live", None) or live_in or entry.get("live_status", None) in live_status
try:
download_dir = calcDownloadPath(basePath=self.config.download_path, folder=folder)
except Exception as e:
LOG.exception(e)
return {'status': 'error', 'msg': str(e)}
return {"status": "error", "msg": str(e)}
dl = ItemDTO(
id=entry.get('id'),
title=entry.get('title'),
url=entry.get('webpage_url') or entry.get('url'),
id=entry.get("id"),
title=entry.get("title"),
url=entry.get("webpage_url") or entry.get("url"),
preset=preset,
thumbnail=entry.get('thumbnail', None),
thumbnail=entry.get("thumbnail", None),
folder=folder,
download_dir=download_dir,
temp_dir=self.config.temp_path,
@ -160,42 +161,37 @@ class DownloadQueue:
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):
dlInfo.info.status = 'not_live'
if dlInfo.info.live_in or "is_upcoming" == entry.get("live_status", None):
dlInfo.info.status = "not_live"
itemDownload = self.done.put(dlInfo)
NotifyEvent = 'completed'
NotifyEvent = "completed"
elif self.config.allow_manifestless is False and is_manifestless is True:
dlInfo.info.status = 'error'
dlInfo.info.error = 'Video is in post-live manifestless mode.'
dlInfo.info.status = "error"
dlInfo.info.error = "Video is in post-live manifestless mode."
itemDownload = self.done.put(dlInfo)
NotifyEvent = 'completed'
NotifyEvent = "completed"
else:
NotifyEvent = 'added'
NotifyEvent = "added"
itemDownload = self.queue.put(dlInfo)
self.event.set()
asyncio.create_task(
self.emitter.emit(NotifyEvent, itemDownload.info),
name=f'notifier-{NotifyEvent}-{itemDownload.info.id}')
self.emitter.emit(NotifyEvent, itemDownload.info), name=f"notifier-{NotifyEvent}-{itemDownload.info.id}"
)
return {
'status': 'ok'
}
elif eventType.startswith('url'):
return {"status": "ok"}
elif eventType.startswith("url"):
return await self.add(
url=entry.get('url'),
url=entry.get("url"),
preset=preset,
folder=folder,
ytdlp_config=ytdlp_config,
ytdlp_cookies=ytdlp_cookies,
output_template=output_template,
already=already
already=already,
)
return {
'status': 'error',
'msg': f'Unsupported resource "{eventType}"'
}
return {"status": "error", "msg": f'Unsupported resource "{eventType}"'}
async def add(
self,
@ -203,26 +199,28 @@ class DownloadQueue:
preset: str,
folder: str,
ytdlp_config: dict = {},
ytdlp_cookies: str = '',
output_template: str = '',
already=None
ytdlp_cookies: str = "",
output_template: str = "",
already=None,
):
ytdlp_config = ytdlp_config if ytdlp_config else {}
LOG.info(f"Adding url '{url}' to folder '{folder}' with the following options 'Preset: {preset}' 'Naming: {output_template}', 'Cookies: {ytdlp_cookies}' 'YTConfig: {ytdlp_config}'.")
LOG.info(
f"Adding url '{url}' to folder '{folder}' with the following options 'Preset: {preset}' 'Naming: {output_template}', 'Cookies: {ytdlp_cookies}' 'YTConfig: {ytdlp_config}'."
)
if isinstance(ytdlp_config, str):
try:
ytdlp_config = json.loads(ytdlp_config)
except Exception as e:
LOG.error(f"Unable to load '{ytdlp_config=}'. {str(e)}")
return {'status': 'error', 'msg': f"Failed to parse json yt-dlp config. {str(e)}"}
return {"status": "error", "msg": f"Failed to parse json yt-dlp config. {str(e)}"}
already = set() if already is None else already
if url in already:
LOG.warning(f"Recursion detected with url '{url}' skipping.")
return {'status': 'ok'}
return {"status": "ok"}
already.add(url)
@ -231,10 +229,10 @@ class DownloadQueue:
if downloaded is True:
message = f"This url with ID '{id_dict.get('id')}' has been downloaded already and recorded in archive."
LOG.info(message)
return {'status': 'error', 'msg': message}
return {"status": "error", "msg": message}
started = time.perf_counter()
LOG.debug(f'extract_info: checking {url=}')
LOG.debug(f"extract_info: checking {url=}")
entry = await asyncio.wait_for(
fut=asyncio.get_running_loop().run_in_executor(
@ -242,24 +240,29 @@ class DownloadQueue:
ExtractInfo,
mergeConfig(self.config.ytdl_options, ytdlp_config),
url,
bool(self.config.ytdl_debug)
bool(self.config.ytdl_debug),
),
timeout=self.config.extract_info_timeout)
timeout=self.config.extract_info_timeout,
)
if not entry:
return {'status': 'error', 'msg': 'Unable to extract info check logs.'}
return {"status": "error", "msg": "Unable to extract info check logs."}
LOG.debug(
f"extract_info: for 'URL: {url}' is done in '{time.perf_counter() - started}'. Length: '{len(entry)}'.")
f"extract_info: for 'URL: {url}' is done in '{time.perf_counter() - started}'. Length: '{len(entry)}'."
)
except yt_dlp.utils.ExistingVideoReached as exc:
LOG.error(f'Video has been downloaded already and recorded in archive.log file. {str(exc)}')
return {'status': 'error', 'msg': 'Video has been downloaded already and recorded in archive.log file.'}
LOG.error(f"Video has been downloaded already and recorded in archive.log file. {str(exc)}")
return {"status": "error", "msg": "Video has been downloaded already and recorded in archive.log file."}
except yt_dlp.utils.YoutubeDLError as exc:
LOG.error(f'YoutubeDLError: Unable to extract info. {str(exc)}')
return {'status': 'error', 'msg': str(exc)}
LOG.error(f"YoutubeDLError: Unable to extract info. {str(exc)}")
return {"status": "error", "msg": str(exc)}
except asyncio.exceptions.TimeoutError as exc:
LOG.error(f'TimeoutError: Unable to extract info. {str(exc)}')
return {'status': 'error', 'msg': f'TimeoutError: {self.config.extract_info_timeout}s reached Unable to extract info.'}
LOG.error(f"TimeoutError: Unable to extract info. {str(exc)}")
return {
"status": "error",
"msg": f"TimeoutError: {self.config.extract_info_timeout}s reached Unable to extract info.",
}
return await self.__add_entry(
entry=entry,
@ -279,27 +282,27 @@ class DownloadQueue:
item = self.queue.get(key=id)
except KeyError as e:
status[id] = str(e)
status['status'] = 'error'
LOG.warning(f'Requested cancel for non-existent download {id=}. {str(e)}')
status["status"] = "error"
LOG.warning(f"Requested cancel for non-existent download {id=}. {str(e)}")
continue
itemMessage = f"{id=} {item.info.id=} {item.info.title=}"
if item.running():
LOG.debug(f'Canceling {itemMessage}')
LOG.debug(f"Canceling {itemMessage}")
item.cancel()
LOG.info(f'Cancelled {itemMessage}')
LOG.info(f"Cancelled {itemMessage}")
await item.close()
else:
await item.close()
LOG.debug(f'Deleting from queue {itemMessage}')
LOG.debug(f"Deleting from queue {itemMessage}")
self.queue.delete(id)
asyncio.create_task(self.emitter.canceled(id=id, dl=item), name=f'notifier-c-{id}')
asyncio.create_task(self.emitter.canceled(id=id, dl=item), name=f"notifier-c-{id}")
self.done.put(item)
asyncio.create_task(self.emitter.completed(dl=item), name=f'notifier-d-{id}')
LOG.info(f'Deleted from queue {itemMessage}')
asyncio.create_task(self.emitter.completed(dl=item), name=f"notifier-d-{id}")
LOG.info(f"Deleted from queue {itemMessage}")
status[id] = 'ok'
status[id] = "ok"
return status
@ -311,35 +314,35 @@ class DownloadQueue:
item = self.done.get(key=id)
except KeyError as e:
status[id] = str(e)
status['status'] = 'error'
LOG.warning(f'Requested delete for non-existent download {id=}. {str(e)}')
status["status"] = "error"
LOG.warning(f"Requested delete for non-existent download {id=}. {str(e)}")
continue
itemMessage = f"{id=} {item.info.id=} {item.info.title=}"
LOG.debug(f'Deleting completed download {itemMessage}')
LOG.debug(f"Deleting completed download {itemMessage}")
self.done.delete(id)
asyncio.create_task(self.emitter.cleared(id, dl=item), name=f'notifier-c-{id}')
LOG.info(f'Deleted completed download {itemMessage}')
status[id] = 'ok'
asyncio.create_task(self.emitter.cleared(id, dl=item), name=f"notifier-c-{id}")
LOG.info(f"Deleted completed download {itemMessage}")
status[id] = "ok"
return status
def get(self) -> dict[str, list[dict[str, ItemDTO]]]:
items = {'queue': {}, 'done': {}}
items = {"queue": {}, "done": {}}
for k, v in self.queue.saved_items():
items['queue'][k] = v
items["queue"][k] = v
for k, v in self.done.saved_items():
items['done'][k] = v
items["done"][k] = v
for k, v in self.queue.items():
if not k in items['queue']:
items['queue'][k] = v.info
if k not in items["queue"]:
items["queue"][k] = v.info
for k, v in self.done.items():
if not k in items['done']:
items['done'][k] = v.info
if k not in items["done"]:
items["done"][k] = v.info
return items
@ -348,8 +351,8 @@ class DownloadQueue:
loop=asyncio.get_running_loop(),
num_workers=self.config.max_workers,
worker_co=self.__downloadFile,
name='download_pool',
logger=logging.getLogger('WorkerPool'),
name="download_pool",
logger=logging.getLogger("WorkerPool"),
)
self.pool.start()
@ -362,14 +365,14 @@ class DownloadQueue:
break
if time.time() - lastLog > 120:
lastLog = time.time()
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)
while not self.queue.hasDownloads():
LOG.info(f"Waiting for item to download. '{self.pool.get_available_workers()}' free workers.")
await self.event.wait()
self.event.clear()
LOG.debug(f"Cleared wait event.")
LOG.debug("Cleared wait event.")
entry = self.queue.getNextDownload()
await asyncio.sleep(0.2)
@ -387,7 +390,7 @@ class DownloadQueue:
async def __download(self):
while True:
while self.queue.empty():
LOG.info('Waiting for item to download.')
LOG.info("Waiting for item to download.")
await self.event.wait()
self.event.clear()
@ -396,34 +399,34 @@ class DownloadQueue:
async def __downloadFile(self, id: str, entry: Download):
LOG.info(
f"Downloading 'id: {id}', 'Title: {entry.info.title}', 'URL: {entry.info.url}' to 'folder: {entry.info.folder}'.")
f"Downloading 'id: {id}', 'Title: {entry.info.title}', 'URL: {entry.info.url}' to 'folder: {entry.info.folder}'."
)
try:
await entry.start(self.emitter)
if entry.info.status != 'finished':
if entry.info.status != "finished":
if entry.tmpfilename and os.path.isfile(entry.tmpfilename):
try:
os.remove(entry.tmpfilename)
entry.tmpfilename = None
except:
except Exception:
pass
entry.info.status = 'error'
entry.info.status = "error"
finally:
await entry.close()
if self.queue.exists(key=id):
self.queue.delete(key=id)
if entry.is_canceled() is True:
asyncio.create_task(self.emitter.canceled(id, dl=entry.info), name=f'notifier-c-{id}')
entry.info.status = 'canceled'
entry.info.error = 'Canceled by user.'
asyncio.create_task(self.emitter.canceled(id, dl=entry.info), name=f"notifier-c-{id}")
entry.info.status = "canceled"
entry.info.error = "Canceled by user."
self.done.put(value=entry)
asyncio.create_task(self.emitter.completed(dl=entry.info), name=f'notifier-d-{id}')
asyncio.create_task(self.emitter.completed(dl=entry.info), name=f"notifier-d-{id}")
self.event.set()
@ -431,4 +434,4 @@ class DownloadQueue:
if not url or not self.config.keep_archive:
return False, None
return isDownloaded(self.config.ytdl_options.get('download_archive', None), url)
return isDownloaded(self.config.ytdl_options.get("download_archive", None), url)

28
app/library/Emitter.py
View file

@ -1,48 +1,48 @@
import asyncio
import logging
LOG = logging.getLogger('Emitter')
LOG = logging.getLogger("Emitter")
class Emitter:
"""
This class is used to emit events to the registered emitters.
"""
emitters: list[callable] = []
def add_emitter(self, emitter: callable):
"""
Add an emitter to the list of emitters.
Add an emitter to the list of emitters.
Args:
emitter (function): The emitter function. The function must return a coroutine or None.
Args:
emitter (function): The emitter function. The function must return a coroutine or None.
"""
self.emitters.append(emitter)
async def added(self, dl: dict, **kwargs):
await self.emit('added', dl, **kwargs)
await self.emit("added", dl, **kwargs)
async def updated(self, dl: dict, **kwargs):
await self.emit('updated', dl, **kwargs)
await self.emit("updated", dl, **kwargs)
async def completed(self, dl: dict, **kwargs):
await self.emit('completed', dl, **kwargs)
await self.emit("completed", dl, **kwargs)
async def canceled(self, id: str, dl: dict = None, **kwargs):
await self.emit('canceled', id, **kwargs)
await self.emit("canceled", id, **kwargs)
async def cleared(self, id: str, dl: dict = None, **kwargs):
await self.emit('cleared', id, **kwargs)
await self.emit("cleared", id, **kwargs)
async def error(self, message: str, data: dict = {}, **kwargs):
msg = {'status': 'error', 'message': message, 'data': {}}
msg = {"status": "error", "message": message, "data": {}}
if data:
msg.update({'data': data})
await self.emit('error', msg, **kwargs)
msg.update({"data": data})
await self.emit("error", msg, **kwargs)
async def warning(self, message: str, **kwargs):
await self.emit('error', message, **kwargs)
await self.emit("error", message, **kwargs)
async def emit(self, event: str, data, **kwargs):
tasks = []

596
app/library/HttpAPI.py
View file

@ -1,44 +1,41 @@
import base64
from datetime import datetime
import functools
import hmac
import json
import logging
import os
import time
from datetime import datetime
from pathlib import Path
import httpx
import magic
from aiohttp import web
from aiohttp.web import Request, RequestHandler, Response
from .common import common
from .config import Config
from .DownloadQueue import DownloadQueue
from aiohttp import web
from aiohttp.web import Request, Response, RequestHandler
from .Playlist import Playlist
from .Emitter import Emitter
from .encoder import Encoder
from .M3u8 import M3u8
from .Playlist import Playlist
from .Segments import Segments
from .Subtitle import Subtitle
import logging
import magic
from .common import common
from pathlib import Path
from .encoder import Encoder
from .Emitter import Emitter
from .Utils import validate_url
from .Utils import validate_url, StreamingError
LOG = logging.getLogger('app')
LOG = logging.getLogger("http_api")
MIME = magic.Magic(mime=True)
class HttpAPI(common):
config: Config = None
encoder: Encoder = None
routes: web.RouteTableDef = None
queue: DownloadQueue = None
rootPath: str = None
staticHolder: dict = {}
extToMime: dict = {
'.html': 'text/html',
'.css': 'text/css',
'.js': 'application/javascript',
'.json': 'application/json',
'.ico': 'image/x-icon',
".html": "text/html",
".css": "text/css",
".js": "application/javascript",
".json": "application/json",
".ico": "image/x-icon",
}
def __init__(self, queue: DownloadQueue, emitter: Emitter, encoder: Encoder):
@ -57,13 +54,16 @@ class HttpAPI(common):
"""
Decorator to mark a method as an HTTP route handler.
"""
def decorator(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
wrapper._http_method = method.upper()
wrapper._http_path = path
return wrapper
return decorator
async def staticFile(self, req: Request) -> Response:
@ -72,22 +72,26 @@ class HttpAPI(common):
"""
path = req.path
if req.path not in self.staticHolder:
return web.json_response({"error": "File not found.", "file": path}, status=404)
return web.json_response({"error": "File not found.", "file": path}, status=web.HTTPNotFound.status_code)
item: dict = self.staticHolder[req.path]
return web.Response(body=item.get('content'), headers={
'Pragma': 'public',
'Cache-Control': 'public, max-age=31536000',
'Content-Type': item.get('content_type'),
'X-Via': 'memory' if not item.get('file', None) else 'disk',
})
return web.Response(
body=item.get("content"),
headers={
"Pragma": "public",
"Cache-Control": "public, max-age=31536000",
"Content-Type": item.get("content_type"),
"X-Via": "memory" if not item.get("file", None) else "disk",
},
status=web.HTTPOk.status_code,
)
def preloadStatic(self, app: web.Application):
"""
Preload static files from the ui/exported folder.
"""
staticDir = os.path.join(self.rootPath, 'ui', 'exported')
staticDir = os.path.join(self.rootPath, "ui", "exported")
if not os.path.exists(staticDir):
raise ValueError(f"Could not find the frontend UI static assets. '{staticDir}'.")
@ -95,25 +99,25 @@ class HttpAPI(common):
for root, _, files in os.walk(staticDir):
for file in files:
if file.endswith('.map'):
if file.endswith(".map"):
continue
file = os.path.join(root, file)
urlPath = f"{self.config.url_prefix}{file.replace(f'{staticDir}/', '')}"
content = open(file, 'rb').read()
content = open(file, "rb").read()
contentType = self.extToMime.get(os.path.splitext(file)[1], MIME.from_file(file))
self.staticHolder[urlPath] = {'content': content, 'content_type': contentType}
self.staticHolder[urlPath] = {"content": content, "content_type": contentType}
LOG.debug(f"Preloading '{urlPath}'.")
app.router.add_get(urlPath, self.staticFile)
preloaded += 1
if urlPath.endswith('/index.html') and urlPath != '/index.html':
parentSlash = urlPath.replace('/index.html', '/')
parentNoSlash = urlPath.replace('/index.html', '')
self.staticHolder[parentSlash] = {'content': content, 'content_type': contentType}
self.staticHolder[parentNoSlash] = {'content': content, 'content_type': contentType}
if urlPath.endswith("/index.html") and urlPath != "/index.html":
parentSlash = urlPath.replace("/index.html", "/")
parentNoSlash = urlPath.replace("/index.html", "")
self.staticHolder[parentSlash] = {"content": content, "content_type": contentType}
self.staticHolder[parentNoSlash] = {"content": content, "content_type": contentType}
app.router.add_get(parentSlash, self.staticFile)
app.router.add_get(parentNoSlash, self.staticFile)
preloaded += 2
@ -121,39 +125,37 @@ class HttpAPI(common):
if preloaded < 1:
raise ValueError(f"Could not find the frontend UI static assets. '{staticDir}'.")
LOG.info(f"Preloaded {preloaded} static files.")
LOG.info(f"Preloaded '{preloaded}' static files.")
def attach(self, app: web.Application):
if self.config.auth_username and self.config.auth_password:
app.middlewares.append(HttpAPI.basic_auth(self.config.auth_username, self.config.auth_password))
self.add_routes(app)
pass
def add_routes(self, app: web.Application):
for attr_name in dir(self):
method = getattr(self, attr_name)
if hasattr(method, '_http_method') and hasattr(method, '_http_path'):
if hasattr(method, "_http_method") and hasattr(method, "_http_path"):
http_path = method._http_path
if http_path.startswith('/') and self.config.url_prefix.endswith('/'):
if http_path.startswith("/") and self.config.url_prefix.endswith("/"):
http_path = method._http_path[1:]
self.routes.route(method._http_method, self.config.url_prefix + http_path)(method)
async def on_prepare(request: Request, response: Response):
if 'Server' in response.headers:
del response.headers['Server']
if "Server" in response.headers:
del response.headers["Server"]
if 'Origin' in request.headers:
response.headers['Access-Control-Allow-Origin'] = request.headers['Origin']
response.headers['Access-Control-Allow-Headers'] = 'Content-Type'
response.headers['Access-Control-Allow-Methods'] = 'PATCH, PUT, POST, DELETE'
if "Origin" in request.headers:
response.headers["Access-Control-Allow-Origin"] = request.headers["Origin"]
response.headers["Access-Control-Allow-Headers"] = "Content-Type"
response.headers["Access-Control-Allow-Methods"] = "PATCH, PUT, POST, DELETE"
if self.config.url_prefix != '/':
self.routes.route('GET', '/')(lambda _: web.HTTPFound(self.config.url_prefix))
if self.config.url_prefix != "/":
self.routes.route("GET", "/")(lambda _: web.HTTPFound(self.config.url_prefix))
self.routes.get(self.config.url_prefix[:-1])(lambda _: web.HTTPFound(self.config.url_prefix))
# add static files.
self.routes.static(f"{self.config.url_prefix}download/", self.config.download_path)
self.preloadStatic(app)
@ -161,54 +163,66 @@ class HttpAPI(common):
app.add_routes(self.routes)
app.on_response_prepare.append(on_prepare)
except ValueError as e:
if 'ui/exported' in str(e):
if "ui/exported" in str(e):
raise RuntimeError(f"Could not find the frontend UI static assets. '{e}'.") from e
raise e
def basic_auth(username: str, password: str):
@web.middleware
async def middleware_handler(request: Request, handler: RequestHandler) -> Response:
auth_header = request.headers.get('Authorization')
auth_header = request.headers.get("Authorization")
if auth_header is None:
return web.Response(status=401, headers={
'WWW-Authenticate': 'Basic realm="Authorization Required."',
}, text='Unauthorized.')
return web.json_response(
status=web.HTTPUnauthorized.status_code,
headers={
"WWW-Authenticate": 'Basic realm="Authorization Required."',
},
data={"error": "Authorization Required."},
)
auth_type, encoded_credentials = auth_header.split(' ', 1)
auth_type, encoded_credentials = auth_header.split(" ", 1)
if 'basic' != auth_type.lower():
return web.Response(status=401, text="Unsupported authentication method.")
if "basic" != auth_type.lower():
return web.json_response(
data={"error": "Unsupported authentication method.", "method": auth_type},
status=web.HTTPUnauthorized.status_code,
)
decoded_credentials = base64.b64decode(encoded_credentials).decode('utf-8')
user_name, _, user_password = decoded_credentials.partition(':')
decoded_credentials = base64.b64decode(encoded_credentials).decode("utf-8")
user_name, _, user_password = decoded_credentials.partition(":")
if user_name != username or user_password != password:
return web.Response(status=401, text='Unauthorized (Invalid credentials).')
user_match = hmac.compare_digest(user_name, username)
pass_match = hmac.compare_digest(user_password, password)
if not (user_match and pass_match):
return web.json_response(
data={"error": "Unauthorized (Invalid credentials)."}, status=web.HTTPUnauthorized.status_code
)
return await handler(request)
return middleware_handler
@route('GET', 'ping')
@route("GET", "ping")
async def ping(self, _) -> Response:
await self.queue.test()
return web.Response(text='pong')
return web.json_response(data={"status": "pong"}, status=web.HTTPOk.status_code)
@route('POST', 'add')
@route("POST", "add")
async def add_url(self, request: Request) -> Response:
post = await request.json()
url: str = post.get('url')
preset: str = post.get('preset', 'default')
url: str = post.get("url")
preset: str = post.get("preset", "default")
if not url:
raise web.HTTPBadRequest()
return web.json_response(data={"error": "url is required."}, status=web.HTTPBadRequest.status_code)
folder: str = post.get('folder')
ytdlp_cookies: str = post.get('ytdlp_cookies')
ytdlp_config: dict = post.get('ytdlp_config')
output_template: str = post.get('output_template')
folder: str = post.get("folder")
ytdlp_cookies: str = post.get("ytdlp_cookies")
ytdlp_config: dict | None = post.get("ytdlp_config")
output_template: str = post.get("output_template")
if ytdlp_config is None:
ytdlp_config = {}
@ -218,83 +232,92 @@ class HttpAPI(common):
folder=folder,
ytdlp_cookies=ytdlp_cookies,
ytdlp_config=ytdlp_config,
output_template=output_template
output_template=output_template,
)
return web.Response(text=self.encoder.encode(status))
return web.json_response(data=status, status=web.HTTPOk.status_code, dumps=self.encoder.encode)
@route('GET', 'tasks')
@route("GET", "tasks")
async def tasks(self, _: Request) -> Response:
tasks_file: str = os.path.join(self.config.config_path, 'tasks.json')
tasks_file: str = os.path.join(self.config.config_path, "tasks.json")
if not os.path.exists(tasks_file):
return web.json_response({"error": "No tasks defined."}, status=404)
return web.json_response({"error": "No tasks defined."}, status=web.HTTPNotFound.status_code)
try:
with open(tasks_file, 'r') as f:
with open(tasks_file, "r") as f:
tasks = json.load(f)
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
LOG.exception(e)
return web.json_response({"error": "Failed to load tasks."}, status=web.HTTPInternalServerError.status_code)
return web.json_response(tasks)
return web.json_response(data=tasks, status=web.HTTPOk.status_code, dumps=self.encoder.encode)
@route('POST', 'add_batch')
@route("POST", "add_batch")
async def add_batch(self, request: Request) -> Response:
status = {}
post = await request.json()
if not isinstance(post, list):
raise web.HTTPBadRequest(text='Invalid request body expecting list with dicts.')
return web.json_response(
data={"error": "Invalid request body expecting list with dicts."},
status=web.HTTPBadRequest.status_code,
)
for item in post:
if not isinstance(item, dict):
raise web.HTTPBadRequest(text='Invalid request body expecting list with dicts.')
return web.json_response(
data={"error": "Invalid request body expecting list with dicts."},
status=web.HTTPBadRequest.status_code,
)
if not item.get('url'):
raise web.HTTPBadRequest(text='url is required.')
if not item.get("url"):
return web.json_response(
data={"error": "url is required.", "data": post}, status=web.HTTPBadRequest.status_code
)
status[item.get('url')] = await self.add(
url=item.get('url'),
preset=item.get('preset', 'default'),
folder=item.get('folder'),
ytdlp_cookies=item.get('ytdlp_cookies'),
ytdlp_config=item.get('ytdlp_config'),
output_template=item.get('output_template')
for item in post:
status[item.get("url")] = await self.add(
url=item.get("url"),
preset=item.get("preset", "default"),
folder=item.get("folder"),
ytdlp_cookies=item.get("ytdlp_cookies"),
ytdlp_config=item.get("ytdlp_config"),
output_template=item.get("output_template"),
)
return web.Response(text=self.encoder.encode(status))
return web.json_response(data=status, status=web.HTTPOk.status_code, dumps=self.encoder.encode)
@route('DELETE', 'delete')
@route("DELETE", "delete")
async def delete(self, request: Request) -> Response:
post = await request.json()
ids = post.get('ids')
where = post.get('where')
ids = post.get("ids")
where = post.get("where")
if not ids or where not in ['queue', 'done']:
raise web.HTTPBadRequest()
if not ids or where not in ["queue", "done"]:
return web.json_response(
data={"error": "ids and where are required."}, status=web.HTTPBadRequest.status_code
)
status: dict[str, str] = {}
return web.json_response(
data=await (self.queue.cancel(ids) if where == "queue" else self.queue.clear(ids)),
status=web.HTTPOk.status_code,
dumps=self.encoder.encode,
)
status = await (self.queue.cancel(ids) if where == 'queue' else self.queue.clear(ids))
return web.Response(text=self.encoder.encode(status))
@route('POST', 'item/{id}')
@route("POST", "item/{id}")
async def update_item(self, request: Request) -> Response:
id: str = request.match_info.get('id')
id: str = request.match_info.get("id")
if not id:
raise web.HTTPBadRequest(text='id is required.')
return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code)
item = self.queue.done.getById(id)
if not item:
raise web.HTTPNotFound(text='Item not found.')
return web.json_response(data={"error": "item not found."}, status=web.HTTPNotFound.status_code)
try:
post = await request.json()
if not post:
raise web.HTTPBadRequest(text='no data provided.')
except Exception as e:
raise web.HTTPBadRequest(text=str(e))
post = await request.json()
if not post:
return web.json_response(data={"error": "no data provided."}, status=web.HTTPBadRequest.status_code)
updated = False
@ -307,30 +330,33 @@ class HttpAPI(common):
updated = True
setattr(item.info, k, v)
LOG.info(f"Updated '{k}' to '{v}' for '{item.info.id}'")
LOG.debug(f"Updated '{k}' to '{v}' for '{item.info.id}'")
status = 200 if updated else 304
if updated:
self.queue.done.put(item)
await self.notifier.emit('update', item.info)
await self.emitter.emit("update", item.info)
return web.Response(text=self.encoder.encode(item.info), status=status)
return web.json_response(
data=item.info,
status=web.HTTPOk.status_code if updated else web.HTTPNotModified.status_code,
dumps=self.encoder.encode,
)
@route('GET', 'history')
@route("GET", "history")
async def history(self, _: Request) -> Response:
history = {'done': [], 'queue': []}
data: dict = {"queue": [], "history": []}
for _, v in self.queue.queue.saved_items():
history['queue'].append(v)
data["queue"].append(v)
for _, v in self.queue.done.saved_items():
history['done'].append(v)
data["history"].append(v)
return web.Response(text=self.encoder.encode(history))
return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode)
@route('GET', 'workers')
@route("GET", "workers")
async def workers(self, _) -> Response:
if self.queue.pool is None:
return web.json_response({"error": "Worker pool not initialized."}, status=404)
return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code)
status = self.queue.pool.get_workers_status()
@ -338,231 +364,271 @@ class HttpAPI(common):
for worker in status:
worker_status = status.get(worker)
data.append(
{
"id": worker,
"data": {"status": "Waiting for download."} if worker_status is None else worker_status,
}
)
data.append({
'id': worker,
'data': {"status": 'Waiting for download.'} if worker_status is None else worker_status,
})
return web.json_response(
data={
"open": self.queue.pool.has_open_workers(),
"count": self.queue.pool.get_available_workers(),
"workers": data,
},
status=web.HTTPOk.status_code,
dumps=lambda obj: json.dumps(obj, default=lambda o: f"<<non-serializable: {type(o).__qualname__}>>"),
)
def safe_serialize(obj):
def default(o): return f"<<non-serializable: {type(o).__qualname__}>>"
return json.dumps(obj, default=default)
return web.Response(text=safe_serialize({
'open': self.queue.pool.has_open_workers(),
'count': self.queue.pool.get_available_workers(),
'workers': data,
}), headers={
'Content-Type': 'application/json',
})
@route('POST', 'workers')
@route("POST", "workers")
async def restart_pool(self, _) -> Response:
if self.queue.pool is None:
return web.json_response({"error": "Worker pool not initialized."}, status=404)
return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code)
self.queue.pool.start()
return web.Response()
return web.json_response({"message": "Workers pool being restarted."}, status=web.HTTPOk.status_code)
@route('PATCH', 'workers/{id}')
@route("PATCH", "workers/{id}")
async def restart_worker(self, request: Request) -> Response:
id: str = request.match_info.get('id')
id: str = request.match_info.get("id")
if not id:
raise web.HTTPBadRequest(text='worker id is required.')
return web.json_response({"error": "worker id is required."}, status=web.HTTPBadRequest.status_code)
if self.queue.pool is None:
return web.json_response({"error": "Worker pool not initialized."}, status=404)
return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code)
status = await self.queue.pool.restart(id, 'requested by user.')
status = await self.queue.pool.restart(id, "requested by user.")
return web.json_response({"status": "restarted" if status else "in_error_state"})
return web.json_response({"status": "restarted" if status else "in_error_state"}, status=web.HTTPOk.status_code)
@route('delete', 'workers/{id}')
@route("delete", "workers/{id}")
async def stop_worker(self, request: Request) -> Response:
id: str = request.match_info.get('id')
id: str = request.match_info.get("id")
if not id:
raise web.HTTPBadRequest(text='worker id is required.')
raise web.HTTPBadRequest(text="worker id is required.")
if self.queue.pool is None:
return web.json_response({"error": "Worker pool not initialized."}, status=404)
return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code)
status = await self.queue.pool.stop(id, 'requested by user.')
status = await self.queue.pool.stop(id, "requested by user.")
return web.json_response({"status": "stopped" if status else "in_error_state"})
return web.json_response({"status": "stopped" if status else "in_error_state"}, status=web.HTTPOk.status_code)
@route('GET', 'player/playlist/{file:.*}.m3u8')
@route("GET", "player/playlist/{file:.*}.m3u8")
async def playlist(self, request: Request) -> Response:
file: str = request.match_info.get('file')
file: str = request.match_info.get("file")
if not file:
raise web.HTTPBadRequest(text='file is required.')
raise web.HTTPBadRequest(text="file is required.")
try:
text = await Playlist(url=f"{self.config.url_host}{self.config.url_prefix}").make(
download_path=self.config.download_path,
file=file
download_path=self.config.download_path, file=file
)
if isinstance(text, Response):
return text
except Exception as e:
return web.HTTPNotFound(text=str(e))
except StreamingError as e:
return web.json_response(data={"error": str(e)}, status=web.HTTPNotFound.status_code)
return web.Response(text=text, headers={
'Content-Type': 'application/x-mpegURL',
'Cache-Control': 'no-cache',
'Access-Control-Max-Age': "300",
})
return web.Response(
text=text,
headers={
"Content-Type": "application/x-mpegURL",
"Cache-Control": "no-cache",
"Access-Control-Max-Age": "300",
},
status=web.HTTPOk.status_code,
)
@route('GET', 'player/m3u8/{mode}/{file:.*}.m3u8')
@route("GET", "player/m3u8/{mode}/{file:.*}.m3u8")
async def m3u8(self, request: Request) -> Response:
file: str = request.match_info.get('file')
mode: str = request.match_info.get('mode')
file: str = request.match_info.get("file")
mode: str = request.match_info.get("mode")
if mode not in ['video', 'subtitle']:
raise web.HTTPBadRequest(text='Only video and subtitle modes are supported.')
if mode not in ["video", "subtitle"]:
return web.json_response(
data={"error": "Only video and subtitle modes are supported."}, status=web.HTTPBadRequest.status_code
)
if not file:
raise web.HTTPBadRequest(text='file is required.')
return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code)
duration = request.query.get('duration', None)
duration = request.query.get("duration", None)
if 'subtitle' in mode:
if "subtitle" in mode:
if not duration:
raise web.HTTPBadRequest(text='duration is required.')
return web.json_response(data={"error": "duration is required."}, status=web.HTTPBadRequest.status_code)
duration = float(duration)
try:
cls = M3u8(f"{self.config.url_host}{self.config.url_prefix}")
if 'subtitle' in mode:
if "subtitle" in mode:
text = await cls.make_subtitle(self.config.download_path, file, duration)
else:
text = await cls.make_stream(self.config.download_path, file)
except Exception as e:
return web.HTTPNotFound(text=str(e))
except StreamingError as e:
LOG.exception(e)
return web.json_response(data={"error": str(e)}, status=web.HTTPNotFound.status_code)
return web.Response(text=text, headers={
'Content-Type': 'application/x-mpegURL',
'Cache-Control': 'no-cache',
'Access-Control-Max-Age': "300",
})
return web.Response(
text=text,
headers={
"Content-Type": "application/x-mpegURL",
"Cache-Control": "no-cache",
"Access-Control-Max-Age": "300",
},
status=web.HTTPOk.status_code,
)
@route('GET', 'player/segments/{segment:\d+}/{file:.*}.ts')
@route("GET", r"player/segments/{segment:\d+}/{file:.*}.ts")
async def segments(self, request: Request) -> Response:
file: str = request.match_info.get('file')
segment: int = request.match_info.get('segment')
sd: int = request.query.get('sd')
vc: int = int(request.query.get('vc', 0))
ac: int = int(request.query.get('ac', 0))
file: str = request.match_info.get("file")
segment: int = request.match_info.get("segment")
sd: int = request.query.get("sd")
vc: int = int(request.query.get("vc", 0))
ac: int = int(request.query.get("ac", 0))
file_path: str = os.path.normpath(os.path.join(self.config.download_path, file))
if not file_path.startswith(self.config.download_path):
raise web.HTTPBadRequest(text='Invalid file path.')
return web.json_response(data={"error": "Invalid file path."}, status=web.HTTPBadRequest.status_code)
if request.if_modified_since:
lastMod = time.strftime(
"%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple()
)
if os.path.exists(file_path) and request.if_modified_since.timestamp() == os.path.getmtime(file_path):
return web.Response(status=304)
return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod})
if not file:
raise web.HTTPBadRequest(text='file is required')
return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code)
if not segment:
raise web.HTTPBadRequest(text='segment is required')
return web.json_response(data={"error": "segment id is required."}, status=web.HTTPBadRequest.status_code)
segmenter = Segments(
index=int(segment),
duration=float('{:.6f}'.format(float(sd if sd else M3u8.duration))),
duration=float("{:.6f}".format(float(sd if sd else M3u8.duration))),
vconvert=True if vc == 1 else False,
aconvert=True if ac == 1 else False
aconvert=True if ac == 1 else False,
)
return web.Response(body=await segmenter.stream(path=self.config.download_path, file=file), headers={
'Content-Type': 'video/mpegts',
'X-Accel-Buffering': 'no',
'Access-Control-Allow-Origin': '*',
'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()),
'Expires': time.strftime('%a, %d %b %Y %H:%M:%S GMT', datetime.fromtimestamp(time.time() + 31536000).timetuple()),
})
return web.Response(
body=await segmenter.stream(path=self.config.download_path, file=file),
headers={
"Content-Type": "video/mpegts",
"X-Accel-Buffering": "no",
"Access-Control-Allow-Origin": "*",
"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()
),
"Expires": time.strftime(
"%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000).timetuple()
),
},
status=web.HTTPOk.status_code,
)
@route('GET', 'player/subtitle/{file:.*}.vtt')
@route("GET", "player/subtitle/{file:.*}.vtt")
async def subtitles(self, request: Request) -> Response:
file: str = request.match_info.get('file')
file: str = request.match_info.get("file")
file_path: str = os.path.normpath(os.path.join(self.config.download_path, file))
if not file_path.startswith(self.config.download_path):
raise web.HTTPBadRequest(text='Invalid file path.')
return web.json_response(data={"error": "Invalid file path."}, status=web.HTTPBadRequest.status_code)
if request.if_modified_since:
lastMod = time.strftime('%a, %d %b %Y %H:%M:%S GMT', datetime.fromtimestamp(
os.path.getmtime(file_path)).timetuple())
lastMod = time.strftime(
"%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple()
)
if os.path.exists(file_path) and request.if_modified_since.timestamp() == os.path.getmtime(file_path):
return web.Response(status=304, headers={'Last-Modified': lastMod})
return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod})
if not file:
raise web.HTTPBadRequest(text='file is required')
return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code)
return web.Response(body=await Subtitle().make(path=self.config.download_path, file=file), headers={
'Content-Type': 'text/vtt; charset=UTF-8',
'X-Accel-Buffering': 'no',
'Access-Control-Allow-Origin': '*',
'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()),
'Expires': time.strftime('%a, %d %b %Y %H:%M:%S GMT', datetime.fromtimestamp(time.time() + 31536000).timetuple()),
})
@route('OPTIONS', '/add')
async def add_cors(self, _: Request) -> Response:
return web.Response(text=self.encoder.encode({"status": "ok"}))
@route('OPTIONS', '/delete')
async def delete_cors(self, _: Request) -> Response:
return web.Response(text=self.encoder.encode({"status": "ok"}))
@route('GET', '/')
async def index(self, _) -> Response:
if '/index.html' not in self.staticHolder:
LOG.error("Static frontend files not found.")
return web.json_response({"error": "File not found.", "file": '/index.html'}, status=404)
data = self.staticHolder['/index.html']
return web.Response(
body=data.get('content'),
content_type=data.get('content_type'),
charset='utf-8',
status=web.HTTPOk.status_code)
body=await Subtitle().make(path=self.config.download_path, file=file),
headers={
"Content-Type": "text/vtt; charset=UTF-8",
"X-Accel-Buffering": "no",
"Access-Control-Allow-Origin": "*",
"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()
),
"Expires": time.strftime(
"%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000).timetuple()
),
},
status=web.HTTPOk.status_code,
)
@route('GET', '/thumbnail')
async def get_thumbnail(self, request: Request) -> dict:
url = request.query.get('url')
@route("OPTIONS", "/add")
async def add_cors(self, _: Request) -> Response:
return web.json_response(data={"status": "ok"}, status=web.HTTPOk.status_code)
@route("OPTIONS", "/delete")
async def delete_cors(self, _: Request) -> Response:
return web.json_response(data={"status": "ok"}, status=web.HTTPOk.status_code)
@route("GET", "/")
async def index(self, _) -> Response:
if "/index.html" not in self.staticHolder:
LOG.error("Static frontend files not found.")
return web.json_response(
data={"error": "File not found.", "file": "/index.html"}, status=web.HTTPNotFound.status_code
)
data = self.staticHolder["/index.html"]
return web.Response(
body=data.get("content"),
content_type=data.get("content_type"),
charset="utf-8",
status=web.HTTPOk.status_code,
)
@route("GET", "/thumbnail")
async def get_thumbnail(self, request: Request) -> Response:
url = request.query.get("url")
if not url:
return web.json_response({"error": "URL is required."}, status=400)
return web.json_response(data={"error": "URL is required."}, status=web.HTTPForbidden.status_code)
try:
validate_url(url)
except Exception as e:
return web.json_response({"error": str(e)}, status=400)
except ValueError as e:
return web.json_response(data={"error": str(e)}, status=web.HTTPForbidden.status_code)
try:
opts = {
'proxy': self.config.ytdl_options.get('proxy', None),
'headers': {
'User-Agent': self.config.ytdl_options.get(
'user_agent',
request.headers.get('User-Agent', f"YTPTube/{self.config.version}")),
"proxy": self.config.ytdl_options.get("proxy", None),
"headers": {
"User-Agent": self.config.ytdl_options.get(
"user_agent", request.headers.get("User-Agent", f"YTPTube/{self.config.version}")
),
},
}
async with httpx.AsyncClient(**opts) as client:
LOG.info(f"Fetching thumbnail from '{url}'.")
response = await client.request(method='GET', url=url)
return web.Response(body=response.content,
headers={'Content-Type': response.headers.get('Content-Type'),
'Pragma': 'public', '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()), })
LOG.debug(f"Fetching thumbnail from '{url}'.")
response = await client.request(method="GET", url=url)
return web.Response(
body=response.content,
headers={
"Content-Type": response.headers.get("Content-Type"),
"Pragma": "public",
"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()
),
},
)
except Exception as e:
LOG.error(f"Error fetching thumbnail from '{url}'. '{e}'")
return web.json_response({"error": str(e)}, status=500)
LOG.error(f"Error fetching thumbnail from '{url}'. '{e}'.")
LOG.exception(e)
return web.json_response(
data={"error": "failed to retrieve the thumbnail."}, status=web.HTTPInternalServerError.status_code
)

136
app/library/HttpSocket.py
View file

@ -1,34 +1,37 @@
import asyncio
from datetime import datetime
import errno
import functools
import logging
import os
import pty
import shlex
from aiohttp import web
from datetime import datetime
import socketio
import logging
from aiohttp import web
from .common import common
from .config import Config
from .DownloadQueue import DownloadQueue
from .common import common
from .Emitter import Emitter
from .encoder import Encoder
from .Utils import isDownloaded
from .Emitter import Emitter
LOG = logging.getLogger('socket')
LOG = logging.getLogger("socket_api")
class HttpSocket(common):
"""
This class is used to handle WebSocket events.
"""
config: Config = None
sio: socketio.AsyncServer = None
def __init__(self, queue: DownloadQueue, emitter: Emitter, encoder: Encoder):
super().__init__(queue=queue, encoder=encoder)
self.sio = socketio.AsyncServer(cors_allowed_origins='*')
self.sio = socketio.AsyncServer(cors_allowed_origins="*")
emitter.add_emitter(lambda event, data, **kwargs: self.sio.emit(event, encoder.encode(data), **kwargs))
self.config = Config.get_instance()
@ -40,41 +43,45 @@ class HttpSocket(common):
"""
Decorator to mark a method as a socket event.
"""
@functools.wraps(func)
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
wrapper._ws_event = func.__name__
return wrapper
def attach(self, app: web.Application):
self.sio.attach(app, socketio_path=self.config.url_prefix + 'socket.io')
self.sio.attach(app, socketio_path=self.config.url_prefix + "socket.io")
for attr_name in dir(self):
method = getattr(self, attr_name)
if hasattr(method, '_ws_event'):
if hasattr(method, "_ws_event"):
event = method._ws_event
self.sio.on(event)(method)
@ws_event
async def cli_post(self, sid: str, data):
if not data:
await self.emitter.emit('cli_close', {'exitcode': 0}, to=sid)
await self.emitter.emit("cli_close", {"exitcode": 0}, to=sid)
return
try:
LOG.info(f"Cli command from client '{sid}'. '{data}'")
args = ['yt-dlp'] + shlex.split(data)
args = ["yt-dlp"] + shlex.split(data)
_env = os.environ.copy()
_env.update({
"TERM": "xterm-256color",
"LANG": "en_US.UTF-8",
"SHELL": "/bin/bash",
"LC_ALL": "en_US.UTF-8",
"PWD": self.config.download_path,
"FORCE_COLOR": "1",
"PYTHONUNBUFFERED": "1",
})
_env.update(
{
"TERM": "xterm-256color",
"LANG": "en_US.UTF-8",
"SHELL": "/bin/bash",
"LC_ALL": "en_US.UTF-8",
"PWD": self.config.download_path,
"FORCE_COLOR": "1",
"PYTHONUNBUFFERED": "1",
}
)
master_fd, slave_fd = pty.openpty()
@ -94,7 +101,7 @@ class HttpSocket(common):
async def read_pty():
loop = asyncio.get_running_loop()
buffer = b''
buffer = b""
while True:
try:
chunk = await loop.run_in_executor(None, lambda: os.read(master_fd, 1024))
@ -108,18 +115,16 @@ class HttpSocket(common):
# No more output
if buffer:
# Emit any remaining partial line
await self.emitter.emit('cli_output', {
'type': 'stdout',
'line': buffer.decode('utf-8', errors='replace')
})
await self.emitter.emit(
"cli_output", {"type": "stdout", "line": buffer.decode("utf-8", errors="replace")}
)
break
buffer += chunk
*lines, buffer = buffer.split(b'\n')
*lines, buffer = buffer.split(b"\n")
for line in lines:
await self.emitter.emit('cli_output', {
'type': 'stdout',
'line': line.decode('utf-8', errors='replace')
})
await self.emitter.emit(
"cli_output", {"type": "stdout", "line": line.decode("utf-8", errors="replace")}
)
try:
os.close(master_fd)
except Exception as e:
@ -134,28 +139,33 @@ class HttpSocket(common):
# Ensure reading is done
await read_task
await self.emitter.emit('cli_close', {'exitcode': returncode}, to=sid)
await self.emitter.emit("cli_close", {"exitcode": returncode}, to=sid)
except Exception as e:
LOG.error(f"CLI execute exception was thrown for client '{sid}'. {str(e)}")
await self.emitter.emit('cli_out1put', {
'type': 'stderr',
'line': str(e),
}, to=sid)
await self.emitter.emit('cli_close', {'exitcode': -1}, to=sid)
LOG.error(f"CLI execute exception was thrown for client '{sid}'.")
LOG.exception(e)
await self.emitter.emit(
"cli_out1put",
{
"type": "stderr",
"line": str(e),
},
to=sid,
)
await self.emitter.emit("cli_close", {"exitcode": -1}, to=sid)
@ws_event
async def add_url(self, sid: str, data: dict):
url: str = data.get('url')
url: str = data.get("url")
if not url:
self.emitter.warning('No URL provided.', to=sid)
self.emitter.warning("No URL provided.", to=sid)
return
preset: str = data.get('preset', 'default')
folder: str = data.get('folder')
ytdlp_cookies: str = data.get('ytdlp_cookies')
ytdlp_config: dict = data.get('ytdlp_config')
output_template: str = data.get('output_template')
preset: str = data.get("preset", "default")
folder: str = data.get("folder")
ytdlp_cookies: str = data.get("ytdlp_cookies")
ytdlp_config: dict | None = data.get("ytdlp_config")
output_template: str = data.get("output_template")
if ytdlp_config is None:
ytdlp_config = {}
@ -165,64 +175,64 @@ class HttpSocket(common):
folder=folder,
ytdlp_cookies=ytdlp_cookies,
ytdlp_config=ytdlp_config,
output_template=output_template
output_template=output_template,
)
await self.emitter.emit('status', status, to=sid)
await self.emitter.emit("status", status, to=sid)
@ws_event
async def item_cancel(self, sid: str, id: str):
if not id:
await self.emitter.warning('Invalid request.', to=sid)
await self.emitter.warning("Invalid request.", to=sid)
return
status: dict[str, str] = {}
status = await self.queue.cancel([id])
status.update({'identifier': id})
status.update({"identifier": id})
await self.emitter.emit('item_cancel', status)
await self.emitter.emit("item_cancel", status)
@ws_event
async def item_delete(self, sid: str, id: str):
if not id:
await self.emitter.warning('Invalid request.', to=sid)
await self.emitter.warning("Invalid request.", to=sid)
return
status: dict[str, str] = {}
status = await self.queue.clear([id])
status.update({'identifier': id})
status.update({"identifier": id})
await self.emitter.emit('item_delete', status)
await self.emitter.emit("item_delete", status)
@ws_event
async def archive_item(self, sid: 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
file: str = self.config.ytdl_options.get('download_archive', None)
file: str = self.config.ytdl_options.get("download_archive", None)
if not file:
return
exists, idDict = isDownloaded(file, data['url'])
if exists or 'archive_id' not in idDict or idDict['archive_id'] is None:
exists, idDict = isDownloaded(file, data["url"])
if exists or "archive_id" not in idDict or idDict["archive_id"] is None:
return
with open(file, 'a') as f:
with open(file, "a") as f:
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:
with open(manual_archive, "r") as f:
for line in f.readlines():
if idDict['archive_id'] in line:
if idDict["archive_id"] in line:
previouslyArchived = True
break
if not previouslyArchived:
with open(manual_archive, 'a') as f:
with open(manual_archive, "a") as f:
f.write(f"{idDict['archive_id']} - at: {datetime.now().isoformat()}\n")
LOG.info(f"Archiving url '{data['url']}' with id '{idDict['archive_id']}'.")
@ -238,8 +248,6 @@ class HttpSocket(common):
# get download folder listing
downloadPath: str = self.config.download_path
data['folders'] = [
name for name in os.listdir(downloadPath) if os.path.isdir(os.path.join(downloadPath, name))
]
data["folders"] = [name for name in os.listdir(downloadPath) if os.path.isdir(os.path.join(downloadPath, name))]
await self.emitter.emit('initial_data', data, to=sid)
await self.emitter.emit("initial_data", data, to=sid)

53
app/library/ItemDTO.py
View file

@ -1,47 +1,52 @@
from email.utils import formatdate
import json
import time
from dataclasses import dataclass, field
import uuid
from dataclasses import dataclass, field
from email.utils import formatdate
@dataclass(kw_only=True)
class ItemDTO:
_id: int = field(default_factory=lambda: str(uuid.uuid4()), init=False)
"""
ItemDTO is a data transfer object that represents a single item in the queue.
It contains all the required information for both the frontend and the backend to process the item.
"""
_id: str = field(default_factory=lambda: str(uuid.uuid4()), init=False)
error: str = None
id: str
title: str
url: str
quality: str = None
format: str = None
thumbnail: str = None
quality: str | None = None
format: str | None = None
thumbnail: str | None = None
preset: str = "default"
folder: str
download_dir: str = None
temp_dir: str = None
status: str = None
ytdlp_cookies: str = None
download_dir: str | None = None
temp_dir: str | None = None
status: str | None = None
ytdlp_cookies: str | None = None
ytdlp_config: dict = field(default_factory=dict)
output_template: str = None
output_template_chapter: str = None
output_template: str | None = None
output_template_chapter: str | None = None
timestamp: float = time.time_ns()
is_live: bool = None
is_live: bool | None = None
datetime: str = field(default_factory=lambda: str(formatdate(time.time())))
live_in: str = None
file_size: int = None
live_in: str | None = None
file_size: int | None = None
options: dict = field(default_factory=dict)
# yt-dlp injected fields.
tmpfilename: str = None
filename: str = None
total_bytes: int = None
total_bytes_estimate: int = None
downloaded_bytes: int = None
msg: str = None
percent: int = None
speed: str = None
eta: str = None
tmpfilename: str | None = None
filename: str | None = None
total_bytes: int | None = None
total_bytes_estimate: int | None = None
downloaded_bytes: int | None = None
msg: str | None = None
percent: int | None = None
speed: str | None = None
eta: str | None = None
def json(self) -> str:
return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)

49
app/library/M3u8.py
View file

@ -1,37 +1,44 @@
import math
import os
from urllib.parse import quote
from .Utils import calcDownloadPath
from .Utils import calcDownloadPath, StreamingError
from .ffprobe import FFProbe
class M3u8:
ok_vcodecs: tuple = ('h264', 'x264', 'avc',)
ok_acodecs: tuple = ('aac', 'm4a', 'mp3',)
url: str = None
duration: float = 6.000000
ok_vcodecs: tuple = (
"h264",
"x264",
"avc",
)
ok_acodecs: tuple = (
"aac",
"m4a",
"mp3",
)
def __init__(self, url: str, segment_duration: float = 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):
async def make_stream(self, download_path: str, file: str) -> str:
realFile: str = calcDownloadPath(basePath=download_path, folder=file, createPath=False)
if not os.path.exists(realFile):
raise Exception(f"File '{realFile}' does not exist.")
raise StreamingError(f"File '{realFile}' does not exist.")
try:
ffprobe = FFProbe(realFile)
await ffprobe.run()
except UnicodeDecodeError as e:
except UnicodeDecodeError:
pass
if not 'duration' in ffprobe.metadata:
raise Exception(f"Unable to get '{realFile}' duration.")
if "duration" not in ffprobe.metadata:
raise StreamingError(f"Unable to get '{realFile}' play duration.")
duration: float = float(ffprobe.metadata.get('duration'))
duration: float = float(ffprobe.metadata.get("duration"))
m3u8 = []
@ -41,7 +48,7 @@ class M3u8:
m3u8.append("#EXT-X-MEDIA-SEQUENCE:0")
m3u8.append("#EXT-X-PLAYLIST-TYPE:VOD")
segmentSize: float = '{:.6f}'.format(self.duration)
segmentSize: float = "{:.6f}".format(self.duration)
splits: int = math.ceil(duration / self.duration)
segmentParams: dict = {}
@ -49,33 +56,33 @@ class M3u8:
for stream in ffprobe.streams:
if stream.is_video():
if stream.codec_name not in self.ok_vcodecs:
segmentParams['vc'] = 1
segmentParams["vc"] = 1
if stream.is_audio():
if stream.codec_name not in self.ok_acodecs:
segmentParams['ac'] = 1
segmentParams["ac"] = 1
for i in range(splits):
if (i + 1) == splits:
segmentSize = '{:.6f}'.format(duration - (i * self.duration))
segmentParams.update({'sd': segmentSize})
segmentSize = "{:.6f}".format(duration - (i * self.duration))
segmentParams.update({"sd": segmentSize})
m3u8.append(f"#EXTINF:{segmentSize},")
url = f"{self.url}player/segments/{i}/{quote(file)}.ts"
if len(segmentParams) > 0:
url += '?'+'&'.join([f"{key}={value}" for key, value in segmentParams.items()])
url += "?" + "&".join([f"{key}={value}" for key, value in segmentParams.items()])
m3u8.append(url)
m3u8.append("#EXT-X-ENDLIST")
return '\n'.join(m3u8)
return "\n".join(m3u8)
async def make_subtitle(self, download_path: str, file: str, duration: float):
async def make_subtitle(self, download_path: str, file: str, duration: float) -> str:
realFile: str = calcDownloadPath(basePath=download_path, folder=file, createPath=False)
if not os.path.exists(realFile):
raise Exception(f"File '{realFile}' does not exist.")
raise StreamingError(f"File '{realFile}' does not exist.")
m3u8 = []
@ -88,4 +95,4 @@ class M3u8:
m3u8.append(f"{self.url}player/subtitle/{quote(file)}.vtt")
m3u8.append("#EXT-X-ENDLIST")
return '\n'.join(m3u8)
return "\n".join(m3u8)

54
app/library/Playlist.py
View file

@ -1,13 +1,15 @@
import glob
import pathlib
import re
from pathlib import Path
from urllib.parse import quote
from library.Utils import calcDownloadPath, checkId
import pathlib
from library.ffprobe import FFProbe
from .Subtitle import Subtitle
from aiohttp.web import Response
from .ffprobe import FFProbe
from .Subtitle import Subtitle
from .Utils import calcDownloadPath, checkId, StreamingError
class Playlist:
_url: str = None
@ -15,27 +17,30 @@ class Playlist:
self.url = url
async def make(self, download_path: str, file: str) -> str | Response:
rFile = pathlib.Path(calcDownloadPath(basePath=download_path, folder=file, createPath=False))
rFile = Path(calcDownloadPath(basePath=download_path, folder=file, createPath=False))
if not rFile.exists():
possibleFile = checkId(download_path, rFile)
if not possibleFile:
raise Exception(f"File '{rFile}' does not exist.")
raise StreamingError(f"File '{rFile}' does not exist.")
return Response(status=302, headers={
'Location': f"{self.url}player/playlist/{quote(str(possibleFile).replace(download_path, '').strip('/'))}.m3u8"
})
return Response(
status=302,
headers={
"Location": f"{self.url}player/playlist/{quote(str(possibleFile).replace(download_path, '').strip('/'))}.m3u8"
},
)
try:
ffprobe = FFProbe(rFile)
await ffprobe.run()
except UnicodeDecodeError as e:
except UnicodeDecodeError:
pass
if not 'duration' in ffprobe.metadata:
raise Exception(f"Unable to get '{rFile}' duration.")
if "duration" not in ffprobe.metadata:
raise StreamingError(f"Unable to get '{rFile}' duration.")
duration: float = float(ffprobe.metadata.get('duration'))
duration: float = float(ffprobe.metadata.get("duration"))
playlist = []
playlist.append("#EXTM3U")
@ -44,28 +49,31 @@ class Playlist:
index = 0
for item in self.getSideCarFiles(rFile):
if not item.suffix in Subtitle.allowedExtensions:
if item.suffix not in Subtitle.allowedExtensions:
continue
index += 1
lang: str = "und"
lg = re.search(r'\.(?P<lang>\w{2,3})\.\w{3}$', item.name)
lg = re.search(r"\.(?P<lang>\w{2,3})\.\w{3}$", item.name)
if lg:
lang = lg.groupdict().get('lang')
lang = lg.groupdict().get("lang")
subs = ',SUBTITLES="subs"'
url = f"{self.url}player/m3u8/subtitle/{quote(str(pathlib.Path(file).with_name(item.name)))}.m3u8?duration={duration}"
url = (
f"{self.url}player/m3u8/subtitle/{quote(str(Path(file).with_name(item.name)))}.m3u8?duration={duration}"
)
name = f"{item.suffix[1:].upper()} ({index}) - {lang}"
playlist.append(
f'#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID="subs",NAME="{name}",DEFAULT=NO,AUTOSELECT=NO,FORCED=NO,LANGUAGE="{lang}",URI="{url}"')
f'#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID="subs",NAME="{name}",DEFAULT=NO,AUTOSELECT=NO,FORCED=NO,LANGUAGE="{lang}",URI="{url}"'
)
playlist.append(f'#EXT-X-STREAM-INF:PROGRAM-ID=1{subs}')
playlist.append(f"#EXT-X-STREAM-INF:PROGRAM-ID=1{subs}")
playlist.append(f"{self.url}player/m3u8/video/{quote(file)}.m3u8")
return '\n'.join(playlist)
return "\n".join(playlist)
def getSideCarFiles(self, file: pathlib.Path) -> list[pathlib.Path]:
def getSideCarFiles(self, file: Path) -> list[Path]:
"""
Get sidecar files for the given file.
@ -75,7 +83,7 @@ class Playlist:
files = []
for sub_file in file.parent.glob(f"{glob.escape(file.stem)}.*"):
if sub_file == file or sub_file.is_file() is False or sub_file.stem.startswith('.'):
if sub_file == file or sub_file.is_file() is False or sub_file.stem.startswith("."):
continue
files.append(sub_file)

96
app/library/Segments.py
View file

@ -3,20 +3,14 @@ import hashlib
import logging
import os
import tempfile
from .Utils import calcDownloadPath
from .config import Config
LOG = logging.getLogger('player.segments')
from .config import Config
from .Utils import calcDownloadPath, StreamingError
LOG = logging.getLogger("player.segments")
class Segments:
duration: int
index: int
vconvert: bool
aconvert: bool
vcodec: str
acodec: str
def __init__(self, index: int, duration: float, vconvert: bool, aconvert: bool):
config = Config.get_instance()
self.index = int(index)
@ -25,13 +19,14 @@ class Segments:
self.aconvert = bool(aconvert)
self.vcodec = config.streamer_vcodec
self.acodec = config.streamer_acodec
# sadly due to unforeseen circumstances, we have to convert the video for now.
self.vconvert = True
async def stream(self, path: str, file: str) -> bytes:
realFile: str = calcDownloadPath(basePath=path, folder=file, createPath=False)
if not os.path.exists(realFile):
raise Exception(f"File {realFile} does not exist.")
raise StreamingError(f"File {realFile} does not exist.")
tmpDir: str = tempfile.gettempdir()
tmpFile = os.path.join(tmpDir, f'ytptube_stream.{hashlib.md5(realFile.encode("utf-8")).hexdigest()}')
@ -40,68 +35,69 @@ class Segments:
os.symlink(realFile, tmpFile)
if self.index == 0:
startTime: float = '{:.6f}'.format(0)
startTime: float = "{:.6f}".format(0)
else:
startTime: float = '{:.6f}'.format((self.duration * self.index))
startTime: float = "{:.6f}".format((self.duration * self.index))
fargs = []
fargs.append('-xerror')
fargs.append('-hide_banner')
fargs.append('-loglevel')
fargs.append('error')
fargs.append("-xerror")
fargs.append("-hide_banner")
fargs.append("-loglevel")
fargs.append("error")
fargs.append('-ss')
fargs.append(str(startTime if startTime else '0.00000'))
fargs.append('-t')
fargs.append(str('{:.6f}'.format(self.duration)))
fargs.append("-ss")
fargs.append(str(startTime if startTime else "0.00000"))
fargs.append("-t")
fargs.append(str("{:.6f}".format(self.duration)))
fargs.append('-copyts')
fargs.append("-copyts")
fargs.append('-i')
fargs.append(f'file:{tmpFile}')
fargs.append('-map_metadata')
fargs.append('-1')
fargs.append("-i")
fargs.append(f"file:{tmpFile}")
fargs.append("-map_metadata")
fargs.append("-1")
fargs.append('-pix_fmt')
fargs.append('yuv420p')
fargs.append('-g')
fargs.append('52')
fargs.append("-pix_fmt")
fargs.append("yuv420p")
fargs.append("-g")
fargs.append("52")
fargs.append('-map')
fargs.append('0:v:0')
fargs.append('-strict')
fargs.append('-2')
fargs.append("-map")
fargs.append("0:v:0")
fargs.append("-strict")
fargs.append("-2")
fargs.append('-codec:v')
fargs.append(self.vcodec if self.vconvert else 'copy')
fargs.append("-codec:v")
fargs.append(self.vcodec if self.vconvert else "copy")
# audio section.
fargs.append('-map')
fargs.append('0:a:0')
fargs.append('-codec:a')
fargs.append(self.acodec if self.aconvert else 'copy')
fargs.append("-map")
fargs.append("0:a:0")
fargs.append("-codec:a")
fargs.append(self.acodec if self.aconvert else "copy")
fargs.append('-sn')
fargs.append("-sn")
fargs.append('-muxdelay')
fargs.append('0')
fargs.append('-f')
fargs.append('mpegts')
fargs.append('pipe:1')
fargs.append("-muxdelay")
fargs.append("0")
fargs.append("-f")
fargs.append("mpegts")
fargs.append("pipe:1")
LOG.debug(f"Streaming '{realFile}' segment '{self.index}'. " + " ".join(fargs))
LOG.debug(f"Streaming '{realFile}' segment '{self.index}'. ffmpeg: {' '.join(fargs)}")
proc = await asyncio.subprocess.create_subprocess_exec(
'ffmpeg', *fargs,
"ffmpeg",
*fargs,
stdin=asyncio.subprocess.DEVNULL,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
stderr=asyncio.subprocess.PIPE,
)
data, err = await proc.communicate()
if 0 != proc.returncode:
LOG.error(f'Failed to stream {realFile} segment {self.index}. {err.decode("utf-8")}.')
raise Exception(f'Failed to stream {realFile} segment {self.index}.')
raise StreamingError(f"Failed to stream {realFile} segment {self.index}.")
return data

22
app/library/Subtitle.py
View file

@ -5,7 +5,7 @@ import pysubs2
from pysubs2.time import ms_to_times
from pysubs2.formats.substation import SubstationFormat
LOG = logging.getLogger('player.subtitle')
LOG = logging.getLogger("player.subtitle")
def ms_to_timestamp(ms: int) -> str:
@ -19,9 +19,13 @@ SubstationFormat.ms_to_timestamp = ms_to_timestamp
class Subtitle:
allowedExtensions: tuple[str] = (".srt", ".vtt", ".ass",)
allowedExtensions: tuple[str] = (
".srt",
".vtt",
".ass",
)
async def make(self, path: str, file: str) -> bytes:
async def make(self, path: str, file: str) -> str:
realFile: str = calcDownloadPath(basePath=path, folder=file, createPath=False)
rFile = pathlib.Path(realFile)
@ -29,12 +33,12 @@ class Subtitle:
if not rFile.exists():
raise Exception(f"File '{file}' does not exist.")
if not rFile.suffix in self.allowedExtensions:
if rFile.suffix not in self.allowedExtensions:
raise Exception(f"File '{file}' subtitle type is not supported.")
if rFile.suffix == ".vtt":
subData = ''
with open(realFile, 'r') as f:
subData = ""
with open(realFile, "r") as f:
subData = f.read()
return subData
@ -45,9 +49,9 @@ class Subtitle:
raise Exception(f"No subtitle events were found in '{rFile}'.")
if len(subs.events) < 2:
return subs.to_string('vtt')
return subs.to_string("vtt")
if subs.events[0].end == subs.events[len(subs.events)-1].end:
if subs.events[0].end == subs.events[len(subs.events) - 1].end:
subs.events.pop(0)
return subs.to_string('vtt')
return subs.to_string("vtt")

218
app/library/Utils.py
View file

@ -1,6 +1,4 @@
import copy
from datetime import datetime, timezone
from functools import lru_cache
import ipaddress
import json
import logging
@ -8,16 +6,31 @@ import os
import pathlib
import re
import socket
from typing import Any
import uuid
from datetime import datetime, timezone
from functools import lru_cache
from typing import Any
import yt_dlp
LOG = logging.getLogger('Utils')
LOG = logging.getLogger("Utils")
IGNORED_KEYS: tuple[str] = ('cookiefile', 'paths', 'outtmpl', 'progress_hooks', 'postprocessor_hooks',)
IGNORED_KEYS: tuple[str] = (
"cookiefile",
"paths",
"outtmpl",
"progress_hooks",
"postprocessor_hooks",
)
YTDLP_INFO_CLS: yt_dlp.YoutubeDL = None
class StreamingError(Exception):
"""Raised when an error occurs during streaming."""
pass
def get_opts(preset: str, ytdl_opts: dict) -> dict:
"""
Returns ytdlp options download options
@ -31,16 +44,17 @@ def get_opts(preset: str, ytdl_opts: dict) -> dict:
"""
opts = copy.deepcopy(ytdl_opts)
if 'default' == preset:
if "default" == preset:
LOG.debug("Using default preset.")
return opts
from .config import Config
presets = Config.get_instance().presets
found = False
for _preset in presets:
if _preset['name'] == preset:
if _preset["name"] == preset:
found = True
preset_opts = _preset
break
@ -49,24 +63,24 @@ def get_opts(preset: str, ytdl_opts: dict) -> dict:
LOG.error(f"Preset '{preset}' is not defined in the presets.")
return opts
opts['format'] = preset_opts.get('format')
opts["format"] = preset_opts.get("format")
if 'postprocessors' in preset_opts:
opts['postprocessors'] = preset_opts['postprocessors']
if "postprocessors" in preset_opts:
opts["postprocessors"] = preset_opts["postprocessors"]
if 'args' in preset_opts:
for key, value in preset_opts['args'].items():
if "args" in preset_opts:
for key, value in preset_opts["args"].items():
opts[key] = value
LOG.debug(f"Using preset '{preset}', altered options: {opts}")
return opts
def getVideoInfo(url: str, ytdlp_opts: dict = None) -> (Any | dict[str, Any] | None):
def getVideoInfo(url: str, ytdlp_opts: dict = None) -> Any | dict[str, Any] | None:
params: dict = {
'quiet': True,
'color': 'no_color',
'extract_flat': True,
"quiet": True,
"color": "no_color",
"extract_flat": True,
}
if ytdlp_opts:
@ -98,30 +112,35 @@ def calcDownloadPath(basePath: str, folder: str = None, createPath: bool = True)
def ExtractInfo(config: dict, url: str, debug: bool = False) -> dict:
params: dict = {
'color': 'no_color',
'extract_flat': True,
'skip_download': True,
'ignoreerrors': True,
'ignore_no_formats_error': True,
"color": "no_color",
"extract_flat": True,
"skip_download": True,
"ignoreerrors": True,
"ignore_no_formats_error": True,
**config,
}
# Remove keys that are not needed for info extraction as those keys generate files when used with extract_info.
for key in ('writeinfojson', 'writethumbnail', 'writedescription', 'writeautomaticsub',):
for key in (
"writeinfojson",
"writethumbnail",
"writedescription",
"writeautomaticsub",
):
if key in params:
params.pop(key)
if debug:
params['verbose'] = True
params['logger'] = logging.getLogger('YTPTube-ytdl')
params["verbose"] = True
params["logger"] = logging.getLogger("YTPTube-ytdl")
else:
params['quiet'] = True
params["quiet"] = True
return yt_dlp.YoutubeDL(params=params).extract_info(url, download=False)
def mergeDict(source: dict, destination: dict) -> dict:
""" Merge data from source into destination """
"""Merge data from source into destination"""
destination_copy = copy.deepcopy(destination)
for key, value in source.items():
@ -137,15 +156,14 @@ def mergeDict(source: dict, destination: dict) -> dict:
def mergeConfig(config: dict, new_config: dict) -> dict:
""" Merge user provided config into default config """
"""Merge user provided config into default config"""
ignored_keys: tuple = (
'cookiefile',
'download_archive'
'paths',
'outtmpl',
'progress_hooks',
'postprocessor_hooks',
"cookiefile",
"download_archive" "paths",
"outtmpl",
"progress_hooks",
"postprocessor_hooks",
)
for key in ignored_keys:
@ -159,23 +177,28 @@ def isDownloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, st
global YTDLP_INFO_CLS
idDict = {
'id': None,
'ie_key': None,
'archive_id': None,
"id": None,
"ie_key": None,
"archive_id": None,
}
if not url or not archive_file or not os.path.exists(archive_file):
return False, idDict,
return (
False,
idDict,
)
if not YTDLP_INFO_CLS:
YTDLP_INFO_CLS = yt_dlp.YoutubeDL(params={
'color': 'no_color',
'extract_flat': True,
'skip_download': True,
'ignoreerrors': True,
'ignore_no_formats_error': True,
'quiet': True,
})
YTDLP_INFO_CLS = yt_dlp.YoutubeDL(
params={
"color": "no_color",
"extract_flat": True,
"skip_download": True,
"ignoreerrors": True,
"ignore_no_formats_error": True,
"quiet": True,
}
)
for key, ie in YTDLP_INFO_CLS._ies.items():
if not ie.suitable(url):
@ -188,26 +211,35 @@ def isDownloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, st
if not temp_id:
break
idDict['id'] = temp_id
idDict['ie_key'] = key
idDict['archive_id'] = YTDLP_INFO_CLS._make_archive_id(idDict)
idDict["id"] = temp_id
idDict["ie_key"] = key
idDict["archive_id"] = YTDLP_INFO_CLS._make_archive_id(idDict)
break
if not idDict['archive_id']:
return False, idDict,
if not idDict["archive_id"]:
return (
False,
idDict,
)
with open(archive_file, 'r') as f:
with open(archive_file, "r") as f:
for line in f.readlines():
if idDict['archive_id'] in line:
return True, idDict,
if idDict["archive_id"] in line:
return (
True,
idDict,
)
return False, idDict,
return (
False,
idDict,
)
def jsonCookie(cookies: dict[dict[str, any]]) -> str | None:
"""Converts JSON cookies to Netscape cookies
Returns None if no cookies are found, otherwise returns a string of cookies in Netscape format.
Returns None if no cookies are found, otherwise returns a string of cookies in Netscape format.
"""
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"
@ -227,19 +259,26 @@ 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)
if 0 == int(cookieDict["expirationDate"]):
cookieDict["expirationDate"] = datetime.now(timezone.utc).timestamp() + (86400 * 1000)
hasCookies = True
netscapeCookies += "\t".join([
cookieDict['domain'] if str(cookieDict['domain']).startswith('.') else '.' + cookieDict['domain'],
'TRUE',
cookieDict['path'],
'TRUE' if cookieDict['secure'] else 'FALSE',
str(int(cookieDict['expirationDate'])),
cookieDict['name'],
cookieDict['value']
])+"\n"
netscapeCookies += (
"\t".join(
[
cookieDict["domain"]
if str(cookieDict["domain"]).startswith(".")
else "." + cookieDict["domain"],
"TRUE",
cookieDict["path"],
"TRUE" if cookieDict["secure"] else "FALSE",
str(int(cookieDict["expirationDate"])),
cookieDict["name"],
cookieDict["value"],
]
)
+ "\n"
)
return netscapeCookies if hasCookies else None
@ -264,21 +303,38 @@ def load_file(file: str, check_type=None) -> tuple[dict | list, bool, str]:
if check_type:
assert isinstance(opts, check_type)
return (opts, True, '',)
return (
opts,
True,
"",
)
except Exception:
with open(file) as json_data:
from pyjson5 import load as json5_load
try:
opts = json5_load(json_data)
if check_type:
assert isinstance(opts, check_type)
return (opts, True, '',)
return (
opts,
True,
"",
)
except AssertionError:
return ({}, False, f"Failed to assert that the contents '{type(opts)}' are of type '{check_type}'.",)
return (
{},
False,
f"Failed to assert that the contents '{type(opts)}' are of type '{check_type}'.",
)
except Exception as e:
return ({}, False, f'{e}',)
return (
{},
False,
f"{e}",
)
def checkId(basePath: str, file: pathlib.Path) -> bool | str:
@ -292,17 +348,26 @@ def checkId(basePath: str, file: pathlib.Path) -> bool | str:
:return: False if no id found, otherwise the id.
"""
match = re.search(r'(?<=\[)(?:youtube-)?(?P<id>[a-zA-Z0-9\-_]{11})(?=\])', file.stem, re.IGNORECASE)
match = re.search(r"(?<=\[)(?:youtube-)?(?P<id>[a-zA-Z0-9\-_]{11})(?=\])", file.stem, re.IGNORECASE)
if not match:
return False
id = match.groupdict().get('id')
id = match.groupdict().get("id")
for f in file.parent.iterdir():
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()
@ -314,7 +379,7 @@ def get_value(value):
return value() if callable(value) else value
def ag(array: dict | list, path: list[str | int] | str | int, default: Any = None, separator: str = '.') -> Any:
def ag(array: dict | list, path: list[str | int] | str | int, default: Any = None, separator: str = ".") -> Any:
"""
dict/array getter: Retrieve a value from a nested dict or object using a path.
@ -328,7 +393,7 @@ def ag(array: dict | list, path: list[str | int] | str | int, default: Any = Non
:return: The found value or the default if not found.
"""
if path is None or path == '':
if path is None or path == "":
return array
if not isinstance(array, dict):
@ -369,7 +434,7 @@ def is_private_address(hostname: str) -> bool:
try:
ip = socket.gethostbyname(hostname)
ip_obj = ipaddress.ip_address(ip)
return (ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_reserved or ip_obj.is_link_local)
return ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_reserved or ip_obj.is_link_local
except socket.gaierror:
# Could not resolve - treat as invalid or restricted
return True
@ -393,6 +458,7 @@ def validate_url(url: str) -> bool:
try:
from yarl import URL
parsed_url = URL(url)
except ValueError:
raise ValueError("Invalid URL.")

59
app/library/Webhooks.py
View file

@ -5,12 +5,17 @@ from .ItemDTO import ItemDTO
import httpx
from .version import APP_VERSION
LOG = logging.getLogger('Webhooks')
LOG = logging.getLogger("Webhooks")
class Webhooks:
targets: list[dict] = []
allowed_events: tuple = ('added', 'completed', 'error', 'not_live',)
allowed_events: tuple = (
"added",
"completed",
"error",
"not_live",
)
def __init__(self, file: str):
if os.path.exists(file):
@ -19,13 +24,14 @@ class Webhooks:
def load(self, file: str):
try:
if os.path.getsize(file) < 3:
raise Exception(f'file is empty.')
raise Exception("file is empty.")
LOG.info(f"Loading webhooks from '{file}'.")
from .Utils import load_file
(target, status, error) = load_file(file, list)
if not status:
raise Exception(f'{error}')
raise Exception(f"{error}")
self.targets = target
except Exception as e:
@ -42,7 +48,7 @@ class Webhooks:
tasks = []
for target in self.targets:
allowed_events = target.get('on') if 'on' in target else []
allowed_events = target.get("on") if "on" in target else []
if len(allowed_events) > 0 and event not in allowed_events:
continue
@ -52,41 +58,36 @@ class Webhooks:
async def __send(self, event: str, target: dict, item: ItemDTO) -> dict:
from .config import Config
req: dict = target.get('request')
req: dict = target.get("request")
try:
LOG.info(f"Sending event '{event}' id '{item.id}' to '{target.get('name')}'.")
async with httpx.AsyncClient() as client:
request_type = req.get('type', 'json')
request_type = req.get("type", "json")
reqBody = {
'method': req.get('method', 'POST'),
'url': req.get('url'),
'headers': {
'User-Agent': f"YTPTube/{APP_VERSION}"
},
"method": req.get("method", "POST"),
"url": req.get("url"),
"headers": {"User-Agent": f"YTPTube/{APP_VERSION}"},
}
if req.get('headers', None):
reqBody['headers'].update(req.get('headers'))
if req.get("headers", None):
reqBody["headers"].update(req.get("headers"))
match(request_type):
case 'json':
reqBody['json'] = item.__dict__
reqBody['headers']['Content-Type'] = 'application/json'
match request_type:
case "json":
reqBody["json"] = item.__dict__
reqBody["headers"]["Content-Type"] = "application/json"
case _:
reqBody['data'] = item.__dict__
reqBody['headers']['Content-Type'] = 'application/x-www-form-urlencoded'
reqBody["data"] = item.__dict__
reqBody["headers"]["Content-Type"] = "application/x-www-form-urlencoded"
response = await client.request(**reqBody)
respData = {
'url': req.get('url'),
'status': response.status_code,
'text': response.text
}
respData = {"url": req.get("url"), "status": response.status_code, "text": response.text}
msg = f"Webhook target '{target.get('name')}' Responded to event '{event}' id '{item.id}' with status '{response.status_code}'."
if Config.get_instance().debug and respData.get('text'):
if Config.get_instance().debug and respData.get("text"):
msg += f" body '{respData.get('text','??')}'."
LOG.info(msg)
@ -94,11 +95,7 @@ class Webhooks:
return respData
except Exception as e:
LOG.error(f"Error sending event '{event}' id '{item.id}' to '{target.get('name')}'. '{e}'")
return {
'url': req.get('url'),
'status': 500,
'text': str(e)
}
return {"url": req.get("url"), "status": 500, "text": str(e)}
def emit(self, event, data, **kwargs):
if len(self.targets) < 1 or event not in self.allowed_events:

15
app/library/common.py
View file

@ -1,14 +1,16 @@
import logging
from .DownloadQueue import DownloadQueue
from .encoder import Encoder
LOG = logging.getLogger('common')
LOG = logging.getLogger("common")
class common():
class common:
"""
This class is used to share common methods between the socket and the API gateways.
"""
queue: DownloadQueue = None
encoder: Encoder = None
@ -17,8 +19,9 @@ class common():
self.queue = queue
self.encoder = encoder
async def add(self, url: str, preset: str, folder: str, ytdlp_cookies: str, ytdlp_config: dict,
output_template: str) -> dict[str, str]:
async def add(
self, url: str, preset: str, folder: str, ytdlp_cookies: str, ytdlp_config: dict, output_template: str
) -> dict[str, str]:
if ytdlp_config is None:
ytdlp_config = {}
@ -27,11 +30,11 @@ class common():
status = await self.queue.add(
url=url,
preset=preset if preset else 'default',
preset=preset if preset else "default",
folder=folder,
ytdlp_cookies=ytdlp_cookies,
ytdlp_config=ytdlp_config,
output_template=output_template
output_template=output_template,
)
return status

213
app/library/config.py
View file

@ -3,41 +3,45 @@ import os
import re
import sys
import time
from pathlib import Path
import coloredlogs
from .version import APP_VERSION
from dotenv import load_dotenv
from yt_dlp.version import __version__ as YTDLP_VERSION
from .Utils import load_file
from pathlib import Path
from .version import APP_VERSION
class Config:
__instance = None
config_path: str = '.'
download_path: str = '.'
temp_path: str = '/tmp'
config_path: str = "."
download_path: str = "."
temp_path: str = "/tmp"
temp_keep: bool = False
db_file: str = '{config_path}/ytptube.db'
manual_archive: str = '{config_path}/archive.manual.log'
db_file: str = "{config_path}/ytptube.db"
manual_archive: str = "{config_path}/archive.manual.log"
url_host: str = ''
url_prefix: str = ''
url_socketio: str = '{url_prefix}socket.io'
url_host: str = ""
url_prefix: str = ""
url_socketio: str = "{url_prefix}socket.io"
output_template: str = '%(title)s.%(ext)s'
output_template_chapter: str = '%(title)s - %(section_number)s %(section_title)s.%(ext)s'
output_template: str = "%(title)s.%(ext)s"
output_template_chapter: str = "%(title)s - %(section_number)s %(section_title)s.%(ext)s"
ytdl_options: dict | str = {}
ytdl_debug: bool = False
host: str = '0.0.0.0'
host: str = "0.0.0.0"
port: int = 8081
keep_archive: bool = True
base_path: str = ''
base_path: str = ""
log_level: str = 'info'
log_level: str = "info"
access_log: bool = True
allow_manifestless: bool = False
@ -57,8 +61,8 @@ class Config:
started: int = 0
streamer_vcodec = 'libx264'
streamer_acodec = 'aac'
streamer_vcodec = "libx264"
streamer_acodec = "aac"
auth_username: str = None
auth_password: str = None
@ -66,73 +70,97 @@ class Config:
ytdlp_version: str = YTDLP_VERSION
tasks: list = []
presets: list = [
{'name': 'default', 'format': 'default', 'postprocessors': [],
'args': {}},
{'name': 'Video - 1080p h264/acc', 'format': 'bv[height<=1080][ext=mp4]+ba[ext=m4a]/b[ext=mp4]/b[ext=webm]',
'args': {'format_sort': ['vcodec:h264'], }, },
{'name': 'Video - 720p h264/acc', 'format': 'bv[height<=720][ext=mp4]+ba[ext=m4a]/b[ext=mp4]/b[ext=webm]',
'args': {'format_sort': ['vcodec:h264'], }, },
{"name": "default", "format": "default", "postprocessors": [], "args": {}},
{
'name': 'Audio - Best audio [MP3]',
'format': 'bestaudio/best',
'postprocessors': [
"name": "Video - 1080p h264/acc",
"format": "bv[height<=1080][ext=mp4]+ba[ext=m4a]/b[ext=mp4]/b[ext=webm]",
"args": {
"format_sort": ["vcodec:h264"],
},
},
{
"name": "Video - 720p h264/acc",
"format": "bv[height<=720][ext=mp4]+ba[ext=m4a]/b[ext=mp4]/b[ext=webm]",
"args": {
"format_sort": ["vcodec:h264"],
},
},
{
"name": "Audio - Best audio [MP3]",
"format": "bestaudio/best",
"postprocessors": [
{
"key": "FFmpegExtractAudio",
"preferredcodec": "mp3",
"preferredquality": "5",
"nopostoverwrites": False
"nopostoverwrites": False,
},
{
"key": "FFmpegMetadata",
"add_chapters": True,
"add_metadata": True,
"add_infojson": "if_exists"
},
{
"key": "EmbedThumbnail",
"already_have_thumbnail": False
},
{
"key": "FFmpegConcat",
"only_multi_video": True,
"when": "playlist"
}
{"key": "FFmpegMetadata", "add_chapters": True, "add_metadata": True, "add_infojson": "if_exists"},
{"key": "EmbedThumbnail", "already_have_thumbnail": False},
{"key": "FFmpegConcat", "only_multi_video": True, "when": "playlist"},
],
'args': {
"outtmpl": {
"pl_thumbnail": ""
},
"args": {
"outtmpl": {"pl_thumbnail": ""},
"ignoreerrors": "only_download",
"retries": 10,
"fragment_retries": 10,
"writethumbnail": True,
"extract_flat": "discard_in_playlist",
"final_ext": "mp3",
}
}
},
},
]
_manual_vars: tuple = ('temp_path', 'config_path', 'download_path',)
_manual_vars: tuple = (
"temp_path",
"config_path",
"download_path",
)
_immutable: tuple = (
'version', '__instance', 'ytdl_options', 'tasks',
'new_version_available', 'ytdlp_version', 'started', 'presets',
"version",
"__instance",
"ytdl_options",
"tasks",
"new_version_available",
"ytdlp_version",
"started",
"presets",
)
_int_vars: tuple = ('port', 'max_workers', 'socket_timeout', 'extract_info_timeout', 'debugpy_port',)
_boolean_vars: tuple = ('keep_archive', 'ytdl_debug', 'debug', 'temp_keep', 'allow_manifestless',)
_int_vars: tuple = (
"port",
"max_workers",
"socket_timeout",
"extract_info_timeout",
"debugpy_port",
)
_boolean_vars: tuple = (
"keep_archive",
"ytdl_debug",
"debug",
"temp_keep",
"allow_manifestless",
"access_log",
)
_frontend_vars: tuple = (
'download_path', 'keep_archive', 'output_template',
'ytdlp_version', 'version', 'url_host', 'started', 'url_prefix',
"download_path",
"keep_archive",
"output_template",
"ytdlp_version",
"version",
"url_host",
"started",
"url_prefix",
)
@ staticmethod
@staticmethod
def get_instance():
""" Static access method. """
"""Static access method."""
return Config() if not Config.__instance else Config.__instance
def __init__(self):
""" Virtually private constructor. """
"""Virtually private constructor."""
if Config.__instance is not None:
raise Exception("This class is a singleton. Use Config.get_instance() instead.")
else:
@ -140,20 +168,20 @@ class Config:
baseDefaultPath: str = str(Path(__file__).parent.parent.parent.absolute())
self.temp_path = os.environ.get('YTP_TEMP_PATH', None) or os.path.join(baseDefaultPath, 'var', 'tmp')
self.config_path = os.environ.get('YTP_CONFIG_PATH', None) or os.path.join(baseDefaultPath, 'var', 'config')
self.download_path = os.environ.get(
'YTP_DOWNLOAD_PATH', None) or os.path.join(
baseDefaultPath, 'var', 'downloads')
self.temp_path = os.environ.get("YTP_TEMP_PATH", None) or os.path.join(baseDefaultPath, "var", "tmp")
self.config_path = os.environ.get("YTP_CONFIG_PATH", None) or os.path.join(baseDefaultPath, "var", "config")
self.download_path = os.environ.get("YTP_DOWNLOAD_PATH", None) or os.path.join(
baseDefaultPath, "var", "downloads"
)
envFile: str = os.path.join(self.config_path, '.env')
envFile: str = os.path.join(self.config_path, ".env")
if os.path.exists(envFile):
logging.info(f"Loading environment variables from '{envFile}'.")
load_dotenv(envFile)
for k, v in self._getAttributes().items():
if k.startswith('_') or k in self._manual_vars:
if k.startswith("_") or k in self._manual_vars:
continue
# If the variable declared as immutable, set it to the default value.
@ -161,15 +189,15 @@ class Config:
setattr(self, k, v)
continue
lookUpKey: str = f'YTP_{k}'.upper()
lookUpKey: str = f"YTP_{k}".upper()
setattr(self, k, os.environ[lookUpKey] if lookUpKey in os.environ else v)
for k, v in self.__dict__.items():
if k.startswith('_') or k in self._immutable or k in self._manual_vars:
if k.startswith("_") or k in self._immutable or k in self._manual_vars:
continue
if isinstance(v, str) and '{' in v and '}' in v:
for key in re.findall(r'\{.*?\}', v):
if isinstance(v, str) and "{" in v and "}" in v:
for key in re.findall(r"\{.*?\}", v):
localKey: str = key[1:-1]
if localKey not in self.__dict__:
logging.error(f"Config variable '{k}' had non-existing config reference '{key}'.")
@ -180,32 +208,31 @@ class Config:
setattr(self, k, v)
if k in self._boolean_vars:
if str(v).lower() not in (True, False, 'true', 'false', 'on', 'off', '1', '0'):
if str(v).lower() not in (True, False, "true", "false", "on", "off", "1", "0"):
raise ValueError(f"Config variable '{k}' is set to a non-boolean value '{v}'.")
setattr(self, k, str(v).lower() in (True, 'true', 'on', '1'))
setattr(self, k, str(v).lower() in (True, "true", "on", "1"))
if k in self._int_vars:
setattr(self, k, int(v))
if not self.url_prefix.endswith('/'):
self.url_prefix += '/'
if not self.url_prefix.endswith("/"):
self.url_prefix += "/"
numeric_level = getattr(logging, self.log_level.upper(), None)
if not isinstance(numeric_level, int):
raise ValueError(f"Invalid log level '{self.log_level}' specified.")
coloredlogs.install(
level=numeric_level,
fmt="%(asctime)s [%(name)s] [%(levelname)-5.5s] %(message)s",
datefmt='%H:%M:%S'
level=numeric_level, fmt="%(asctime)s [%(name)s] [%(levelname)-5.5s] %(message)s", datefmt="%H:%M:%S"
)
LOG = logging.getLogger('config')
LOG = logging.getLogger("config")
if self.debug:
try:
import debugpy
debugpy.listen(("0.0.0.0", self.debugpy_port))
LOG.info(f"starting debugpy server on '0.0.0.0:{self.debugpy_port}'.")
except ImportError:
@ -213,7 +240,7 @@ class Config:
except Exception as e:
LOG.error(f"Error starting debugpy server at '0.0.0.0:{self.debugpy_port}'. {e}")
optsFile: str = os.path.join(self.config_path, 'ytdlp.json')
optsFile: str = os.path.join(self.config_path, "ytdlp.json")
if os.path.exists(optsFile) and os.path.getsize(optsFile) > 4:
LOG.info(f"Loading yt-dlp custom options from '{optsFile}'.")
@ -226,7 +253,7 @@ class Config:
else:
LOG.info(f"No yt-dlp custom options found at '{optsFile}'.")
tasksFile = os.path.join(self.config_path, 'tasks.json')
tasksFile = os.path.join(self.config_path, "tasks.json")
if os.path.exists(tasksFile) and os.path.getsize(tasksFile) > 0:
LOG.info(f"Loading tasks from '{tasksFile}'.")
try:
@ -235,10 +262,10 @@ class Config:
LOG.error(f"Could not load tasks file from '{tasksFile}'. '{error}'.")
sys.exit(1)
self.tasks.extend(tasks)
except Exception as e:
except Exception:
pass
presetsFile = os.path.join(self.config_path, 'presets.json')
presetsFile = os.path.join(self.config_path, "presets.json")
if os.path.exists(presetsFile) and os.path.getsize(presetsFile) > 0:
LOG.info(f"Loading extra presets from '{presetsFile}'.")
try:
@ -248,32 +275,32 @@ class Config:
sys.exit(1)
for preset in presets:
if 'name' not in preset:
if "name" not in preset:
LOG.error(f"Missing 'name' key in preset '{preset}'.")
continue
if 'format' not in preset:
if "format" not in preset:
LOG.error(f"Missing 'format' key in preset '{preset}'.")
continue
if 'args' not in preset:
preset['args'] = {}
if "args" not in preset:
preset["args"] = {}
if 'postprocessors' not in preset:
preset['postprocessors'] = []
if "postprocessors" not in preset:
preset["postprocessors"] = []
self.presets.append(preset)
except Exception as e:
except Exception:
pass
self.ytdl_options['socket_timeout'] = self.socket_timeout
self.ytdl_options["socket_timeout"] = self.socket_timeout
if self.keep_archive:
LOG.info(f"keep archive option is enabled.")
self.ytdl_options['download_archive'] = os.path.join(self.config_path, 'archive.log')
LOG.info("keep archive option is enabled.")
self.ytdl_options["download_archive"] = os.path.join(self.config_path, "archive.log")
if self.temp_keep:
LOG.info(f'Keep temp files option is enabled.')
LOG.info("Keep temp files option is enabled.")
if self.auth_password and self.auth_username:
LOG.warning(f"Basic authentication enabled with username '{self.auth_username}'.")
@ -285,7 +312,7 @@ class Config:
vClass: str = self.__class__
for attribute in vClass.__dict__.keys():
if attribute.startswith('_'):
if attribute.startswith("_"):
continue
value = getattr(vClass, attribute)

2
app/library/encoder.py
View file

@ -9,7 +9,7 @@ class Encoder(json.JSONEncoder):
"""
def default(self, o):
if isinstance(o, object) and hasattr(o, '__dict__'):
if isinstance(o, object) and hasattr(o, "__dict__"):
return o.__dict__
return json.JSONEncoder.default(self, o)

84
app/library/ffprobe.py
View file

@ -1,6 +1,7 @@
"""
Python wrapper for ffprobe command line tool. ffprobe must exist in the path.
"""
import asyncio
import functools
import json
@ -22,25 +23,26 @@ class FFStream:
setattr(self, key, val)
try:
self.__dict__['framerate'] = round(
functools.reduce(
operator.truediv, map(int, self.__dict__.get('avg_frame_rate', '').split('/'))
)
self.__dict__["framerate"] = round(
functools.reduce(operator.truediv, map(int, self.__dict__.get("avg_frame_rate", "").split("/")))
)
except ValueError:
self.__dict__['framerate'] = None
self.__dict__["framerate"] = None
except ZeroDivisionError:
self.__dict__['framerate'] = 0
self.__dict__["framerate"] = 0
def __repr__(self):
if not 'codec_long_name' in self.__dict__:
self.codec_long_name = self.__dict__.get('codec_name', '')
if "codec_long_name" not in self.__dict__:
self.codec_long_name = self.__dict__.get("codec_name", "")
if self.is_video():
return f"<Stream: #{self.index} [{self.codec_type}] {self.codec_long_name}, {self.framerate}, ({self.width}x{self.height})>"
if self.is_audio():
return f"<Stream: #{self.index} [{self.codec_type}] {self.codec_long_name}, channels: {self.channels} ({self.channel_layout}), " "{sample_rate}Hz> "
return (
f"<Stream: #{self.index} [{self.codec_type}] {self.codec_long_name}, channels: {self.channels} ({self.channel_layout}), "
"{sample_rate}Hz> "
)
if self.is_subtitle() or self.is_attachment():
return f"<Stream: #{self.index} [{self.codec_type}] {self.codec_long_name}>"
@ -51,25 +53,25 @@ class FFStream:
"""
Is this stream labelled as an audio stream?
"""
return self.__dict__.get('codec_type', None) == 'audio'
return self.__dict__.get("codec_type", None) == "audio"
def is_video(self):
"""
Is the stream labelled as a video stream.
"""
return self.__dict__.get('codec_type', None) == 'video'
return self.__dict__.get("codec_type", None) == "video"
def is_subtitle(self):
"""
Is the stream labelled as a subtitle stream.
"""
return self.__dict__.get('codec_type', None) == 'subtitle'
return self.__dict__.get("codec_type", None) == "subtitle"
def is_attachment(self):
"""
Is the stream labelled as a attachment stream.
"""
return self.__dict__.get('codec_type', None) == 'attachment'
return self.__dict__.get("codec_type", None) == "attachment"
def frame_size(self):
"""
@ -78,8 +80,8 @@ class FFStream:
"""
size = None
if self.is_video():
width = self.__dict__['width']
height = self.__dict__['height']
width = self.__dict__["width"]
height = self.__dict__["height"]
if width and height:
try:
@ -96,7 +98,7 @@ class FFStream:
Returns a string representing the pixel format of the video stream. e.g. yuv420p.
Returns none is it is not a video stream.
"""
return self.__dict__.get('pix_fmt', None)
return self.__dict__.get("pix_fmt", None)
def frames(self):
"""
@ -104,9 +106,9 @@ class FFStream:
"""
if self.is_video() or self.is_audio():
try:
frame_count = int(self.__dict__.get('nb_frames', ''))
frame_count = int(self.__dict__.get("nb_frames", ""))
except ValueError:
raise FFProbeError('None integer frame count')
raise FFProbeError("None integer frame count")
else:
frame_count = 0
@ -119,9 +121,9 @@ class FFStream:
"""
if self.is_video() or self.is_audio():
try:
duration = float(self.__dict__.get('duration', ''))
duration = float(self.__dict__.get("duration", ""))
except ValueError:
raise FFProbeError('None numeric duration')
raise FFProbeError("None numeric duration")
else:
duration = 0.0
@ -131,34 +133,34 @@ class FFStream:
"""
Returns language tag of stream. e.g. eng
"""
return self.__dict__.get('TAG:language', None)
return self.__dict__.get("TAG:language", None)
def codec(self):
"""
Returns a string representation of the stream codec.
"""
return self.__dict__.get('codec_name', None)
return self.__dict__.get("codec_name", None)
def codec_description(self):
"""
Returns a long representation of the stream codec.
"""
return self.__dict__.get('codec_long_name', None)
return self.__dict__.get("codec_long_name", None)
def codec_tag(self):
"""
Returns a short representative tag of the stream codec.
"""
return self.__dict__.get('codec_tag_string', None)
return self.__dict__.get("codec_tag_string", None)
def bit_rate(self):
"""
Returns bit_rate as an integer in bps
"""
try:
return int(self.__dict__.get('bit_rate', ''))
return int(self.__dict__.get("bit_rate", ""))
except ValueError:
raise FFProbeError('None integer bit_rate')
raise FFProbeError("None integer bit_rate")
class FFProbe:
@ -166,38 +168,40 @@ class FFProbe:
FFProbe wraps the ffprobe command and pulls the data into an object form::
metadata=FFProbe('multimedia-file.mov')
"""
audio: list[FFStream] = []
attachment: list[FFStream] = []
streams: list[FFStream] = []
subtitle: list[FFStream] = []
video: list[FFStream] = []
metadata: dict = {}
path_to_video: str = ''
path_to_video: str = ""
def __init__(self, path_to_video):
self.path_to_video = path_to_video
async def run(self):
try:
with open(os.devnull, 'w') as tempf:
await asyncio.create_subprocess_exec(
"ffprobe", "-h", stdout=tempf, stderr=tempf
)
with open(os.devnull, "w") as tempf:
await asyncio.create_subprocess_exec("ffprobe", "-h", stdout=tempf, stderr=tempf)
except FileNotFoundError:
raise IOError('ffprobe not found.')
raise IOError("ffprobe not found.")
if not os.path.isfile(self.path_to_video):
raise IOError(f"No such media file '{self.path_to_video}'.")
args = [
'-v', 'quiet',
'-of', 'json',
'-show_streams',
'-show_format',
"-v",
"quiet",
"-of",
"json",
"-show_streams",
"-show_format",
self.path_to_video,
]
p = await asyncio.create_subprocess_exec(
'ffprobe', *args,
"ffprobe",
*args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
@ -206,7 +210,7 @@ class FFProbe:
data, err = await p.communicate()
if 0 == exitCode:
parsed: dict = json.loads(data.decode('utf-8'))
parsed: dict = json.loads(data.decode("utf-8"))
else:
raise FFProbeError(f"FFProbe return with non-0 exit code. '{err.decode('utf-8')}'")
@ -217,10 +221,10 @@ class FFProbe:
self.subtitle = []
self.attachment = []
for stream in parsed.get('streams', []):
for stream in parsed.get("streams", []):
self.streams.append(FFStream(stream))
self.metadata = parsed.get('format', {})
self.metadata = parsed.get("format", {})
for stream in self.streams:
if stream.is_audio():

1
app/library/version.py
View file

@ -2,5 +2,6 @@
Version of the application,
This file is updated by the CI/CD pipeline, do not edit it manually.
"""
APP_VERSION = "dev-master"
"Application version"

57
app/main.py
View file

@ -1,33 +1,34 @@
#!/usr/bin/env python3
import asyncio
import logging
import os
import random
import logging
import caribou
import sqlite3
import magic
from datetime import datetime
from aiohttp import web
from pathlib import Path
from library.Emitter import Emitter
from library.config import Config
from library.encoder import Encoder
from library.DownloadQueue import DownloadQueue
from library.Webhooks import Webhooks
from library.HttpSocket import HttpSocket
from library.HttpAPI import HttpAPI
import caribou
import magic
from aiocron import crontab
from aiohttp import web
from library.config import Config
from library.DownloadQueue import DownloadQueue
from library.Emitter import Emitter
from library.encoder import Encoder
from library.HttpAPI import HttpAPI, LOG as http_logger
from library.HttpSocket import HttpSocket
from library.Webhooks import Webhooks
LOG = logging.getLogger("app")
MIME = magic.Magic(mime=True)
class main:
config: Config = None
app: web.Application = None
http: HttpAPI = None
socket: HttpSocket = None
class Main:
config: Config
app: web.Application
http: HttpAPI
socket: HttpSocket
def __init__(self):
self.config = Config.get_instance()
@ -61,9 +62,7 @@ class main:
LOG.info(f"Creating download folder at '{self.config.download_path}'.")
os.makedirs(self.config.download_path, exist_ok=True)
except OSError as e:
LOG.error(
f"Could not create download folder at '{self.config.download_path}'."
)
LOG.error(f"Could not create download folder at '{self.config.download_path}'.")
raise e
try:
@ -111,9 +110,7 @@ class main:
LOG.info(f"Completed 'Task: {taskName}' at '{timeNow}'.")
except Exception as e:
timeNow = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
LOG.error(
f"Failed 'Task: {taskName}' at '{timeNow}'. Error message '{str(e)}'."
)
LOG.error(f"Failed 'Task: {taskName}' at '{timeNow}'. Error message '{str(e)}'.")
def load_tasks(self):
for task in self.config.tasks:
@ -131,9 +128,7 @@ class main:
loop=asyncio.get_event_loop(),
)
LOG.info(
f"Added 'Task: {task.get('name', task.get('url'))}' to be executed every '{cron_timer}'."
)
LOG.info(f"Added 'Task: {task.get('name', task.get('url'))}' to be executed every '{cron_timer}'.")
def start(self):
self.socket.attach(self.app)
@ -141,18 +136,22 @@ class main:
self.load_tasks()
start: str = f"YTPTube v{self.config.version} - started on http://{self.config.host}:{self.config.port}"
def started(_):
LOG.info("=" * 40)
LOG.info(f"YTPTube v{self.config.version} - started on http://{self.config.host}:{self.config.port}")
LOG.info("=" * 40)
web.run_app(
self.app,
host=self.config.host,
port=self.config.port,
reuse_port=True,
loop=asyncio.get_event_loop(),
access_log=None,
print=lambda _: LOG.info(start),
access_log=http_logger if self.config.access_log else None,
print=started,
)
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
main().start()
Main().start()

6
entrypoint.sh
View file

@ -1,15 +1,15 @@
#!/usr/bin/env bash
set -e
echo "Setting umask to ${UMASK}, to change it set the UMASK environment variable"
echo "Setting umask to '${UMASK}', to change it set the UMASK environment variable"
umask ${UMASK}
echo_err() { cat <<< "$@" 1>&2; }
echo_err() { cat <<<"$@" 1>&2; }
if [ ! -w "${YTP_CONFIG_PATH}" ]; then
CH_USER=$(stat -c "%u" "${YTP_CONFIG_PATH}")
CH_GRP=$(stat -c "%g" "${YTP_CONFIG_PATH}")
echo_err "ERROR: Unable to write to [${YTP_CONFIG_PATH}] data directory. Current user id [${UID}] while directory owner is [${CH_USER}]"
echo_err "ERROR: Unable to write to '${YTP_CONFIG_PATH}' data directory. Current user id '${UID}' while directory owner is '${CH_USER}'."
echo_err "[Running under docker]"
echo_err "change docker-compose.yaml user: to user:\"${CH_USER}:${CH_GRP}\""
echo_err "Run the following command to change the directory ownership"

4
healthcheck.sh
View file

@ -2,4 +2,6 @@
set -eu pipefail
/usr/bin/curl -f http://localhost:${YTP_PORT:-8081}/ping
LOCAL_PORT="${YTP_PORT:-8081}"
/usr/bin/curl -f "http://localhost:${LOCAL_PORT}/ping"

78
pyproject.toml
View file

@ -1,2 +1,80 @@
[tool.ruff]
# Exclude a variety of commonly ignored directories.
exclude = [
".bzr",
".direnv",
".eggs",
".git",
".git-rewrite",
".hg",
".mypy_cache",
".nox",
".pants.d",
".pyenv",
".pytest_cache",
".pytype",
".ruff_cache",
".svn",
".tox",
".venv",
".vscode",
"__pypackages__",
"_build",
"buck-out",
"build",
"dist",
"node_modules",
"site-packages",
"venv",
]
# Same as Black.
line-length = 120
indent-width = 4
# Assume Python 3.11
target-version = "py311"
[tool.ruff.lint]
# Enable Pyflakes (`F`) and a subset of the pycodestyle (`E`) codes by default.
# Unlike Flake8, Ruff doesn't enable pycodestyle warnings (`W`) or
# McCabe complexity (`C901`) by default.
select = ["E4", "E7", "E9", "F", "G", "W"]
ignore = ["PTH", "G0", "G1", "G201"]
# Allow fix for all enabled rules (when `--fix`) is provided.
fixable = ["ALL"]
unfixable = []
# Allow unused variables when underscore-prefixed.
dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
[tool.ruff.format]
# Like Black, use double quotes for strings.
quote-style = "double"
# Like Black, indent with spaces, rather than tabs.
indent-style = "space"
# Like Black, respect magic trailing commas.
skip-magic-trailing-comma = false
# Like Black, automatically detect the appropriate line ending.
line-ending = "auto"
# Enable auto-formatting of code examples in docstrings. Markdown,
# reStructuredText code/literal blocks and doctests are all supported.
#
# This is currently disabled by default, but it is planned for this
# to be opt-out in the future.
docstring-code-format = false
# Set the line length limit used when formatting code snippets in
# docstrings.
#
# This only has an effect when the `docstring-code-format` setting is
# enabled.
docstring-code-line-length = "dynamic"
[tool.autopep8]
--max-line-length = 120

2
ui/components/History.vue
View file

@ -273,7 +273,7 @@ const socket = useSocketStore()
const selectedElms = ref([]);
const masterSelectAll = ref(false);
const showCompleted = useStorage('showCompleted', true)
const hideThumbnail = useStorage('hideThumbnail', false)
const hideThumbnail = useStorage('hideThumbnailHistory', false)
const direction = useStorage('sortCompleted', 'desc')
const video_link = ref('')

2
ui/components/Queue.vue
View file

@ -154,7 +154,7 @@ const socket = useSocketStore();
const selectedElms = ref([]);
const masterSelectAll = ref(false);
const showQueue = useStorage('showQueue', true)
const hideThumbnail = useStorage('hideThumbnail', false)
const hideThumbnail = useStorage('hideThumbnailQueue', false)
watch(masterSelectAll, (value) => {
for (const key in stateStore.queue) {