database.py - master - op1x/MPPlus

database.py
81 строка3 KB
Op1x
финалочка
13 ноя 2025, 01:01
13 ноя 2025, 01:01aae47e2
import sqlite3
import os
from typing import List


class TrackDatabase:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.init_database()

    def init_database(self):
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS tracks (
                    path TEXT PRIMARY KEY,
                    filename TEXT NOT NULL,
                    liked INTEGER DEFAULT 0
                )
            ''')
            conn.commit()

    def add_tracks(self, track_paths: List[str]):
        tracks_data = []
        for path in track_paths:
            if os.path.isfile(path):
                filename = os.path.basename(path)
                tracks_data.append((path, filename))

        if not tracks_data:
            return

        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.executemany('''
                INSERT OR IGNORE INTO tracks (path, filename, liked) VALUES (?, ?, 0)
            ''', tracks_data)
            conn.commit()

    def toggle_like(self, track_path: str) -> int:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT liked FROM tracks WHERE path = ?', (track_path,))
            row = cursor.fetchone()

            if row is None:
                filename = os.path.basename(track_path)
                cursor.execute('''
                    INSERT INTO tracks (path, filename, liked) VALUES (?, ?, 1)
                ''', (track_path, filename))
                return 1
            else:
                # Меняем статус
                new_like = 0 if row[0] else 1
                cursor.execute('UPDATE tracks SET liked = ? WHERE path = ?', (new_like, track_path))
                return new_like

    def is_liked(self, track_path: str) -> int:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT liked FROM tracks WHERE path = ?', (track_path,))
            row = cursor.fetchone()
            return row[0] if row else 0

    def get_liked_tracks(self) -> List[str]:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT path FROM tracks WHERE liked = 1')
            rows = cursor.fetchall()
            # Фильтруем только существующие файлы
            return [row[0] for row in rows if os.path.isfile(row[0])]

    def get_all_tracks(self) -> List[dict]:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT path, filename, liked FROM tracks')
            rows = cursor.fetchall()
            return [
                {'path': row[0], 'filename': row[1], 'liked': row[2]}
                for row in rows if os.path.isfile(row[0])
            ]