Offload source search to the media worker

Author: Michal Warda

deploy/pipeline/sam-cinematic-continuous.env

@@ -42,7 +42,7 @@ SAM_CONTINUOUS_SOURCE_STAGING_DIR=/mnt/sam-source-staging
 # private NFS mount). Tasks left out of this list always execute locally.
 # SAM_MEDIA_WORKER_SSH_TARGET=ubuntu@172.31.0.10
 # SAM_MEDIA_WORKER_SSH_IDENTITY=/home/ubuntu/.ssh/sam-media-worker
-# SAM_MEDIA_WORKER_REMOTE_TASKS=download,ffmpeg
+# SAM_MEDIA_WORKER_REMOTE_TASKS=download,ffmpeg,search
 SAM_CONTINUOUS_DOWNLOADED_HIGH_WATER=512
 SAM_CONTINUOUS_DOWNLOADED_HIGH_WATER_BYTES=34359738368
 SAM_CONTINUOUS_PROVIDER_CIRCUIT_FAILURE_THRESHOLD=12

deploy/pipeline/sam-source-discovery.service

@@ -10,6 +10,7 @@ Group=ubuntu
 WorkingDirectory=/home/ubuntu/sam-audio-deploy/pipeline
 EnvironmentFile=/etc/sam-cinematic-continuous.env
 ExecStart=/home/ubuntu/sam-audio-deploy/deploy/pipeline/run-source-discovery.sh
+ExecStopPost=/home/ubuntu/sam-audio-deploy/pipeline/.venv/bin/python -m sam_audio_pipeline.remote_media cleanup search
 Restart=always
 RestartSec=5
 TimeoutStopSec=15

pipeline/src/sam_audio_pipeline/youtube_random.py

@@ -507,6 +507,18 @@ def _run(command: list[str], *, timeout: float) -> subprocess.CompletedProcess[s
     return completed
 
 
+def _run_search_command(
+    command: list[str], *, timeout: float
+) -> subprocess.CompletedProcess[str]:
+    """Run yt-dlp discovery work on the media host when one is configured."""
+    wrapped, wrapped_timeout = command_for_media_worker(
+        command,
+        task="search",
+        timeout=timeout,
+    )
+    return _run(wrapped, timeout=wrapped_timeout)
+
+
 def _terminate_process_group(process: subprocess.Popen[str]) -> tuple[str, str]:
     """Stop and reap a downloader plus every ffmpeg/ffprobe descendant."""
     try:
@@ -709,7 +721,7 @@ def _cinematic_candidate_priority(item: dict[str, Any]) -> int:
 
 
 def _search_youtube(query: str, results: int, profile: str) -> list[dict[str, Any]]:
-    response = _run(
+    response = _run_search_command(
         [
             YTDLP_PYTHON,
             "-m",
@@ -749,7 +761,7 @@ def _hydrate_search_result(
     affinity_key: str,
     timeout: float = 90,
 ) -> dict[str, Any]:
-    response = _run(
+    response = _run_search_command(
         [
             YTDLP_PYTHON,
             "-m",
@@ -786,7 +798,7 @@ def _search_ytdlp_provider(
     if provider.get("site"):
         provider_query = f"site:{provider['site']} {provider_query}"
     target = f"{provider['search_key']}{limited_results}:{provider_query}"
-    response = _run(
+    response = _run_search_command(
         [
             YTDLP_PYTHON,
             "-m",