89 lines
2.5 KiB
Python
89 lines
2.5 KiB
Python
import sqlite3
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Optional, List, Dict, Any
|
|
from app.config import settings
|
|
|
|
DB_PATH = os.path.join(settings.DATA_DIR, "sync.db")
|
|
|
|
def get_db():
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def init_db():
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
# Tracks table
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS downloaded_tracks (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
service TEXT NOT NULL,
|
|
service_id TEXT NOT NULL,
|
|
artist TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
album TEXT,
|
|
file_path TEXT NOT NULL,
|
|
downloaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE(service, service_id)
|
|
)
|
|
""")
|
|
|
|
# Task table
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS tasks (
|
|
id TEXT PRIMARY KEY,
|
|
type TEXT NOT NULL,
|
|
target TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
progress INTEGER DEFAULT 0,
|
|
total INTEGER DEFAULT 0,
|
|
message TEXT,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def is_track_downloaded(service: str, service_id: str) -> bool:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT 1 FROM downloaded_tracks WHERE service = ? AND service_id = ?", (service, str(service_id)))
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
return row is not None
|
|
|
|
def record_downloaded_track(service: str, service_id: str, artist: str, title: str, album: str, file_path: str):
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT OR REPLACE INTO downloaded_tracks (service, service_id, artist, title, album, file_path, downloaded_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""", (service, str(service_id), artist, title, album, file_path, datetime.utcnow()))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def get_recent_tracks(limit: int = 50) -> List[Dict[str, Any]]:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT id, service, service_id, artist, title, album, file_path, downloaded_at
|
|
FROM downloaded_tracks
|
|
ORDER BY downloaded_at DESC
|
|
LIMIT ?
|
|
""", (limit,))
|
|
rows = [dict(row) for row in cursor.fetchall()]
|
|
conn.close()
|
|
return rows
|
|
|
|
def get_tracks_count() -> int:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT COUNT(*) FROM downloaded_tracks")
|
|
count = cursor.fetchone()[0]
|
|
conn.close()
|
|
return count
|