feat(rec): audio analyzer and My Wave v1 engine (M4)

- migration 00006: track features and embeddings (pgvector, HNSW), analysis queue filled by a trigger on new tracks, wave sessions and impressions, recommender profiles, model versions - analyzer basic-1 (ADR-014): ffmpeg decode, three 30 s excerpts, 256-d timbre/rhythm embedding, tempo by onset autocorrelation, key, loudness, energy, danceability, acousticness, intro length; worker with SKIP LOCKED claims, retries and re-analysis on version change - wave engine v1: candidates (session/taste similarity, fresh, liked artists, rediscover, popular, exploration, seed), filters (dislikes, played in 3 h, already served), heuristic score, activity and mood fit from contexts.yaml with relaxing thresholds, MMR with one track per artist, exploration share, session vector updated by reactions and a change of course after three skips - POST /internal/wave/batch with the catalog index cached in memory and the session kept in Redis Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-20 12:20:55 +03:00
1 parent 976ffb574b
commit ef29b85320
@@ -0,0 +1,204 @@
+"""Воркер аудиоанализа: берёт задачи из analysis_jobs, считает признаки и пишет их в БД."""
+
+from __future__ import annotations
+
+import logging
+import subprocess
+import time
+from concurrent.futures import Future, ProcessPoolExecutor
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+import psycopg
+
+from pipu_rec.analysis.features import ANALYZER_VER, SAMPLE_RATE, Features, analyze
+
+log = logging.getLogger(__name__)
+
+MAX_ATTEMPTS = 3
+MAX_DURATION_S = 15 * 60
+IDLE_SLEEP_S = 10.0
+
+
+@dataclass(frozen=True)
+class Job:
+    track_id: str
+    path: str
+
+
+def decode(path: str, ffmpeg: str = "ffmpeg") -> np.ndarray:
+    """Моно float32 22 050 Гц; путь — отдельный аргумент, без shell."""
+    out = subprocess.run(
+        [
+            ffmpeg,
+            "-v",
+            "error",
+            "-t",
+            str(MAX_DURATION_S),
+            "-i",
+            path,
+            "-ac",
+            "1",
+            "-ar",
+            str(SAMPLE_RATE),
+            "-f",
+            "f32le",
+            "-",
+        ],
+        capture_output=True,
+        check=False,
+        timeout=300,
+    )
+    if out.returncode != 0:
+        raise RuntimeError(out.stderr.decode(errors="replace").strip()[:500] or "ffmpeg failed")
+    samples = np.frombuffer(out.stdout, dtype=np.float32)
+    if samples.size < SAMPLE_RATE:
+        raise RuntimeError("audio is shorter than 1 s")
+    return samples
+
+
+def analyze_file(path: str, ffmpeg: str) -> Features:
+    """Выполняется в процессе пула."""
+    return analyze(decode(path, ffmpeg))
+
+
+def vector_literal(v: np.ndarray) -> str:
+    return "[" + ",".join(f"{x:.6f}" for x in v.tolist()) + "]"
+
+
+class Worker:
+    def __init__(
+        self, dsn: str, library_root: str, host_root: str, workers: int, ffmpeg: str = "ffmpeg"
+    ) -> None:
+        self.dsn = dsn
+        self.library_root = Path(library_root)
+        self.host_root = Path(host_root)
+        self.workers = max(1, workers)
+        self.ffmpeg = ffmpeg
+
+    def resolve(self, source_id: str, source_type: str, local_path: str | None, rel_path: str) -> str:
+        """Путь к файлу внутри контейнера анализатора — те же точки монтирования, что у api."""
+        if source_type == "local":
+            return str(self.host_root.joinpath(local_path or "", rel_path))
+        return str(self.library_root.joinpath(source_id, rel_path))
+
+    def claim(self, conn: psycopg.Connection[Any], limit: int) -> list[Job]:
+        """Забирает задачи; SKIP LOCKED — несколько анализаторов не возьмут одну и ту же."""
+        rows = conn.execute(
+            """
+            UPDATE analysis_jobs j SET status = 'running', attempts = attempts + 1, updated_at = now()
+            WHERE j.track_id IN (
+                SELECT a.track_id FROM analysis_jobs a
+                JOIN tracks t ON t.id = a.track_id
+                JOIN media_files f ON f.id = t.primary_file_id
+                JOIN library_sources s ON s.id = f.source_id AND s.status = 'mounted'
+                WHERE a.status = 'queued'
+                ORDER BY a.updated_at
+                LIMIT %s
+                FOR UPDATE OF a SKIP LOCKED)
+            RETURNING j.track_id::text
+            """,
+            (limit,),
+        ).fetchall()
+        conn.commit()
+        if not rows:
+            return []
+        files = conn.execute(
+            """
+            SELECT t.id::text, s.id::text, s.type::text, s.local_path, f.rel_path
+            FROM tracks t
+            JOIN media_files f ON f.id = t.primary_file_id
+            JOIN library_sources s ON s.id = f.source_id
+            WHERE t.id = ANY(%s::uuid[])
+            """,
+            ([r[0] for r in rows],),
+        ).fetchall()
+        return [Job(track_id=r[0], path=self.resolve(r[1], r[2], r[3], r[4])) for r in files]
+
+    def save(self, conn: psycopg.Connection[Any], track_id: str, f: Features) -> None:
+        conn.execute(
+            """
+            INSERT INTO track_features (track_id, analyzer_ver, bpm, bpm_confidence, key, scale,
+                loudness_lufs, energy, danceability, valence, acousticness, intro_ms, analyzed_at)
+            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now())
+            ON CONFLICT (track_id) DO UPDATE SET analyzer_ver = EXCLUDED.analyzer_ver, bpm = EXCLUDED.bpm,
+                bpm_confidence = EXCLUDED.bpm_confidence, key = EXCLUDED.key, scale = EXCLUDED.scale,
+                loudness_lufs = EXCLUDED.loudness_lufs, energy = EXCLUDED.energy,
+                danceability = EXCLUDED.danceability, valence = EXCLUDED.valence,
+                acousticness = EXCLUDED.acousticness, intro_ms = EXCLUDED.intro_ms, analyzed_at = now()
+            """,
+            (
+                track_id,
+                ANALYZER_VER,
+                f.bpm,
+                f.bpm_confidence,
+                f.key,
+                f.scale,
+                f.loudness_lufs,
+                f.energy,
+                f.danceability,
+                f.valence,
+                f.acousticness,
+                f.intro_ms,
+            ),
+        )
+        conn.execute(
+            """
+            INSERT INTO track_embeddings (track_id, audio, updated_at) VALUES (%s, %s::vector, now())
+            ON CONFLICT (track_id) DO UPDATE SET audio = EXCLUDED.audio, updated_at = now()
+            """,
+            (track_id, vector_literal(f.embedding)),
+        )
+        conn.execute(
+            "UPDATE analysis_jobs SET status = 'done', error = NULL, updated_at = now() WHERE track_id = %s",
+            (track_id,),
+        )
+        conn.commit()
+
+    def fail(self, conn: psycopg.Connection[Any], track_id: str, err: str) -> None:
+        conn.execute(
+            """
+            UPDATE analysis_jobs SET status = CASE WHEN attempts >= %s THEN 'failed' ELSE 'queued' END,
+                error = %s, updated_at = now()
+            WHERE track_id = %s
+            """,
+            (MAX_ATTEMPTS, err[:500], track_id),
+        )
+        conn.commit()
+
+    def requeue_stale(self, conn: psycopg.Connection[Any]) -> None:
+        """Задачи, брошенные упавшим воркером, и признаки старой версии анализатора — снова в очередь."""
+        conn.execute(
+            "UPDATE analysis_jobs SET status = 'queued' "
+            "WHERE status = 'running' AND updated_at < now() - interval '30 minutes'"
+        )
+        conn.execute(
+            """
+            UPDATE analysis_jobs a SET status = 'queued', attempts = 0, updated_at = now()
+            FROM track_features f WHERE f.track_id = a.track_id AND f.analyzer_ver <> %s AND a.status = 'done'
+            """,
+            (ANALYZER_VER,),
+        )
+        conn.commit()
+
+    def run_forever(self) -> None:
+        log.info("analyzer started: workers=%d version=%s", self.workers, ANALYZER_VER)
+        with psycopg.connect(self.dsn) as conn, ProcessPoolExecutor(self.workers) as pool:
+            self.requeue_stale(conn)
+            while True:
+                jobs = self.claim(conn, self.workers * 2)
+                if not jobs:
+                    time.sleep(IDLE_SLEEP_S)
+                    continue
+                futures: list[tuple[Job, Future[Features]]] = [
+                    (j, pool.submit(analyze_file, j.path, self.ffmpeg)) for j in jobs
+                ]
+                for job, fut in futures:
+                    try:
+                        self.save(conn, job.track_id, fut.result(timeout=600))
+                    except Exception as exc:
+                        log.warning("analysis failed: track=%s err=%s", job.track_id, exc)
+                        conn.rollback()
+                        self.fail(conn, job.track_id, str(exc))