mirror of
https://github.com/Dvorinka/swingmusic-extended.git
synced 2026-06-03 20:13:02 +00:00
remove deprecated db mappings
+ fix: cli password reset + delete old migrations
This commit is contained in:
cwilvx
@@ -1,21 +1,3 @@
|
||||
"""
|
||||
This module contains the functions to interact with the SQLite database.
|
||||
"""
|
||||
|
||||
import sqlite3
|
||||
from sqlite3 import Connection as SqlConn
|
||||
|
||||
|
||||
def create_connection(db_file: str) -> SqlConn:
|
||||
"""
|
||||
Creates a connection to the specified database.
|
||||
"""
|
||||
conn = sqlite3.connect(db_file)
|
||||
return conn
|
||||
|
||||
|
||||
def create_tables(conn: SqlConn, sql_query: str):
|
||||
"""
|
||||
Executes the specifiend SQL file to create database tables.
|
||||
"""
|
||||
conn.executescript(sql_query)
|
||||
|
||||
@@ -1,66 +0,0 @@
|
||||
from sqlite3 import Cursor
|
||||
|
||||
from .utils import SQLiteManager, tuples_to_albums
|
||||
|
||||
|
||||
class SQLiteAlbumMethods:
|
||||
@classmethod
|
||||
def insert_one_album(cls, cur: Cursor, albumhash: str, colors: str):
|
||||
"""
|
||||
Inserts one album into the database
|
||||
"""
|
||||
|
||||
sql = """INSERT OR REPLACE INTO albums(
|
||||
albumhash,
|
||||
colors
|
||||
) VALUES(?,?)
|
||||
"""
|
||||
|
||||
cur.execute(sql, (albumhash, colors))
|
||||
lastrowid = cur.lastrowid
|
||||
|
||||
return lastrowid
|
||||
|
||||
@classmethod
|
||||
def get_all_albums(cls):
|
||||
with SQLiteManager() as cur:
|
||||
cur.execute("SELECT * FROM albums")
|
||||
albums = cur.fetchall()
|
||||
cur.close()
|
||||
|
||||
if albums is not None:
|
||||
return albums
|
||||
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def get_albums_by_albumartist(albumartist: str):
|
||||
with SQLiteManager() as cur:
|
||||
cur.execute("SELECT * FROM albums WHERE albumartist=?", (albumartist,))
|
||||
albums = cur.fetchall()
|
||||
cur.close()
|
||||
|
||||
if albums is not None:
|
||||
return tuples_to_albums(albums)
|
||||
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def exists(albumhash: str, cur: Cursor = None):
|
||||
"""
|
||||
Checks if an album exists in the database.
|
||||
"""
|
||||
|
||||
sql = "SELECT COUNT(1) FROM albums WHERE albumhash = ?"
|
||||
|
||||
def _exists(cur: Cursor):
|
||||
cur.execute(sql, (albumhash,))
|
||||
count = cur.fetchone()[0]
|
||||
|
||||
return count != 0
|
||||
|
||||
if cur:
|
||||
return _exists(cur)
|
||||
|
||||
with SQLiteManager() as cur:
|
||||
return _exists(cur)
|
||||
@@ -1,64 +0,0 @@
|
||||
"""
|
||||
Contains methods for reading and writing to the sqlite artists database.
|
||||
"""
|
||||
|
||||
import json
|
||||
from sqlite3 import Cursor
|
||||
|
||||
from .utils import SQLiteManager
|
||||
|
||||
|
||||
class SQLiteArtistMethods:
|
||||
@staticmethod
|
||||
def insert_one_artist(cur: Cursor, artisthash: str, colors: list[str]):
|
||||
"""
|
||||
Inserts a single artist into the database.
|
||||
"""
|
||||
sql = """INSERT OR REPLACE INTO artists(
|
||||
artisthash,
|
||||
colors
|
||||
) VALUES(?,?)
|
||||
"""
|
||||
colors = json.dumps(colors)
|
||||
cur.execute(sql, (artisthash, colors))
|
||||
|
||||
@staticmethod
|
||||
def get_all_artists(cur_: Cursor = None):
|
||||
"""
|
||||
Get all artists from the database and return a generator of Artist objects
|
||||
"""
|
||||
sql = """SELECT * FROM artists"""
|
||||
|
||||
if not cur_:
|
||||
with SQLiteManager() as cur:
|
||||
cur.execute(sql)
|
||||
|
||||
for artist in cur.fetchall():
|
||||
yield artist
|
||||
|
||||
cur.close()
|
||||
|
||||
else:
|
||||
cur_.execute(sql)
|
||||
|
||||
for artist in cur_.fetchall():
|
||||
yield artist
|
||||
|
||||
@staticmethod
|
||||
def exists(artisthash: str, cur: Cursor = None):
|
||||
"""
|
||||
Checks if an artist exists in the database.
|
||||
"""
|
||||
sql = "SELECT COUNT(1) FROM artists WHERE artisthash = ?"
|
||||
|
||||
def _exists(cur: Cursor):
|
||||
cur.execute(sql, (artisthash,))
|
||||
count = cur.fetchone()[0]
|
||||
|
||||
return count != 0
|
||||
|
||||
if cur:
|
||||
return _exists(cur)
|
||||
|
||||
with SQLiteManager() as cur:
|
||||
return _exists(cur)
|
||||
@@ -1,144 +0,0 @@
|
||||
import json
|
||||
from app.models.user import User
|
||||
from app.utils.auth import hash_password
|
||||
from app.db.sqlite.utils import SQLiteManager
|
||||
|
||||
|
||||
class SQLiteAuthMethods:
|
||||
"""
|
||||
Methods for authenticating users.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def insert_user(user: dict[str, str]):
|
||||
"""
|
||||
Insert a user into the database.
|
||||
|
||||
:param user: A dict with the username, password and roles.
|
||||
"""
|
||||
sql = """INSERT INTO users(
|
||||
username,
|
||||
password,
|
||||
roles
|
||||
) VALUES(:username, :password, :roles)
|
||||
"""
|
||||
|
||||
user_tuple = tuple(user.values())
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur = cur.execute(sql, user_tuple)
|
||||
userid = cur.lastrowid
|
||||
return userid
|
||||
# if userid:
|
||||
# # sleep
|
||||
# user = SQLiteAuthMethods.get_user_by_id(userid).todict_simplified()
|
||||
# cur.close()
|
||||
# return user
|
||||
|
||||
raise Exception(f"Failed to insert user: {user}")
|
||||
|
||||
@staticmethod
|
||||
def insert_default_user():
|
||||
"""
|
||||
Inserts the default admin user.
|
||||
"""
|
||||
user = {
|
||||
"username": "admin",
|
||||
"password": hash_password("admin"),
|
||||
"roles": json.dumps(["admin"]),
|
||||
}
|
||||
return SQLiteAuthMethods.insert_user(user)
|
||||
|
||||
@staticmethod
|
||||
def insert_guest_user():
|
||||
"""
|
||||
Inserts the default guest user.
|
||||
"""
|
||||
user = {
|
||||
"username": "guest",
|
||||
"password": hash_password("guest"),
|
||||
"roles": json.dumps(["guest"]),
|
||||
}
|
||||
|
||||
return SQLiteAuthMethods.insert_user(user)
|
||||
|
||||
@staticmethod
|
||||
def update_user(user: dict[str, str]):
|
||||
"""
|
||||
Update a user in the database.
|
||||
|
||||
:param user: A dict with the user id and the fields to update. Ommited fields will not be updated.
|
||||
"""
|
||||
# get all user dict keys
|
||||
keys = list(user.keys())
|
||||
sql = f"""UPDATE users SET
|
||||
{', '.join([f"{key} = :{key}" for key in keys if key != 'id'])}
|
||||
WHERE id = :id
|
||||
"""
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, user)
|
||||
cur.close()
|
||||
|
||||
return SQLiteAuthMethods.get_user_by_id(user["id"]).todict()
|
||||
|
||||
@staticmethod
|
||||
def get_all_users():
|
||||
"""
|
||||
Check if there are any users in the database.
|
||||
"""
|
||||
sql = "SELECT * FROM users"
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql)
|
||||
|
||||
data = cur.fetchall()
|
||||
cur.close()
|
||||
|
||||
return [User(*user) for user in data]
|
||||
|
||||
@staticmethod
|
||||
def get_user_by_username(username: str):
|
||||
"""
|
||||
Get a user by username.
|
||||
"""
|
||||
sql = "SELECT * FROM users WHERE username = ?"
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (username,))
|
||||
|
||||
data = cur.fetchone()
|
||||
cur.close()
|
||||
|
||||
if data is not None:
|
||||
return User(*data)
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def get_user_by_id(userid: int):
|
||||
"""
|
||||
Get a user by id.
|
||||
"""
|
||||
sql = "SELECT * FROM users WHERE id = ?"
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (userid,))
|
||||
|
||||
data = cur.fetchone()
|
||||
cur.close()
|
||||
|
||||
if data is not None:
|
||||
return User(*data)
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def delete_user_by_username(username: str):
|
||||
"""
|
||||
Delete a user by username.
|
||||
"""
|
||||
sql = "DELETE FROM users WHERE id = ?"
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (3,))
|
||||
cur.close()
|
||||
@@ -1,121 +0,0 @@
|
||||
from datetime import datetime
|
||||
|
||||
from flask_jwt_extended import current_user
|
||||
from app.models import FavType
|
||||
from .utils import SQLiteManager
|
||||
|
||||
|
||||
class SQLiteFavoriteMethods:
|
||||
"""THis class contains methods for interacting with the favorites table."""
|
||||
|
||||
@classmethod
|
||||
def check_is_favorite(cls, itemhash: str, fav_type: str):
|
||||
"""
|
||||
Checks if an item is favorited.
|
||||
"""
|
||||
userid = current_user["id"]
|
||||
|
||||
sql = """SELECT * FROM favorites WHERE hash = ? AND type = ? AND userid = ?"""
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (itemhash, fav_type, userid))
|
||||
item = cur.fetchone()
|
||||
cur.close()
|
||||
return item is not None
|
||||
|
||||
@classmethod
|
||||
def insert_one_favorite(cls, fav_type: str, fav_hash: str):
|
||||
"""
|
||||
Inserts a single favorite into the database.
|
||||
"""
|
||||
# try to find the favorite in the database, if it exists, don't insert it
|
||||
if cls.check_is_favorite(fav_hash, fav_type):
|
||||
return
|
||||
|
||||
sql = """INSERT INTO favorites(type, hash, timestamp, userid) VALUES(?,?,?,?)"""
|
||||
current_timestamp = int(datetime.now().timestamp())
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
userid = current_user["id"]
|
||||
cur.execute(sql, (fav_type, fav_hash, current_timestamp, userid))
|
||||
cur.close()
|
||||
|
||||
@classmethod
|
||||
def get_all(cls) -> list[tuple]:
|
||||
"""
|
||||
Returns a list of all favorites.
|
||||
"""
|
||||
sql = """SELECT * FROM favorites WHERE userid = ?"""
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
userid = current_user["id"]
|
||||
cur.execute(sql, (userid,))
|
||||
favs = cur.fetchall()
|
||||
cur.close()
|
||||
return [fav for fav in favs if fav[1] != ""]
|
||||
|
||||
@classmethod
|
||||
def get_favorites(cls, fav_type: str, userid: int = None) -> list[tuple]:
|
||||
"""
|
||||
Returns a list of favorite tracks.
|
||||
|
||||
If userid is None, all favorites are returned.
|
||||
"""
|
||||
sql = """SELECT * FROM favorites WHERE type = ?"""
|
||||
params = (fav_type,)
|
||||
|
||||
if not userid:
|
||||
sql += " AND userid = ?"
|
||||
params = (fav_type, userid)
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, params)
|
||||
all_favs = cur.fetchall()
|
||||
cur.close()
|
||||
return all_favs
|
||||
|
||||
@classmethod
|
||||
def get_fav_tracks(cls, userid: int = None) -> list[tuple]:
|
||||
"""
|
||||
Returns a list of favorite tracks.
|
||||
"""
|
||||
return cls.get_favorites(FavType.track, userid)
|
||||
|
||||
@classmethod
|
||||
def get_fav_albums(cls) -> list[tuple]:
|
||||
"""
|
||||
Returns a list of favorite albums.
|
||||
"""
|
||||
userid = current_user["id"]
|
||||
return cls.get_favorites(FavType.album, userid)
|
||||
|
||||
@classmethod
|
||||
def get_fav_artists(cls) -> list[tuple]:
|
||||
"""
|
||||
Returns a list of favorite artists.
|
||||
"""
|
||||
userid = current_user["id"]
|
||||
return cls.get_favorites(FavType.artist, userid)
|
||||
|
||||
@classmethod
|
||||
def delete_favorite(cls, fav_type: str, fav_hash: str):
|
||||
"""
|
||||
Deletes a favorite from the database.
|
||||
"""
|
||||
sql = """DELETE FROM favorites WHERE hash = ? AND type = ? AND userid = ?"""
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
userid = current_user["id"]
|
||||
cur.execute(sql, (fav_hash, fav_type, userid))
|
||||
cur.close()
|
||||
|
||||
@classmethod
|
||||
def get_track_count(cls) -> int:
|
||||
"""
|
||||
Returns the number of favorite tracks.
|
||||
"""
|
||||
sql = """SELECT COUNT(*) FROM favorites WHERE type = ? AND userid = ?"""
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
userid = current_user["id"]
|
||||
cur.execute(sql, (FavType.track, userid))
|
||||
count = cur.fetchone()[0]
|
||||
cur.close()
|
||||
return count
|
||||
@@ -1,62 +0,0 @@
|
||||
from app.models.lastfm import SimilarArtist
|
||||
|
||||
from ..utils import SQLiteManager
|
||||
|
||||
|
||||
class SQLiteLastFMSimilarArtists:
|
||||
"""
|
||||
This class contains methods for interacting with the lastfm_similar_artists table.
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def insert_one(cls, artist: SimilarArtist):
|
||||
"""
|
||||
Inserts a single artist into the database.
|
||||
"""
|
||||
sql = """INSERT OR REPLACE INTO lastfm_similar_artists(artisthash, similar_artists) VALUES(?,?)"""
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (artist.artisthash, artist.similar_artists))
|
||||
cur.close()
|
||||
|
||||
@classmethod
|
||||
def get_similar_artists_for(cls, artisthash: str):
|
||||
"""
|
||||
Returns a list of similar artists.
|
||||
"""
|
||||
sql = """SELECT * FROM lastfm_similar_artists WHERE artisthash = ?"""
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (artisthash,))
|
||||
similar_artists = cur.fetchone()
|
||||
cur.close()
|
||||
|
||||
if similar_artists is None:
|
||||
return None
|
||||
|
||||
return SimilarArtist(artisthash, similar_artists[2])
|
||||
|
||||
@classmethod
|
||||
def get_all(cls):
|
||||
"""
|
||||
Returns a list of all similar artists.
|
||||
"""
|
||||
sql = """SELECT * FROM lastfm_similar_artists"""
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql)
|
||||
similar_artists = cur.fetchall()
|
||||
cur.close()
|
||||
|
||||
for a in similar_artists:
|
||||
yield SimilarArtist(a[1], a[2])
|
||||
|
||||
@classmethod
|
||||
def exists(cls, artisthash: str):
|
||||
"""
|
||||
Checks if an artist exists in the database by counting the number of rows
|
||||
"""
|
||||
sql = """SELECT COUNT(*) FROM lastfm_similar_artists WHERE artisthash = ?"""
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (artisthash,))
|
||||
count = cur.fetchone()[0]
|
||||
cur.close()
|
||||
return count > 0
|
||||
@@ -1,59 +0,0 @@
|
||||
from flask_jwt_extended import current_user
|
||||
from app.db.sqlite.utils import SQLiteManager
|
||||
from app.models.logger import TrackLog as TrackLog
|
||||
from app.utils.auth import get_current_userid
|
||||
|
||||
|
||||
class SQLiteTrackLogger:
|
||||
@classmethod
|
||||
def insert_track(cls, trackhash: str, duration: int, source: str, timestamp: int):
|
||||
"""
|
||||
Inserts a track play record into the database
|
||||
"""
|
||||
|
||||
userid = get_current_userid()
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
sql = """INSERT OR REPLACE INTO track_logger(
|
||||
trackhash,
|
||||
duration,
|
||||
timestamp,
|
||||
source,
|
||||
userid
|
||||
) VALUES(?,?,?,?,?)
|
||||
"""
|
||||
|
||||
cur.execute(
|
||||
sql, (trackhash, duration, timestamp, source, userid)
|
||||
)
|
||||
lastrowid = cur.lastrowid
|
||||
|
||||
return lastrowid
|
||||
|
||||
@classmethod
|
||||
def get_all(cls):
|
||||
"""
|
||||
Returns all track play records from the database
|
||||
"""
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
userid = get_current_userid()
|
||||
sql = f"""SELECT * FROM track_logger WHERE userid = {userid} ORDER BY timestamp DESC"""
|
||||
|
||||
cur.execute(sql)
|
||||
rows = cur.fetchall()
|
||||
|
||||
return rows
|
||||
|
||||
@classmethod
|
||||
def get_recently_played(cls, start: int = 0, limit: int = 100):
|
||||
"""
|
||||
Returns a list of recently played tracks
|
||||
"""
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
sql = f"""SELECT * FROM track_logger WHERE userid = {current_user['id']} ORDER BY timestamp DESC LIMIT ?,?"""
|
||||
|
||||
cur.execute(sql, (start, limit))
|
||||
rows = cur.fetchall()
|
||||
|
||||
return [TrackLog(*row) for row in rows]
|
||||
@@ -1,228 +0,0 @@
|
||||
import json
|
||||
from collections import OrderedDict
|
||||
|
||||
from flask_jwt_extended import current_user
|
||||
|
||||
from app.db.sqlite.utils import SQLiteManager, tuple_to_playlist, tuples_to_playlists
|
||||
from app.utils.dates import create_new_date
|
||||
|
||||
|
||||
class SQLitePlaylistMethods:
|
||||
"""
|
||||
This class contains methods for interacting with the playlists table.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def update_last_updated(playlist_id: int):
|
||||
"""Updates the last updated date of a playlist."""
|
||||
sql = """UPDATE playlists SET last_updated = ? WHERE id = ?"""
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (create_new_date(), playlist_id))
|
||||
|
||||
@staticmethod
|
||||
def insert_one_playlist(playlist: dict):
|
||||
# banner_pos,
|
||||
# has_gif,
|
||||
sql = """INSERT INTO playlists(
|
||||
image,
|
||||
last_updated,
|
||||
name,
|
||||
settings,
|
||||
trackhashes,
|
||||
userid
|
||||
) VALUES(:image, :last_updated, :name, :settings, :trackhashes, :userid)
|
||||
"""
|
||||
|
||||
playlist["userid"] = current_user["id"]
|
||||
playlist = OrderedDict(sorted(playlist.items()))
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, playlist)
|
||||
pid = cur.lastrowid
|
||||
cur.close()
|
||||
|
||||
p_tuple = (pid, *playlist.values())
|
||||
return tuple_to_playlist(p_tuple)
|
||||
|
||||
@staticmethod
|
||||
def count_playlist_by_name(name: str):
|
||||
sql = f"SELECT COUNT(*) FROM playlists WHERE name = ? and userid = {current_user['id']}"
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (name,))
|
||||
|
||||
data = cur.fetchone()
|
||||
cur.close()
|
||||
|
||||
return int(data[0])
|
||||
|
||||
@staticmethod
|
||||
def get_all_playlists():
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
userid = 1
|
||||
|
||||
try:
|
||||
userid = current_user["id"]
|
||||
except RuntimeError:
|
||||
# Catch this error raised during migration execution
|
||||
pass
|
||||
|
||||
cur.execute(f"SELECT * FROM playlists WHERE userid = {userid}")
|
||||
playlists = cur.fetchall()
|
||||
cur.close()
|
||||
|
||||
if playlists is not None:
|
||||
return tuples_to_playlists(playlists)
|
||||
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def get_playlist_by_id(playlist_id: int):
|
||||
sql = f"SELECT * FROM playlists WHERE id = ? and userid = {current_user['id']}"
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (playlist_id,))
|
||||
|
||||
data = cur.fetchone()
|
||||
cur.close()
|
||||
|
||||
if data is not None:
|
||||
return tuple_to_playlist(data)
|
||||
|
||||
return None
|
||||
|
||||
# FIXME: Extract the "add_track_to_playlist" method to use it for both the artisthash and trackhash lists.
|
||||
|
||||
@classmethod
|
||||
def add_item_to_json_list(cls, playlist_id: int, field: str, items: set[str]):
|
||||
"""
|
||||
Adds a string item to a json dumped list using a playlist id and field name.
|
||||
Takes the playlist ID, a field name, an item to add to the field.
|
||||
"""
|
||||
userid = 1
|
||||
|
||||
try:
|
||||
userid = current_user["id"]
|
||||
except RuntimeError:
|
||||
# Catch this error raised during migration execution
|
||||
pass
|
||||
|
||||
sql = f"SELECT {field} FROM playlists WHERE id = ? and userid = {userid}"
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (playlist_id,))
|
||||
data = cur.fetchone()
|
||||
|
||||
if data is not None:
|
||||
db_items: list[str] = json.loads(data[0])
|
||||
|
||||
# Remove duplicates, without changing the order.
|
||||
for item in items:
|
||||
if item in db_items:
|
||||
items.remove(item)
|
||||
|
||||
db_items.extend(items)
|
||||
|
||||
sql = f"UPDATE playlists SET {field} = ? WHERE id = ?"
|
||||
cur.execute(sql, (json.dumps(db_items), playlist_id))
|
||||
return len(items)
|
||||
|
||||
cls.update_last_updated(playlist_id)
|
||||
|
||||
@classmethod
|
||||
def add_tracks_to_playlist(cls, playlist_id: int, trackhashes: list[str]):
|
||||
"""
|
||||
Adds trackhashes to a playlist
|
||||
"""
|
||||
return cls.add_item_to_json_list(playlist_id, "trackhashes", trackhashes)
|
||||
|
||||
@classmethod
|
||||
def update_playlist(cls, playlist_id: int, playlist: dict):
|
||||
sql = f"""UPDATE playlists SET
|
||||
image = ?,
|
||||
last_updated = ?,
|
||||
name = ?,
|
||||
settings = ?
|
||||
WHERE id = ? and userid = {current_user['id']}
|
||||
"""
|
||||
|
||||
del playlist["id"]
|
||||
del playlist["trackhashes"]
|
||||
playlist["settings"] = json.dumps(playlist["settings"])
|
||||
|
||||
playlist = OrderedDict(sorted(playlist.items()))
|
||||
params = (*playlist.values(), playlist_id)
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, params)
|
||||
|
||||
cls.update_last_updated(playlist_id)
|
||||
|
||||
@classmethod
|
||||
def update_settings(cls, playlist_id: int, settings: dict):
|
||||
sql = f"""UPDATE playlists SET settings = ? WHERE id = ? and userid = {current_user['id']}"""
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (json.dumps(settings), playlist_id))
|
||||
|
||||
cls.update_last_updated(playlist_id)
|
||||
|
||||
@staticmethod
|
||||
def delete_playlist(pid: str):
|
||||
sql = f"DELETE FROM playlists WHERE id = ? and userid = {current_user['id']}"
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (pid,))
|
||||
|
||||
@staticmethod
|
||||
def remove_banner(playlistid: int):
|
||||
sql = f"""UPDATE playlists SET image = NULL WHERE id = ? and userid = {current_user['id']}"""
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(sql, (playlistid,))
|
||||
|
||||
@classmethod
|
||||
def remove_tracks_from_playlist(cls, playlistid: int, tracks: list[dict[str, int]]):
|
||||
"""
|
||||
Removes tracks from a playlist by trackhash and position.
|
||||
"""
|
||||
|
||||
sql = """UPDATE playlists SET trackhashes = ? WHERE id = ?"""
|
||||
userid = 1
|
||||
|
||||
try:
|
||||
userid = current_user["id"]
|
||||
except RuntimeError:
|
||||
# Catch this error raised during migration execution
|
||||
pass
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(
|
||||
f"SELECT trackhashes FROM playlists WHERE id = ? and userid = {userid}",
|
||||
(playlistid,),
|
||||
)
|
||||
data = cur.fetchone()
|
||||
|
||||
if data is None:
|
||||
return
|
||||
|
||||
trackhashes: list[str] = json.loads(data[0])
|
||||
to_remove = []
|
||||
|
||||
for track in tracks:
|
||||
# {
|
||||
# trackhash: str;
|
||||
# index: int;
|
||||
# }
|
||||
index = trackhashes.index(track["trackhash"])
|
||||
|
||||
if index == track["index"]:
|
||||
to_remove.append(track["trackhash"])
|
||||
|
||||
for trackhash in to_remove:
|
||||
trackhashes.remove(trackhash)
|
||||
|
||||
cur.execute(sql, (json.dumps(trackhashes), playlistid))
|
||||
|
||||
cls.update_last_updated(playlistid)
|
||||
@@ -1,83 +0,0 @@
|
||||
import json
|
||||
|
||||
from app.models.plugins import Plugin
|
||||
|
||||
from ..utils import SQLiteManager
|
||||
|
||||
|
||||
def plugin_tuple_to_obj(plugin_tuple: tuple) -> Plugin:
|
||||
return Plugin(
|
||||
name=plugin_tuple[1],
|
||||
active=bool(plugin_tuple[3]),
|
||||
settings=json.loads(plugin_tuple[4]),
|
||||
)
|
||||
|
||||
|
||||
class PluginsMethods:
|
||||
@classmethod
|
||||
def insert_plugin(cls, plugin: Plugin):
|
||||
"""
|
||||
Inserts one plugin into the database
|
||||
"""
|
||||
|
||||
sql = """INSERT OR IGNORE INTO plugins(
|
||||
name,
|
||||
description,
|
||||
active,
|
||||
settings
|
||||
) VALUES(?,?,?,?)
|
||||
"""
|
||||
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(
|
||||
sql,
|
||||
(
|
||||
plugin.name,
|
||||
plugin.description,
|
||||
int(plugin.active),
|
||||
json.dumps(plugin.settings),
|
||||
),
|
||||
)
|
||||
lastrowid = cur.lastrowid
|
||||
|
||||
return lastrowid
|
||||
|
||||
|
||||
@classmethod
|
||||
def get_all_plugins(cls):
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute("SELECT * FROM plugins")
|
||||
plugins = cur.fetchall()
|
||||
cur.close()
|
||||
|
||||
if plugins is not None:
|
||||
return [plugin_tuple_to_obj(plugin) for plugin in plugins]
|
||||
|
||||
return []
|
||||
|
||||
@classmethod
|
||||
def plugin_set_active(cls, name: str, active: int):
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute("UPDATE plugins SET active=? WHERE name=?", (active, name))
|
||||
cur.close()
|
||||
|
||||
@classmethod
|
||||
def update_plugin_settings(cls, plugin_name: str, settings: dict):
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute(
|
||||
"UPDATE plugins SET settings=? WHERE name=?",
|
||||
(json.dumps(settings), plugin_name),
|
||||
)
|
||||
cur.close()
|
||||
|
||||
@classmethod
|
||||
def get_plugin_by_name(cls, name: str):
|
||||
with SQLiteManager(userdata_db=True) as cur:
|
||||
cur.execute("SELECT * FROM plugins WHERE name=?", (name,))
|
||||
plugin = cur.fetchone()
|
||||
cur.close()
|
||||
|
||||
if plugin is not None:
|
||||
return plugin_tuple_to_obj(plugin)
|
||||
|
||||
return None
|
||||
@@ -1,123 +0,0 @@
|
||||
"""
|
||||
This file contains the SQL queries to create the database tables.
|
||||
"""
|
||||
|
||||
CREATE_USERDATA_TABLES = """
|
||||
CREATE TABLE IF NOT EXISTS playlists (
|
||||
id integer PRIMARY KEY,
|
||||
image text,
|
||||
last_updated text not null,
|
||||
name text not null,
|
||||
settings text,
|
||||
trackhashes text,
|
||||
userid integer not null,
|
||||
constraint fk_users foreign key (userid) references users(id) on delete cascade
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS settings (
|
||||
id integer PRIMARY KEY,
|
||||
root_dirs text NOT NULL,
|
||||
exclude_dirs text,
|
||||
artist_separators text NOT NULL default '/,;',
|
||||
extract_feat integer NOT NULL DEFAULT 1,
|
||||
remove_prod integer NOT NULL DEFAULT 1,
|
||||
clean_album_title integer NOT NULL DEFAULT 1,
|
||||
remove_remaster integer NOT NULL DEFAULT 1,
|
||||
merge_albums integer NOT NULL DEFAULT 0,
|
||||
show_albums_as_singles integer NOT NULL DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS lastfm_similar_artists (
|
||||
id integer PRIMARY KEY,
|
||||
artisthash text NOT NULL,
|
||||
similar_artists text NOT NULL,
|
||||
UNIQUE (artisthash)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS plugins (
|
||||
id integer PRIMARY KEY,
|
||||
name text NOT NULL UNIQUE,
|
||||
description text NOT NULL,
|
||||
active integer NOT NULL DEFAULT 0,
|
||||
settings text
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS track_logger (
|
||||
id integer PRIMARY KEY,
|
||||
trackhash text NOT NULL,
|
||||
duration integer NOT NULL,
|
||||
timestamp integer NOT NULL,
|
||||
source text,
|
||||
userid integer NOT NULL DEFAULT 1,
|
||||
constraint fk_users foreign key (userid) references users(id) on delete cascade
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
id integer PRIMARY KEY,
|
||||
username text NOT NULL UNIQUE,
|
||||
firstname text,
|
||||
lastname text,
|
||||
password text NOT NULL,
|
||||
email text,
|
||||
image text,
|
||||
roles text NOT NULL DEFAULT '["user"]'
|
||||
)
|
||||
"""
|
||||
|
||||
CREATE_APPDB_TABLES = """
|
||||
CREATE TABLE IF NOT EXISTS tracks (
|
||||
id integer PRIMARY KEY,
|
||||
album text NOT NULL,
|
||||
albumartist text NOT NULL,
|
||||
albumhash text NOT NULL,
|
||||
artist text NOT NULL,
|
||||
bitrate integer NOT NULL,
|
||||
copyright text,
|
||||
date integer NOT NULL,
|
||||
disc integer NOT NULL,
|
||||
duration integer NOT NULL,
|
||||
filepath text NOT NULL,
|
||||
folder text NOT NULL,
|
||||
genre text,
|
||||
title text NOT NULL,
|
||||
track integer NOT NULL,
|
||||
trackhash text NOT NULL,
|
||||
last_mod float NOT NULL,
|
||||
UNIQUE (filepath)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS albums (
|
||||
id integer PRIMARY KEY,
|
||||
albumhash text NOT NULL,
|
||||
colors text NOT NULL,
|
||||
UNIQUE (albumhash)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS artists (
|
||||
id integer PRIMARY KEY,
|
||||
artisthash text NOT NULL,
|
||||
colors text,
|
||||
bio text,
|
||||
UNIQUE (artisthash)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS folders (
|
||||
id integer PRIMARY KEY,
|
||||
path text NOT NULL,
|
||||
trackcount integer NOT NULL
|
||||
);
|
||||
"""
|
||||
|
||||
# changed from migrations to dbmigrations in v1.3.0
|
||||
# to avoid conflicts with the previous migrations.
|
||||
|
||||
CREATE_MIGRATIONS_TABLE = """
|
||||
CREATE TABLE IF NOT EXISTS dbmigrations (
|
||||
id integer PRIMARY KEY,
|
||||
version integer NOT NULL DEFAULT 0
|
||||
);
|
||||
|
||||
INSERT INTO dbmigrations (version)
|
||||
SELECT -1
|
||||
WHERE NOT EXISTS (SELECT 1 FROM dbmigrations);
|
||||
"""
|
||||
@@ -1,151 +0,0 @@
|
||||
from pprint import pprint
|
||||
from typing import Any
|
||||
|
||||
from app.config import UserConfig
|
||||
from app.db.sqlite.utils import SQLiteManager
|
||||
from app.utils.wintools import win_replace_slash
|
||||
|
||||
|
||||
# class SettingsSQLMethods:
|
||||
# """
|
||||
# Methods for interacting with the settings table.
|
||||
# """
|
||||
|
||||
# @staticmethod
|
||||
# def get_all_settings():
|
||||
# """
|
||||
# Gets all settings from the database.
|
||||
# """
|
||||
|
||||
# sql = "SELECT * FROM settings WHERE id = 1"
|
||||
|
||||
# with SQLiteManager(userdata_db=True) as cur:
|
||||
# cur.execute(sql)
|
||||
# settings = cur.fetchone()
|
||||
# cur.close()
|
||||
|
||||
# # if root_dirs not set
|
||||
# if settings is None:
|
||||
# return []
|
||||
|
||||
# # omit id, root_dirs, and exclude_dirs
|
||||
# return settings[3:]
|
||||
|
||||
# @staticmethod
|
||||
# def get_root_dirs() -> list[str]:
|
||||
# """
|
||||
# Gets custom root directories from the database.
|
||||
# """
|
||||
|
||||
# sql = "SELECT root_dirs FROM settings"
|
||||
|
||||
# with SQLiteManager(userdata_db=True) as cur:
|
||||
# cur.execute(sql)
|
||||
# dirs = cur.fetchall()
|
||||
# cur.close()
|
||||
|
||||
# dirs = [_dir[0] for _dir in dirs]
|
||||
# return [win_replace_slash(d) for d in dirs]
|
||||
|
||||
# @staticmethod
|
||||
# def add_root_dirs(dirs: list[str]):
|
||||
# """
|
||||
# Add custom root directories to the database.
|
||||
# """
|
||||
|
||||
# sql = "INSERT INTO settings (root_dirs) VALUES (?)"
|
||||
# existing_dirs = SettingsSQLMethods.get_root_dirs()
|
||||
|
||||
# dirs = [_dir for _dir in dirs if _dir not in existing_dirs]
|
||||
|
||||
# if len(dirs) == 0:
|
||||
# return
|
||||
|
||||
# with SQLiteManager(userdata_db=True) as cur:
|
||||
# for _dir in dirs:
|
||||
# cur.execute(sql, (_dir,))
|
||||
|
||||
# @staticmethod
|
||||
# def remove_root_dirs(dirs: list[str]):
|
||||
# """
|
||||
# Remove custom root directories from the database.
|
||||
# """
|
||||
|
||||
# sql = "DELETE FROM settings WHERE root_dirs = ?"
|
||||
|
||||
# with SQLiteManager(userdata_db=True) as cur:
|
||||
# for _dir in dirs:
|
||||
# cur.execute(sql, (_dir,))
|
||||
|
||||
# # Not currently used anywhere, to be used later
|
||||
# @staticmethod
|
||||
# def add_excluded_dirs(dirs: list[str]):
|
||||
# """
|
||||
# Add custom exclude directories to the database.
|
||||
# """
|
||||
|
||||
# sql = "INSERT INTO settings (exclude_dirs) VALUES (?)"
|
||||
|
||||
# with SQLiteManager(userdata_db=True) as cur:
|
||||
# cur.executemany(sql, dirs)
|
||||
|
||||
# @staticmethod
|
||||
# def remove_excluded_dirs(dirs: list[str]):
|
||||
# """
|
||||
# Remove custom exclude directories from the database.
|
||||
# """
|
||||
|
||||
# sql = "DELETE FROM settings WHERE exclude_dirs = ?"
|
||||
|
||||
# with SQLiteManager(userdata_db=True) as cur:
|
||||
# cur.executemany(sql, dirs)
|
||||
|
||||
# @staticmethod
|
||||
# def get_excluded_dirs() -> list[str]:
|
||||
# """
|
||||
# Gets custom exclude directories from the database.
|
||||
# """
|
||||
|
||||
# sql = "SELECT exclude_dirs FROM settings"
|
||||
|
||||
# with SQLiteManager(userdata_db=True) as cur:
|
||||
# cur.execute(sql)
|
||||
# dirs = cur.fetchall()
|
||||
# return [_dir[0] for _dir in dirs]
|
||||
|
||||
# @staticmethod
|
||||
# def get_settings() -> dict[str, Any]:
|
||||
# pass
|
||||
|
||||
# @staticmethod
|
||||
# def set_setting(key: str, value: Any):
|
||||
# sql = f"UPDATE settings SET {key} = :value WHERE id = 1"
|
||||
|
||||
# if type(value) == bool:
|
||||
# value = str(int(value))
|
||||
|
||||
# with SQLiteManager(userdata_db=True) as cur:
|
||||
# cur.execute(sql, {"value": value})
|
||||
|
||||
|
||||
# def load_settings():
|
||||
# # s = SettingsSQLMethods.get_all_settings()
|
||||
# config = UserConfig()
|
||||
|
||||
# try:
|
||||
# db_separators: str = s[0]
|
||||
# db_separators = db_separators.replace(" ", "")
|
||||
# separators = db_separators.split(",")
|
||||
# separators = set(separators)
|
||||
# except IndexError:
|
||||
# separators = {";", "/"}
|
||||
|
||||
# SessionVars.ARTIST_SEPARATORS = config.artistSeparators
|
||||
|
||||
# # boolean settings
|
||||
# SessionVars.EXTRACT_FEAT = bool(s[1])
|
||||
# SessionVars.REMOVE_PROD = bool(s[2])
|
||||
# SessionVars.CLEAN_ALBUM_TITLE = bool(s[3])
|
||||
# SessionVars.REMOVE_REMASTER_FROM_TRACK = bool(s[4])
|
||||
# SessionVars.MERGE_ALBUM_VERSIONS = bool(s[5])
|
||||
# SessionVars.SHOW_ALBUMS_AS_SINGLES = bool(s[6])
|
||||
@@ -1,135 +0,0 @@
|
||||
"""
|
||||
Contains the SQLiteTrackMethods class which contains methods for
|
||||
interacting with the tracks table.
|
||||
"""
|
||||
|
||||
from collections import OrderedDict
|
||||
from sqlite3 import Cursor
|
||||
|
||||
from app.db.sqlite.utils import tuple_to_track, tuples_to_tracks
|
||||
|
||||
from .utils import SQLiteManager
|
||||
from app.utils.unicode import handle_unicode
|
||||
|
||||
|
||||
class SQLiteTrackMethods:
|
||||
"""
|
||||
This class contains all methods for interacting with the tracks table.
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def insert_one_track(cls, track: dict, cur: Cursor):
|
||||
"""
|
||||
Inserts a single track into the database.
|
||||
"""
|
||||
sql = """INSERT OR REPLACE INTO tracks(
|
||||
album,
|
||||
albumartist,
|
||||
albumhash,
|
||||
artist,
|
||||
bitrate,
|
||||
copyright,
|
||||
date,
|
||||
disc,
|
||||
duration,
|
||||
filepath,
|
||||
folder,
|
||||
genre,
|
||||
last_mod,
|
||||
title,
|
||||
track,
|
||||
trackhash
|
||||
) VALUES(:album, :albumartist, :albumhash, :artist, :bitrate, :copyright,
|
||||
:date, :disc, :duration, :filepath, :folder, :genre, :last_mod, :title, :track, :trackhash)
|
||||
"""
|
||||
|
||||
track = OrderedDict(sorted(track.items()))
|
||||
|
||||
track["artist"] = track["artists"]
|
||||
track["albumartist"] = track["albumartists"]
|
||||
|
||||
del track["artists"]
|
||||
del track["albumartists"]
|
||||
|
||||
try:
|
||||
cur.execute(sql, track)
|
||||
except UnicodeEncodeError:
|
||||
# for each of the values in the track, call handle_unicode on it
|
||||
for key, value in track.items():
|
||||
track[key] = handle_unicode(value)
|
||||
|
||||
cur.execute(sql, track)
|
||||
|
||||
@classmethod
|
||||
def insert_many_tracks(cls, tracks: list[dict]):
|
||||
"""
|
||||
Inserts a list of tracks into the database.
|
||||
"""
|
||||
|
||||
with SQLiteManager() as cur:
|
||||
for track in tracks:
|
||||
cls.insert_one_track(track, cur)
|
||||
|
||||
@staticmethod
|
||||
def get_all_tracks():
|
||||
"""
|
||||
Get all tracks from the database and return a generator of Track objects
|
||||
or an empty list.
|
||||
"""
|
||||
with SQLiteManager() as cur:
|
||||
cur.execute("SELECT * FROM tracks")
|
||||
rows = cur.fetchall()
|
||||
|
||||
if rows is not None:
|
||||
return tuples_to_tracks(rows)
|
||||
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def get_track_by_trackhash(trackhash: str):
|
||||
"""
|
||||
Gets a track using its trackhash. Returns a Track object or None.
|
||||
"""
|
||||
with SQLiteManager() as cur:
|
||||
cur.execute("SELECT * FROM tracks WHERE trackhash=?", (trackhash,))
|
||||
row = cur.fetchone()
|
||||
|
||||
if row is not None:
|
||||
return tuple_to_track(row)
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def get_track_by_albumhash(albumhash: str):
|
||||
"""
|
||||
Gets a track using its albumhash. Returns a Track object or None.
|
||||
"""
|
||||
with SQLiteManager() as cur:
|
||||
cur.execute("SELECT * FROM tracks WHERE albumhash=?", (albumhash,))
|
||||
row = cur.fetchone()
|
||||
|
||||
if row is not None:
|
||||
return tuple_to_track(row)
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def remove_tracks_by_filepaths(filepaths: str | set[str]):
|
||||
"""
|
||||
Removes a track or tracks from the database using their filepaths.
|
||||
"""
|
||||
if isinstance(filepaths, str):
|
||||
filepaths = {filepaths}
|
||||
|
||||
with SQLiteManager() as cur:
|
||||
for filepath in filepaths:
|
||||
cur.execute("DELETE FROM tracks WHERE filepath=?", (filepath,))
|
||||
|
||||
@staticmethod
|
||||
def remove_tracks_not_in_folders(folders: set[str]):
|
||||
sql = "DELETE FROM tracks WHERE folder NOT IN ({})".format(
|
||||
",".join("?" * len(folders))
|
||||
)
|
||||
|
||||
with SQLiteManager() as cur:
|
||||
cur.execute(sql, tuple(folders))
|
||||
Reference in New Issue
Block a user