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

Merge pull request #82 from arabcoders/dev

Browse source 
Made the video player async to not block some requests.
This commit is contained in:
Abdulmohsen 2024-03-25 23:48:09 +03:00 committed by GitHub
commit 97a9cd0751
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 208 additions and 167 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
app/DataStore.py
View file

@ -28,13 +28,7 @@ class DataStore:
def load(self) -> None:
for id, item in self.saved_items():
self.dict.update({id: Download(
info=item,
download_dir=calcDownloadPath(basePath=self.config.download_path, folder=item.folder),
temp_dir=self.config.temp_path,
output_template_chapter=self.config.output_template_chapter,
default_ytdl_opts=self.config.ytdl_options)
})
self.dict.update({id: Download(info=item)})
def exists(self, key: str = None, url: str = None) -> bool:
if not key and not url:

87
app/Download.py
View file

@ -6,12 +6,12 @@ import os
import re
import shutil
import yt_dlp
import hashlib
from Utils import Notifier, get_format, get_opts, jsonCookie, mergeConfig
from ItemDTO import ItemDTO
from Config import Config
import hashlib
LOG = logging.getLogger('download')
@ -60,25 +60,18 @@ class Download:
tempKeep: bool = False
"Keep temp directory after download."
def __init__(
self,
info: ItemDTO,
download_dir: str,
temp_dir: str,
output_template_chapter: str,
default_ytdl_opts: dict,
info_dict: dict = None,
debug: bool = False
):
self.download_dir = download_dir
self.temp_dir = temp_dir
self.output_template_chapter = output_template_chapter
def __init__(self, info: ItemDTO, info_dict: dict = None, debug: bool = False):
config = Config.get_instance()
self.download_dir = info.download_dir
self.temp_dir = info.temp_dir
self.output_template_chapter = info.output_template_chapter
self.output_template = info.output_template
self.format = get_format(info.format, info.quality)
self.ytdl_opts = get_opts(info.format, info.quality, info.ytdlp_config if info.ytdlp_config else {})
self.info = info
self.id = info._id
self.default_ytdl_opts = default_ytdl_opts
self.default_ytdl_opts = config.ytdl_options
self.debug = debug
self.canceled = False
@ -87,8 +80,8 @@ class Download:
self.proc = None
self.loop = None
self.notifier = None
self.max_workers = int(Config.get_instance().max_workers)
self.tempKeep = bool(Config.get_instance().temp_keep)
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.info_dict = info_dict
@ -190,9 +183,7 @@ class Download:
LOG.info(f'Finished {os.getpid()=} id="{self.info.id}" title="{self.info.title}".')
async def start(self, notifier: Notifier):
self.manager = multiprocessing.Manager() if self.manager is None else self.manager
self.status_queue = self.manager.Queue()
self.status_queue = multiprocessing.Manager().Queue()
self.loop = asyncio.get_running_loop()
self.notifier = notifier
@ -205,33 +196,59 @@ class Download:
self.proc = multiprocessing.Process(target=self._download)
self.proc.start()
self.info.status = 'preparing'
await self.notifier.updated(self.info)
asyncio.create_task(self.notifier.updated(self.info))
asyncio.create_task(self.progress_update())
return await self.loop.run_in_executor(None, self.proc.join)
def started(self) -> bool:
return self.proc is not None
def cancel(self):
self.kill()
self.canceled = True
def cancel(self) -> bool:
if not self.started():
return False
def close(self):
if self.started():
if self.kill():
self.canceled = True
return True
def close(self) -> bool:
if not self.started():
return False
try:
LOG.info(f"Closing download process: '{self.proc.ident}'.")
self.proc.close()
self.delete_temp()
return True
except Exception as e:
LOG.error(f"Failed to close process: '{self.proc.ident}'. {e}")
self.delete_temp()
return False
def running(self) -> bool:
return self.started() and self.proc.is_alive()
try:
return self.proc is not None and self.proc.is_alive()
except ValueError:
return False
def is_canceled(self) -> bool:
return self.canceled
def kill(self):
if self.running():
LOG.info(f'Killing download process: {self.proc.ident}')
def kill(self) -> bool:
if not self.started():
return False
try:
LOG.info(f"Killing download process: '{self.proc.ident}'.")
self.proc.kill()
return True
except Exception as e:
LOG.error(f"Failed to kill process: '{self.proc.ident}'. {e}")
return False
def delete_temp(self):
if self.tempKeep is True or not self.tempPath:
@ -266,7 +283,7 @@ class Download:
LOG.debug(f'Status Update: {self.info._id=} {status=}')
if isinstance(status, str):
await self.notifier.updated(self.info)
asyncio.create_task(self.notifier.updated(self.info))
return
self.tmpfilename = status.get('tmpfilename')
@ -286,7 +303,7 @@ class Download:
if self.info.status == 'error' and 'error' in status:
self.info.error = status.get('error')
await self.notifier.error(self.info, self.info.error)
asyncio.create_task(self.notifier.error(self.info, self.info.error))
if 'downloaded_bytes' in status:
total = status.get('total_bytes') or status.get('total_bytes_estimate')
@ -305,4 +322,4 @@ class Download:
self.info.file_size = None
pass
await self.notifier.updated(self.info)
asyncio.create_task(self.notifier.updated(self.info))

62
app/DownloadQueue.py
View file

@ -126,6 +126,12 @@ class DownloadQueue:
live_status: list = ['is_live', 'is_upcoming']
is_live = entry.get('is_live', None) or live_in or entry.get('live_status', None) in live_status
try:
download_dir = calcDownloadPath(basePath=self.config.download_path, folder=folder)
except Exception as e:
LOG.exception(e)
return {'status': 'error', 'msg': str(e)}
dl = ItemDTO(
id=entry.get('id'),
title=entry.get('title'),
@ -133,37 +139,24 @@ class DownloadQueue:
quality=quality,
format=format,
folder=folder,
download_dir=download_dir,
temp_dir=self.config.temp_path,
ytdlp_cookies=ytdlp_cookies,
ytdlp_config=ytdlp_config,
output_template=output_template if output_template else self.config.output_template,
output_template_chapter=self.config.output_template_chapter,
datetime=formatdate(time.time()),
error=error,
is_live=is_live,
live_in=live_in,
options=options
options=options,
)
try:
download_dir = calcDownloadPath(basePath=self.config.download_path, folder=folder)
except Exception as e:
LOG.exception(e)
return {'status': 'error', 'msg': str(e)}
output_chapter = self.config.output_template_chapter
for property, value in entry.items():
if property.startswith("playlist"):
dl.output_template = dl.output_template.replace(f"%({property})s", str(value))
dlInfo: Download = Download(
info=dl,
download_dir=download_dir,
temp_dir=self.config.temp_path,
output_template_chapter=output_chapter,
default_ytdl_opts=self.config.ytdl_options,
info_dict=entry,
debug=bool(self.config.ytdl_debug)
)
dlInfo: Download = Download(info=dl, info_dict=entry, debug=bool(self.config.ytdl_debug))
if dlInfo.info.live_in or 'is_upcoming' == entry.get('live_status', None):
dlInfo.info.status = 'not_live'
@ -179,7 +172,7 @@ class DownloadQueue:
itemDownload = self.queue.put(dlInfo)
self.event.set()
await self.notifier.emit(NotifyEvent, itemDownload.info)
asyncio.create_task(self.notifier.emit(NotifyEvent, itemDownload.info))
return {
'status': 'ok'
@ -270,11 +263,15 @@ class DownloadQueue:
already=already,
)
async def cancel(self, ids):
async def cancel(self, ids: list[str]) -> dict[str, str]:
status: dict[str, str] = {"status": "ok"}
for id in ids:
try:
item = self.queue.get(key=id)
except KeyError as e:
status[id] = str(e)
status['status'] = 'error'
LOG.warning(f'Requested cancel for non-existent download {id=}. {str(e)}')
continue
@ -284,33 +281,39 @@ class DownloadQueue:
LOG.debug(f'Canceling {itemMessage}')
item.cancel()
LOG.info(f'Cancelled {itemMessage}')
else:
item.close()
LOG.debug(f'Deleting from queue {itemMessage}')
self.queue.delete(id)
await self.notifier.canceled(id)
asyncio.create_task(self.notifier.canceled(id))
self.done.put(item)
await self.notifier.completed(item)
asyncio.create_task(self.notifier.completed(item))
LOG.info(f'Deleted from queue {itemMessage}')
return {'status': 'ok'}
status[id] = 'ok'
return status
async def clear(self, ids: list[str]) -> dict[str, str]:
status: dict[str, str] = {"status": "ok"}
async def clear(self, ids):
for id in ids:
try:
item = self.done.get(key=id)
except KeyError as e:
status[id] = str(e)
status['status'] = 'error'
LOG.warning(f'Requested delete for non-existent download {id=}. {str(e)}')
continue
itemMessage = f"{id=} {item.info.id=} {item.info.title=}"
LOG.debug(f'Deleting completed download {itemMessage}')
self.done.delete(id)
await self.notifier.cleared(id)
asyncio.create_task(self.notifier.cleared(id))
LOG.info(f'Deleted completed download {itemMessage}')
status[id] = 'ok'
return {'status': 'ok'}
return status
def get(self) -> dict[str, list[dict[str, ItemDTO]]]:
items = {'queue': {}, 'done': {}}
@ -391,6 +394,7 @@ class DownloadQueue:
if entry.tmpfilename and os.path.isfile(entry.tmpfilename):
try:
os.remove(entry.tmpfilename)
entry.tmpfilename = None
except:
pass
@ -403,12 +407,12 @@ class DownloadQueue:
self.queue.delete(key=id)
if entry.is_canceled() is True:
await self.notifier.canceled(id)
asyncio.create_task(self.notifier.canceled(id))
entry.info.status = 'canceled'
entry.info.error = 'Canceled by user.'
self.done.put(value=entry)
await self.notifier.completed(entry.info)
asyncio.create_task(self.notifier.completed(entry.info))
self.event.set()

3
app/ItemDTO.py
View file

@ -16,10 +16,13 @@ class ItemDTO:
quality: str
format: str
folder: str
download_dir: str = None
temp_dir: str = None
status: str = None
ytdlp_cookies: str = None
ytdlp_config: dict = field(default_factory=dict)
output_template: str = None
output_template_chapter: str = None
timestamp: float = time.time_ns()
is_live: bool = None
datetime: str = field(default_factory=lambda: str(formatdate(time.time())))

4
app/main.py
View file

@ -291,6 +291,8 @@ class Main:
if not ids or where not in ['queue', 'done']:
raise web.HTTPBadRequest()
status: dict[str, str] = {}
status = await (self.queue.cancel(ids) if where == 'queue' else self.queue.clear(ids))
return web.Response(text=self.serializer.encode(status))
@ -415,7 +417,7 @@ class Main:
raise web.HTTPBadRequest(reason='file is required.')
return web.Response(
text=M3u8(url=f"{self.config.url_host}{self.config.url_prefix}").make_stream(
text=await M3u8(url=f"{self.config.url_host}{self.config.url_prefix}").make_stream(
download_path=self.config.download_path,
file=file
),

7
app/player/M3u8.py
View file

@ -16,19 +16,20 @@ class M3u8:
self.url = url
self.segment_duration = float(segment_duration)
def make_stream(self, download_path: str, file: str):
async def make_stream(self, download_path: str, file: 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 Exception(f"File '{realFile}' does not exist.")
try:
ffprobe = FFProbe(realFile)
await ffprobe.run()
except UnicodeDecodeError as e:
pass
if not 'duration' in ffprobe.metadata:
raise Exception(f"Unable to get {realFile} duration.")
raise Exception(f"Unable to get '{realFile}' duration.")
duration: float = float(ffprobe.metadata.get('duration'))

116
app/player/Segments.py
View file

@ -1,3 +1,4 @@
import asyncio
import hashlib
import logging
import os
@ -37,69 +38,80 @@ class Segments:
else:
startTime: float = '{:.6f}'.format((self.segment_duration * self.segment_index))
ffmpegCmd = []
ffmpegCmd.append('ffmpeg')
ffmpegCmd.append('-xerror')
ffmpegCmd.append('-hide_banner')
ffmpegCmd.append('-loglevel')
ffmpegCmd.append('error')
fargs = []
fargs.append('-xerror')
fargs.append('-hide_banner')
fargs.append('-loglevel')
fargs.append('error')
ffmpegCmd.append('-ss')
ffmpegCmd.append(str(startTime if startTime else '0.00000'))
ffmpegCmd.append('-t')
ffmpegCmd.append(str('{:.6f}'.format(self.segment_duration)))
fargs.append('-ss')
fargs.append(str(startTime if startTime else '0.00000'))
fargs.append('-t')
fargs.append(str('{:.6f}'.format(self.segment_duration)))
ffmpegCmd.append('-copyts')
fargs.append('-copyts')
ffmpegCmd.append('-i')
ffmpegCmd.append(f'file:{tmpFile}')
ffmpegCmd.append('-map_metadata')
ffmpegCmd.append('-1')
fargs.append('-i')
fargs.append(f'file:{tmpFile}')
fargs.append('-map_metadata')
fargs.append('-1')
ffmpegCmd.append('-pix_fmt')
ffmpegCmd.append('yuv420p')
ffmpegCmd.append('-g')
ffmpegCmd.append('52')
fargs.append('-pix_fmt')
fargs.append('yuv420p')
fargs.append('-g')
fargs.append('52')
ffmpegCmd.append('-map')
ffmpegCmd.append('0:v:0')
ffmpegCmd.append('-strict')
ffmpegCmd.append('-2')
fargs.append('-map')
fargs.append('0:v:0')
fargs.append('-strict')
fargs.append('-2')
ffmpegCmd.append('-codec:v')
ffmpegCmd.append('libx264' if self.vconvert else 'copy')
fargs.append('-codec:v')
fargs.append('libx264' if self.vconvert else 'copy')
if self.vconvert:
ffmpegCmd.append('-crf')
ffmpegCmd.append('23')
ffmpegCmd.append('-preset:v')
ffmpegCmd.append('fast')
ffmpegCmd.append('-level')
ffmpegCmd.append('4.1')
ffmpegCmd.append('-profile:v')
ffmpegCmd.append('baseline')
fargs.append('-crf')
fargs.append('23')
fargs.append('-preset:v')
fargs.append('fast')
fargs.append('-level')
fargs.append('4.1')
fargs.append('-profile:v')
fargs.append('baseline')
# audio section.
ffmpegCmd.append('-map')
ffmpegCmd.append('0:a:0')
ffmpegCmd.append('-codec:a')
ffmpegCmd.append('aac' if self.aconvert else 'copy')
fargs.append('-map')
fargs.append('0:a:0')
fargs.append('-codec:a')
fargs.append('aac' if self.aconvert else 'copy')
if self.aconvert:
ffmpegCmd.append('-b:a')
ffmpegCmd.append('192k')
ffmpegCmd.append('-ar')
ffmpegCmd.append('22050')
ffmpegCmd.append('-ac')
ffmpegCmd.append('2')
fargs.append('-b:a')
fargs.append('192k')
fargs.append('-ar')
fargs.append('22050')
fargs.append('-ac')
fargs.append('2')
ffmpegCmd.append('-sn')
fargs.append('-sn')
ffmpegCmd.append('-muxdelay')
ffmpegCmd.append('0')
ffmpegCmd.append('-f')
ffmpegCmd.append('mpegts')
ffmpegCmd.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.segment_index}.' + ' '.join(ffmpegCmd))
proc = subprocess.run(ffmpegCmd, stdout=subprocess.PIPE)
LOG.debug(f"Streaming '{realFile}' segment '{self.segment_index}'. " + " ".join(fargs))
return proc.stdout
proc = await asyncio.subprocess.create_subprocess_exec(
'ffmpeg', *fargs,
stdin=asyncio.subprocess.DEVNULL,
stdout=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.segment_index}. {err.decode("utf-8")}')
raise Exception(f'Failed to stream {realFile} segment {self.segment_index}.')
return data

87
app/player/ffprobe.py
View file

@ -1,8 +1,10 @@
"""
Python wrapper for ffprobe command line tool. ffprobe must exist in the path.
"""
import asyncio
import functools
import json
import logging
import operator
import os
import pipes
@ -178,55 +180,60 @@ class FFProbe:
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:
subprocess.check_call(["ffprobe", "-h"], stdout=tempf, stderr=tempf)
await asyncio.create_subprocess_exec(
"ffprobe", "-h", stdout=tempf, stderr=tempf
)
except FileNotFoundError:
raise IOError('ffprobe not found.')
if os.path.isfile(self.path_to_video):
cmd: list = [
"ffprobe -v quiet -of json -show_format -show_streams " +
pipes.quote(self.path_to_video)
]
if not os.path.isfile(self.path_to_video):
raise IOError(f"No such media file '{self.path_to_video}'.")
p = subprocess.Popen(
args=cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
shell=True
)
p.wait()
data, err = p.communicate()
if p.returncode == 0:
parsed: dict = json.loads(data.decode('utf-8'))
else:
raise FFProbeError(f"FFProbe error: {err}")
args = [
'-v', 'quiet',
'-of', 'json',
'-show_streams',
'-show_format',
self.path_to_video,
]
p = await asyncio.create_subprocess_exec(
'ffprobe', *args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stream = False
self.streams = []
self.video = []
self.audio = []
self.subtitle = []
self.attachment = []
exitCode = await p.wait()
for stream in parsed['streams'] if 'streams' in parsed else []:
self.streams.append(FFStream(stream))
self.metadata = parsed['format'] if 'format' in parsed else {}
for stream in self.streams:
if stream.is_audio():
self.audio.append(stream)
elif stream.is_video():
self.video.append(stream)
elif stream.is_subtitle():
self.subtitle.append(stream)
elif stream.is_attachment():
self.attachment.append(stream)
data, err = await p.communicate()
if 0 == exitCode:
parsed: dict = json.loads(data.decode('utf-8'))
else:
raise IOError('No such media file ' + self.path_to_video)
raise FFProbeError(f"FFProbe return with non-0 exit code. '{err.decode('utf-8')}'")
stream = False
self.streams = []
self.video = []
self.audio = []
self.subtitle = []
self.attachment = []
for stream in parsed.get('streams', []):
self.streams.append(FFStream(stream))
self.metadata = parsed.get('format', {})
for stream in self.streams:
if stream.is_audio():
self.audio.append(stream)
elif stream.is_video():
self.video.append(stream)
elif stream.is_subtitle():
self.subtitle.append(stream)
elif stream.is_attachment():
self.attachment.append(stream)
def __repr__(self):
return "<FFprobe: {metadata}, {video}, {audio}, {subtitle}, {attachment}>".format(**vars(self))

1
frontend/src/components/Video-Player.vue
View file

@ -72,6 +72,7 @@ const prepareVideoPlayer = () => {
enableWorker: true,
lowLatencyMode: true,
backBufferLength: 90,
fragLoadingTimeOut: 200000,
});
hls.value.loadSource(props.link)