71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
import subprocess
|
|
from typing import Literal
|
|
from dataclasses import dataclass
|
|
|
|
from app.utils.uploader import ChunkUploadBackend
|
|
|
|
|
|
@dataclass
|
|
class MediaContent:
|
|
url: str
|
|
headers: dict
|
|
chunk_size: int
|
|
|
|
|
|
class YtDlpManager:
|
|
def __init__(self, url: str, backend: ChunkUploadBackend):
|
|
self.url = url
|
|
self.backend = backend
|
|
self.info = None
|
|
self._extract_info()
|
|
|
|
def _extract_info(self):
|
|
import yt_dlp
|
|
|
|
ydl_opts = {
|
|
"quiet": True,
|
|
"skip_download": True,
|
|
"noplaylist": True,
|
|
"js-runtimes": "deno",
|
|
}
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
self.info = ydl.extract_info(self.url, download=False)
|
|
|
|
@property
|
|
def title(self) -> str:
|
|
"""Возвращает title видео"""
|
|
return self.info.get("title", "unknown")
|
|
|
|
def download_video(self, size: Literal[240, 360, 480, 720, 1080, 1440, 2160]):
|
|
command = [
|
|
"yt-dlp",
|
|
"-f",
|
|
f"bestvideo[height<={size}]",
|
|
"--no-part",
|
|
"--quiet",
|
|
"--no-warnings",
|
|
"-o",
|
|
"-",
|
|
self.url
|
|
]
|
|
print("Start processing")
|
|
process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=0)
|
|
print("Write filename to upload backend")
|
|
self.backend.start(self.title + ".mp4")
|
|
print("Start write chunk to upload backend")
|
|
chunk_size = 1024 ** 2
|
|
length = 0
|
|
while True:
|
|
chunk = process.stdout.read(chunk_size)
|
|
if not chunk:
|
|
break
|
|
length += chunk_size
|
|
print("Write chunk to backend", length)
|
|
self.backend.upload_chunk(chunk)
|
|
print("End writing to backend")
|
|
ret = process.wait()
|
|
print("Check ret status")
|
|
if ret != 0:
|
|
self.backend.abort()
|
|
raise RuntimeError(f"yt-dlp failed, status code: {ret}")
|