Collin Kasbergen
70 lines
2.0 KiB
Python
70 lines
2.0 KiB
Python
import os
|
|
from sqlalchemy import create_engine, Column, String, Float, Boolean, Integer
|
|
from sqlalchemy.orm import declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
DATA_DIR = os.getenv("DATA_DIR", "/data")
|
|
SQLALCHEMY_DATABASE_URL = f"sqlite:///{os.path.join(DATA_DIR, 'navifm.db')}"
|
|
|
|
engine = create_engine(
|
|
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
|
|
)
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, expire_on_commit=False)
|
|
|
|
Base = declarative_base()
|
|
|
|
class Track(Base):
|
|
__tablename__ = "tracks"
|
|
|
|
id = Column(String, primary_key=True, index=True) # Navidrome Track ID
|
|
title = Column(String, index=True)
|
|
artist = Column(String, index=True)
|
|
album = Column(String)
|
|
duration = Column(Integer)
|
|
path = Column(String) # Navidrome file path or relative path
|
|
|
|
# Analysis Status
|
|
analyzed = Column(Boolean, default=False)
|
|
|
|
# Analyzed Metadata
|
|
bpm = Column(Float, nullable=True)
|
|
key = Column(String, nullable=True)
|
|
lufs = Column(Float, nullable=True)
|
|
energy = Column(Float, nullable=True)
|
|
spectral_centroid = Column(Float, nullable=True)
|
|
onset_density = Column(Float, nullable=True)
|
|
|
|
# Play stats
|
|
last_played = Column(Integer, default=0) # timestamp
|
|
play_count = Column(Integer, default=0)
|
|
|
|
class Setting(Base):
|
|
__tablename__ = "settings"
|
|
|
|
key = Column(String, primary_key=True)
|
|
value = Column(String)
|
|
|
|
def get_setting(db, key: str, default: str = None):
|
|
setting = db.query(Setting).filter(Setting.key == key).first()
|
|
return setting.value if setting else default
|
|
|
|
def set_setting(db, key: str, value: str):
|
|
setting = db.query(Setting).filter(Setting.key == key).first()
|
|
if setting:
|
|
setting.value = value
|
|
else:
|
|
setting = Setting(key=key, value=value)
|
|
db.add(setting)
|
|
db.commit()
|
|
|
|
def init_db():
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|