einkaufsapp/backend/app/prices.py

"""Preiserfassung und Preisvergleich.

Ein Preis wird festgehalten, sobald er an einem Eintrag steht, der einem
Markt zugeordnet ist. Ohne Markt ergibt er keinen Vergleichswert und wird
deshalb nicht aufgenommen.
"""

from datetime import timedelta
from decimal import Decimal

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import Article, ListItem, Market, PricePoint
from app.schemas_shopping import (
    ArticlePrices,
    MarketPrice,
    PriceEntry,
    PriceHint,
    PriceOverview,
    PriceOverviewRow,
)
from app.security import utcnow

# Innerhalb dieser Spanne gilt ein gleicher Preis am selben Markt als
# derselbe Beobachtungswert. Ohne das entstünde bei jedem Tippen im
# Preisfeld ein neuer Eintrag.
DEDUPE_WINDOW = timedelta(hours=12)


def record_price(db: Session, item: ListItem) -> None:
    """Hält den Preis eines Eintrags fest, sofern sinnvoll.

    Wird nach jeder Änderung an einem Eintrag aufgerufen; die Prüfungen
    hier entscheiden, ob daraus wirklich ein Datenpunkt wird.
    """
    if not item.price_cents or not item.market_id:
        return

    latest = db.scalar(
        select(PricePoint)
        .where(
            PricePoint.article_id == item.article_id,
            PricePoint.market_id == item.market_id,
        )
        .order_by(PricePoint.recorded_at.desc())
        .limit(1)
    )

    if latest is not None:
        same_price = latest.price_cents == item.price_cents
        same_size = (latest.pack_size == item.pack_size
                     and latest.pack_unit == item.pack_unit)
        if same_price and same_size and utcnow() - latest.recorded_at < DEDUPE_WINDOW:
            return

    db.add(
        PricePoint(
            list_id=item.list_id,
            article_id=item.article_id,
            market_id=item.market_id,
            price_cents=item.price_cents,
            pack_size=item.pack_size,
            pack_unit=item.pack_unit,
        )
    )


def _unit_price(price_cents: int, pack_size: Decimal | None) -> int | None:
    """Preis je Mengeneinheit des Gebindes, in Zehntelcent - sonst gingen
    bei kleinen Mengen zu viele Stellen verloren. Ohne Gebindeangabe
    nicht bestimmbar."""
    if not pack_size or pack_size <= 0:
        return None
    return int(round(price_cents * 10 / float(pack_size)))


def article_prices(db: Session, list_id: str, article: Article) -> ArticlePrices:
    """Preisverlauf eines Artikels, gegliedert nach Markt."""
    points = db.scalars(
        select(PricePoint)
        .where(PricePoint.article_id == article.id)
        .order_by(PricePoint.recorded_at.desc())
        .limit(500)
    ).all()

    markets = {
        m.id: m
        for m in db.scalars(
            select(Market).where(Market.list_id == list_id)
        ).all()
    }

    grouped: dict[str, list[PricePoint]] = {}
    for point in points:
        grouped.setdefault(point.market_id, []).append(point)

    rows: list[MarketPrice] = []
    for market_id, entries in grouped.items():
        market = markets.get(market_id)
        if market is None or market.deleted_at is not None:
            continue

        amounts = [e.price_cents for e in entries]
        newest = entries[0]
        rows.append(
            MarketPrice(
                market_id=market_id,
                market_name=market.name,
                latest_cents=newest.price_cents,
                latest_at=newest.recorded_at,
                latest_pack_size=newest.pack_size,
                latest_pack_unit=newest.pack_unit,
                unit_price_deci=_unit_price(newest.price_cents, newest.pack_size),
                min_cents=min(amounts),
                max_cents=max(amounts),
                observations=len(entries),
            )
        )

    rows.sort(key=lambda r: r.latest_cents)

    history = [
        PriceEntry(
            market_id=p.market_id,
            market_name=markets[p.market_id].name if p.market_id in markets else "?",
            price_cents=p.price_cents,
            pack_size=p.pack_size,
            pack_unit=p.pack_unit,
            recorded_at=p.recorded_at,
        )
        for p in points[:100]
        if p.market_id in markets
    ]

    return ArticlePrices(
        article_id=article.id,
        article_name=article.name,
        markets=rows,
        history=history,
    )


def overview(db: Session, list_id: str) -> PriceOverview:
    """Vergleichstabelle über alle Artikel mit erfassten Preisen."""
    points = db.scalars(
        select(PricePoint)
        .where(PricePoint.list_id == list_id)
        .order_by(PricePoint.recorded_at.desc())
        .limit(5000)
    ).all()

    markets = {
        m.id: m
        for m in db.scalars(
            select(Market).where(
                Market.list_id == list_id, Market.deleted_at.is_(None)
            ).order_by(Market.sort_order, Market.name)
        ).all()
    }
    articles = {
        a.id: a
        for a in db.scalars(
            select(Article).where(
                Article.list_id == list_id, Article.deleted_at.is_(None)
            )
        ).all()
    }

    # Je Artikel und Markt nur der jüngste Wert - die Liste ist bereits
    # absteigend sortiert, der erste Treffer gewinnt.
    latest: dict[tuple[str, str], PricePoint] = {}
    for point in points:
        key = (point.article_id, point.market_id)
        if key not in latest:
            latest[key] = point

    rows: list[PriceOverviewRow] = []
    for article_id, article in articles.items():
        prices = {
            market_id: point.price_cents
            for (aid, market_id), point in latest.items()
            if aid == article_id and market_id in markets
        }
        if not prices:
            continue

        cheapest = min(prices.values())
        best = [mid for mid, cents in prices.items() if cents == cheapest]
        dearest = max(prices.values())

        rows.append(
            PriceOverviewRow(
                article_id=article_id,
                article_name=article.name,
                prices=prices,
                best_market_ids=best,
                best_cents=cheapest,
                spread_cents=dearest - cheapest,
            )
        )

    rows.sort(key=lambda r: r.article_name.casefold())

    return PriceOverview(
        markets=[
            MarketPrice(
                market_id=m.id, market_name=m.name, latest_cents=0,
                latest_at=utcnow(), min_cents=0, max_cents=0, observations=0,
            )
            for m in markets.values()
        ],
        market_names={m.id: m.name for m in markets.values()},
        rows=rows,
    )


def hints(db: Session, list_id: str) -> list[PriceHint]:
    """Kurzhinweise für die Listenansicht: Wo war dieser Artikel zuletzt
    am günstigsten?"""
    data = overview(db, list_id)
    return [
        PriceHint(
            article_id=row.article_id,
            best_market_ids=row.best_market_ids,
            best_cents=row.best_cents,
            spread_cents=row.spread_cents,
            prices=row.prices,
        )
        for row in data.rows
        if row.spread_cents > 0
    ]