"""Datenbank-Operationen (CRUD) für Quivio.""" # SPDX-License-Identifier: Apache-2.0 from __future__ import annotations from typing import Iterable, Optional, Sequence from sqlalchemy import and_, func, or_, select from sqlalchemy.orm import Session, selectinload from quivio import models, schemas # ---------- Hilfsfunktionen -------------------------------------------------- def _get_or_create_named(db: Session, model, names: Iterable[str]) -> list: """Holt oder erstellt Genres/Tags anhand ihrer Namen.""" norm = sorted({n.strip() for n in names if n and n.strip()}) if not norm: return [] existing = {x.name: x for x in db.scalars(select(model).where(model.name.in_(norm)))} out = list(existing.values()) for n in norm: if n not in existing: obj = model(name=n) db.add(obj) out.append(obj) db.flush() return out def _attach_relations(media: models.Media, genres: list[str], tags: list[str], db: Session) -> None: media.genres = _get_or_create_named(db, models.Genre, genres) media.tags = _get_or_create_named(db, models.Tag, tags) # ---------- Lese-Operationen ------------------------------------------------- def list_media( db: Session, *, kind: Optional[str] = None, status: Optional[str] = None, search: Optional[str] = None, genre: Optional[str] = None, tag: Optional[str] = None, sort: str = "updated_desc", ) -> Sequence[models.Media]: stmt = select(models.Media).options(selectinload(models.Media.genres), selectinload(models.Media.tags)) conds = [] if kind: conds.append(models.Media.kind == kind) if status: conds.append(models.Media.status == status) if search: like = f"%{search.lower()}%" conds.append( or_( func.lower(models.Media.title).like(like), func.lower(func.coalesce(models.Media.original_title, "")).like(like), func.lower(func.coalesce(models.Media.author, "")).like(like), ) ) if genre: conds.append(models.Media.genres.any(models.Genre.name == genre)) if tag: conds.append(models.Media.tags.any(models.Tag.name == tag)) if conds: stmt = stmt.where(and_(*conds)) sort_map = { "updated_desc": models.Media.updated_at.desc(), "updated_asc": models.Media.updated_at.asc(), "title_asc": models.Media.title.asc(), "title_desc": models.Media.title.desc(), "rating_desc": models.Media.rating.desc().nulls_last(), "rating_asc": models.Media.rating.asc().nulls_last(), "year_desc": models.Media.release_year.desc().nulls_last(), "year_asc": models.Media.release_year.asc().nulls_last(), } stmt = stmt.order_by(sort_map.get(sort, models.Media.updated_at.desc())) return db.scalars(stmt).all() def get_media(db: Session, media_id: int) -> Optional[models.Media]: stmt = ( select(models.Media) .options(selectinload(models.Media.genres), selectinload(models.Media.tags)) .where(models.Media.id == media_id) ) return db.scalars(stmt).first() # ---------- Schreib-Operationen ---------------------------------------------- def create_media(db: Session, payload: schemas.MediaCreate) -> models.Media: media = models.Media(**payload.model_dump(exclude={"genres", "tags"})) db.add(media) db.flush() _attach_relations(media, payload.genres, payload.tags, db) db.commit() db.refresh(media) return media def update_media(db: Session, media: models.Media, payload: schemas.MediaUpdate) -> models.Media: data = payload.model_dump(exclude_unset=True) rel_fields = {"genres", "tags"} for k, v in data.items(): if k in rel_fields: continue setattr(media, k, v) if "genres" in data: media.genres = _get_or_create_named(db, models.Genre, data["genres"] or []) if "tags" in data: media.tags = _get_or_create_named(db, models.Tag, data["tags"] or []) db.commit() db.refresh(media) return media def delete_media(db: Session, media: models.Media) -> None: db.delete(media) db.commit() def bump_progress(db: Session, media: models.Media, payload: schemas.ProgressUpdate) -> models.Media: """Setzt/erhöht Fortschritt, je nach Medium.""" if media.kind == "book": total = media.total_chapters or media.total_pages or media.total_volumes attr = ( "chapters_read" if media.total_chapters else "pages_read" if media.total_pages else "volumes_read" ) else: total = media.total_episodes attr = "episodes_watched" current = getattr(media, attr) or 0 if payload.set_to is not None: new_value = payload.set_to else: new_value = current + payload.delta if total is not None and total >= 0: new_value = max(0, min(new_value, total)) setattr(media, attr, new_value) # Auto-Status: 0 -> plan, total -> done, sonst reading if total and new_value == 0: media.status = "plan" elif total and new_value >= total: media.status = "done" if not media.end_date: from datetime import date media.end_date = date.today() elif media.status == "plan": media.status = "reading" db.commit() db.refresh(media) return media # ---------- Statistik -------------------------------------------------------- def _avg(values: Sequence[Optional[float]]) -> Optional[float]: """Mittelwert ohne None, gerundet auf 2 Stellen; None wenn leer.""" rated = [v for v in values if v is not None] if not rated: return None return round(sum(rated) / len(rated), 2) def stats(db: Session) -> schemas.StatsResponse: def per_kind(kind: str) -> schemas.StatsPerKind: items = db.scalars(select(models.Media).where(models.Media.kind == kind)).all() by_status: dict[str, int] = {s: 0 for s in models.STATUS_VALUES} for m in items: by_status[m.status] = by_status.get(m.status, 0) + 1 return schemas.StatsPerKind( kind=kind, total=len(items), avg_rating=_avg([m.rating for m in items]), by_status=[schemas.StatsBucket(status=k, count=v) for k, v in by_status.items()], ) books = per_kind("book") series = per_kind("series") return schemas.StatsResponse( books=books, series=series, total_entries=books.total + series.total, ) # ---------- Hilfslisten ------------------------------------------------------ def list_genres(db: Session) -> Sequence[models.Genre]: return db.scalars(select(models.Genre).order_by(models.Genre.name)).all() def list_tags(db: Session) -> Sequence[models.Tag]: return db.scalars(select(models.Tag).order_by(models.Tag.name)).all()