food-market/app/stats.py

"""Статистика: сводные цифры по периодам, товарам и людям.

Чистые функции без Qt, работают поверх готового отчёта из ledger.

**Важное про базу расчёта.** Смешивать «когда продал» и «когда получил
деньги» нельзя — это разные вопросы, и ответы на них расходятся, как только
появляется хоть один долг. Поэтому здесь два основания одновременно, и каждая
цифра честно приписана своему:

* по дате **продажи** — выручка, себестоимость проданного, наценка, съеденное
  и чаевые. Это «сколько я наторговал за месяц»;
* по дате **платежа** — собранные деньги. Это «сколько денег реально пришло
  в кассу за месяц», и в другой месяц они попадут, если долг вернули позже;
* по дате **закупки** — взятое у пекарни;
* по дате **платежа пекарне** — отданное ей.

У разреза по партиям основание третье: там всё привязано к самой партии,
потому что вопрос стоит иначе — «покрыл ли я эту партию к сроку».
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import date, timedelta
from decimal import Decimal

from . import ledger
from . import money as m
from .models import CONSUMPTION_KINDS, Document

GRAIN_PERIOD = "period"
GRAIN_WEEK = "week"
GRAIN_MONTH = "month"
GRAIN_YEAR = "year"

GRAIN_LABELS = {
    GRAIN_PERIOD: "Периоды",
    GRAIN_WEEK: "Недели",
    GRAIN_MONTH: "Месяцы",
    GRAIN_YEAR: "Годы",
}

MONTHS_NOMINATIVE = (
    "Январь", "Февраль", "Март", "Апрель", "Май", "Июнь",
    "Июль", "Август", "Сентябрь", "Октябрь", "Ноябрь", "Декабрь",
)
MONTHS_GENITIVE = (
    "января", "февраля", "марта", "апреля", "мая", "июня",
    "июля", "августа", "сентября", "октября", "ноября", "декабря",
)


# --- границы и подписи периодов -------------------------------------------


def week_bounds(day: date) -> tuple[date, date]:
    """Понедельник и воскресенье недели по ISO."""
    start = day - timedelta(days=day.isoweekday() - 1)
    return start, start + timedelta(days=6)


def month_bounds(day: date) -> tuple[date, date]:
    start = day.replace(day=1)
    end = (start.replace(year=start.year + 1, month=1) if start.month == 12
           else start.replace(month=start.month + 1)) - timedelta(days=1)
    return start, end


def year_bounds(day: date) -> tuple[date, date]:
    return day.replace(month=1, day=1), day.replace(month=12, day=31)


def bucket_of(day: date, grain: str) -> tuple[str, str, date, date]:
    """Ключ, подпись и границы корзины, в которую попадает дата."""
    if grain == GRAIN_WEEK:
        start, end = week_bounds(day)
        iso_year, iso_week, _ = day.isocalendar()
        label = f"Неделя {iso_week} · {_span(start, end)}"
        return f"{iso_year}-W{iso_week:02d}", label, start, end

    if grain == GRAIN_MONTH:
        start, end = month_bounds(day)
        return f"{day.year}-{day.month:02d}", f"{MONTHS_NOMINATIVE[day.month - 1]} {day.year}", start, end

    if grain == GRAIN_YEAR:
        start, end = year_bounds(day)
        return str(day.year), f"{day.year} год", start, end

    raise ValueError(f"Неизвестная детализация: {grain}")


def _span(start: date, end: date) -> str:
    """«17–23 августа», «29 сентября – 5 октября 2026»."""
    if start.month == end.month:
        return f"{start.day}–{end.day} {MONTHS_GENITIVE[end.month - 1]} {end.year}"
    left = f"{start.day} {MONTHS_GENITIVE[start.month - 1]}"
    if start.year != end.year:
        left += f" {start.year}"
    return f"{left} – {end.day} {MONTHS_GENITIVE[end.month - 1]} {end.year}"


# --- корзины --------------------------------------------------------------


def _units_text(quantities: dict[str, Decimal]) -> str:
    """Количества по единицам: «12 шт · 3,5 кг».

    Единицы принципиально не складываются: сумма литров с килограммами —
    бессмысленное число, которое выглядит как настоящее.
    """
    return " · ".join(
        f"{m.fmt_qty(qty)} {unit}"
        for unit, qty in sorted(quantities.items())
        if qty > 0
    )


@dataclass
class Bucket:
    """Сводка за отрезок времени."""

    key: str
    label: str
    start: date
    end: date

    # По дате продажи.
    revenue: Decimal = m.ZERO
    cogs: Decimal = m.ZERO
    margin: Decimal = m.ZERO
    tips: Decimal = m.ZERO
    consumed_cost: Decimal = m.ZERO
    sales_count: int = 0
    qty_by_unit: dict[str, Decimal] = field(default_factory=dict)

    # По дате платежа.
    collected: Decimal = m.ZERO

    # По датам закупки и расчёта с пекарней.
    purchased: Decimal = m.ZERO
    paid_to_bakery: Decimal = m.ZERO
    # Прощённый пекарней остаток. Отдельно от paid_to_bakery: долг закрыт,
    # но денег не отдавали.
    bakery_discount: Decimal = m.ZERO

    @property
    def profit(self) -> Decimal:
        """Наценка, чаевые и прощённое пекарней минус съеденное и подаренное."""
        return m.money(self.margin + self.tips + self.bakery_discount - self.consumed_cost)

    @property
    def margin_pct(self) -> Decimal:
        """Наценка в процентах от выручки."""
        if self.revenue <= 0:
            return m.ZERO
        return m.money(self.margin / self.revenue * 100)

    @property
    def qty_text(self) -> str:
        return _units_text(self.qty_by_unit)

    def add_qty(self, unit: str, qty: Decimal) -> None:
        self.qty_by_unit[unit] = self.qty_by_unit.get(unit, m.ZERO) + qty


def build_buckets(doc: Document, report: ledger.Report, grain: str) -> list[Bucket]:
    """Разложить всё движение по корзинам выбранной детализации.

    Пустые корзины между непустыми не создаются: если в марте не торговали,
    строки за март не будет. Дырка в таблице честнее выдуманного нуля.
    """
    buckets: dict[str, Bucket] = {}

    def bucket_for(day: date) -> Bucket:
        key, label, start, end = bucket_of(day, grain)
        if key not in buckets:
            buckets[key] = Bucket(key=key, label=label, start=start, end=end)
        return buckets[key]

    sales = {sale.id: sale for sale in doc.sales}

    # Начисления — по дате продажи.
    for allocation in report.allocations:
        sale = sales.get(allocation.sale_id)
        if sale is None:
            continue

        bucket = bucket_for(sale.date)
        if allocation.kind in CONSUMPTION_KINDS:
            bucket.consumed_cost += allocation.cost
        else:
            bucket.revenue += allocation.revenue
            bucket.cogs += allocation.cost
            bucket.margin += allocation.revenue - allocation.cost

    for sale in doc.sales:
        bucket = bucket_for(sale.date)
        bucket.sales_count += 1
        bucket.tips += sale.tip
        for line in sale.lines:
            product = doc.product(line.product_id)
            bucket.add_qty(line.unit_name(product), line.base_qty)

        # Деньги — по дате платежа: долг, возвращённый в сентябре, относится
        # к сентябрю, а не к августовской продаже.
        for payment in sale.payments:
            bucket_for(payment.date).collected += payment.amount
        # Чаевые отдают из рук в руки в момент продажи, своей даты у них нет.
        bucket.collected += sale.tip

    for batch in doc.batches:
        bucket_for(batch.date).purchased += batch.cost_total

    for payment in doc.bakery_payments:
        bucket = bucket_for(payment.date)
        if payment.is_discount:
            bucket.bakery_discount += payment.amount
        else:
            bucket.paid_to_bakery += payment.amount

    return [_rounded(b) for b in sorted(buckets.values(), key=lambda b: b.start)]


def _rounded(bucket: Bucket) -> Bucket:
    """Округлить до копеек один раз в конце, а не на каждом слагаемом."""
    for name in ("revenue", "cogs", "margin", "tips", "consumed_cost",
                 "collected", "purchased", "paid_to_bakery", "bakery_discount"):
        setattr(bucket, name, m.money(getattr(bucket, name)))
    bucket.qty_by_unit = {unit: m.qty(qty) for unit, qty in bucket.qty_by_unit.items()}
    return bucket


def totals(buckets: list[Bucket], label: str = "Итого") -> Bucket:
    """Свернуть корзины в одну итоговую строку."""
    if not buckets:
        return Bucket(key="total", label=label, start=date.today(), end=date.today())

    total = Bucket(key="total", label=label, start=buckets[0].start, end=buckets[-1].end)
    for bucket in buckets:
        total.revenue += bucket.revenue
        total.cogs += bucket.cogs
        total.margin += bucket.margin
        total.tips += bucket.tips
        total.consumed_cost += bucket.consumed_cost
        total.collected += bucket.collected
        total.purchased += bucket.purchased
        total.paid_to_bakery += bucket.paid_to_bakery
        total.bakery_discount += bucket.bakery_discount
        total.sales_count += bucket.sales_count
        for unit, qty in bucket.qty_by_unit.items():
            total.add_qty(unit, qty)
    return _rounded(total)


# --- разрез по партиям ----------------------------------------------------


@dataclass
class PeriodStat:
    """Партия как период: взял булки — верни себестоимость к сроку."""

    report: ledger.BatchReport
    margin: Decimal = m.ZERO
    tips: Decimal = m.ZERO
    # По единицам, а не одним числом: в партии может быть и сок в литрах,
    # и булки в штуках, а их сумма — бессмыслица.
    sold_by_unit: dict[str, Decimal] = field(default_factory=dict)
    left_by_unit: dict[str, Decimal] = field(default_factory=dict)

    @property
    def batch(self):
        return self.report.batch

    @property
    def sold_text(self) -> str:
        return _units_text(self.sold_by_unit)

    @property
    def left_text(self) -> str:
        return _units_text(self.left_by_unit)

    @property
    def label(self) -> str:
        return self.report.label

    @property
    def profit(self) -> Decimal:
        """Прощённый по этой партии остаток — тоже её заработок."""
        return m.money(
            self.margin + self.tips + self.report.discount - self.report.consumed_cost
        )

    @property
    def days_total(self) -> int:
        return (self.batch.due_date - self.batch.date).days


def build_periods(doc: Document, report: ledger.Report) -> list[PeriodStat]:
    """Статистика по заданным вручную периодам — то есть по партиям."""
    margins: dict[str, Decimal] = {}
    tips: dict[str, Decimal] = {}
    sold: dict[str, dict[str, Decimal]] = {}

    def unit_of(product_id: str) -> str:
        product = doc.product(product_id)
        return product.unit if product else "?"

    for allocation in report.allocations:
        by_unit = sold.setdefault(allocation.batch_id, {})
        unit = unit_of(allocation.product_id)
        by_unit[unit] = by_unit.get(unit, m.ZERO) + allocation.qty

        if allocation.kind in CONSUMPTION_KINDS:
            continue
        margins[allocation.batch_id] = (
            margins.get(allocation.batch_id, m.ZERO) + allocation.revenue - allocation.cost
        )
        tips[allocation.batch_id] = tips.get(allocation.batch_id, m.ZERO) + allocation.tip_share

    result = []
    for item in report.batches:
        left: dict[str, Decimal] = {}
        for product_id, qty in item.stock.items():
            unit = unit_of(product_id)
            left[unit] = left.get(unit, m.ZERO) + qty

        result.append(
            PeriodStat(
                report=item,
                margin=m.money(margins.get(item.batch.id, m.ZERO)),
                tips=m.money(tips.get(item.batch.id, m.ZERO)),
                sold_by_unit={u: m.qty(q) for u, q in sold.get(item.batch.id, {}).items()},
                left_by_unit={u: m.qty(q) for u, q in left.items()},
            )
        )
    return result


# --- разрез по товарам ----------------------------------------------------


@dataclass
class ProductStat:
    product_id: str
    name: str
    unit: str
    sold_qty: Decimal = m.ZERO
    revenue: Decimal = m.ZERO
    cogs: Decimal = m.ZERO
    consumed_qty: Decimal = m.ZERO
    consumed_cost: Decimal = m.ZERO
    stock_qty: Decimal = m.ZERO
    stock_cost: Decimal = m.ZERO

    @property
    def margin(self) -> Decimal:
        return m.money(self.revenue - self.cogs)

    @property
    def margin_pct(self) -> Decimal:
        if self.revenue <= 0:
            return m.ZERO
        return m.money(self.margin / self.revenue * 100)


def build_products(doc: Document, report: ledger.Report) -> list[ProductStat]:
    """Кто из булок кормит, а кто лежит мёртвым грузом."""
    stats: dict[str, ProductStat] = {}

    def stat_for(product_id: str) -> ProductStat:
        if product_id not in stats:
            product = doc.product(product_id)
            stats[product_id] = ProductStat(
                product_id=product_id,
                name=doc.product_name(product_id),
                unit=product.unit if product else "",
            )
        return stats[product_id]

    for allocation in report.allocations:
        stat = stat_for(allocation.product_id)
        if allocation.kind in CONSUMPTION_KINDS:
            stat.consumed_qty += allocation.qty
            stat.consumed_cost += allocation.cost
        else:
            stat.sold_qty += allocation.qty
            stat.revenue += allocation.revenue
            stat.cogs += allocation.cost

    for row in report.stock:
        stat = stat_for(row.product_id)
        stat.stock_qty = row.qty
        stat.stock_cost = row.cost

    for stat in stats.values():
        stat.sold_qty = m.qty(stat.sold_qty)
        stat.consumed_qty = m.qty(stat.consumed_qty)
        stat.revenue = m.money(stat.revenue)
        stat.cogs = m.money(stat.cogs)
        stat.consumed_cost = m.money(stat.consumed_cost)

    return sorted(stats.values(), key=lambda s: (-s.margin, s.name))


# --- разрез по людям ------------------------------------------------------


@dataclass
class PersonStat:
    counterparty_id: str | None
    name: str
    sales_count: int = 0
    revenue: Decimal = m.ZERO
    paid: Decimal = m.ZERO
    tips: Decimal = m.ZERO
    debt: Decimal = m.ZERO


def build_people(doc: Document) -> list[PersonStat]:
    """Кто сколько взял, сколько занёс и сколько остался должен.

    Продажи без контрагента идут одной строкой: это розница «с прилавка»,
    и она тоже часть картины.
    """
    stats: dict[str | None, PersonStat] = {}

    for sale in doc.sales:
        if sale.is_consumption:
            continue

        key = sale.counterparty_id
        if key not in stats:
            stats[key] = PersonStat(
                counterparty_id=key,
                name=doc.counterparty_name(key) if key else "Без контрагента",
            )
        stat = stats[key]
        stat.sales_count += 1
        stat.revenue += sale.total
        stat.paid += sale.paid
        stat.tips += sale.tip
        stat.debt += sale.debt

    for stat in stats.values():
        stat.revenue = m.money(stat.revenue)
        stat.paid = m.money(stat.paid)
        stat.tips = m.money(stat.tips)
        stat.debt = m.money(stat.debt)

    return sorted(stats.values(), key=lambda s: (-s.revenue, s.name))