Probe YouTube imports through the dedicated API network

app.py

@@ -1,7 +1,9 @@
 """The dedicated ZeroGPU transcription API for Score Studio."""
-import logging
-import os
-from pathlib import Path
+import logging
+import json
+import os
+from pathlib import Path
+import re
 import shutil
 import subprocess
 import sys
@@ -10,8 +12,9 @@ import time
 
 import gradio as gr
 import spaces
-import torch
-from transformers import AutoModel
+import torch
+from transformers import AutoModel
+from yt_dlp import YoutubeDL
 
 MODEL_ID = "m-a-p/SheetSage2"
 REVISION = "eab522a8168e8b8b8c4856bf8609cd86198f01fe"
@@ -36,7 +39,7 @@ def gpu_seconds(audio, start, duration, melody_only, piano, progress=None, **kwa
 
 
 @spaces.GPU(duration=gpu_seconds)
-def transcribe(audio, start, duration, melody_only, piano, progress=gr.Progress()):
+def transcribe(audio, start, duration, melody_only, piano, progress=gr.Progress()):
     if not audio:
         raise gr.Error("Upload an audio file first.")
     if start < 0 or not 5 <= duration <= 600:
@@ -111,7 +114,46 @@ def transcribe(audio, start, duration, melody_only, piano, progress=gr.Progress(
         raise
     except Exception:
         logging.exception("Transcription failed")
-        raise gr.Error("Conversion failed. Try a shorter excerpt. Details are available in the Space logs.")
+        raise gr.Error("Conversion failed. Try a shorter excerpt. Details are available in the Space logs.")
+
+
+def import_youtube(url):
+    match = re.fullmatch(r"https?://(?:www\.|m\.|music\.)?(?:youtube\.com/(?:watch\?v=|shorts/|live/)|youtu\.be/)([A-Za-z0-9_-]{11})(?:[?&/].*)?", str(url).strip())
+    if not match:
+        raise gr.Error("Enter a valid YouTube video link.")
+    video_id = match.group(1)
+    job = Path(tempfile.mkdtemp(prefix="youtube-", dir=ROOT))
+    try:
+        options = {
+            "quiet": True, "no_warnings": True, "noplaylist": True,
+            "socket_timeout": 15, "retries": 1, "extractor_retries": 1,
+            "fragment_retries": 1, "js_runtimes": {"node": {}},
+            "format": "bestaudio/best", "outtmpl": str(job / "audio.%(ext)s"),
+            "max_filesize": 100 * 1024 * 1024,
+        }
+        with YoutubeDL(options) as ydl:
+            source = "https://www.youtube.com/watch?v=" + video_id
+            info = ydl.extract_info(source, download=False)
+            duration = float(info.get("duration") or 0)
+            if info.get("is_live") or info.get("age_limit", 0) > 0:
+                raise gr.Error("Live or age-restricted videos are not supported.")
+            if not 5 <= duration <= 600:
+                raise gr.Error("Choose a video between five seconds and ten minutes.")
+            ydl.process_info(info)
+        audio = next((p for p in job.glob("audio.*") if p.is_file() and p.stat().st_size > 2048), None)
+        if not audio:
+            raise gr.Error("No audio could be read from this video.")
+        metadata = json.dumps({"title": info.get("title", "YouTube audio"),
+            "artist": info.get("uploader", ""), "duration": duration,
+            "thumbnail": f"https://i.ytimg.com/vi/{video_id}/hqdefault.jpg", "url": source})
+        return str(audio), metadata
+    except gr.Error:
+        raise
+    except Exception as exc:
+        logging.warning("YouTube API import failed: %s", str(exc)[:800])
+        if "not a bot" in str(exc).lower() or "confirm you" in str(exc).lower():
+            raise gr.Error("YouTube blocked this server address.")
+        raise gr.Error("The API could not download this YouTube audio.")
 
 
 with gr.Blocks(title="Score Studio API", delete_cache=(3600, 86400)) as demo:
@@ -129,9 +171,15 @@ with gr.Blocks(title="Score Studio API", delete_cache=(3600, 86400)) as demo:
     preview = gr.Audio(label="Piano preview")
     downloads = gr.File(label="Downloads", file_count="multiple")
     abc = gr.Code(label="ABC notation", language=None)
-    run.click(transcribe, [audio, start, duration, melody, piano],
+    run.click(transcribe, [audio, start, duration, melody, piano],
               [status, gallery, pdf, preview, downloads, abc],
-              api_name="transcribe", concurrency_limit=1)
+              api_name="transcribe", concurrency_limit=1)
+    youtube_url = gr.Textbox(visible=False)
+    youtube_run = gr.Button(visible=False)
+    youtube_audio = gr.File(visible=False)
+    youtube_metadata = gr.Textbox(visible=False)
+    youtube_run.click(import_youtube, youtube_url, [youtube_audio, youtube_metadata],
+                      api_name="youtube_import", concurrency_limit=2)
     gr.Markdown("Powered by [SheetSage2](https://huggingface.co/m-a-p/SheetSage2), "
                 "CC BY-NC 4.0. Temporary files expire after 24 hours.")
 

packages.txt

@@ -1,4 +1,5 @@
 ffmpeg
+nodejs
 libsndfile1
 libglib2.0-0
 libnss3

requirements.txt

@@ -10,3 +10,4 @@ pretty_midi==0.2.10
 mido==1.3.3
 setuptools==78.1.1
 playwright==1.58.0
+yt-dlp[default]==2026.8.19

test_gpu_duration.py

@@ -0,0 +1,22 @@
+"""Check the actual Gradio-injected call signature without loading CUDA weights."""
+import ast
+from pathlib import Path
+import unittest
+
+tree = ast.parse(Path(__file__).with_name("app.py").read_text(encoding="utf-8"))
+definition = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "gpu_seconds")
+namespace = {}
+exec(compile(ast.Module(body=[definition], type_ignores=[]), "gpu-duration", "exec"), namespace)
+
+class DurationTests(unittest.TestCase):
+    def test_gradio_injects_progress_as_a_sixth_argument(self):
+        reserve = namespace["gpu_seconds"]("audio.wav", 0, 275, False, True, object())
+        self.assertGreaterEqual(reserve, 45)
+        self.assertLessEqual(reserve, 300)
+
+    def test_short_recording_does_not_reserve_full_song_budget(self):
+        function = namespace["gpu_seconds"]
+        self.assertLess(function("audio.wav", 0, 10, False, True), function("audio.wav", 0, 600, False, True))
+
+if __name__ == "__main__":
+    unittest.main()