init

app/downloader.py

@@ -0,0 +1,45 @@
+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")