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

mostly code cleanup

Browse source
This commit is contained in:
ArabCoders 2024-02-02 18:07:24 +03:00
parent b102267983
commit 0cd3e15f34
13 changed files with 268 additions and 373 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -18,6 +18,7 @@
!.vscode/tasks.json !.vscode/tasks.json
!.vscode/launch.json !.vscode/launch.json
!.vscode/extensions.json !.vscode/extensions.json
!.vscode/settings.json
.history/* .history/*
# Keep Var dir # Keep Var dir

34
.vscode/settings.json vendored Normal file
View file

@ -0,0 +1,34 @@
{
"files.exclude": {
"**/__pycache__": true
},
"autopep8.args": [
"--max-line-length",
"120",
"--experimental"
],
"editor.rulers": [
120
],
"cSpell.words": [
"aconvert",
"aiocron",
"copyts",
"dotenv",
"finaldir",
"getpid",
"libcurl",
"libx",
"mpegts",
"muxdelay",
"nodesc",
"noprogress",
"postprocessor",
"preferredcodec",
"preferredquality",
"tmpfilename",
"vconvert",
"writedescription",
"xerror"
],
}

2
app/AsyncPool.py
View file

@ -103,7 +103,7 @@ class AsyncPool:
""" """
return self._active_workers < self._num_workers return self._active_workers < self._num_workers
def get_avalialbe_workers(self) -> int: def get_available_workers(self) -> int:
""" """
:return: number of available workers. :return: number of available workers.
""" """

68
app/Config.py
View file

@ -41,12 +41,11 @@ class Config:
version: str = APP_VERSION version: str = APP_VERSION
_boolean_vars: tuple = ( new_version_available: bool = False
'keep_archive', 'ytdl_debug',
'temp_keep', 'allow_manifestless',
)
_int_vars: tuple = ('port', 'max_workers',) _int_vars: tuple = ('port', 'max_workers',)
_immutable: tuple = ('version', '__instance', 'ytdl_options',) _immutable: tuple = ('version', '__instance', 'ytdl_options', 'new_version_available')
_boolean_vars: tuple = ('keep_archive', 'ytdl_debug', 'temp_keep', 'allow_manifestless',)
@staticmethod @staticmethod
def get_instance(): def get_instance():
@ -56,19 +55,15 @@ class Config:
def __init__(self): def __init__(self):
""" Virtually private constructor. """ """ Virtually private constructor. """
if Config.__instance is not None: if Config.__instance is not None:
raise Exception( raise Exception("This class is a singleton!. Use Config.getInstance() instead.")
"This class is a singleton!. Use Config.getInstance() instead.")
else: else:
Config.__instance = self Config.__instance = self
baseDefualtPath: str = os.path.dirname(__file__) baseDefaultPath: str = os.path.dirname(__file__)
self.config_path = os.environ.get('YTP_CONFIG_PATH', None) or os.path.join( self.temp_path = os.environ.get('YTP_TEMP_PATH', None) or os.path.join(baseDefaultPath, 'var', 'tmp')
baseDefualtPath, 'var', 'config') self.config_path = os.environ.get('YTP_CONFIG_PATH', None) or os.path.join(baseDefaultPath, 'var', 'config')
self.download_path = os.environ.get('YTP_DOWNLOAD_PATH', None) or os.path.join( self.download_path = os.environ.get('YTP_DOWNLOAD_PATH', None) or os.path.join(baseDefaultPath, 'var', 'downloads')
baseDefualtPath, 'var', 'downloads')
self.temp_path = os.environ.get('YTP_TEMP_PATH', None) or os.path.join(
baseDefualtPath, 'var', 'tmp')
envFile: str = os.path.join(self.config_path, '.env') envFile: str = os.path.join(self.config_path, '.env')
@ -80,16 +75,13 @@ class Config:
if k.startswith('_'): if k.startswith('_'):
continue continue
# If the variable is not updateable, set it to the default value. # If the variable declared as immutable, set it to the default value.
if k in self._immutable: if k in self._immutable:
setattr(self, k, v) setattr(self, k, v)
continue continue
lookUpKey: str = f'YTP_{k}'.upper() lookUpKey: str = f'YTP_{k}'.upper()
setattr( setattr(self, k, os.environ[lookUpKey] if lookUpKey in os.environ else v)
self, k,
os.environ[lookUpKey] if lookUpKey in os.environ else v
)
for k, v in self.__dict__.items(): for k, v in self.__dict__.items():
if k.startswith('_') or k in self._immutable: if k.startswith('_') or k in self._immutable:
@ -99,17 +91,16 @@ class Config:
for key in re.findall(r'\{.*?\}', v): for key in re.findall(r'\{.*?\}', v):
localKey: str = key[1:-1] localKey: str = key[1:-1]
if localKey not in self.__dict__: if localKey not in self.__dict__:
logging.error( logging.error(f'Config variable "{k}" had non existing config reference "{key}"')
f'Config variable "{k}" had non exisitng config reference "{key}"')
sys.exit(1) sys.exit(1)
v = v.replace(key, getattr(self, localKey)) v = v.replace(key, getattr(self, localKey))
setattr(self, k, v) setattr(self, k, v)
if k in self._boolean_vars: if k in self._boolean_vars:
if str(v).lower() not in (True, False, 'true', 'false', 'on', 'off', '1', '0'): if str(v).lower() not in (True, False, 'true', 'false', 'on', 'off', '1', '0'):
raise ValueError( raise ValueError(f'Config variable "{k}" is set to a non-boolean value "{v}".')
f'Config variable "{k}" is set to a non-boolean value "{v}".')
setattr(self, k, str(v).lower() in (True, 'true', 'on', '1')) setattr(self, k, str(v).lower() in (True, 'true', 'on', '1'))
@ -119,25 +110,21 @@ class Config:
if not self.url_prefix.endswith('/'): if not self.url_prefix.endswith('/'):
self.url_prefix += '/' self.url_prefix += '/'
numeric_level = getattr( numeric_level = getattr(logging, self.logging_level.upper(), None)
logging, self.logging_level.upper(), None)
if not isinstance(numeric_level, int): if not isinstance(numeric_level, int):
raise ValueError(f"Invalid log level: {self.logging_level}") raise ValueError(f"Invalid log level: {self.logging_level}")
logging.basicConfig( coloredlogs.install(
force=True,
level=numeric_level, level=numeric_level,
format="%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s] %(message)s" fmt="%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s] %(message)s",
datefmt='%H:%M:%S'
) )
log = logging.getLogger('config') LOG = logging.getLogger('config')
coloredlogs.install()
optsFile: str = os.path.join(self.config_path, 'ytdlp.json') optsFile: str = os.path.join(self.config_path, 'ytdlp.json')
if os.path.exists(optsFile) and os.path.getsize(optsFile) > 0: if os.path.exists(optsFile) and os.path.getsize(optsFile) > 0:
log.info(f'Loading yt-dlp custom options from "{optsFile}"') LOG.info(f'Loading yt-dlp custom options from "{optsFile}"')
try: try:
with open(optsFile) as json_data: with open(optsFile) as json_data:
@ -145,28 +132,27 @@ class Config:
assert isinstance(opts, dict) assert isinstance(opts, dict)
self.ytdl_options.update(opts) self.ytdl_options.update(opts)
except (json.decoder.JSONDecodeError, AssertionError) as e: except (json.decoder.JSONDecodeError, AssertionError) as e:
log.error( LOG.error(f'JSON error in "{optsFile}": {e}')
f'JSON error in "{optsFile}": {e}')
sys.exit(1) sys.exit(1)
else: else:
log.info(f'No custom yt-dlp options found in "{self.config_path}"') LOG.info(f'No custom yt-dlp options found in "{self.config_path}"')
if self.keep_archive: if self.keep_archive:
log.info(f'keep archive: {self.keep_archive}') LOG.info(f'keep archive: {self.keep_archive}')
self.ytdl_options['download_archive'] = os.path.join( self.ytdl_options['download_archive'] = os.path.join(
self.config_path, 'archive.log') self.config_path, 'archive.log')
log.info(f'Keep temp: {self.temp_keep}') LOG.info(f'Keep temp: {self.temp_keep}')
def _getAttributes(self) -> dict: def _getAttributes(self) -> dict:
attrs: dict = {} attrs: dict = {}
vclass: str = self.__class__ vClass: str = self.__class__
for attribute in vclass.__dict__.keys(): for attribute in vClass.__dict__.keys():
if attribute.startswith('_'): if attribute.startswith('_'):
continue continue
value = getattr(vclass, attribute) value = getattr(vClass, attribute)
if not callable(value): if not callable(value):
attrs[attribute] = value attrs[attribute] = value

72
app/DataStore.py
View file

@ -3,7 +3,7 @@ import copy
from datetime import datetime, timezone from datetime import datetime, timezone
from email.utils import formatdate from email.utils import formatdate
import json import json
import sqlite3 from sqlite3 import Connection
from Utils import calcDownloadPath from Utils import calcDownloadPath
from Config import Config from Config import Config
from Download import Download from Download import Download
@ -15,24 +15,22 @@ class DataStore:
Persistent queue. Persistent queue.
""" """
type: str = None type: str = None
db_file: str = None
dict: OrderedDict[str, Download] = None dict: OrderedDict[str, Download] = None
config: Config = None config: Config = None
def __init__(self, type: str, config: Config): connection: Connection
def __init__(self, type: str, connection: Connection):
self.dict = OrderedDict() self.dict = OrderedDict()
self.type = type self.type = type
self.config = config self.config = Config.get_instance()
self.db_file = self.config.db_file self.connection = connection
def load(self) -> None: def load(self) -> None:
for id, item in self.saved_items(): for id, item in self.saved_items():
self.dict.update({id: Download( self.dict.update({id: Download(
info=item, info=item,
download_dir=calcDownloadPath( download_dir=calcDownloadPath(basePath=self.config.download_path, folder=item.folder),
basePath=self.config.download_path,
folder=item.folder
),
temp_dir=self.config.temp_path, temp_dir=self.config.temp_path,
output_template_chapter=self.config.output_template_chapter, output_template_chapter=self.config.output_template_chapter,
default_ytdl_opts=self.config.ytdl_options) default_ytdl_opts=self.config.ytdl_options)
@ -40,7 +38,7 @@ class DataStore:
def exists(self, key: str = None, url: str = None) -> bool: def exists(self, key: str = None, url: str = None) -> bool:
if not key and not url: 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: if key and key in self.dict:
return True return True
@ -66,31 +64,24 @@ class DataStore:
def saved_items(self) -> list[tuple[str, ItemDTO]]: def saved_items(self) -> list[tuple[str, ItemDTO]]:
items: list[tuple[str, ItemDTO]] = [] items: list[tuple[str, ItemDTO]] = []
with sqlite3.connect(self.db_file) as db:
db.row_factory = sqlite3.Row
cursor = db.execute(
f'SELECT "id", "data", "created_at" FROM "history" WHERE "type" = ? ORDER BY "created_at" ASC',
(self.type,)
)
for row in cursor: cursor = self.connection.execute(
data: dict = json.loads(row['data']) f'SELECT "id", "data", "created_at" FROM "history" WHERE "type" = ? ORDER BY "created_at" ASC',
key: str = data.pop('_id') (self.type,)
item: ItemDTO = ItemDTO(**data) )
item._id = key
item.datetime = formatdate(datetime.strptime( for row in cursor:
row['created_at'], '%Y-%m-%d %H:%M:%S').replace(tzinfo=timezone.utc).timestamp() rowDate = datetime.strptime(row['created_at'], '%Y-%m-%d %H:%M:%S')
) data: dict = json.loads(row['data'])
items.append((row['id'], item)) 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))
return items return items
def put(self, value: Download) -> Download: def put(self, value: Download) -> Download:
# for key in self.dict:
# if self.dict[key].info.url == value.info.url:
# value.info._id = key
# return
self.dict.update({value.info._id: value}) self.dict.update({value.info._id: value})
self._updateStoreItem(self.type, value.info) self._updateStoreItem(self.type, value.info)
@ -144,21 +135,10 @@ class DataStore:
except AttributeError: except AttributeError:
pass pass
with sqlite3.connect(self.db_file) as db: self.connection.execute(sqlStatement.strip(), (
db.execute(sqlStatement.strip(), ( stored._id, type, stored.url, stored.json(), type, stored.url,
stored._id, stored.json(), datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')
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: def _deleteStoreItem(self, key: str) -> None:
with sqlite3.connect(self.db_file) as db: self.connection.execute('DELETE FROM "history" WHERE "type" = ? AND "id" = ?', (self.type, key,))
db.execute(
'DELETE FROM "history" WHERE "type" = ? AND "id" = ?',
(self.type, key,)
)

80
app/Download.py
View file

@ -11,7 +11,7 @@ from ItemDTO import ItemDTO
from Config import Config from Config import Config
import hashlib import hashlib
log = logging.getLogger('download') LOG = logging.getLogger('download')
class Download: class Download:
@ -52,7 +52,10 @@ class Download:
'speed', 'speed',
'eta', 'eta',
) )
"Fields to be extracted from yt-dlp progress hook."
tempKeep: bool = False tempKeep: bool = False
"Keep temp directory after download."
def __init__( def __init__(
self, self,
@ -68,8 +71,7 @@ class Download:
self.output_template_chapter = output_template_chapter self.output_template_chapter = output_template_chapter
self.output_template = info.output_template self.output_template = info.output_template
self.format = get_format(info.format, info.quality) self.format = get_format(info.format, info.quality)
self.ytdl_opts = get_opts( self.ytdl_opts = get_opts(info.format, info.quality, info.ytdlp_config if info.ytdlp_config else {})
info.format, info.quality, info.ytdlp_config if info.ytdlp_config else {})
self.info = info self.info = info
self.id = info._id self.id = info._id
self.default_ytdl_opts = default_ytdl_opts self.default_ytdl_opts = default_ytdl_opts
@ -81,26 +83,21 @@ class Download:
self.proc = None self.proc = None
self.loop = None self.loop = None
self.notifier = None self.notifier = None
config = Config.get_instance() self.max_workers = int(Config.get_instance().max_workers)
self.max_workers = int(config.max_workers) self.tempKeep = bool(Config.get_instance().temp_keep)
self.tempKeep = bool(config.temp_keep)
self.is_live = bool(info.is_live) or info.live_in is not None self.is_live = bool(info.is_live) or info.live_in is not None
self.is_manifestless = 'is_manifestless' in self.info.options and self.info.options[ self.is_manifestless = 'is_manifestless' in self.info.options and self.info.options['is_manifestless'] is True
'is_manifestless'] is True
def _progress_hook(self, data: dict): def _progress_hook(self, data: dict):
dicto = {k: v for k, v in data.items() if k in self._ytdlp_fields} dataDict = {k: v for k, v in data.items() if k in self._ytdlp_fields}
self.status_queue.put({'id': self.id, **dicto}) self.status_queue.put({'id': self.id, **dataDict})
def _postprocessor_hook(self, data: dict): 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 return
if '__finaldir' in data['info_dict']: if '__finaldir' in data['info_dict']:
filename = os.path.join( filename = os.path.join(data['info_dict']['__finaldir'], os.path.basename(data['info_dict']['filepath']))
data['info_dict']['__finaldir'],
os.path.basename(data['info_dict']['filepath'])
)
else: else:
filename = data['info_dict']['filepath'] filename = data['info_dict']['filepath']
@ -141,14 +138,14 @@ class Download:
try: try:
data = jsonCookie(json.loads(self.info.ytdlp_cookies)) data = jsonCookie(json.loads(self.info.ytdlp_cookies))
if not data: if not data:
log.warning( LOG.warning(
f'The cookie string that was provided for {self.info.title} is empty or not in expected spec.') f'The cookie string that was provided for {self.info.title} is empty or not in expected spec.')
with open(os.path.join(self.tempPath, f'cookie_{self.info._id}.txt'), 'w') as f: with open(os.path.join(self.tempPath, f'cookie_{self.info._id}.txt'), 'w') as f:
f.write(data) f.write(data)
params['cookiefile'] = f.name params['cookiefile'] = f.name
except ValueError as e: except ValueError as e:
log.error( LOG.error(
f'Invalid cookies: was provided for {self.info.title} - {str(e)}') f'Invalid cookies: was provided for {self.info.title} - {str(e)}')
if self.is_live or self.is_manifestless: if self.is_live or self.is_manifestless:
@ -161,18 +158,14 @@ class Download:
deletedOpts.append(opt) deletedOpts.append(opt)
if hasDeletedOptions: if hasDeletedOptions:
log.warning( 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( LOG.info(f'Downloading {os.getpid()=} id="{self.info.id}" title="{self.info.title}".')
f'Downloading {os.getpid()=} id="{self.info.id}" title="{self.info.title}".')
ret = yt_dlp.YoutubeDL(params=params).download([self.info.url]) ret = yt_dlp.YoutubeDL(params=params).download([self.info.url])
self.status_queue.put({ self.status_queue.put({'id': self.id, 'status': 'finished' if ret == 0 else 'error'})
'id': self.id,
'status': 'finished' if ret == 0 else 'error'
})
except Exception as exc: except Exception as exc:
self.status_queue.put({ self.status_queue.put({
'id': self.id, 'id': self.id,
@ -181,22 +174,17 @@ class Download:
'error': str(exc) 'error': str(exc)
}) })
log.info( LOG.info(f'Finished {os.getpid()=} id="{self.info.id}" title="{self.info.title}".')
f'Finished {os.getpid()=} id="{self.info.id}" title="{self.info.title}".')
async def start(self, notifier: Notifier): async def start(self, notifier: Notifier):
if self.manager is None: self.manager = multiprocessing.Manager() if self.manager is None else self.manager
self.manager = multiprocessing.Manager()
self.status_queue = self.manager.Queue() self.status_queue = self.manager.Queue()
self.loop = asyncio.get_running_loop() self.loop = asyncio.get_running_loop()
self.notifier = notifier self.notifier = notifier
# Create temp dir for each download. # Create temp dir for each download.
self.tempPath = os.path.join( self.tempPath = os.path.join(self.temp_dir, hashlib.shake_256(self.info.id.encode('utf-8')).hexdigest(5))
self.temp_dir,
hashlib.shake_256(self.info.id.encode('utf-8')).hexdigest(5)
)
if not os.path.exists(self.tempPath): if not os.path.exists(self.tempPath):
os.makedirs(self.tempPath, exist_ok=True) os.makedirs(self.tempPath, exist_ok=True)
@ -229,7 +217,7 @@ class Download:
def kill(self): def kill(self):
if self.running(): if self.running():
log.info(f'Killing download process: {self.proc.ident}') LOG.info(f'Killing download process: {self.proc.ident}')
self.proc.kill() self.proc.kill()
def delete_temp(self): def delete_temp(self):
@ -237,19 +225,19 @@ class Download:
return return
if self.info.status != 'finished' and self.is_live: if self.info.status != 'finished' and self.is_live:
log.warning( LOG.warning(
f'Keeping live temp directory: {self.tempPath}, as the reported status is not finished [{self.info.status}].') f'Keeping live temp directory [{self.tempPath}], as the reported status is not finished [{self.info.status}].')
return return
if not os.path.exists(self.tempPath): if not os.path.exists(self.tempPath):
return return
if self.tempPath == self.temp_dir: if self.tempPath == self.temp_dir:
log.warning( LOG.warning(
f'Attempted to delete video temp directory: {self.tempPath}, but it is the same as main temp directory.') f'Attempted to delete video temp directory: {self.tempPath}, but it is the same as main temp directory.')
return return
log.info(f'Deleting Temp directory: {self.tempPath}') LOG.info(f'Deleting Temp directory: {self.tempPath}')
shutil.rmtree(self.tempPath, ignore_errors=True) shutil.rmtree(self.tempPath, ignore_errors=True)
async def progress_update(self): async def progress_update(self):
@ -262,7 +250,7 @@ class Download:
return return
if self.debug: 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): if isinstance(status, str):
await self.notifier.updated(self.info) await self.notifier.updated(self.info)
@ -271,13 +259,11 @@ class Download:
self.tmpfilename = status.get('tmpfilename') self.tmpfilename = status.get('tmpfilename')
if 'filename' in status: if 'filename' in status:
self.info.filename = os.path.relpath( self.info.filename = os.path.relpath(status.get('filename'), self.download_dir)
status.get('filename'), self.download_dir)
# Set correct file extension for thumbnails # Set correct file extension for thumbnails
if self.info.format == 'thumbnail': if self.info.format == 'thumbnail':
self.info.filename = re.sub( self.info.filename = re.sub(r'\.webm$', '.jpg', self.info.filename)
r'\.webm$', '.jpg', self.info.filename)
self.info.status = status.get('status', self.info.status) self.info.status = status.get('status', self.info.status)
self.info.msg = status.get('msg') self.info.msg = status.get('msg')
@ -287,11 +273,9 @@ class Download:
await self.notifier.error(self.info, self.info.error) await self.notifier.error(self.info, self.info.error)
if 'downloaded_bytes' in status: if 'downloaded_bytes' in status:
total = status.get('total_bytes') or status.get( total = status.get('total_bytes') or status.get('total_bytes_estimate')
'total_bytes_estimate')
if total: if total:
perc = status['downloaded_bytes'] / total * 100 self.info.percent = status['downloaded_bytes'] / total * 100
self.info.percent = perc
self.info.total_bytes = total self.info.total_bytes = total
self.info.speed = status.get('speed') self.info.speed = status.get('speed')
@ -299,11 +283,9 @@ class Download:
if self.info.status == 'finished' and 'filename' in status and self.info.format != 'thumbnail': if self.info.status == 'finished' and 'filename' in status and self.info.format != 'thumbnail':
try: try:
self.info.file_size = os.path.getsize( self.info.file_size = os.path.getsize(status.get('filename'))
status.get('filename'))
except FileNotFoundError: except FileNotFoundError:
log.warning( LOG.warning(f'File not found: {status.get("filename")}')
f'File not found: {status.get("filename")}')
self.info.file_size = None self.info.file_size = None
pass pass

98
app/DownloadQueue.py
View file

@ -12,32 +12,31 @@ from DataStore import DataStore
from Utils import Notifier, ObjectSerializer, calcDownloadPath, ExtractInfo, isDownloaded, mergeConfig from Utils import Notifier, ObjectSerializer, calcDownloadPath, ExtractInfo, isDownloaded, mergeConfig
from AsyncPool import AsyncPool from AsyncPool import AsyncPool
log = logging.getLogger('DownloadQueue') LOG = logging.getLogger('DownloadQueue')
dl_queue = asyncio.Queue() TYPE_DONE: str = 'done'
TYPE_QUEUE: str = 'queue'
class DownloadQueue: class DownloadQueue:
config: Config = None config: Config = None
notifier: Notifier = None notifier: Notifier = None
connection: Connection = None
serializer: ObjectSerializer = None serializer: ObjectSerializer = None
queue: DataStore = None queue: DataStore = None
done: DataStore = None done: DataStore = None
event: asyncio.Event = None event: asyncio.Event = None
def __init__(self, config: Config, notifier: Notifier, connection: Connection, serializer: ObjectSerializer): def __init__(self, notifier: Notifier, connection: Connection, serializer: ObjectSerializer):
self.config = config self.config = Config.get_instance()
self.notifier = notifier self.notifier = notifier
self.connection = connection
self.serializer = serializer self.serializer = serializer
self.done = DataStore(type='done', config=self.config) self.done = DataStore(type=TYPE_DONE, connection=connection)
self.queue = DataStore(type='queue', config=self.config) self.queue = DataStore(type=TYPE_QUEUE, connection=connection)
self.done.load() self.done.load()
self.queue.load() self.queue.load()
async def initialize(self): async def initialize(self):
self.event = asyncio.Event() self.event = asyncio.Event()
log.info(f'Using {self.config.max_workers} workers for downloading.') LOG.info(f'Using {self.config.max_workers} workers for downloading.')
asyncio.create_task( asyncio.create_task(
self.__download_pool() if self.config.max_workers > 1 else self.__download() self.__download_pool() if self.config.max_workers > 1 else self.__download()
@ -62,15 +61,14 @@ class DownloadQueue:
error: str = None error: str = None
live_in: str = None live_in: str = None
etype = entry.get('_type') or 'video' eventType = entry.get('_type') or 'video'
if etype == 'playlist': if eventType == 'playlist':
entries = entry['entries'] entries = entry['entries']
playlist_index_digits = len(str(len(entries))) playlist_index_digits = len(str(len(entries)))
results = [] results = []
for index, etr in enumerate(entries, start=1): for index, etr in enumerate(entries, start=1):
etr['playlist'] = entry.get('id') etr['playlist'] = entry.get('id')
etr['playlist_index'] = '{{0:0{0:d}d}}'.format( etr['playlist_index'] = '{{0:0{0:d}d}}'.format(playlist_index_digits).format(index)
playlist_index_digits).format(index)
for property in ('id', 'title', 'uploader', 'uploader_id'): for property in ('id', 'title', 'uploader', 'uploader_id'):
if property in entry: if property in entry:
etr[f'playlist_{property}'] = entry.get(property) etr[f'playlist_{property}'] = entry.get(property)
@ -92,7 +90,7 @@ class DownloadQueue:
return {'status': 'error', 'msg': ', '.join(res['msg'] for res in results if res['status'] == 'error' and 'msg' in res)} return {'status': 'error', 'msg': ', '.join(res['msg'] for res in results if res['status'] == 'error' and 'msg' in res)}
return {'status': 'ok'} return {'status': 'ok'}
elif (etype == 'video' or etype.startswith('url')) and 'id' in entry and 'title' in entry: elif (eventType == 'video' or eventType.startswith('url')) and 'id' in entry and 'title' in entry:
# check if the video is live stream. # check if the video is live stream.
if 'live_status' in entry and entry.get('live_status') == 'is_upcoming': if 'live_status' in entry and entry.get('live_status') == 'is_upcoming':
if 'release_timestamp' in entry and entry.get('release_timestamp'): if 'release_timestamp' in entry and entry.get('release_timestamp'):
@ -102,30 +100,25 @@ class DownloadQueue:
else: else:
error = entry.get('msg', None) error = entry.get('msg', None)
log.debug( LOG.debug(f"entry: {entry.get('id', None)} - {entry.get('webpage_url', None)} - {entry.get('url', None)}")
f"entry: {entry.get('id', None)} - {entry.get('webpage_url', None)} - {entry.get('url', None) }")
if self.done.exists(key=entry['id'], url=entry.get('webpage_url') or entry.get('url')): if self.done.exists(key=entry['id'], url=entry.get('webpage_url') or entry.get('url')):
item = self.done.get(key=entry['id'], url=entry.get( item = self.done.get(key=entry['id'], url=entry.get('webpage_url') or entry['url'])
'webpage_url') or entry['url'])
log.debug( LOG.debug(f'Item [{item.info.title}] already downloaded. Removing from history.')
f'Item [{item.info.title}] already downloaded. Removing from history.')
await self.clear([item.info._id]) await self.clear([item.info._id])
try: try:
item = self.queue.get(key=entry.get('id'), url=entry.get( item = self.queue.get(key=entry.get('id'), url=entry.get('webpage_url') or entry.get('url'))
'webpage_url') or entry.get('url'))
if item is not None: if item is not None:
err_message = f'Item [{item.info.id}: {item.info.title}] already in download queue.' err_message = f'Item [{item.info.id}: {item.info.title}] already in download queue.'
log.info(err_message) LOG.info(err_message)
return {'status': 'error', 'msg': err_message} return {'status': 'error', 'msg': err_message}
except KeyError: except KeyError:
pass pass
is_manifestless = 'post_live' == entry.get( is_manifestless = 'post_live' == entry.get('live_status') if 'live_status' in entry else None
'live_status') if 'live_status' in entry else None
options.update({'is_manifestless': is_manifestless}) options.update({'is_manifestless': is_manifestless})
@ -147,23 +140,20 @@ class DownloadQueue:
) )
try: try:
dldirectory = calcDownloadPath( download_dir = calcDownloadPath(basePath=self.config.download_path, folder=folder)
basePath=self.config.download_path,
folder=folder
)
except Exception as e: except Exception as e:
LOG.exception(e)
return {'status': 'error', 'msg': str(e)} return {'status': 'error', 'msg': str(e)}
output_chapter = self.config.output_template_chapter output_chapter = self.config.output_template_chapter
for property, value in entry.items(): for property, value in entry.items():
if property.startswith("playlist"): if property.startswith("playlist"):
dl.output_template = dl.output_template.replace( dl.output_template = dl.output_template.replace(f"%({property})s", str(value))
f"%({property})s", str(value))
dlInfo: Download = Download( dlInfo: Download = Download(
info=dl, info=dl,
download_dir=dldirectory, download_dir=download_dir,
temp_dir=self.config.temp_path, temp_dir=self.config.temp_path,
output_template_chapter=output_chapter, output_template_chapter=output_chapter,
default_ytdl_opts=self.config.ytdl_options, default_ytdl_opts=self.config.ytdl_options,
@ -173,23 +163,23 @@ class DownloadQueue:
if dlInfo.info.live_in: if dlInfo.info.live_in:
dlInfo.info.status = 'not_live' dlInfo.info.status = 'not_live'
itemDownload = self.done.put(dlInfo) itemDownload = self.done.put(dlInfo)
NotifiyEvent = 'completed' NotifyEvent = 'completed'
elif self.config.allow_manifestless is False and is_manifestless is True: elif self.config.allow_manifestless is False and is_manifestless is True:
dlInfo.info.status = 'error' dlInfo.info.status = 'error'
dlInfo.info.error = 'Video is in Post-Live Manifestless mode.' dlInfo.info.error = 'Video is in Post-Live Manifestless mode.'
itemDownload = self.done.put(dlInfo) itemDownload = self.done.put(dlInfo)
NotifiyEvent = 'completed' NotifyEvent = 'completed'
else: else:
NotifiyEvent = 'added' NotifyEvent = 'added'
itemDownload = self.queue.put(dlInfo) itemDownload = self.queue.put(dlInfo)
self.event.set() self.event.set()
await self.notifier.emit(NotifiyEvent, itemDownload.info) await self.notifier.emit(NotifyEvent, itemDownload.info)
return { return {
'status': 'ok' 'status': 'ok'
} }
elif etype.startswith('url'): elif eventType.startswith('url'):
return await self.add( return await self.add(
entry=entry.get('url'), entry=entry.get('url'),
quality=quality, quality=quality,
@ -203,7 +193,7 @@ class DownloadQueue:
return { return {
'status': 'error', 'status': 'error',
'msg': f'Unsupported resource "{etype}"' 'msg': f'Unsupported resource "{eventType}"'
} }
async def add( async def add(
@ -219,12 +209,11 @@ class DownloadQueue:
): ):
ytdlp_config = ytdlp_config if ytdlp_config else {} ytdlp_config = ytdlp_config if ytdlp_config else {}
log.info( LOG.info(f'Adding {url=} {quality=} {format=} {folder=} {output_template=} {ytdlp_cookies=} {ytdlp_config=}')
f'Adding {url=} {quality=} {format=} {folder=} {output_template=} {ytdlp_cookies=} {ytdlp_config=}')
already = set() if already is None else already already = set() if already is None else already
if url in already: if url in already:
log.info('recursion detected, skipping') LOG.info('recursion detected, skipping')
return {'status': 'ok'} return {'status': 'ok'}
else: else:
already.add(url) already.add(url)
@ -252,8 +241,7 @@ class DownloadQueue:
return {'status': 'error', 'msg': 'Unable to extract info check logs.'} return {'status': 'error', 'msg': 'Unable to extract info check logs.'}
if self.isDownloaded(entry): if self.isDownloaded(entry):
log.info( LOG.info(f'[{entry.get("id")}: {entry.get("title")}]: has been downloaded already.')
f'[{entry.get("id")}: {entry.get("title")}]: has been downloaded already.')
return { return {
'status': 'error', 'status': 'error',
'msg': f'[{entry.get("id")}: {entry.get("title")}]: has been downloaded already.' 'msg': f'[{entry.get("id")}: {entry.get("title")}]: has been downloaded already.'
@ -267,7 +255,7 @@ class DownloadQueue:
if self.isDownloaded(entry): if self.isDownloaded(entry):
raise yt_dlp.utils.ExistingVideoReached() raise yt_dlp.utils.ExistingVideoReached()
log.debug(f'entry: extract info says: {entry}') LOG.debug(f'entry: extract info says: {entry}')
except yt_dlp.utils.ExistingVideoReached: except yt_dlp.utils.ExistingVideoReached:
return {'status': 'error', 'msg': 'Video has been downloaded already and recorded in archive.log file.'} return {'status': 'error', 'msg': 'Video has been downloaded already and recorded in archive.log file.'}
except yt_dlp.utils.YoutubeDLError as exc: except yt_dlp.utils.YoutubeDLError as exc:
@ -289,16 +277,15 @@ class DownloadQueue:
try: try:
item = self.queue.get(key=id) item = self.queue.get(key=id)
except KeyError as e: except KeyError as e:
log.warn( LOG.warning(f'Requested cancel for non-existent download {id=}. {str(e)}')
f'Requested cancel for non-existent download {id=}. {str(e)}')
continue continue
if item.started() is True: if item.started() is True:
log.info(f'Canceling {id=} {item.info.title=}') LOG.info(f'Canceling {id=} {item.info.title=}')
item.cancel() item.cancel()
else: else:
item.close() item.close()
log.info(f'Deleting from queue {id=} {item.info.title=}') LOG.info(f'Deleting from queue {id=} {item.info.title=}')
self.queue.delete(id) self.queue.delete(id)
await self.notifier.canceled(id) await self.notifier.canceled(id)
self.done.put(item) self.done.put(item)
@ -312,11 +299,10 @@ class DownloadQueue:
try: try:
item = self.done.get(key=id) item = self.done.get(key=id)
except KeyError as e: except KeyError as e:
log.warn( LOG.warning(f'Requested delete for non-existent download {id=}. {str(e)}')
f'Requested delete for non-existent download {id=}. {str(e)}')
continue continue
log.info(f'Deleting completed download {id=} {item.info.title=}') LOG.info(f'Deleting completed download {id=} {item.info.title=}')
self.done.delete(id) self.done.delete(id)
await self.notifier.cleared(id) await self.notifier.cleared(id)
@ -347,7 +333,7 @@ class DownloadQueue:
num_workers=self.config.max_workers, num_workers=self.config.max_workers,
worker_co=self.__downloadFile, worker_co=self.__downloadFile,
name='WorkerPool', name='WorkerPool',
logger=log, logger=logging.getLogger('WorkerPool'),
) as executor: ) as executor:
while True: while True:
while True: while True:
@ -357,8 +343,7 @@ class DownloadQueue:
break break
while not self.queue.hasDownloads(): while not self.queue.hasDownloads():
log.info( LOG.info(f'Waiting for item to download. [{executor.get_available_workers()}] workers available.')
f'Waiting for item to download. [{executor.get_avalialbe_workers()}] Workers available.')
await self.event.wait() await self.event.wait()
self.event.clear() self.event.clear()
@ -372,7 +357,7 @@ class DownloadQueue:
async def __download(self): async def __download(self):
while True: while True:
while self.queue.empty(): while self.queue.empty():
log.info('waiting for item to download.') LOG.info('waiting for item to download.')
await self.event.wait() await self.event.wait()
self.event.clear() self.event.clear()
@ -380,8 +365,7 @@ class DownloadQueue:
await self.__downloadFile(id, entry) await self.__downloadFile(id, entry)
async def __downloadFile(self, id: str, entry: Download): async def __downloadFile(self, id: str, entry: Download):
log.info( LOG.info(f'Queuing {id=} - {entry.info.title=} - {entry.info.url=} - {entry.info.folder=}.')
f'Queuing {id=} - {entry.info.title=} - {entry.info.url=} - {entry.info.folder=}.')
await entry.start(self.notifier) await entry.start(self.notifier)

85
app/Utils.py
View file

@ -9,19 +9,10 @@ import yt_dlp
from socketio import AsyncServer from socketio import AsyncServer
from Webhooks import Webhooks from Webhooks import Webhooks
log = logging.getLogger('Utils') LOG = logging.getLogger('Utils')
IGNORED_KEYS: tuple = ( AUDIO_FORMATS: tuple = ('m4a', 'mp3', 'opus', 'wav')
'cookiefile', IGNORED_KEYS: tuple = ('cookiefile', 'paths', 'outtmpl', 'progress_hooks', 'postprocessor_hooks',)
'paths',
'outtmpl',
'progress_hooks',
'postprocessor_hooks',
)
AUDIO_FORMATS: tuple = (
'm4a', 'mp3', 'opus', 'wav'
)
def get_format(format: str, quality: str) -> str: def get_format(format: str, quality: str) -> str:
@ -55,16 +46,15 @@ def get_format(format: str, quality: str) -> str:
if quality == 'audio': if quality == 'audio':
return "bestaudio/best" return "bestaudio/best"
vfmt, afmt = ( videoFormat, audioFormat = ('[ext=mp4]', '[ext=m4a]') if format == "mp4" else ('', '')
'[ext=mp4]', '[ext=m4a]') if format == "mp4" else ('', '')
vres = f'[height<={quality}]' if quality != 'best' else '' videoResolution = f'[height<={quality}]' if quality != 'best' else ''
vcombo = vres + vfmt videoCombo = videoResolution + videoFormat
return f'bestvideo{vcombo}+bestaudio{afmt}/best{vcombo}' return f'bestvideo{videoCombo}+bestaudio{audioFormat}/best{videoCombo}'
raise Exception(f"Unkown format {format}") raise Exception(f"Unknown format {format}")
def get_opts(format: str, quality: str, ytdl_opts: dict) -> dict: def get_opts(format: str, quality: str, ytdl_opts: dict) -> dict:
@ -98,19 +88,16 @@ def get_opts(format: str, quality: str, ytdl_opts: dict) -> dict:
# Audio formats without thumbnail # Audio formats without thumbnail
if format not in ("wav") and "writethumbnail" not in opts: if format not in ("wav") and "writethumbnail" not in opts:
opts["writethumbnail"] = True opts["writethumbnail"] = True
postprocessors.append( postprocessors.append({"key": "FFmpegThumbnailsConvertor", "format": "jpg", "when": "before_dl"})
{"key": "FFmpegThumbnailsConvertor", "format": "jpg", "when": "before_dl"})
postprocessors.append({"key": "FFmpegMetadata"}) postprocessors.append({"key": "FFmpegMetadata"})
postprocessors.append({"key": "EmbedThumbnail"}) postprocessors.append({"key": "EmbedThumbnail"})
if format == "thumbnail": if format == "thumbnail":
opts["skip_download"] = True opts["skip_download"] = True
opts["writethumbnail"] = True opts["writethumbnail"] = True
postprocessors.append( postprocessors.append({"key": "FFmpegThumbnailsConvertor", "format": "jpg", "when": "before_dl"})
{"key": "FFmpegThumbnailsConvertor", "format": "jpg", "when": "before_dl"})
opts["postprocessors"] = postprocessors + \ opts["postprocessors"] = postprocessors + (opts["postprocessors"] if "postprocessors" in opts else [])
(opts["postprocessors"] if "postprocessors" in opts else [])
return opts return opts
@ -127,26 +114,11 @@ def getVideoInfo(url: str, ytdlp_opts: dict = None) -> (Any | dict[str, Any] | N
return yt_dlp.YoutubeDL().extract_info(url, download=False) return yt_dlp.YoutubeDL().extract_info(url, download=False)
def getAttributes(vclass: str | type) -> dict:
attrs: dict = {}
if not isinstance(vclass, type):
vclass = vclass.__class__
for attribute in vclass.__dict__.keys():
if not attribute.startswith('_'):
value = getattr(vclass, attribute)
if not callable(value):
attrs[attribute] = value
return attrs
def calcDownloadPath(basePath: str, folder: str = None, createPath: bool = True) -> str: def calcDownloadPath(basePath: str, folder: str = None, createPath: bool = True) -> str:
"""Calculates download path and prevents directory traversal. """Calculates download path and prevents directory traversal.
Returns: Returns:
Dir with base dir factored in. Download path with base directory factored in.
""" """
if not folder: if not folder:
return basePath return basePath
@ -155,8 +127,7 @@ def calcDownloadPath(basePath: str, folder: str = None, createPath: bool = True)
download_path = os.path.realpath(os.path.join(basePath, folder)) download_path = os.path.realpath(os.path.join(basePath, folder))
if not download_path.startswith(realBasePath): if not download_path.startswith(realBasePath):
raise Exception( raise Exception(f'Folder "{folder}" must resolve inside the base download directory "{realBasePath}".')
f'Folder "{folder}" must resolve inside the base download directory "{realBasePath}"')
if not os.path.isdir(download_path) and createPath: if not os.path.isdir(download_path) and createPath:
os.makedirs(download_path, exist_ok=True) os.makedirs(download_path, exist_ok=True)
@ -198,10 +169,7 @@ def mergeDict(source: dict, destination: dict) -> dict:
for key, value in source.items(): for key, value in source.items():
destination_key_value = destination_copy.get(key) destination_key_value = destination_copy.get(key)
if isinstance(value, dict) and isinstance(destination_key_value, dict): if isinstance(value, dict) and isinstance(destination_key_value, dict):
destination_copy[key] = mergeDict( destination_copy[key] = mergeDict(source=value, destination=destination_copy.setdefault(key, {}))
source=value,
destination=destination_copy.setdefault(key, {})
)
elif isinstance(value, list) and isinstance(destination_key_value, list): elif isinstance(value, list) and isinstance(destination_key_value, list):
destination_copy[key] = destination_key_value + value destination_copy[key] = destination_key_value + value
else: else:
@ -238,7 +206,7 @@ def isDownloaded(archive_file: str, info: dict) -> bool:
if not id: if not id:
return False return False
except Exception as e: except Exception as e:
log.error(f'Error generating archive id: {e}') LOG.error(f'Error generating archive id: {e}')
return False return False
with open(archive_file, 'r') as f: with open(archive_file, 'r') as f:
@ -270,22 +238,20 @@ def jsonCookie(cookies: dict[dict[str, any]]) -> str | None:
if not isinstance(cookies[domain][subDomain][cookie], dict): if not isinstance(cookies[domain][subDomain][cookie], dict):
continue continue
dicto = cookies[domain][subDomain][cookie] cookieDict = cookies[domain][subDomain][cookie]
if 0 == int(dicto['expirationDate']): if 0 == int(cookieDict['expirationDate']):
dicto['expirationDate']: float = datetime.now( cookieDict['expirationDate']: float = datetime.now(timezone.utc).timestamp() + (86400 * 1000)
timezone.utc).timestamp() + (86400 * 1000)
hasCookies = True hasCookies = True
netscapeCookies += "\t".join([ netscapeCookies += "\t".join([
dicto['domain'] if str(dicto['domain']).startswith( cookieDict['domain'] if str(cookieDict['domain']).startswith('.') else '.' + cookieDict['domain'],
'.') else '.' + dicto['domain'],
'TRUE', 'TRUE',
dicto['path'], cookieDict['path'],
'TRUE' if dicto['secure'] else 'FALSE', 'TRUE' if cookieDict['secure'] else 'FALSE',
str(int(dicto['expirationDate'])), str(int(cookieDict['expirationDate'])),
dicto['name'], cookieDict['name'],
dicto['value'] cookieDict['value']
])+"\n" ])+"\n"
return netscapeCookies if hasCookies else None return netscapeCookies if hasCookies else None
@ -309,10 +275,13 @@ class Notifier:
sio: AsyncServer = None sio: AsyncServer = None
"SocketIO server instance" "SocketIO server instance"
serializer: ObjectSerializer = None serializer: ObjectSerializer = None
"Serializer used to serialize objects to JSON" "Serializer used to serialize objects to JSON"
webhooks: Webhooks = None webhooks: Webhooks = None
"Send webhooks events." "Send webhooks events."
webhooks_allowed_events: tuple = ( webhooks_allowed_events: tuple = (
'added', 'completed', 'error', 'not_live' 'added', 'completed', 'error', 'not_live'
) )

27
app/Webhooks.py
View file

@ -5,7 +5,7 @@ import os
from ItemDTO import ItemDTO from ItemDTO import ItemDTO
from aiohttp import client from aiohttp import client
log = logging.getLogger('Webhooks') LOG = logging.getLogger('Webhooks')
class Webhooks: class Webhooks:
@ -13,16 +13,19 @@ class Webhooks:
def __init__(self, file: str): def __init__(self, file: str):
if os.path.exists(file): if os.path.exists(file):
try: self.load(file)
if os.path.getsize(file) < 2:
raise Exception(f'file is empty.')
log.info(f'Loading webhooks from {file}') def load(self, file: str):
with open(file, 'r') as f: try:
self.targets = json.load(f) if os.path.getsize(file) < 2:
except Exception as e: raise Exception(f'file is empty.')
log.error(f'Error loading webhooks from {file}: {e}')
pass LOG.info(f'Loading webhooks from {file}')
with open(file, 'r') as f:
self.targets = json.load(f)
except Exception as e:
LOG.error(f'Error loading webhooks from {file}: {e}')
pass
async def send(self, event: str, item: ItemDTO) -> list[dict]: async def send(self, event: str, item: ItemDTO) -> list[dict]:
if len(self.targets) == 0: if len(self.targets) == 0:
@ -45,11 +48,11 @@ class Webhooks:
async def __send(self, event: str, target: dict, item: ItemDTO) -> dict: async def __send(self, event: str, target: dict, item: ItemDTO) -> dict:
req: dict = target.get('request') req: dict = target.get('request')
try: try:
log.info(f"Sending {event=} {item.id=} to [{target.get('name')}]") LOG.info(f"Sending {event=} {item.id=} to [{target.get('name')}]")
async with client.ClientSession() as session: async with client.ClientSession() as session:
headers = req.get('headers', {}) if 'headers' in req else {} headers = req.get('headers', {}) if 'headers' in req else {}
async with session.request(method=req.get('method', 'POST'), url=req.get('url'), json=item.__dict__, headers=headers) as response: async with session.request(method=req.get('method', 'POST'), url=req.get('url'), json=item.__dict__, headers=headers) as response:
log.info(f"Sent {event=} {item.id=} to [{target.get('name')}] and responsed with [status: {response.status}].") LOG.info(f"[{target.get('name')}] Response to [{event=} {item.id=}] [status: {response.status}].")
return { return {
'url': req.get('url'), 'url': req.get('url'),
'status': response.status, 'status': response.status,

110
app/main.py
View file

@ -19,6 +19,8 @@ import caribou
import sqlite3 import sqlite3
from aiocron import crontab from aiocron import crontab
LOG = logging.getLogger('app')
class Main: class Main:
config: Config = None config: Config = None
@ -27,75 +29,68 @@ class Main:
sio: socketio.AsyncServer = None sio: socketio.AsyncServer = None
routes: web.RouteTableDef = None routes: web.RouteTableDef = None
connection: sqlite3.Connection = None connection: sqlite3.Connection = None
dqueue: DownloadQueue = None queue: DownloadQueue = None
loop: asyncio.AbstractEventLoop = None loop: asyncio.AbstractEventLoop = None
logger: logging.Logger = None
def __init__(self): def __init__(self):
self.config = Config.get_instance() self.config = Config.get_instance()
self.logger = logging.getLogger('main')
try: try:
if not os.path.exists(self.config.download_path): if not os.path.exists(self.config.download_path):
self.logger.info( LOG.info(f'Creating download folder at [{self.config.download_path}]')
f'Creating download folder at {self.config.download_path}')
os.makedirs(self.config.download_path, exist_ok=True) os.makedirs(self.config.download_path, exist_ok=True)
except OSError as e: except OSError as e:
self.logger.error( LOG.error(f'Could not create download folder at [{self.config.download_path}]')
f'Could not create download folder at {self.config.download_path}')
raise e raise e
try: try:
if not os.path.exists(self.config.temp_path): if not os.path.exists(self.config.temp_path):
self.logger.info( LOG.info(f'Creating temp folder at [{self.config.temp_path}]')
f'Creating temp folder at {self.config.temp_path}')
os.makedirs(self.config.temp_path, exist_ok=True) os.makedirs(self.config.temp_path, exist_ok=True)
except OSError as e: except OSError as e:
self.logger.error( LOG.error(f'Could not create temp folder at [{self.config.temp_path}]')
f'Could not create temp folder at {self.config.temp_path}')
raise e raise e
try: try:
if not os.path.exists(self.config.config_path): if not os.path.exists(self.config.config_path):
self.logger.info( LOG.info(f'Creating config folder at [{self.config.config_path}]')
f'Creating config folder at {self.config.config_path}')
os.makedirs(self.config.config_path, exist_ok=True) os.makedirs(self.config.config_path, exist_ok=True)
except OSError as e: except OSError as e:
self.logger.error( LOG.error(f'Could not create config folder at [{self.config.config_path}]')
f'Could not create config folder at {self.config.config_path}')
raise e raise e
try: try:
if not os.path.exists(self.config.db_file): if not os.path.exists(self.config.db_file):
self.logger.info( LOG.info(f'Creating database file at [{self.config.db_file}]')
f'Creating database file at {self.config.db_file}')
with open(self.config.db_file, 'w') as _: with open(self.config.db_file, 'w') as _:
pass pass
except OSError as e: except OSError as e:
self.logger.error( LOG.error(f'Could not create database file at [{self.config.db_file}]')
f'Could not create database file at {self.config.db_file}')
raise e raise e
caribou.upgrade(self.config.db_file, os.path.join( caribou.upgrade(self.config.db_file, os.path.join(os.path.realpath(os.path.dirname(__file__)), 'migrations'))
os.path.realpath(os.path.dirname(__file__)), 'migrations'))
self.webhooks = Webhooks(os.path.join( self.webhooks = Webhooks(os.path.join(self.config.config_path, 'webhooks.json'))
self.config.config_path, 'webhooks.json'))
self.loop = asyncio.get_event_loop() self.loop = asyncio.get_event_loop()
self.serializer = ObjectSerializer() self.serializer = ObjectSerializer()
self.app = web.Application() self.app = web.Application()
self.sio = socketio.AsyncServer(cors_allowed_origins='*') self.sio = socketio.AsyncServer(cors_allowed_origins='*')
self.sio.attach( self.sio.attach(self.app, socketio_path=self.config.url_prefix + 'socket.io')
self.app, socketio_path=self.config.url_prefix + 'socket.io')
self.routes = web.RouteTableDef() self.routes = web.RouteTableDef()
self.connection = sqlite3.connect(self.config.db_file)
self.dqueue = DownloadQueue( self.connection = sqlite3.connect(self.config.db_file, isolation_level=None)
config=self.config, self.connection.row_factory = sqlite3.Row
notifier=Notifier( self.connection.execute('PRAGMA journal_mode=wal')
sio=self.sio, serializer=self.serializer, webhooks=self.webhooks),
self.queue = DownloadQueue(
notifier=Notifier(sio=self.sio, serializer=self.serializer, webhooks=self.webhooks),
connection=self.connection, connection=self.connection,
serializer=self.serializer, serializer=self.serializer,
) )
self.app.on_startup.append(lambda _: self.dqueue.initialize())
self.app.on_startup.append(lambda _: self.queue.initialize())
self.addRoutes() self.addRoutes()
self.addTasks() self.addTasks()
self.start() self.start()
@ -114,7 +109,7 @@ class Main:
async def connect(self, sid, _): async def connect(self, sid, _):
data: dict = { data: dict = {
**self.dqueue.get(), **self.queue.get(),
"config": self.config, "config": self.config,
"tasks": [], "tasks": [],
} }
@ -145,17 +140,15 @@ class Main:
cvDate = self.config.version.split('-') cvDate = self.config.version.split('-')
if tagName > cvDate[0]: if tagName > cvDate[0]:
self.config.new_version_available = True self.config.new_version_available = True
self.logger.info( LOG.info(f'New version [{tagName}] is available.')
f'New version {tagName} available at')
break break
except Exception as e: except Exception as e:
pass pass
def addTasks(self): def addTasks(self):
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): if not os.path.exists(tasks_file):
self.logger.info( LOG.info(
f'No tasks file found at {tasks_file}. Skipping Tasks.') f'No tasks file found at {tasks_file}. Skipping Tasks.')
return return
@ -163,21 +156,20 @@ class Main:
with open(tasks_file, 'r') as f: with open(tasks_file, 'r') as f:
tasks = json.load(f) tasks = json.load(f)
except Exception as e: except Exception as e:
self.logger.error( LOG.error(f'Could not load tasks file [{tasks_file}]. Error message [{str(e)}]. Skipping Tasks.')
f'Could not load tasks file [{tasks_file}]. Error message [{str(e)}]. Skipping Tasks.')
return return
for task in tasks: for task in tasks:
if not task.get('url'): if not task.get('url'):
self.logger.warning(f'Invalid task {task}.') LOG.warning(f'Invalid task {task}.')
continue continue
cron_timer: str = task.get( cron_timer: str = task.get('timer', f'{random.randint(1,59)} */1 * * *')
'timer', f'{random.randint(1,59)} */1 * * *')
async def cron_runner(task: dict): async def cron_runner(task: dict):
self.logger.info( taskName = task.get("name", task.get("url"))
f'Running task [{task.get("name",task.get("url"))}] at [{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}].') timeNow = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
LOG.info(f'Started [Task: {taskName}] at [{timeNow}].')
await self.add( await self.add(
url=task.get('url'), url=task.get('url'),
@ -189,8 +181,8 @@ class Main:
output_template=task.get('output_template') output_template=task.get('output_template')
) )
self.logger.info( timeNow = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
f'Finished Running task [{task.get("name",task.get("url"))}] at [{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}].') LOG.info(f'Completed [Task: {taskName}] at [{timeNow}].')
crontab( crontab(
spec=cron_timer, spec=cron_timer,
@ -200,12 +192,12 @@ class Main:
loop=self.loop loop=self.loop
) )
self.logger.info( LOG.info(f'Added [Task: {task.get("name", task.get("url"))}] executed every [{cron_timer}].')
f'Added task to grab {task.get("name",task.get("url"))} content every [{cron_timer}].')
def addRoutes(self): def addRoutes(self):
@self.routes.get(self.config.url_prefix + 'ping') @self.routes.get(self.config.url_prefix + 'ping')
async def ping(_) -> Response: async def ping(_) -> Response:
self.connection.execute('SELECT "id" FROM "history" LIMIT 1').fetchone()
return web.Response(text='pong') return web.Response(text='pong')
@self.routes.post(self.config.url_prefix + 'add') @self.routes.post(self.config.url_prefix + 'add')
@ -240,8 +232,7 @@ class Main:
@self.routes.get(self.config.url_prefix + 'tasks') @self.routes.get(self.config.url_prefix + 'tasks')
async def tasks(_: Request) -> Response: async def tasks(_: Request) -> Response:
tasks_file: str = os.path.join( tasks_file: str = os.path.join(self.config.config_path, 'tasks.json')
self.config.config_path, 'tasks.json')
if not os.path.exists(tasks_file): 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=404)
@ -264,10 +255,9 @@ class Main:
for item in post: for item in post:
if not isinstance(item, dict): if not isinstance(item, dict):
raise web.HTTPBadRequest( raise web.HTTPBadRequest(reason='Invalid request body expecting list with dicts.')
'Invalid request body expecting list with dicts.')
if not item.get('url'): if not item.get('url'):
raise web.HTTPBadRequest(reason='url is required') raise web.HTTPBadRequest(reason='url is required.')
status[item.get('url')] = await self.add( status[item.get('url')] = await self.add(
url=item.get('url'), url=item.get('url'),
@ -290,7 +280,7 @@ class Main:
if not ids or where not in ['queue', 'done']: if not ids or where not in ['queue', 'done']:
raise web.HTTPBadRequest() raise web.HTTPBadRequest()
status = await (self.dqueue.cancel(ids) if where == 'queue' else self.dqueue.clear(ids)) status = await (self.queue.cancel(ids) if where == 'queue' else self.queue.clear(ids))
return web.Response(text=self.serializer.encode(status)) return web.Response(text=self.serializer.encode(status))
@ -298,9 +288,9 @@ class Main:
async def history(_) -> Response: async def history(_) -> Response:
history = {'done': [], 'queue': []} history = {'done': [], 'queue': []}
for _, v in self.dqueue.queue.saved_items(): for _, v in self.queue.queue.saved_items():
history['queue'].append(v) history['queue'].append(v)
for _, v in self.dqueue.done.saved_items(): for _, v in self.queue.done.saved_items():
history['done'].append(v) history['done'].append(v)
return web.Response(text=self.serializer.encode(history)) return web.Response(text=self.serializer.encode(history))
@ -310,7 +300,7 @@ class Main:
file: str = request.match_info.get('file') file: str = request.match_info.get('file')
if not file: if not file:
raise web.HTTPBadRequest(reason='file is required') raise web.HTTPBadRequest(reason='file is required.')
return web.Response( return web.Response(
text=M3u8(url=f"{self.config.url_host}{self.config.url_prefix}").make_stream( text=M3u8(url=f"{self.config.url_host}{self.config.url_prefix}").make_stream(
@ -340,8 +330,7 @@ class Main:
segmenter = Segments( segmenter = Segments(
segment_index=int(segment), segment_index=int(segment),
segment_duration=float('{:.6f}'.format( segment_duration=float('{:.6f}'.format(float(sd if sd else M3u8.segment_duration))),
float(sd if sd else M3u8.segment_duration))),
vconvert=True if vc == 1 else False, vconvert=True if vc == 1 else False,
aconvert=True if ac == 1 else False aconvert=True if ac == 1 else False
) )
@ -408,8 +397,7 @@ class Main:
self.app.on_response_prepare.append(on_prepare) self.app.on_response_prepare.append(on_prepare)
except ValueError as e: except ValueError as e:
if 'frontend/dist' in str(e): if 'frontend/dist' in str(e):
raise RuntimeError( raise RuntimeError('Could not find the frontend UI static assets.') from e
'Could not find the frontend UI static assets. Please run `node_modules/.bin/ng build` inside the frontend folder.') from e
raise e raise e
async def add( async def add(
@ -428,7 +416,7 @@ class Main:
if ytdlp_cookies and isinstance(ytdlp_cookies, dict): if ytdlp_cookies and isinstance(ytdlp_cookies, dict):
ytdlp_cookies = self.serializer.encode(ytdlp_cookies) ytdlp_cookies = self.serializer.encode(ytdlp_cookies)
status = await self.dqueue.add( status = await self.queue.add(
url=url, url=url,
format=format if format else 'any', format=format if format else 'any',
quality=quality if quality else 'best', quality=quality if quality else 'best',

19
app/player/M3u8.py
View file

@ -17,11 +17,7 @@ class M3u8:
self.segment_duration = float(segment_duration) self.segment_duration = float(segment_duration)
def make_stream(self, download_path: str, file: str): def make_stream(self, download_path: str, file: str):
realFile: str = calcDownloadPath( realFile: str = calcDownloadPath(basePath=download_path, folder=file, createPath=False)
basePath=download_path,
folder=file,
createPath=False
)
if not os.path.exists(realFile): if not os.path.exists(realFile):
raise Exception(f"File {realFile} does not exist.") raise Exception(f"File {realFile} does not exist.")
@ -45,8 +41,7 @@ class M3u8:
segmentSize: float = '{:.6f}'.format(self.segment_duration) segmentSize: float = '{:.6f}'.format(self.segment_duration)
splits: int = math.ceil(duration / self.segment_duration) splits: int = math.ceil(duration / self.segment_duration)
segmentParams = { segmentParams: dict = {}
}
for stream in ffprobe.streams: for stream in ffprobe.streams:
if stream.is_video(): if stream.is_video():
@ -58,17 +53,14 @@ class M3u8:
for i in range(splits): for i in range(splits):
if (i + 1) == splits: if (i + 1) == splits:
segmentParams.update({ segmentParams.update({'sd': '{:.6f}'.format(duration - (i * self.segment_duration))})
'sd': '{:.6f}'.format(duration - (i * self.segment_duration))
})
m3u8 += f"#EXTINF:{segmentParams['sd']}, nodesc\n" m3u8 += f"#EXTINF:{segmentParams['sd']}, nodesc\n"
else: else:
m3u8 += f"#EXTINF:{segmentSize}, nodesc\n" m3u8 += f"#EXTINF:{segmentSize}, nodesc\n"
m3u8 += f"{self.url}segments/{i}/{quote(file)}" m3u8 += f"{self.url}segments/{i}/{quote(file)}"
if len(segmentParams) > 0: if len(segmentParams) > 0:
m3u8 += '?'+'&'.join([f"{key}={value}" for key, m3u8 += '?'+'&'.join([f"{key}={value}" for key, value in segmentParams.items()])
value in segmentParams.items()])
m3u8 += "\n" m3u8 += "\n"
m3u8 += "#EXT-X-ENDLIST\n" m3u8 += "#EXT-X-ENDLIST\n"
@ -79,8 +71,7 @@ class M3u8:
if duration.find(':') > -1: if duration.find(':') > -1:
duration = duration.split(':') duration = duration.split(':')
duration.reverse() duration.reverse()
duration = sum([float(duration[i]) * (60 ** i) duration = sum([float(duration[i]) * (60 ** i) for i in range(len(duration))])
for i in range(len(duration))])
else: else:
duration = float(duration) duration = float(duration)

19
app/player/Segments.py
View file

@ -4,9 +4,8 @@ import os
import subprocess import subprocess
import tempfile import tempfile
from Utils import calcDownloadPath from Utils import calcDownloadPath
from Config import Config
log = logging.getLogger('segments') LOG = logging.getLogger('segments')
class Segments: class Segments:
@ -22,18 +21,13 @@ class Segments:
self.aconvert = bool(aconvert) self.aconvert = bool(aconvert)
async def stream(self, download_path: str, file: str) -> bytes: async def stream(self, download_path: str, file: str) -> bytes:
realFile: str = calcDownloadPath( realFile: str = calcDownloadPath(basePath=download_path, folder=file, createPath=False)
basePath=download_path,
folder=file,
createPath=False
)
if not os.path.exists(realFile): if not os.path.exists(realFile):
raise Exception(f"File {realFile} does not exist.") raise Exception(f"File {realFile} does not exist.")
tmpDir: str = tempfile.gettempdir() tmpDir: str = tempfile.gettempdir()
tmpFile = os.path.join( tmpFile = os.path.join(tmpDir, f'ytptube_stream.{hashlib.md5(realFile.encode("utf-8")).hexdigest()}')
tmpDir, f'ytptube_stream.{hashlib.md5(realFile.encode("utf-8")).hexdigest()}')
if not os.path.exists(tmpFile): if not os.path.exists(tmpFile):
os.symlink(realFile, tmpFile) os.symlink(realFile, tmpFile)
@ -41,9 +35,7 @@ class Segments:
if self.segment_index == 0: if self.segment_index == 0:
startTime: float = '{:.6f}'.format(0) startTime: float = '{:.6f}'.format(0)
else: else:
startTime: float = '{:.6f}'.format( startTime: float = '{:.6f}'.format((self.segment_duration * self.segment_index))
(self.segment_duration * self.segment_index
))
ffmpegCmd = [] ffmpegCmd = []
ffmpegCmd.append('ffmpeg') ffmpegCmd.append('ffmpeg')
@ -107,8 +99,7 @@ class Segments:
ffmpegCmd.append('mpegts') ffmpegCmd.append('mpegts')
ffmpegCmd.append('pipe:1') ffmpegCmd.append('pipe:1')
log.debug( LOG.debug(f'Streaming {realFile} segment {self.segment_index}.' + ' '.join(ffmpegCmd))
f'Streaming {realFile} segment {self.segment_index}.' + ' '.join(ffmpegCmd))
proc = subprocess.run(ffmpegCmd, stdout=subprocess.PIPE) proc = subprocess.run(ffmpegCmd, stdout=subprocess.PIPE)
return proc.stdout return proc.stdout

26
app/player/ffprobe.py
View file

@ -12,6 +12,7 @@ import subprocess
class FFProbeError(Exception): class FFProbeError(Exception):
pass pass
class FFStream: class FFStream:
""" """
An object representation of an individual stream in a multimedia file. An object representation of an individual stream in a multimedia file.
@ -24,8 +25,7 @@ class FFStream:
try: try:
self.__dict__['framerate'] = round( self.__dict__['framerate'] = round(
functools.reduce( functools.reduce(
operator.truediv, map(int, self.__dict__.get( operator.truediv, map(int, self.__dict__.get('avg_frame_rate', '').split('/'))
'avg_frame_rate', '').split('/'))
) )
) )
except ValueError: except ValueError:
@ -41,8 +41,7 @@ class FFStream:
return f"<Stream: #{self.index} [{self.codec_type}] {self.codec_long_name}, {self.framerate}, ({self.width}x{self.height})>" return f"<Stream: #{self.index} [{self.codec_type}] {self.codec_long_name}, {self.framerate}, ({self.width}x{self.height})>"
if self.is_audio(): if self.is_audio():
return f"<Stream: #{self.index} [{self.codec_type}] {self.codec_long_name}, channels: {self.channels} ({self.channel_layout}), " \ return f"<Stream: #{self.index} [{self.codec_type}] {self.codec_long_name}, channels: {self.channels} ({self.channel_layout}), " "{sample_rate}Hz> "
"{sample_rate}Hz> "
if self.is_subtitle() or self.is_attachment(): if self.is_subtitle() or self.is_attachment():
return f"<Stream: #{self.index} [{self.codec_type}] {self.codec_long_name}>" return f"<Stream: #{self.index} [{self.codec_type}] {self.codec_long_name}>"
@ -87,8 +86,7 @@ class FFStream:
try: try:
size = (int(width), int(height)) size = (int(width), int(height))
except ValueError: except ValueError:
raise FFProbeError( raise FFProbeError("None integer size {}:{}".format(width, height))
"None integer size {}:{}".format(width, height))
else: else:
return None return None
@ -163,6 +161,7 @@ class FFStream:
except ValueError: except ValueError:
raise FFProbeError('None integer bit_rate') raise FFProbeError('None integer bit_rate')
class FFProbe: class FFProbe:
""" """
FFProbe wraps the ffprobe command and pulls the data into an object form:: FFProbe wraps the ffprobe command and pulls the data into an object form::
@ -181,8 +180,7 @@ class FFProbe:
try: try:
with open(os.devnull, 'w') as tempf: with open(os.devnull, 'w') as tempf:
subprocess.check_call( subprocess.check_call(["ffprobe", "-h"], stdout=tempf, stderr=tempf)
["ffprobe", "-h"], stdout=tempf, stderr=tempf)
except FileNotFoundError: except FileNotFoundError:
raise IOError('ffprobe not found.') raise IOError('ffprobe not found.')
@ -232,15 +230,3 @@ class FFProbe:
def __repr__(self): def __repr__(self):
return "<FFprobe: {metadata}, {video}, {audio}, {subtitle}, {attachment}>".format(**vars(self)) return "<FFprobe: {metadata}, {video}, {audio}, {subtitle}, {attachment}>".format(**vars(self))
if __name__ == '__main__':
import sys
if len(sys.argv) < 2:
print("Usage: ffprobe.py <path to media file>")
sys.exit(1)
ff = FFProbe(sys.argv[1])
print(ff)