food-market/app/ledger.py

"""Расчёты по партиям, долгам и остаткам. Чистые функции, без Qt.

Главный вопрос, на который отвечает модуль: успеваю ли я собрать себестоимость
конкретной партии до её дедлайна. Чтобы на него ответить, недостаточно знать
общую выручку — нужно понимать, деньги за какую именно партию уже пришли.
Поэтому продажи распределяются по партиям методом FIFO.

Цены у поставщика фиксированные, так что на саму себестоимость выбор FIFO
не влияет. Он нужен ровно для привязки выручки к периоду.
"""

from __future__ import annotations

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

from . import money as m
from .models import (
    Batch,
    CONSUMPTION_KINDS,
    Document,
    Sale,
)

# За сколько дней до дедлайна партия начинает считаться «горящей».
DUE_SOON_DAYS = 3

STATUS_SETTLED = "settled"
STATUS_OVERDUE = "overdue"
STATUS_DUE_SOON = "due_soon"
STATUS_OPEN = "open"

STATUS_LABELS = {
    STATUS_SETTLED: "Закрыта",
    STATUS_OVERDUE: "Просрочена",
    STATUS_DUE_SOON: "Горит",
    STATUS_OPEN: "В работе",
}


@dataclass
class Allocation:
    """Кусок продажи, списанный с конкретной партии."""

    sale_id: str
    line_index: int
    product_id: str
    batch_id: str
    qty: Decimal
    unit_cost: Decimal
    unit_price: Decimal
    kind: str
    paid_fraction: Decimal

    @property
    def cost(self) -> Decimal:
        return self.qty * self.unit_cost

    @property
    def revenue(self) -> Decimal:
        """Начислено — включая то, что ещё не оплачено."""
        return self.qty * self.unit_price

    @property
    def cash(self) -> Decimal:
        """Реально полученные деньги с учётом частичной оплаты."""
        return self.revenue * self.paid_fraction


@dataclass
class Shortfall:
    """Продали больше, чем закупали: столько-то штук взялись из ниоткуда."""

    sale_id: str
    product_id: str
    qty: Decimal


@dataclass
class BatchReport:
    batch: Batch
    cost_total: Decimal
    qty_total: Decimal
    qty_sold: Decimal
    qty_left: Decimal
    revenue: Decimal
    cash_collected: Decimal
    consumed_cost: Decimal
    paid_to_bakery: Decimal
    days_left: int
    status: str
    stock: dict[str, Decimal] = field(default_factory=dict)
    stock_cost: Decimal = m.ZERO

    @property
    def remaining_to_bakery(self) -> Decimal:
        return max(m.ZERO, m.money(self.cost_total - self.paid_to_bakery))

    @property
    def coverage_pct(self) -> Decimal:
        """Какая доля себестоимости уже собрана деньгами."""
        if self.cost_total <= 0:
            return Decimal(100)
        return m.money(self.cash_collected / self.cost_total * 100)

    @property
    def still_to_collect(self) -> Decimal:
        """Сколько денег ещё надо собрать, чтобы выйти в ноль по этой партии."""
        return max(m.ZERO, m.money(self.cost_total - self.cash_collected))

    @property
    def label(self) -> str:
        return f"№{self.batch.number} от {self.batch.date.strftime('%d.%m.%Y')}"


@dataclass
class DebtorReport:
    counterparty_id: str | None
    name: str
    debt: Decimal
    sales: list[Sale] = field(default_factory=list)


@dataclass
class StockRow:
    product_id: str
    name: str
    unit: str
    qty: Decimal
    cost: Decimal


@dataclass
class Summary:
    owed_to_bakery: Decimal
    receivable: Decimal
    revenue: Decimal
    cash_collected: Decimal
    gross_margin: Decimal
    consumed_cost: Decimal
    # Осторожно: сумма по всем товарам сразу. Осмысленна, только когда единица
    # измерения у всех одна. Литры с килограммами складывать нельзя, поэтому
    # в сводке показывается не она, а stock_cost и stock_positions.
    stock_qty: Decimal
    stock_cost: Decimal
    stock_positions: int
    overdue_count: int
    next_due: BatchReport | None

    @property
    def profit(self) -> Decimal:
        """Что осталось тебе.

        Съеденное и подаренное вычитается: пекарне за эти булки всё равно
        платить, и покрывается это из наценки на проданных.
        """
        return m.money(self.gross_margin - self.consumed_cost)


@dataclass
class Report:
    today: date
    batches: list[BatchReport]
    allocations: list[Allocation]
    shortfalls: list[Shortfall]
    debtors: list[DebtorReport]
    stock: list[StockRow]
    summary: Summary
    unassigned_bakery_payment: Decimal
    warnings: list[str]

    def batch_report(self, batch_id: str) -> BatchReport | None:
        return next((r for r in self.batches if r.batch.id == batch_id), None)

    @property
    def open_batches(self) -> list[BatchReport]:
        return [r for r in self.batches if r.status != STATUS_SETTLED]


# --- Распределение продаж по партиям --------------------------------------


def _paid_fraction(sale: Sale) -> Decimal:
    """Какая доля продажи оплачена. Переплата не даёт больше единицы."""
    total = sale.total
    if total <= 0:
        return m.ZERO
    return min(Decimal(1), sale.paid / total)


def _ordered_batches(doc: Document) -> list[tuple[int, Batch]]:
    return sorted(enumerate(doc.batches), key=lambda t: (t[1].date, t[1].number, t[0]))


def _ordered_sales(doc: Document) -> list[tuple[int, Sale]]:
    # Порядок ввода — вторичный ключ: две продажи одной датой должны
    # распределяться одинаково при каждом пересчёте.
    return sorted(enumerate(doc.sales), key=lambda t: (t[1].date, t[0]))


def batch_unit_costs(doc: Document) -> dict[tuple[str, str], Decimal]:
    """Себестоимость базовой единицы по каждой паре (партия, товар).

    Себестоимость — слепок цены на момент закупки. Если один товар попал
    в партию несколькими строками (например, пачками и поштучно), берём
    средневзвешенную: остаток по товару всё равно единый.
    """
    totals: dict[tuple[str, str], list[Decimal]] = {}
    for batch in doc.batches:
        for line in batch.lines:
            key = (batch.id, line.product_id)
            acc = totals.setdefault(key, [m.ZERO, m.ZERO])
            acc[0] += line.base_qty * line.base_unit_cost
            acc[1] += line.base_qty

    return {
        key: (cost / qty if qty > 0 else m.ZERO) for key, (cost, qty) in totals.items()
    }


def allocate(
    doc: Document,
) -> tuple[list[Allocation], list[Shortfall], dict[str, dict[str, Decimal]]]:
    """Разложить продажи по партиям методом FIFO.

    Возвращает аллокации, нехватку товара и остаток по каждой партии.

    Намеренно не проверяем, что дата продажи позже даты партии: люди заводят
    документы в произвольном порядке — особенно когда вносят историю задним
    числом, — и жаловаться на это было бы шумом. Значение имеет только очередь
    партий.
    """
    # Всё внутри — в базовых единицах. Иначе поштучная продажа не вычиталась бы
    # из купленной пачки: это один товар, а не два разных.
    remaining: dict[str, dict[str, Decimal]] = {}
    for _, batch in _ordered_batches(doc):
        per_product = remaining.setdefault(batch.id, {})
        for line in batch.lines:
            per_product[line.product_id] = per_product.get(line.product_id, m.ZERO) + line.base_qty

    unit_costs = batch_unit_costs(doc)
    order = [batch.id for _, batch in _ordered_batches(doc)]

    allocations: list[Allocation] = []
    shortfalls: list[Shortfall] = []

    for _, sale in _ordered_sales(doc):
        fraction = _paid_fraction(sale)
        for line_index, line in enumerate(sale.lines):
            need = line.base_qty
            for batch_id in order:
                if need <= 0:
                    break
                available = remaining.get(batch_id, {}).get(line.product_id, m.ZERO)
                if available <= 0:
                    continue

                take = min(available, need)
                remaining[batch_id][line.product_id] = available - take
                need -= take
                allocations.append(
                    Allocation(
                        sale_id=sale.id,
                        line_index=line_index,
                        product_id=line.product_id,
                        batch_id=batch_id,
                        qty=take,
                        unit_cost=unit_costs.get((batch_id, line.product_id), m.ZERO),
                        unit_price=line.base_unit_price,
                        kind=sale.kind,
                        paid_fraction=fraction,
                    )
                )

            if need > 0:
                shortfalls.append(
                    Shortfall(sale_id=sale.id, product_id=line.product_id, qty=need)
                )

    return allocations, shortfalls, remaining


# --- Разнесение платежей пекарне ------------------------------------------


def _spread_bakery_payments(doc: Document, cost_totals: dict[str, Decimal]) -> tuple[dict[str, Decimal], Decimal]:
    """Разложить платежи пекарне по партиям.

    Платёж с указанной партией гасит её; излишек не превращается в
    отрицательный долг, а переливается в общий котёл. Платежи без привязки
    гасят старейшие непокрытые партии — так же, как это происходит в жизни.
    """
    paid: dict[str, Decimal] = {batch_id: m.ZERO for batch_id in cost_totals}
    pool = m.ZERO

    for payment in doc.bakery_payments:
        batch_id = payment.batch_id
        if batch_id and batch_id in paid:
            room = max(m.ZERO, cost_totals[batch_id] - paid[batch_id])
            applied = min(room, payment.amount)
            paid[batch_id] += applied
            pool += payment.amount - applied
        else:
            pool += payment.amount

    for _, batch in _ordered_batches(doc):
        if pool <= 0:
            break
        room = max(m.ZERO, cost_totals.get(batch.id, m.ZERO) - paid.get(batch.id, m.ZERO))
        applied = min(room, pool)
        paid[batch.id] = paid.get(batch.id, m.ZERO) + applied
        pool -= applied

    return paid, pool


# --- Сборка отчёта --------------------------------------------------------


def _status(batch: Batch, remaining_to_bakery: Decimal, days_left: int) -> str:
    if batch.closed or remaining_to_bakery <= 0:
        return STATUS_SETTLED
    if days_left < 0:
        return STATUS_OVERDUE
    if days_left <= DUE_SOON_DAYS:
        return STATUS_DUE_SOON
    return STATUS_OPEN


def build(doc: Document, today: date | None = None) -> Report:
    """Пересчитать всё. Дешевле, чем поддерживать инкрементальное состояние."""
    today = today or date.today()
    allocations, shortfalls, remaining = allocate(doc)
    unit_costs = batch_unit_costs(doc)

    cost_totals = {batch.id: batch.cost_total for batch in doc.batches}
    paid_to_bakery, unassigned = _spread_bakery_payments(doc, cost_totals)

    per_batch: dict[str, dict[str, Decimal]] = {
        batch.id: {"qty_sold": m.ZERO, "revenue": m.ZERO, "cash": m.ZERO, "consumed": m.ZERO}
        for batch in doc.batches
    }
    for a in allocations:
        bucket = per_batch.get(a.batch_id)
        if bucket is None:
            continue
        bucket["qty_sold"] += a.qty
        if a.kind in CONSUMPTION_KINDS:
            bucket["consumed"] += a.cost
        else:
            bucket["revenue"] += a.revenue
            bucket["cash"] += a.cash

    batch_reports: list[BatchReport] = []
    for _, batch in _ordered_batches(doc):
        bucket = per_batch[batch.id]
        cost_total = cost_totals[batch.id]
        paid = m.money(paid_to_bakery.get(batch.id, m.ZERO))
        days_left = (batch.due_date - today).days
        stock = {
            pid: qty
            for pid, qty in remaining.get(batch.id, {}).items()
            if qty > 0
        }
        batch_reports.append(
            BatchReport(
                batch=batch,
                cost_total=cost_total,
                qty_total=batch.qty_total,
                qty_sold=m.qty(bucket["qty_sold"]),
                qty_left=m.qty(sum(stock.values(), m.ZERO)),
                revenue=m.money(bucket["revenue"]),
                cash_collected=m.money(bucket["cash"]),
                consumed_cost=m.money(bucket["consumed"]),
                paid_to_bakery=paid,
                days_left=days_left,
                status=_status(batch, max(m.ZERO, cost_total - paid), days_left),
                stock=stock,
                stock_cost=m.money(
                    sum(
                        (
                            qty * unit_costs.get((batch.id, pid), m.ZERO)
                            for pid, qty in stock.items()
                        ),
                        m.ZERO,
                    )
                ),
            )
        )

    return Report(
        today=today,
        batches=batch_reports,
        allocations=allocations,
        shortfalls=shortfalls,
        debtors=_debtors(doc),
        stock=_stock(doc, batch_reports),
        summary=_summary(doc, batch_reports, allocations, today),
        unassigned_bakery_payment=m.money(unassigned),
        warnings=_warnings(doc, shortfalls, unassigned),
    )


def _debtors(doc: Document) -> list[DebtorReport]:
    """Кто и сколько должен. Продажи без контрагента идут отдельной строкой."""
    buckets: dict[str | None, DebtorReport] = {}
    for sale in doc.sales:
        if sale.debt <= 0:
            continue
        key = sale.counterparty_id
        report = buckets.get(key)
        if report is None:
            report = DebtorReport(
                counterparty_id=key,
                name=doc.counterparty_name(key) if key else "Без контрагента",
                debt=m.ZERO,
            )
            buckets[key] = report
        report.debt = m.money(report.debt + sale.debt)
        report.sales.append(sale)

    return sorted(buckets.values(), key=lambda r: (-r.debt, r.name))


def _stock(doc: Document, batch_reports: list[BatchReport]) -> list[StockRow]:
    unit_costs = batch_unit_costs(doc)
    qty_by_product: dict[str, Decimal] = {}
    cost_by_product: dict[str, Decimal] = {}
    for report in batch_reports:
        for product_id, qty in report.stock.items():
            unit_cost = unit_costs.get((report.batch.id, product_id), m.ZERO)
            qty_by_product[product_id] = qty_by_product.get(product_id, m.ZERO) + qty
            cost_by_product[product_id] = cost_by_product.get(product_id, m.ZERO) + qty * unit_cost

    rows = []
    for product in doc.products:
        qty = qty_by_product.get(product.id, m.ZERO)
        if qty <= 0 and product.archived:
            continue
        rows.append(
            StockRow(
                product_id=product.id,
                name=product.name,
                unit=product.unit,
                qty=m.qty(qty),
                cost=m.money(cost_by_product.get(product.id, m.ZERO)),
            )
        )
    return sorted(rows, key=lambda r: r.name)


def _summary(
    doc: Document,
    batch_reports: list[BatchReport],
    allocations: list[Allocation],
    today: date,
) -> Summary:
    revenue = m.ZERO
    cash = m.ZERO
    margin = m.ZERO
    consumed = m.ZERO
    for a in allocations:
        if a.kind in CONSUMPTION_KINDS:
            consumed += a.cost
        else:
            revenue += a.revenue
            cash += a.cash
            margin += a.revenue - a.cost

    open_reports = [r for r in batch_reports if r.status != STATUS_SETTLED]
    next_due = min(open_reports, key=lambda r: r.batch.due_date, default=None)

    return Summary(
        owed_to_bakery=m.money(sum((r.remaining_to_bakery for r in batch_reports), m.ZERO)),
        receivable=m.money(sum((s.debt for s in doc.sales), m.ZERO)),
        revenue=m.money(revenue),
        cash_collected=m.money(cash),
        gross_margin=m.money(margin),
        consumed_cost=m.money(consumed),
        stock_qty=m.qty(sum((r.qty_left for r in batch_reports), m.ZERO)),
        stock_cost=m.money(sum((r.stock_cost for r in batch_reports), m.ZERO)),
        stock_positions=len({pid for r in batch_reports for pid in r.stock}),
        overdue_count=sum(1 for r in batch_reports if r.status == STATUS_OVERDUE),
        next_due=next_due,
    )


def _warnings(doc: Document, shortfalls: list[Shortfall], unassigned: Decimal) -> list[str]:
    """То, что стоит показать пользователю, но что не должно ломать расчёт."""
    warnings: list[str] = []

    by_product: dict[str, Decimal] = {}
    for s in shortfalls:
        by_product[s.product_id] = by_product.get(s.product_id, m.ZERO) + s.qty
    for product_id, qty in by_product.items():
        warnings.append(
            f"«{doc.product_name(product_id)}»: продано на {m.fmt_qty(qty)} больше, "
            "чем закуплено — проверь закупки."
        )

    if unassigned > 0:
        warnings.append(
            f"Пекарне переплачено {m.fmt_money(unassigned, doc.settings.currency)} — "
            "это аванс под будущие партии."
        )

    return warnings