Spaces:
Running
Running
File size: 2,160 Bytes
4ae09a8 ce489e3 4ae09a8 ce489e3 4ae09a8 ce489e3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | from datetime import datetime, timezone
from typing import Any
from slugify import slugify as _slugify
from supabase import Client
def slugify(title: str, artist: str, max_len: int = 80) -> str:
raw = _slugify(f"{title} {artist}")
return raw[:max_len].rstrip("-")
_TERMINAL = {"ready", "failed_fetch", "failed_separation", "failed_other"}
class SongRepo:
def __init__(self, client: Client):
self._c = client
# songs ---------------------------------------------------------------
def get_song(self, youtube_id: str) -> dict | None:
res = (
self._c.table("songs").select("*")
.eq("youtube_id", youtube_id).maybe_single().execute()
)
return res.data if res else None
def get_song_by_slug(self, slug: str) -> dict | None:
res = (
self._c.table("songs").select("*")
.eq("slug", slug).eq("status", "ready").maybe_single().execute()
)
return res.data if res else None
def upsert_song(self, row: dict[str, Any]) -> None:
self._c.table("songs").upsert(row, on_conflict="youtube_id").execute()
def increment_play_count(self, youtube_id: str) -> None:
# Best-effort, ignore the row if missing.
self._c.rpc("increment_play_count", {"yt_id": youtube_id}).execute()
# jobs ----------------------------------------------------------------
def create_job(self, youtube_id: str) -> str:
res = self._c.table("jobs").insert(
{"youtube_id": youtube_id, "status": "queued"}
).execute()
return res.data[0]["id"]
def update_job(self, job_id: str, *, status: str, error: str | None = None) -> None:
patch: dict[str, Any] = {"status": status, "error": error}
if status in _TERMINAL:
patch["finished_at"] = datetime.now(timezone.utc).isoformat()
self._c.table("jobs").update(patch).eq("id", job_id).execute()
def get_job(self, job_id: str) -> dict | None:
res = (
self._c.table("jobs").select("*")
.eq("id", job_id).maybe_single().execute()
)
return res.data if res else None
|