338 lines
10 KiB
Python
338 lines
10 KiB
Python
# Standard library imports
|
|
from typing import Optional
|
|
from dataclasses import asdict
|
|
import datetime as dt
|
|
import json
|
|
|
|
# PyQt imports
|
|
|
|
# Third party imports
|
|
from alchemical import Model # type: ignore
|
|
from sqlalchemy import (
|
|
Boolean,
|
|
DateTime,
|
|
ForeignKey,
|
|
String,
|
|
)
|
|
from sqlalchemy.ext.associationproxy import association_proxy
|
|
from sqlalchemy.engine.interfaces import Dialect
|
|
from sqlalchemy.orm import (
|
|
Mapped,
|
|
mapped_column,
|
|
relationship,
|
|
)
|
|
from sqlalchemy.orm.session import Session
|
|
from sqlalchemy.types import TypeDecorator, TEXT
|
|
|
|
# App imports
|
|
from classes import Filter
|
|
|
|
|
|
class JSONEncodedDict(TypeDecorator):
|
|
"""
|
|
Custom JSON Type for MariaDB (since native JSON type is just LONGTEXT)
|
|
"""
|
|
|
|
impl = TEXT
|
|
|
|
def process_bind_param(self, value: dict | None, dialect: Dialect) -> str | None:
|
|
"""Convert Python dictionary to JSON string before saving."""
|
|
if value is None:
|
|
return None
|
|
return json.dumps(value, default=lambda o: o.__dict__)
|
|
|
|
def process_result_value(self, value: str | None, dialect: Dialect) -> dict | None:
|
|
"""Convert JSON string back to Python dictionary after retrieval."""
|
|
if value is None:
|
|
return None
|
|
return json.loads(value)
|
|
|
|
|
|
# Database classes
|
|
class NoteColours(Model):
|
|
__tablename__ = "notecolours"
|
|
|
|
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
|
|
substring: Mapped[str] = mapped_column(String(256), index=True, unique=True)
|
|
colour: Mapped[str] = mapped_column(String(21), index=False)
|
|
enabled: Mapped[bool] = mapped_column(default=True, index=True)
|
|
foreground: Mapped[Optional[str]] = mapped_column(String(21), index=False)
|
|
is_regex: Mapped[bool] = mapped_column(default=False, index=False)
|
|
is_casesensitive: Mapped[bool] = mapped_column(default=False, index=False)
|
|
order: Mapped[Optional[int]] = mapped_column(index=True)
|
|
strip_substring: Mapped[bool] = mapped_column(default=True, index=False)
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<NoteColours(id={self.id}, substring={self.substring}, "
|
|
f"colour={self.colour}>"
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
session: Session,
|
|
substring: str,
|
|
colour: str,
|
|
enabled: bool = True,
|
|
is_regex: bool = False,
|
|
is_casesensitive: bool = False,
|
|
order: Optional[int] = 0,
|
|
) -> None:
|
|
self.substring = substring
|
|
self.colour = colour
|
|
self.enabled = enabled
|
|
self.is_regex = is_regex
|
|
self.is_casesensitive = is_casesensitive
|
|
self.order = order
|
|
|
|
session.add(self)
|
|
session.commit()
|
|
|
|
|
|
class Playdates(Model):
|
|
__tablename__ = "playdates"
|
|
|
|
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
|
|
lastplayed: Mapped[dt.datetime] = mapped_column(index=True)
|
|
track_id: Mapped[int] = mapped_column(ForeignKey("tracks.id", ondelete="CASCADE"))
|
|
track: Mapped["Tracks"] = relationship(
|
|
"Tracks",
|
|
back_populates="playdates",
|
|
)
|
|
|
|
def __init__(
|
|
self, session: Session, track_id: int, when: dt.datetime | None = None
|
|
) -> None:
|
|
"""Record that track was played"""
|
|
|
|
if not when:
|
|
self.lastplayed = dt.datetime.now()
|
|
else:
|
|
self.lastplayed = when
|
|
self.track_id = track_id
|
|
|
|
session.add(self)
|
|
session.commit()
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<Playdates(id={self.id}, track_id={self.track_id} "
|
|
f"lastplayed={self.lastplayed}>"
|
|
)
|
|
|
|
|
|
class Playlists(Model):
|
|
"""
|
|
Manage playlists
|
|
"""
|
|
|
|
__tablename__ = "playlists"
|
|
|
|
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
|
|
name: Mapped[str] = mapped_column(String(32), unique=True)
|
|
last_used: Mapped[Optional[dt.datetime]] = mapped_column(DateTime, default=None)
|
|
tab: Mapped[Optional[int]] = mapped_column(default=None)
|
|
open: Mapped[bool] = mapped_column(default=False)
|
|
is_template: Mapped[bool] = mapped_column(default=False)
|
|
rows: Mapped[list["PlaylistRows"]] = relationship(
|
|
"PlaylistRows",
|
|
back_populates="playlist",
|
|
cascade="all, delete-orphan",
|
|
order_by="PlaylistRows.row_number",
|
|
)
|
|
favourite: Mapped[bool] = mapped_column(
|
|
Boolean, nullable=False, index=False, default=False
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<Playlists(id={self.id}, name={self.name}, "
|
|
f"is_templatee={self.is_template}, open={self.open}>"
|
|
)
|
|
|
|
def __init__(self, session: Session, name: str, template_id: int) -> None:
|
|
"""Create playlist with passed name"""
|
|
|
|
self.name = name
|
|
self.last_used = dt.datetime.now()
|
|
|
|
session.add(self)
|
|
session.commit()
|
|
|
|
# If a template is specified, copy from it
|
|
if template_id:
|
|
PlaylistRows.copy_playlist(session, template_id, self.id)
|
|
|
|
|
|
class PlaylistRows(Model):
|
|
__tablename__ = "playlist_rows"
|
|
|
|
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
|
|
row_number: Mapped[int] = mapped_column(index=True)
|
|
note: Mapped[str] = mapped_column(
|
|
String(2048), index=False, default="", nullable=False
|
|
)
|
|
playlist_id: Mapped[int] = mapped_column(
|
|
ForeignKey("playlists.id", ondelete="CASCADE"), index=True
|
|
)
|
|
|
|
playlist: Mapped[Playlists] = relationship(back_populates="rows")
|
|
track_id: Mapped[Optional[int]] = mapped_column(
|
|
ForeignKey("tracks.id", ondelete="CASCADE")
|
|
)
|
|
track: Mapped["Tracks"] = relationship(
|
|
"Tracks",
|
|
back_populates="playlistrows",
|
|
)
|
|
played: Mapped[bool] = mapped_column(
|
|
Boolean, nullable=False, index=False, default=False
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<PlaylistRows(id={self.id}, playlist_id={self.playlist_id}, "
|
|
f"track_id={self.track_id}, "
|
|
f"note={self.note}, row_number={self.row_number}>"
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
session: Session,
|
|
playlist_id: int,
|
|
row_number: int,
|
|
note: str = "",
|
|
track_id: Optional[int] = None,
|
|
) -> None:
|
|
"""Create PlaylistRows object"""
|
|
|
|
self.playlist_id = playlist_id
|
|
self.track_id = track_id
|
|
self.row_number = row_number
|
|
self.note = note
|
|
|
|
session.add(self)
|
|
session.commit()
|
|
|
|
|
|
class Queries(Model):
|
|
__tablename__ = "queries"
|
|
|
|
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
|
|
name: Mapped[str] = mapped_column(String(128), nullable=False)
|
|
_filter_data: Mapped[dict | None] = mapped_column("filter_data", JSONEncodedDict, nullable=False)
|
|
favourite: Mapped[bool] = mapped_column(Boolean, nullable=False, index=False, default=False)
|
|
|
|
def _get_filter(self) -> Filter:
|
|
"""Convert stored JSON dictionary to a Filter object."""
|
|
if isinstance(self._filter_data, dict):
|
|
return Filter(**self._filter_data)
|
|
return Filter() # Default object if None or invalid data
|
|
|
|
def _set_filter(self, value: Filter | None) -> None:
|
|
"""Convert a Filter object to JSON before storing."""
|
|
self._filter_data = asdict(value) if isinstance(value, Filter) else None
|
|
|
|
# Single definition of `filter`
|
|
filter = property(_get_filter, _set_filter)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<Queries(id={self.id}, name={self.name}, filter={self.filter})>"
|
|
|
|
def __init__(
|
|
self,
|
|
session: Session,
|
|
name: str,
|
|
filter: Filter,
|
|
favourite: bool = False,
|
|
) -> None:
|
|
"""Create new query"""
|
|
|
|
self.name = name
|
|
self.filter = filter
|
|
self.favourite = favourite
|
|
|
|
session.add(self)
|
|
session.commit()
|
|
|
|
|
|
class Settings(Model):
|
|
"""Manage settings"""
|
|
|
|
__tablename__ = "settings"
|
|
|
|
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
|
|
name: Mapped[str] = mapped_column(String(64), unique=True)
|
|
f_datetime: Mapped[Optional[dt.datetime]] = mapped_column(default=None)
|
|
f_int: Mapped[Optional[int]] = mapped_column(default=None)
|
|
f_string: Mapped[Optional[str]] = mapped_column(String(128), default=None)
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<Settings(id={self.id}, name={self.name}, "
|
|
f"f_datetime={self.f_datetime}, f_int={self.f_int}, f_string={self.f_string}>"
|
|
)
|
|
|
|
def __init__(self, session: Session, name: str) -> None:
|
|
self.name = name
|
|
|
|
session.add(self)
|
|
session.commit()
|
|
|
|
|
|
class Tracks(Model):
|
|
__tablename__ = "tracks"
|
|
|
|
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
|
|
artist: Mapped[str] = mapped_column(String(256), index=True)
|
|
bitrate: Mapped[int] = mapped_column(default=None)
|
|
duration: Mapped[int] = mapped_column(index=True)
|
|
fade_at: Mapped[int] = mapped_column(index=False)
|
|
intro: Mapped[Optional[int]] = mapped_column(default=None)
|
|
path: Mapped[str] = mapped_column(String(2048), index=False, unique=True)
|
|
silence_at: Mapped[int] = mapped_column(index=False)
|
|
start_gap: Mapped[int] = mapped_column(index=False)
|
|
title: Mapped[str] = mapped_column(String(256), index=True)
|
|
|
|
playlistrows: Mapped[list[PlaylistRows]] = relationship(
|
|
"PlaylistRows",
|
|
back_populates="track",
|
|
cascade="all, delete-orphan",
|
|
)
|
|
playlists = association_proxy("playlistrows", "playlist")
|
|
playdates: Mapped[list[Playdates]] = relationship(
|
|
"Playdates",
|
|
back_populates="track",
|
|
cascade="all, delete-orphan",
|
|
lazy="joined",
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<Track(id={self.id}, title={self.title}, "
|
|
f"artist={self.artist}, path={self.path}>"
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
session: Session,
|
|
path: str,
|
|
title: str,
|
|
artist: str,
|
|
duration: int,
|
|
start_gap: int,
|
|
fade_at: int,
|
|
silence_at: int,
|
|
bitrate: int,
|
|
) -> None:
|
|
self.path = path
|
|
self.title = title
|
|
self.artist = artist
|
|
self.bitrate = bitrate
|
|
self.duration = duration
|
|
self.start_gap = start_gap
|
|
self.fade_at = fade_at
|
|
self.silence_at = silence_at
|
|
|
|
session.add(self)
|
|
session.commit()
|