134 lines
3.8 KiB
Python
134 lines
3.8 KiB
Python
#!/usr/bin/python3
|
|
|
|
import logging
|
|
import os
|
|
import sqlalchemy
|
|
|
|
import pytiger.logging.config
|
|
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy import Column, Float, DateTime, Integer, String
|
|
from sqlalchemy.orm.exc import NoResultFound
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from config import Config
|
|
|
|
from log import ERROR, DEBUG
|
|
# from log import INFO, ERROR, DEBUG
|
|
|
|
# "Constants"
|
|
# Instantiate logging
|
|
pytiger.logging.config.basic_config(stderr=False, level=logging.INFO)
|
|
log = logging.getLogger(__name__)
|
|
log.info("Starting")
|
|
|
|
# Create session at the global level as per
|
|
# https://docs.sqlalchemy.org/en/13/orm/session_basics.html
|
|
|
|
# Set up database connection
|
|
log.info("Connect to database")
|
|
engine = sqlalchemy.create_engine(f"{Config.MYSQL_CONNECT}?charset=utf8",
|
|
encoding='utf-8',
|
|
echo=Config.DISPLAY_SQL)
|
|
Base = declarative_base()
|
|
Base.metadata.create_all(engine)
|
|
|
|
# Create a Session
|
|
Session = sessionmaker(bind=engine)
|
|
session = Session()
|
|
|
|
|
|
# Database classes
|
|
class Settings(Base):
|
|
__tablename__ = 'settings'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
name = Column(String(32), nullable=False, unique=True)
|
|
f_datetime = Column(DateTime, default=None, nullable=True)
|
|
f_int = Column(Integer, default=None, nullable=True)
|
|
f_string = Column(String(128), default=None, nullable=True)
|
|
|
|
@classmethod
|
|
def get_int(cls, name):
|
|
try:
|
|
int_setting = session.query(cls).filter(
|
|
cls.name == name).one()
|
|
except NoResultFound:
|
|
int_setting = Settings()
|
|
int_setting.name = name
|
|
int_setting.f_int = None
|
|
session.add(int_setting)
|
|
session.commit()
|
|
return int_setting
|
|
|
|
def update(self, data):
|
|
for key, value in data.items():
|
|
assert hasattr(self, key)
|
|
setattr(self, key, value)
|
|
session.commit()
|
|
|
|
|
|
class Tracks(Base):
|
|
__tablename__ = 'tracks'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
title = Column(String(256), index=True)
|
|
artist = Column(String(256), index=True)
|
|
duration = Column(Integer, index=True)
|
|
start_gap = Column(Integer, index=False)
|
|
fade_at = Column(Integer, index=False)
|
|
silence_at = Column(Integer, index=False)
|
|
path = Column(String(2048), index=False, nullable=False)
|
|
|
|
mtime = Column(Float, index=True)
|
|
lastplayed = Column(DateTime, index=True, default=None)
|
|
|
|
def __repr__(self):
|
|
return (
|
|
f"<Track(id={self.id}, title={self.title}, "
|
|
f"artist={self.artist}, path={self.path}>"
|
|
)
|
|
|
|
@classmethod
|
|
def get_or_create(cls, path):
|
|
DEBUG(f"get_or_create(cls, {path})")
|
|
try:
|
|
track = session.query(cls).filter(cls.path == path).one()
|
|
except NoResultFound:
|
|
track = Tracks()
|
|
track.path = path
|
|
session.add(track)
|
|
return track
|
|
|
|
@staticmethod
|
|
def get_path(id):
|
|
try:
|
|
path = session.query(Tracks.path).filter(Tracks.id == id).one()[0]
|
|
return os.path.join(Config.ROOT, path)
|
|
except NoResultFound:
|
|
print(f"Can't find track id {id}")
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_track(id):
|
|
try:
|
|
DEBUG(f"get_track({id})")
|
|
track = session.query(Tracks).filter(Tracks.id == id).one()
|
|
return track
|
|
except NoResultFound:
|
|
ERROR(f"get_track({id}): not found")
|
|
return None
|
|
|
|
@staticmethod
|
|
def search_titles(text):
|
|
return (
|
|
session.query(Tracks)
|
|
.filter(Tracks.title.ilike(f"%{text}%"))
|
|
.order_by(Tracks.title)
|
|
).all()
|
|
|
|
@staticmethod
|
|
def track_from_id(id):
|
|
return session.query(Tracks).filter(
|
|
Tracks.id == id).one()
|