Initial commit: WatchStack v0.1.0
- FastAPI + SQLite backend with REST API - Polymorphic Media model (book + series) - Cover lookup via Open Library - Vanilla HTML/CSS/JS frontend (MAL-inspired) - Dark theme, status tabs, drawer detail, stats modal - 8 seed demo entries
This commit is contained in:
ki
+205
@@ -0,0 +1,205 @@
|
||||
"""Datenbank-Operationen (CRUD) für WatchStack."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Iterable, Optional, Sequence
|
||||
|
||||
from sqlalchemy import and_, func, or_, select
|
||||
from sqlalchemy.orm import Session, selectinload
|
||||
|
||||
from app 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 media.kind == "series" and 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 stats(db: Session) -> schemas.StatsResponse:
|
||||
def per_kind(kind: str) -> schemas.StatsPerKind:
|
||||
items = db.scalars(select(models.Media).where(models.Media.kind == kind)).all()
|
||||
total = len(items)
|
||||
avg = (
|
||||
round(sum(m.rating for m in items if m.rating is not None) /
|
||||
max(1, sum(1 for m in items if m.rating is not None)), 2)
|
||||
if any(m.rating is not None for m in items)
|
||||
else None
|
||||
)
|
||||
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=total,
|
||||
avg_rating=avg,
|
||||
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()
|
||||
Reference in New Issue
Block a user