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

updating python code to be little more clean and implements more type checks.

Browse source
This commit is contained in:
ArabCoders 2024-12-20 21:20:51 +03:00
parent 1b9a4e528b
commit 2d70de2032
23 changed files with 970 additions and 783 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

24
app/library/DownloadQueue.py
View file

@ -1,18 +1,20 @@
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 calcDownloadPath, ExtractInfo, isDownloaded, mergeConfig
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"
@ -20,12 +22,8 @@ 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()

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', r'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

9
app/library/config.py
View file

@ -3,12 +3,14 @@ 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:
@ -39,6 +41,8 @@ class Config:
log_level: str = "info"
access_log: bool = True
allow_manifestless: bool = False
max_workers: int = 1
@ -136,6 +140,7 @@ class Config:
"debug",
"temp_keep",
"allow_manifestless",
"access_log",
)
_frontend_vars: tuple = (

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"

12
app/main.py
View file

@ -16,7 +16,7 @@ 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
from library.HttpAPI import HttpAPI, LOG as http_logger
from library.HttpSocket import HttpSocket
from library.Webhooks import Webhooks
@ -136,15 +136,19 @@ 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,
)

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"