wordarr/wordarr/db.py

import datetime

from sqlalchemy import ForeignKey, String, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, sessionmaker

from . import config


class Base(DeclarativeBase):
    pass


class Library(Base):
    __tablename__ = "libraries"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String, unique=True)
    media_type: Mapped[str] = mapped_column(String)  # ebook | audiobook | comic
    root_path: Mapped[str] = mapped_column(String)
    folder_template: Mapped[str] = mapped_column(String)
    file_template: Mapped[str] = mapped_column(String)
    # "english", "german", … - empty means no restriction
    language: Mapped[str] = mapped_column(String, default="")

    requests: Mapped[list["BookRequest"]] = relationship(back_populates="library")
    watches: Mapped[list["WatchedSeries"]] = relationship(
        back_populates="library", cascade="all, delete-orphan")


class BookRequest(Base):
    __tablename__ = "requests"

    id: Mapped[int] = mapped_column(primary_key=True)
    media_type: Mapped[str] = mapped_column(String)
    library_id: Mapped[int] = mapped_column(ForeignKey("libraries.id"))
    title: Mapped[str] = mapped_column(String)
    authors: Mapped[str] = mapped_column(String, default="")  # comma-separated
    narrator: Mapped[str] = mapped_column(String, default="")
    external_id: Mapped[str] = mapped_column(String, default="")  # ISBN / ASIN / AniList id
    year: Mapped[int | None] = mapped_column(default=None)
    series: Mapped[str] = mapped_column(String, default="")
    volume: Mapped[int | None] = mapped_column(default=None)
    cover_url: Mapped[str] = mapped_column(String, default="")
    status: Mapped[str] = mapped_column(String, default="missing")  # missing | imported
    created_at: Mapped[datetime.datetime] = mapped_column(
        default=lambda: datetime.datetime.now(datetime.UTC)
    )
    imported_path: Mapped[str] = mapped_column(String, default="")

    library: Mapped[Library] = relationship(back_populates="requests")


class History(Base):
    """One row per thing done to a request. Title and type are copied, and
    there is no foreign key, so an entry outlives a deleted request."""

    __tablename__ = "history"

    id: Mapped[int] = mapped_column(primary_key=True)
    at: Mapped[datetime.datetime] = mapped_column(
        default=lambda: datetime.datetime.now(datetime.UTC)
    )
    # imported | appended | relocated | retagged | deleted | failed
    event: Mapped[str] = mapped_column(String)
    request_id: Mapped[int | None] = mapped_column(default=None)
    title: Mapped[str] = mapped_column(String)
    media_type: Mapped[str] = mapped_column(String)
    source: Mapped[str] = mapped_column(String, default="")
    dest: Mapped[str] = mapped_column(String, default="")
    message: Mapped[str] = mapped_column(String, default="")


class WatchedSeries(Base):
    """A series looked up again now and then; new parts become suggestions."""

    __tablename__ = "watched_series"

    id: Mapped[int] = mapped_column(primary_key=True)
    media_type: Mapped[str] = mapped_column(String)
    library_id: Mapped[int] = mapped_column(ForeignKey("libraries.id"))
    series_id: Mapped[str] = mapped_column(String)
    source: Mapped[str] = mapped_column(String, default="")
    title: Mapped[str] = mapped_column(String)
    language: Mapped[str] = mapped_column(String, default="")
    checked_at: Mapped[datetime.datetime | None] = mapped_column(default=None)
    error: Mapped[str] = mapped_column(String, default="")

    library: Mapped[Library] = relationship(back_populates="watches")
    parts: Mapped[list["SeriesPart"]] = relationship(
        back_populates="watch", cascade="all, delete-orphan")


class SeriesPart(Base):
    """Every part a watch has seen, so a later lookup can tell what is new."""

    __tablename__ = "series_parts"

    id: Mapped[int] = mapped_column(primary_key=True)
    watch_id: Mapped[int] = mapped_column(ForeignKey("watched_series.id"))
    key: Mapped[str] = mapped_column(String)
    state: Mapped[str] = mapped_column(String)  # known | new | requested | ignored
    title: Mapped[str] = mapped_column(String)
    authors: Mapped[str] = mapped_column(String, default="")
    narrator: Mapped[str] = mapped_column(String, default="")
    external_id: Mapped[str] = mapped_column(String, default="")
    year: Mapped[int | None] = mapped_column(default=None)
    series: Mapped[str] = mapped_column(String, default="")
    volume: Mapped[int | None] = mapped_column(default=None)
    cover_url: Mapped[str] = mapped_column(String, default="")

    watch: Mapped[WatchedSeries] = relationship(back_populates="parts")


def record(session, event: str, req: BookRequest, **fields: str) -> History:
    """Add a history row; it is saved with the caller's next commit."""
    row = History(event=event, request_id=req.id, title=req.title,
                  media_type=req.media_type, **fields)
    session.add(row)
    return row


_engine = None
SessionLocal = None


def init_db(db_path=None):
    global _engine, SessionLocal
    path = db_path or config.DB_PATH
    path.parent.mkdir(parents=True, exist_ok=True)
    _engine = create_engine(f"sqlite:///{path}", connect_args={"check_same_thread": False})
    Base.metadata.create_all(_engine)
    # lightweight migration: create_all doesn't add new columns to existing tables
    with _engine.begin() as conn:
        def add_column(table: str, column: str, ddl: str) -> None:
            cols = [row[1] for row in conn.exec_driver_sql(f"PRAGMA table_info({table})")]
            if column not in cols:
                conn.exec_driver_sql(f"ALTER TABLE {table} ADD COLUMN {ddl}")

        add_column("requests", "narrator", "narrator VARCHAR NOT NULL DEFAULT ''")
        add_column("libraries", "language", "language VARCHAR NOT NULL DEFAULT ''")
    SessionLocal = sessionmaker(bind=_engine, expire_on_commit=False)
    return _engine


def get_session():
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()