Spaces:
Running
Running
File size: 1,830 Bytes
037f340 | 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 | import sqlite3
import json
import os
from pathlib import Path
# The cache database will be stored in backend/data/cache.db
DATA_DIR = Path(__file__).parent.parent / "data"
DB_PATH = DATA_DIR / "cache.db"
class LyricsEmotionCache:
"""SQLite cache for storing DeBERTa lyrical emotion scores."""
def __init__(self):
import threading
self.lock = threading.Lock()
os.makedirs(DATA_DIR, exist_ok=True)
self.conn = sqlite3.connect(DB_PATH, check_same_thread=False)
self._init_db()
def _init_db(self):
with self.conn:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS lyrical_scores (
artist TEXT,
title TEXT,
probabilities JSON,
PRIMARY KEY (artist, title)
)
""")
def get_score(self, title: str, artist: str) -> dict:
"""Fetch cached probabilities for a track, if they exist."""
with self.lock:
cursor = self.conn.cursor()
cursor.execute(
"SELECT probabilities FROM lyrical_scores WHERE title = ? AND artist = ?",
(title, artist)
)
row = cursor.fetchone()
if row:
return json.loads(row[0])
return None
def set_score(self, title: str, artist: str, probabilities: dict):
"""Save probabilities to cache."""
with self.lock:
with self.conn:
self.conn.execute(
"INSERT OR REPLACE INTO lyrical_scores (artist, title, probabilities) VALUES (?, ?, ?)",
(artist, title, json.dumps(probabilities))
)
def close(self):
self.conn.close()
# Singleton instance
cache = LyricsEmotionCache()
|