"""Статистика: сводные цифры по периодам, товарам и людям. Чистые функции без 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))