feat(music-lab): music_batch_jobs 테이블 + 장르별 랜덤 풀

music-lab/app/db.py

@@ -264,6 +264,27 @@ def init_db() -> None:
             )
         """)
 
+        # ── music_batch_jobs 테이블 ──────────────────────────────────────
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS music_batch_jobs (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                genre TEXT NOT NULL,
+                count INTEGER NOT NULL,
+                target_duration_sec INTEGER NOT NULL DEFAULT 180,
+                auto_pipeline INTEGER NOT NULL DEFAULT 1,
+                completed INTEGER NOT NULL DEFAULT 0,
+                track_ids_json TEXT NOT NULL DEFAULT '[]',
+                current_track_index INTEGER NOT NULL DEFAULT 0,
+                current_track_status TEXT,
+                status TEXT NOT NULL DEFAULT 'queued',
+                error TEXT,
+                compile_job_id INTEGER,
+                pipeline_id INTEGER,
+                created_at TEXT NOT NULL,
+                updated_at TEXT NOT NULL
+            )
+        """)
+
         # ── YouTube pipeline 테이블 (5개) ─────────────────────────────────
         # track_id는 nullable: compile_job_id로 입력하는 essential mix 모드 지원
         conn.execute("""
@@ -1257,3 +1278,85 @@ def get_oauth_token() -> Optional[Dict[str, Any]]:
 def delete_oauth_token() -> None:
     with _conn() as conn:
         conn.execute("DELETE FROM youtube_oauth_tokens")
+
+
+# ── music_batch_jobs CRUD ─────────────────────────────────────────────────────
+
+_BATCH_ALLOWED_COLS = frozenset([
+    "completed", "track_ids_json", "current_track_index",
+    "current_track_status", "status", "error",
+    "compile_job_id", "pipeline_id",
+])
+
+
+def create_batch_job(genre: str, count: int, target_duration_sec: int = 180,
+                     auto_pipeline: bool = True) -> int:
+    with _conn() as conn:
+        now = _now()
+        cur = conn.cursor()
+        cur.execute("""
+            INSERT INTO music_batch_jobs
+              (genre, count, target_duration_sec, auto_pipeline,
+               status, created_at, updated_at)
+            VALUES (?, ?, ?, ?, 'queued', ?, ?)
+        """, (genre, count, target_duration_sec, 1 if auto_pipeline else 0, now, now))
+        return cur.lastrowid
+
+
+def get_batch_job(batch_id: int) -> dict | None:
+    with _conn() as conn:
+        row = conn.execute(
+            "SELECT * FROM music_batch_jobs WHERE id = ?", (batch_id,)
+        ).fetchone()
+    if not row:
+        return None
+    d = dict(row)
+    d["track_ids"] = json.loads(d.get("track_ids_json") or "[]")
+    return d
+
+
+def update_batch_job(batch_id: int, **fields) -> None:
+    unknown = set(fields) - _BATCH_ALLOWED_COLS
+    if unknown:
+        raise ValueError(f"unknown batch job columns: {unknown}")
+    if not fields:
+        return
+    cols = ", ".join(f"{k} = ?" for k in fields)
+    vals = list(fields.values()) + [_now(), batch_id]
+    with _conn() as conn:
+        conn.execute(
+            f"UPDATE music_batch_jobs SET {cols}, updated_at = ? WHERE id = ?",
+            vals,
+        )
+
+
+def append_batch_track(batch_id: int, track_id: int) -> None:
+    """track_ids_json에 새 track_id 추가 + completed 증가 (atomic)."""
+    with _conn() as conn:
+        row = conn.execute(
+            "SELECT track_ids_json, completed FROM music_batch_jobs WHERE id = ?",
+            (batch_id,),
+        ).fetchone()
+        if not row:
+            return
+        ids = json.loads(row["track_ids_json"] or "[]")
+        ids.append(track_id)
+        conn.execute(
+            "UPDATE music_batch_jobs SET track_ids_json = ?, completed = ?, updated_at = ? WHERE id = ?",
+            (json.dumps(ids), row["completed"] + 1, _now(), batch_id),
+        )
+
+
+def list_batch_jobs(active_only: bool = False) -> list[dict]:
+    sql = "SELECT * FROM music_batch_jobs"
+    if active_only:
+        sql += " WHERE status NOT IN ('failed','cancelled','piped')"
+    sql += " ORDER BY created_at DESC"
+    with _conn() as conn:
+        rows = conn.execute(sql).fetchall()
+    out = []
+    for r in rows:
+        d = dict(r)
+        d["track_ids"] = json.loads(d.get("track_ids_json") or "[]")
+        out.append(d)
+    return out