725 lines
20 KiB
Python
725 lines
20 KiB
Python
# Standard library imports
|
|
from typing import List, Optional, Sequence
|
|
import datetime as dt
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
# PyQt imports
|
|
|
|
# Third party imports
|
|
from alchemical import Alchemical # type:ignore
|
|
from sqlalchemy import (
|
|
bindparam,
|
|
delete,
|
|
func,
|
|
select,
|
|
update,
|
|
)
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm.exc import NoResultFound
|
|
from sqlalchemy.orm import joinedload
|
|
from sqlalchemy.orm.session import Session
|
|
|
|
# App imports
|
|
import dbtables
|
|
from config import Config
|
|
from log import log
|
|
|
|
|
|
# Establish database connection
|
|
ALCHEMICAL_DATABASE_URI = os.environ.get("ALCHEMICAL_DATABASE_URI")
|
|
if ALCHEMICAL_DATABASE_URI is None:
|
|
raise ValueError("ALCHEMICAL_DATABASE_URI is undefined")
|
|
if "unittest" in sys.modules and "sqlite" not in ALCHEMICAL_DATABASE_URI:
|
|
raise ValueError("Unit tests running on non-Sqlite database")
|
|
db = Alchemical(ALCHEMICAL_DATABASE_URI, engine_options=Config.ENGINE_OPTIONS)
|
|
|
|
|
|
# Database classes
|
|
class NoteColours(dbtables.NoteColoursTable):
|
|
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()
|
|
|
|
@classmethod
|
|
def get_all(cls, session: Session) -> Sequence["NoteColours"]:
|
|
"""
|
|
Return all records
|
|
"""
|
|
|
|
return session.scalars(select(cls)).all()
|
|
|
|
@staticmethod
|
|
def get_colour(session: Session, text: str) -> Optional[str]:
|
|
"""
|
|
Parse text and return colour string if matched, else empty string
|
|
"""
|
|
|
|
if not text:
|
|
return None
|
|
|
|
for rec in session.scalars(
|
|
select(NoteColours)
|
|
.filter(NoteColours.enabled.is_(True))
|
|
.order_by(NoteColours.order)
|
|
).all():
|
|
if rec.is_regex:
|
|
flags = re.UNICODE
|
|
if not rec.is_casesensitive:
|
|
flags |= re.IGNORECASE
|
|
p = re.compile(rec.substring, flags)
|
|
if p.match(text):
|
|
return rec.colour
|
|
else:
|
|
if rec.is_casesensitive:
|
|
if rec.substring in text:
|
|
return rec.colour
|
|
else:
|
|
if rec.substring.lower() in text.lower():
|
|
return rec.colour
|
|
|
|
return None
|
|
|
|
|
|
class Playdates(dbtables.PlaydatesTable):
|
|
def __init__(self, session: Session, track_id: int) -> None:
|
|
"""Record that track was played"""
|
|
|
|
self.lastplayed = dt.datetime.now()
|
|
self.track_id = track_id
|
|
session.add(self)
|
|
session.commit()
|
|
|
|
@staticmethod
|
|
def last_playdates(
|
|
session: Session, track_id: int, limit=5
|
|
) -> Sequence["Playdates"]:
|
|
"""
|
|
Return a list of the last limit playdates for this track, sorted
|
|
earliest to latest.
|
|
"""
|
|
|
|
return session.scalars(
|
|
Playdates.select()
|
|
.where(Playdates.track_id == track_id)
|
|
.order_by(Playdates.lastplayed.asc())
|
|
.limit(limit)
|
|
).all()
|
|
|
|
@staticmethod
|
|
def last_played(session: Session, track_id: int) -> dt.datetime:
|
|
"""Return datetime track last played or None"""
|
|
|
|
last_played = session.execute(
|
|
select(Playdates.lastplayed)
|
|
.where(Playdates.track_id == track_id)
|
|
.order_by(Playdates.lastplayed.desc())
|
|
.limit(1)
|
|
).first()
|
|
|
|
if last_played:
|
|
return last_played[0]
|
|
else:
|
|
# Should never be reached as we create record with a
|
|
# last_played value
|
|
return Config.EPOCH # pragma: no cover
|
|
|
|
@staticmethod
|
|
def last_played_tracks(session: Session, limit=5) -> Sequence["Playdates"]:
|
|
"""
|
|
Return a list of the last limit tracks played, sorted
|
|
earliest to latest.
|
|
"""
|
|
|
|
return session.scalars(
|
|
Playdates.select().order_by(Playdates.lastplayed.desc()).limit(limit)
|
|
).all()
|
|
|
|
@staticmethod
|
|
def played_after(session: Session, since: dt.datetime) -> Sequence["Playdates"]:
|
|
"""Return a list of Playdates objects since passed time"""
|
|
|
|
return session.scalars(
|
|
select(Playdates)
|
|
.where(Playdates.lastplayed >= since)
|
|
.order_by(Playdates.lastplayed)
|
|
).all()
|
|
|
|
|
|
class Playlists(dbtables.PlaylistsTable):
|
|
def __init__(self, session: Session, name: str):
|
|
self.name = name
|
|
session.add(self)
|
|
session.commit()
|
|
|
|
@staticmethod
|
|
def clear_tabs(session: Session, playlist_ids: List[int]) -> None:
|
|
"""
|
|
Make all tab records NULL
|
|
"""
|
|
|
|
session.execute(
|
|
update(Playlists).where((Playlists.id.in_(playlist_ids))).values(tab=None)
|
|
)
|
|
|
|
def close(self, session: Session) -> None:
|
|
"""Mark playlist as unloaded"""
|
|
|
|
self.open = False
|
|
session.commit()
|
|
|
|
@classmethod
|
|
def create_playlist_from_template(
|
|
cls, session: Session, template: "Playlists", playlist_name: str
|
|
) -> Optional["Playlists"]:
|
|
"""Create a new playlist from template"""
|
|
|
|
playlist = cls(session, playlist_name)
|
|
|
|
# Sanity / mypy checks
|
|
if not playlist or not playlist.id or not template.id:
|
|
return None
|
|
|
|
PlaylistRows.copy_playlist(session, template.id, playlist.id)
|
|
|
|
return playlist
|
|
|
|
def delete(self, session: Session) -> None:
|
|
"""
|
|
Mark as deleted
|
|
"""
|
|
|
|
self.deleted = True
|
|
session.commit()
|
|
|
|
@classmethod
|
|
def get_all(cls, session: Session) -> Sequence["Playlists"]:
|
|
"""Returns a list of all playlists ordered by last use"""
|
|
|
|
return session.scalars(
|
|
select(cls)
|
|
.filter(cls.is_template.is_(False))
|
|
.order_by(cls.last_used.desc())
|
|
).all()
|
|
|
|
@classmethod
|
|
def get_all_templates(cls, session: Session) -> Sequence["Playlists"]:
|
|
"""Returns a list of all templates ordered by name"""
|
|
|
|
return session.scalars(
|
|
select(cls).filter(cls.is_template.is_(True)).order_by(cls.name)
|
|
).all()
|
|
|
|
@classmethod
|
|
def get_closed(cls, session: Session) -> Sequence["Playlists"]:
|
|
"""Returns a list of all closed playlists ordered by last use"""
|
|
|
|
return session.scalars(
|
|
select(cls)
|
|
.filter(
|
|
cls.open.is_(False),
|
|
cls.is_template.is_(False),
|
|
cls.deleted.is_(False),
|
|
)
|
|
.order_by(cls.last_used.desc())
|
|
).all()
|
|
|
|
@classmethod
|
|
def get_open(cls, session: Session) -> Sequence[Optional["Playlists"]]:
|
|
"""
|
|
Return a list of loaded playlists ordered by tab.
|
|
"""
|
|
|
|
return session.scalars(
|
|
select(cls).where(cls.open.is_(True)).order_by(cls.tab)
|
|
).all()
|
|
|
|
def mark_open(self) -> None:
|
|
"""Mark playlist as loaded and used now"""
|
|
|
|
self.open = True
|
|
self.last_used = dt.datetime.now()
|
|
|
|
@staticmethod
|
|
def name_is_available(session: Session, name: str) -> bool:
|
|
"""
|
|
Return True if no playlist of this name exists else false.
|
|
"""
|
|
|
|
return (
|
|
session.execute(select(Playlists).where(Playlists.name == name)).first()
|
|
is None
|
|
)
|
|
|
|
def rename(self, session: Session, new_name: str) -> None:
|
|
"""
|
|
Rename playlist
|
|
"""
|
|
|
|
self.name = new_name
|
|
session.commit()
|
|
|
|
@staticmethod
|
|
def save_as_template(
|
|
session: Session, playlist_id: int, template_name: str
|
|
) -> None:
|
|
"""Save passed playlist as new template"""
|
|
|
|
template = Playlists(session, template_name)
|
|
if not template or not template.id:
|
|
return
|
|
|
|
template.is_template = True
|
|
session.commit()
|
|
|
|
PlaylistRows.copy_playlist(session, playlist_id, template.id)
|
|
|
|
|
|
class PlaylistRows(dbtables.PlaylistRowsTable):
|
|
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.plr_rownum = row_number
|
|
self.note = note
|
|
session.add(self)
|
|
session.commit()
|
|
|
|
def append_note(self, extra_note: str) -> None:
|
|
"""Append passed note to any existing note"""
|
|
|
|
current_note = self.note
|
|
if current_note:
|
|
self.note = current_note + "\n" + extra_note
|
|
else:
|
|
self.note = extra_note
|
|
|
|
@staticmethod
|
|
def copy_playlist(session: Session, src_id: int, dst_id: int) -> None:
|
|
"""Copy playlist entries"""
|
|
|
|
src_rows = session.scalars(
|
|
select(PlaylistRows).filter(PlaylistRows.playlist_id == src_id)
|
|
).all()
|
|
|
|
for plr in src_rows:
|
|
PlaylistRows(
|
|
session=session,
|
|
playlist_id=dst_id,
|
|
row_number=plr.plr_rownum,
|
|
note=plr.note,
|
|
track_id=plr.track_id,
|
|
)
|
|
|
|
@classmethod
|
|
def deep_row(
|
|
cls, session: Session, playlist_id: int, row_number: int
|
|
) -> "PlaylistRows":
|
|
"""
|
|
Return a playlist row that includes full track and lastplayed data for
|
|
given playlist_id and row
|
|
"""
|
|
|
|
stmt = (
|
|
select(PlaylistRows)
|
|
.options(joinedload(cls.track))
|
|
.where(
|
|
PlaylistRows.playlist_id == playlist_id,
|
|
PlaylistRows.plr_rownum == row_number,
|
|
)
|
|
# .options(joinedload(Tracks.playdates))
|
|
)
|
|
|
|
return session.execute(stmt).unique().scalar_one()
|
|
|
|
@classmethod
|
|
def deep_rows(cls, session: Session, playlist_id: int) -> Sequence["PlaylistRows"]:
|
|
"""
|
|
Return a list of playlist rows that include full track and lastplayed data for
|
|
given playlist_id., Sequence
|
|
"""
|
|
|
|
stmt = (
|
|
select(PlaylistRows)
|
|
.options(joinedload(cls.track))
|
|
.where(PlaylistRows.playlist_id == playlist_id)
|
|
.order_by(PlaylistRows.plr_rownum)
|
|
# .options(joinedload(Tracks.playdates))
|
|
)
|
|
|
|
return session.scalars(stmt).unique().all()
|
|
|
|
@staticmethod
|
|
def delete_higher_rows(session: Session, playlist_id: int, maxrow: int) -> None:
|
|
"""
|
|
Delete rows in given playlist that have a higher row number
|
|
than 'maxrow'
|
|
"""
|
|
|
|
session.execute(
|
|
delete(PlaylistRows).where(
|
|
PlaylistRows.playlist_id == playlist_id,
|
|
PlaylistRows.plr_rownum > maxrow,
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
@staticmethod
|
|
def delete_row(session: Session, playlist_id: int, row_number: int) -> None:
|
|
"""
|
|
Delete passed row in given playlist.
|
|
"""
|
|
|
|
session.execute(
|
|
delete(PlaylistRows).where(
|
|
PlaylistRows.playlist_id == playlist_id,
|
|
PlaylistRows.plr_rownum == row_number,
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def fixup_rownumbers(session: Session, playlist_id: int) -> None:
|
|
"""
|
|
Ensure the row numbers for passed playlist have no gaps
|
|
"""
|
|
|
|
plrs = session.scalars(
|
|
select(PlaylistRows)
|
|
.where(PlaylistRows.playlist_id == playlist_id)
|
|
.order_by(PlaylistRows.plr_rownum)
|
|
).all()
|
|
|
|
for i, plr in enumerate(plrs):
|
|
plr.plr_rownum = i
|
|
|
|
# Ensure new row numbers are available to the caller
|
|
session.commit()
|
|
|
|
@classmethod
|
|
def plrids_to_plrs(
|
|
cls, session: Session, playlist_id: int, plr_ids: List[int]
|
|
) -> Sequence["PlaylistRows"]:
|
|
"""
|
|
Take a list of PlaylistRows ids and return a list of corresponding
|
|
PlaylistRows objects
|
|
"""
|
|
|
|
plrs = session.scalars(
|
|
select(cls)
|
|
.where(cls.playlist_id == playlist_id, cls.id.in_(plr_ids))
|
|
.order_by(cls.plr_rownum)
|
|
).all()
|
|
|
|
return plrs
|
|
|
|
@staticmethod
|
|
def get_last_used_row(session: Session, playlist_id: int) -> Optional[int]:
|
|
"""Return the last used row for playlist, or None if no rows"""
|
|
|
|
return session.execute(
|
|
select(func.max(PlaylistRows.plr_rownum)).where(
|
|
PlaylistRows.playlist_id == playlist_id
|
|
)
|
|
).scalar_one()
|
|
|
|
@staticmethod
|
|
def get_track_plr(
|
|
session: Session, track_id: int, playlist_id: int
|
|
) -> Optional["PlaylistRows"]:
|
|
"""Return first matching PlaylistRows object or None"""
|
|
|
|
return session.scalars(
|
|
select(PlaylistRows)
|
|
.where(
|
|
PlaylistRows.track_id == track_id,
|
|
PlaylistRows.playlist_id == playlist_id,
|
|
)
|
|
.limit(1)
|
|
).first()
|
|
|
|
@classmethod
|
|
def get_played_rows(
|
|
cls, session: Session, playlist_id: int
|
|
) -> Sequence["PlaylistRows"]:
|
|
"""
|
|
For passed playlist, return a list of rows that
|
|
have been played.
|
|
"""
|
|
|
|
plrs = session.scalars(
|
|
select(cls)
|
|
.where(cls.playlist_id == playlist_id, cls.played.is_(True))
|
|
.order_by(cls.plr_rownum)
|
|
).all()
|
|
|
|
return plrs
|
|
|
|
@classmethod
|
|
def get_rows_with_tracks(
|
|
cls,
|
|
session: Session,
|
|
playlist_id: int,
|
|
) -> Sequence["PlaylistRows"]:
|
|
"""
|
|
For passed playlist, return a list of rows that
|
|
contain tracks
|
|
"""
|
|
|
|
query = select(cls).where(
|
|
cls.playlist_id == playlist_id, cls.track_id.is_not(None)
|
|
)
|
|
plrs = session.scalars((query).order_by(cls.plr_rownum)).all()
|
|
|
|
return plrs
|
|
|
|
@classmethod
|
|
def get_unplayed_rows(
|
|
cls, session: Session, playlist_id: int
|
|
) -> Sequence["PlaylistRows"]:
|
|
"""
|
|
For passed playlist, return a list of playlist rows that
|
|
have not been played.
|
|
"""
|
|
|
|
plrs = session.scalars(
|
|
select(cls)
|
|
.where(
|
|
cls.playlist_id == playlist_id,
|
|
cls.track_id.is_not(None),
|
|
cls.played.is_(False),
|
|
)
|
|
.order_by(cls.plr_rownum)
|
|
).all()
|
|
|
|
return plrs
|
|
|
|
@classmethod
|
|
def insert_row(
|
|
cls,
|
|
session: Session,
|
|
playlist_id: int,
|
|
new_row_number: int,
|
|
note: str = "",
|
|
track_id: Optional[int] = None,
|
|
) -> "PlaylistRows":
|
|
cls.move_rows_down(session, playlist_id, new_row_number, 1)
|
|
return cls(
|
|
session,
|
|
playlist_id=playlist_id,
|
|
row_number=new_row_number,
|
|
note=note,
|
|
track_id=track_id,
|
|
)
|
|
|
|
@staticmethod
|
|
def move_rows_down(
|
|
session: Session, playlist_id: int, starting_row: int, move_by: int
|
|
) -> None:
|
|
"""
|
|
Create space to insert move_by additional rows by incremented row
|
|
number from starting_row to end of playlist
|
|
"""
|
|
|
|
log.debug(f"(move_rows_down({playlist_id=}, {starting_row=}, {move_by=}")
|
|
|
|
session.execute(
|
|
update(PlaylistRows)
|
|
.where(
|
|
(PlaylistRows.playlist_id == playlist_id),
|
|
(PlaylistRows.plr_rownum >= starting_row),
|
|
)
|
|
.values(plr_rownum=PlaylistRows.plr_rownum + move_by)
|
|
)
|
|
|
|
@staticmethod
|
|
def update_plr_rownumbers(
|
|
session: Session, playlist_id: int, sqla_map: List[dict[str, int]]
|
|
) -> None:
|
|
"""
|
|
Take a {plrid: plr_rownum} dictionary and update the row numbers accordingly
|
|
"""
|
|
|
|
# Update database. Ref:
|
|
# https://docs.sqlalchemy.org/en/20/tutorial/data_update.html#the-update-sql-expression-construct
|
|
stmt = (
|
|
update(PlaylistRows)
|
|
.where(
|
|
PlaylistRows.playlist_id == playlist_id,
|
|
PlaylistRows.id == bindparam("plrid"),
|
|
)
|
|
.values(plr_rownum=bindparam("plr_rownum"))
|
|
)
|
|
session.connection().execute(stmt, sqla_map)
|
|
|
|
|
|
class Settings(dbtables.SettingsTable):
|
|
def __init__(self, session: Session, name: str):
|
|
self.name = name
|
|
session.add(self)
|
|
session.commit()
|
|
|
|
@classmethod
|
|
def all_as_dict(cls, session):
|
|
"""
|
|
Return all setting in a dictionary keyed by name
|
|
"""
|
|
|
|
result = {}
|
|
|
|
settings = session.scalars(select(cls)).all()
|
|
for setting in settings:
|
|
result[setting.name] = setting
|
|
|
|
return result
|
|
|
|
@classmethod
|
|
def get_int_settings(cls, session: Session, name: str) -> "Settings":
|
|
"""Get setting for an integer or return new setting record"""
|
|
|
|
try:
|
|
return session.execute(select(cls).where(cls.name == name)).scalar_one()
|
|
|
|
except NoResultFound:
|
|
return Settings(session, name)
|
|
|
|
def update(self, session: Session, data: dict) -> None:
|
|
for key, value in data.items():
|
|
assert hasattr(self, key)
|
|
setattr(self, key, value)
|
|
session.commit()
|
|
|
|
|
|
class Tracks(dbtables.TracksTable):
|
|
def __init__(
|
|
self,
|
|
session: Session,
|
|
path: str,
|
|
title: str,
|
|
artist: str,
|
|
duration: int,
|
|
start_gap: int,
|
|
fade_at: int,
|
|
silence_at: int,
|
|
mtime: int,
|
|
bitrate: int,
|
|
):
|
|
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
|
|
self.mtime = mtime
|
|
|
|
try:
|
|
session.add(self)
|
|
session.commit()
|
|
except IntegrityError as error:
|
|
session.rollback()
|
|
log.error(f"Error ({error=}) importing track ({path=})")
|
|
raise ValueError(error)
|
|
|
|
@classmethod
|
|
def get_all(cls, session) -> List["Tracks"]:
|
|
"""Return a list of all tracks"""
|
|
|
|
return session.scalars(select(cls)).unique().all()
|
|
|
|
@classmethod
|
|
def get_by_basename(
|
|
cls, session: Session, basename: str
|
|
) -> Optional[Sequence["Tracks"]]:
|
|
"""
|
|
Return track(s) with passed basename, or None.
|
|
"""
|
|
|
|
try:
|
|
return session.scalars(
|
|
Tracks.select().where(Tracks.path.like("%/" + basename))
|
|
).all()
|
|
except NoResultFound:
|
|
return None
|
|
|
|
@classmethod
|
|
def get_by_path(cls, session: Session, path: str) -> Optional["Tracks"]:
|
|
"""
|
|
Return track with passed path, or None.
|
|
"""
|
|
|
|
try:
|
|
return (
|
|
session.execute(select(Tracks).where(Tracks.path == path))
|
|
.unique()
|
|
.scalar_one()
|
|
)
|
|
except NoResultFound:
|
|
return None
|
|
|
|
@classmethod
|
|
def search_artists(cls, session: Session, text: str) -> Sequence["Tracks"]:
|
|
"""
|
|
Search case-insenstively for artists containing str
|
|
|
|
The query performs an outer join with 'joinedload' to populate the results
|
|
from the Playdates table at the same time. unique() needed; see
|
|
https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#joined-eager-loading
|
|
"""
|
|
|
|
return (
|
|
session.scalars(
|
|
select(cls)
|
|
.options(joinedload(Tracks.playdates))
|
|
.where(cls.artist.ilike(f"%{text}%"))
|
|
.order_by(cls.title)
|
|
)
|
|
.unique()
|
|
.all()
|
|
)
|
|
|
|
@classmethod
|
|
def search_titles(cls, session: Session, text: str) -> Sequence["Tracks"]:
|
|
"""
|
|
Search case-insenstively for titles containing str
|
|
|
|
The query performs an outer join with 'joinedload' to populate the results
|
|
from the Playdates table at the same time. unique() needed; see
|
|
https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#joined-eager-loading
|
|
"""
|
|
return (
|
|
session.scalars(
|
|
select(cls)
|
|
.options(joinedload(Tracks.playdates))
|
|
.where(cls.title.like(f"{text}%"))
|
|
.order_by(cls.title)
|
|
)
|
|
.unique()
|
|
.all()
|
|
)
|