50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
import asyncio
|
|
import tempfile
|
|
import signal
|
|
import os
|
|
from pathlib import Path
|
|
from dataclasses import dataclass
|
|
|
|
class Downloader:
|
|
"""Class that performs content download."""
|
|
|
|
#
|
|
# PUBLIC
|
|
#
|
|
def __init__(self, url: str):
|
|
self._url = url
|
|
|
|
async def download_video(self, directory: str | Path) -> Path:
|
|
directory = Path(directory).resolve()
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
|
|
with tempfile.TemporaryDirectory(dir=directory, prefix=".ytdlp-") as tmp:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"yt-dlp",
|
|
"-f", "bv*+ba/b",
|
|
"-P", f"home:{directory}",
|
|
"-P", f"temp:{tmp}",
|
|
"--no-playlist",
|
|
"--print", "after_move:filepath",
|
|
self._url,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
start_new_session=True,
|
|
)
|
|
|
|
try:
|
|
stdout, stderr = await proc.communicate()
|
|
|
|
except asyncio.CancelledError:
|
|
try:
|
|
os.killpg(proc.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
await asyncio.shield(proc.wait())
|
|
raise
|
|
|
|
if proc.returncode != 0:
|
|
raise RuntimeError(stderr.decode(errors="replace"))
|
|
|
|
return Path(stdout.decode().strip()) |