fix(worker): retrieve Soulseek downloads from slskd's dir into staging

SlskdClient enqueued the transfer and polled to Completed but never moved any
files into `dest` — slskd saves them to its OWN downloads dir, so staging stayed
empty and every Soulseek job failed "incomplete download". Qobuz/YouTube write
straight into staging and were unaffected.

- SlskdClient._retrieve() moves the finished files out of slskd's downloads dir
  into staging after completion, matching tolerantly by basename and preferring
  a parent-folder + size match (slskd's on-disk layout varies by version).
- download() returns the moved count as track_count so the pipeline's
  completeness check reflects what actually landed; raises loudly if nothing was
  found (misconfigured/unmounted downloads dir).
- Downloads dir resolves slskd.downloadsDir Config > SLSKD_DOWNLOADS_ROOT env >
  /slskd-downloads default; docker-compose mounts ${SLSKD_DOWNLOADS_DIR} there.
- .env.example documents SLSKD_DOWNLOADS_DIR; retrieval unit-tested offline.

slskd runs as an external daemon, so the shared dir must be a host path both
slskd and the worker mount — set once Lyra runs on the slskd host. Not yet
verified live end-to-end (no slskd access from this VM).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jonathan
2026-07-14 09:59:16 +02:00
parent c918cba654
commit f74fd5ab88
5 changed files with 163 additions and 1 deletions
+74 -1
View File
@@ -1,10 +1,13 @@
import json
import os
import shutil
import time
from typing import Callable
import requests
_DEFAULT_DOWNLOADS_ROOT = "/slskd-downloads"
_AUDIO_EXT = {"flac", "mp3", "m4a", "ogg", "opus", "wav", "aac"}
_LOSSLESS_EXT = {"flac", "wav"}
_SEARCH_POLLS = 40 # * 3s ≈ 2 min
@@ -34,6 +37,15 @@ class SlskdClient:
def __init__(self, config: dict):
self._url = (config.get("slskd.url", "") or "").rstrip("/")
self._key = config.get("slskd.api_key", "")
# Where slskd (an external daemon) writes finished downloads. slskd saves into ITS
# OWN directory, so this path must be a share/mount both slskd and this worker see.
# Precedence: the slskd.downloadsDir Config row (Settings-overridable) > the
# SLSKD_DOWNLOADS_ROOT env (the container mount point) > a sensible default.
self._downloads_dir = (
config.get("slskd.downloadsDir")
or os.environ.get("SLSKD_DOWNLOADS_ROOT")
or _DEFAULT_DOWNLOADS_ROOT
)
def is_configured(self) -> bool:
return bool(self._url and self._key)
@@ -138,4 +150,65 @@ class SlskdClient:
if not completed:
raise TimeoutError(f"soulseek download did not complete for {username}")
return {"track_count": total, "path": dest}
# slskd saved the files into its OWN downloads dir — retrieve them into `dest` so the
# pipeline (which verifies + imports against `dest`/staging) actually sees the audio.
moved = self._retrieve(files, dest)
if moved == 0:
raise RuntimeError(
f"soulseek transfer completed but no files were found under "
f"{self._downloads_dir!r} to move into staging — check that the slskd "
f"downloads dir is shared/mounted (SLSKD_DOWNLOADS_ROOT / slskd.downloadsDir)"
)
return {"track_count": moved, "path": dest}
def _retrieve(self, files: list[dict], dest: str) -> int:
"""Move the just-finished slskd files out of its downloads dir into `dest`.
slskd's on-disk layout can vary by version/config, so match tolerantly: for each
wanted file, find on-disk candidates by basename anywhere under the downloads root,
then prefer the one whose parent-folder name matches the remote directory and whose
size matches. Returns the number of files actually moved. Offline-testable.
"""
root = self._downloads_dir
if not root or not os.path.isdir(root):
raise RuntimeError(
f"slskd downloads dir not found: {root!r} — mount slskd's downloads "
f"directory into the worker (SLSKD_DOWNLOADS_ROOT / slskd.downloadsDir)"
)
os.makedirs(dest, exist_ok=True)
wanted: dict[str, dict] = {} # basename(lower) -> {parent, size}
for f in files:
name = f.get("filename", "")
wanted[_basename(name).lower()] = {
"parent": _basename(_dirname(name)).lower(),
"size": f.get("size"),
}
found: dict[str, list[str]] = {b: [] for b in wanted}
for cur_root, _dirs, fnames in os.walk(root):
for fn in fnames:
key = fn.lower()
if key in found:
found[key].append(os.path.join(cur_root, fn))
moved = 0
for basename, paths in found.items():
if not paths:
continue
want = wanted[basename]
def _score(p: str) -> tuple[int, int]:
parent_ok = int(os.path.basename(os.path.dirname(p)).lower() == want["parent"])
size_ok = 0
try:
size_ok = int(bool(want["size"]) and os.path.getsize(p) == want["size"])
except OSError:
pass
return (parent_ok, size_ok)
best = max(paths, key=_score)
shutil.move(best, os.path.join(dest, os.path.basename(best)))
moved += 1
return moved