Spaces:
Running
Running
| 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() | |