Spaces:
Running
Running
| import sqlite3 | |
| import json | |
| import os | |
| import threading | |
| from datetime import datetime | |
| from pathlib import Path | |
| # Save logs to the same data folder as the lyrical cache | |
| DATA_DIR = Path(__file__).parent.parent / "data" | |
| DB_PATH = DATA_DIR / "tracker.db" | |
| class TrackerDB: | |
| """SQLite database for tracking all stages of the recommendation pipeline.""" | |
| def __init__(self): | |
| 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): | |
| """Create the tracking table if it doesn't exist.""" | |
| with self.lock: | |
| with self.conn: | |
| self.conn.execute(''' | |
| CREATE TABLE IF NOT EXISTS requests_log ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, | |
| timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, | |
| user_text TEXT, | |
| emotion TEXT, | |
| confidence REAL, | |
| initial_100_tracks TEXT, | |
| scored_ranking TEXT, | |
| final_top_10 TEXT | |
| ) | |
| ''') | |
| def log_request(self, user_text: str, emotion: str, confidence: float, initial_100: list, scored_ranking: list, final_top_10: list): | |
| """Log a complete pipeline trace to the database.""" | |
| with self.lock: | |
| with self.conn: | |
| self.conn.execute( | |
| ''' | |
| INSERT INTO requests_log (user_text, emotion, confidence, initial_100_tracks, scored_ranking, final_top_10) | |
| VALUES (?, ?, ?, ?, ?, ?) | |
| ''', | |
| ( | |
| user_text, | |
| emotion, | |
| confidence, | |
| json.dumps(initial_100, ensure_ascii=False), | |
| json.dumps(scored_ranking, ensure_ascii=False), | |
| json.dumps(final_top_10, ensure_ascii=False) | |
| ) | |
| ) | |
| def close(self): | |
| self.conn.close() | |
| # Global singleton logger | |
| tracker = TrackerDB() | |