46 lines
1.2 KiB
Python
46 lines
1.2 KiB
Python
import asyncio
|
|
import uuid
|
|
import os
|
|
from pathlib import Path
|
|
|
|
AUDIO_TMP_DIR = "/tmp/apoena-audio"
|
|
|
|
|
|
async def extract_audio(url: str) -> Path:
|
|
"""Download audio-only from a URL using yt-dlp. Returns path to an mp3 temp file."""
|
|
job_id = str(uuid.uuid4())
|
|
outtmpl = f"{AUDIO_TMP_DIR}/{job_id}.%(ext)s"
|
|
expected = Path(f"{AUDIO_TMP_DIR}/{job_id}.mp3")
|
|
|
|
cmd = [
|
|
"yt-dlp",
|
|
"--no-warnings",
|
|
"--quiet",
|
|
"--extract-audio",
|
|
"--audio-format", "mp3",
|
|
"--audio-quality", "128K",
|
|
"--format", "bestaudio/best",
|
|
"--output", outtmpl,
|
|
url,
|
|
]
|
|
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
_, stderr = await proc.communicate()
|
|
|
|
if proc.returncode != 0:
|
|
raise RuntimeError(stderr.decode().strip() or "yt-dlp failed with no output")
|
|
|
|
if expected.exists():
|
|
return expected
|
|
|
|
# yt-dlp sometimes keeps the original extension even with --audio-format mp3
|
|
# Find whatever file was created with this job_id prefix
|
|
for f in Path(AUDIO_TMP_DIR).glob(f"{job_id}.*"):
|
|
return f
|
|
|
|
raise RuntimeError("yt-dlp produced no output file")
|