#!/usr/bin/env python3 """Маржинальность направлений Марса. См. docs/margins-design-2026-07-13.md Клиенты API, .env и календарь года живут в core.py — здесь только бизнес-логика. Заливка в Google Sheet — push_to_gsheet.py (считает из .sdd/full_year_matrix.json). """ import calendar import core from core import (HERE, BRANCHES, FIN_YEAR, ftget, ftmut, ft_all, ft_transactions, # noqa: F401 ft_categories, alfa, alfa_login, alfa_all, months_to_now, month_key, month_range_alfa, year_range_alfa, money, norm as _norm) DIRECTIONS = ["1-1", "Групповые", "Хор", "B2B", "Внутренние мероприятия", "Съёмки", "Аренда"] # --- маппинг доходов AlfaCRM: pay_item name -> направление (или None) --- INCOME_MAP = { "Групповой": "Групповые", "Групповой абонемент": "Групповые", "Групповое пробное": "Групповые", "Хор (курс)": "Хор", "Хор (разовый)": "Хор", "Квартирник": "Внутренние мероприятия", "Концерт": "Внутренние мероприятия", "Прочие внутренние ивенты": "Внутренние мероприятия", "Cover session": "Съёмки", "Сover session": "Съёмки", # в CRM встречается с лат. С "Аренда": "Аренда", "Абонемент оффлайн": "1-1", "Абонемент онлайн": "1-1", "Разовое": "1-1", "Разовое онлайн": "1-1", "Пробное оффлайн": "1-1", "Пробное онлайн": "1-1", "Первый абонемент оффлайн": "1-1", "Первый абонемент онлайн": "1-1", "Первое разовое": "1-1", "Предоплата": "1-1", "оплата с сайта": "1-1", "Парный абонемент": "1-1", "Парное пробное": "1-1", "Парное разовое": "1-1", "Фитмост разовое": "1-1", "Подарочный сертификат": "1-1", "Абонемент ЧП 2024": "1-1", } # Явно НЕ направление: B2B* / b2b (доход B2B из Финтабло), Мерч, Мероприятия внешние на привлечение. def income_direction(pay_item_name): """Статья CRM → направление. B2B-статьи не перечислены поимённо (их много) — фолбэк ниже.""" d = INCOME_MAP.get((pay_item_name or "").strip()) if not d and pay_item_name and "b2b" in pay_item_name.strip().lower(): return "B2B" return d # Намеренно НЕ направления (income_direction → None ожидаемо): мерч, внешние мероприятия — # они попадают в «прочий доход CRM» (см. _collect_income), это не ошибка маппинга. INCOME_EXCLUDE_NAMES = {"мерч", "мероприятия внешние на привлечение"} # прямые статьи расходов: нормализованное имя -> направление EXPENSE_DIRECT = { _norm("Расходы на хоровые курсы"): "Хор", _norm("Расходы на съемки"): "Съёмки", _norm("Расходы на внутренние ивенты"): "Внутренние мероприятия", _norm("Расходы на b2b-ивенты"): "B2B", } LUMP_NAME = _norm("Зарплаты педагогов: Расходы на занятия 1-1") def expense_direction(category_name): n = _norm(category_name) if n == LUMP_NAME: return "__lump__" return EXPENSE_DIRECT.get(n, "overhead") # --- разбивка зарплаты-лампы по бакетам (1-1, Групповые, B2B) --- # Ставки — ЕДИНЫЙ источник pay_config.LESSON_RATES (раньше была своя копия, разъезжались) from pay_config import LESSON_RATES as LESSON_TYPE_RATE # noqa: E402 LESSON_TYPE_BUCKET = { "Индивидуальный": "1-1", "Пробный": "1-1", "Вписанный": "1-1", "ФИТМОСТ": "1-1", "Групповой": "Групповые", "Групповой пробный": "Групповые", "B2B Индивидуальный": "B2B", "B2B Пробный": "B2B", "B2B Групповой": "B2B", } # Типы вне карт (Квартирник, Концерт, Аренда-как-урок) в разбивке НЕ участвуют. def split_lump(lump_amount, lesson_counts): """ЗП-лампу делим между 1-1 / Групповые / B2B пропорционально «уроки × ставка». Если уроков нет вообще (CRM не ответил, месяц ещё пустой) — возвращаем НУЛИ, а не сваливаем всю сумму на «1-1»: раньше сбой CRM молча топил маржу одного направления на миллионы. Неразбитый остаток вызывающий кладёт в overhead (см. month_pnl). """ weights = {"1-1": 0.0, "Групповые": 0.0, "B2B": 0.0} for ltype, cnt in lesson_counts.items(): bucket = LESSON_TYPE_BUCKET.get(ltype) if not bucket: continue weights[bucket] += LESSON_TYPE_RATE.get(ltype, 0) * cnt total = sum(weights.values()) if total == 0: return {"1-1": 0.0, "Групповые": 0.0, "B2B": 0.0} return {k: lump_amount * w / total for k, w in weights.items()} # --- AlfaCRM: выручка по направлениям --- def _alfa_pay_items(token): return {x['id']: x['name'] for x in alfa('/v2api/3/pay-item/index', {'page': 0}, token).get('items', [])} def _collect_income(months, token, pay_items, date_from, date_to): """Платежи CRM за период → выручка по направлениям для каждого месяца из months. Единственное место, где живёт маппинг платёж → направление (раньше эта логика была скопирована в income_by_direction и расходилась с income_all_months). """ res = {m: {d: 0.0 for d in DIRECTIONS} for m in months} crm_other = {m: {} for m in months} # прочий CRM-доход: не направление (мерч, внешние ивенты…) inc_detail = {m: {d: {} for d in DIRECTIONS} for m in months} # {направление: {статья CRM: сумма}} seen = set() for br in BRANCHES: for x in alfa_all(f'/v2api/{br}/pay/index', {'pay_type_id': 1, 'date_from': date_from, 'date_to': date_to}, token): pid = x.get('id') if pid in seen: continue seen.add(pid) dd = (x.get('document_date') or '') # dd.mm.yyyy if len(dd) < 10: continue mo = dd[3:] if mo not in res: continue name = pay_items.get(x.get('pay_item_id')) val = float(x.get('income') or 0) d = income_direction(name) if d: res[mo][d] += val key = (name or "").strip() inc_detail[mo][d][key] = inc_detail[mo][d].get(key, 0.0) + val else: key = (name or "").strip() or "Прочий доход CRM" crm_other[mo][key] = crm_other[mo].get(key, 0.0) + val for mo in months: res[mo]["__crm_other__"] = crm_other[mo] res[mo]["__income_detail__"] = inc_detail[mo] return res def income_all_months(months, token, pay_items): """Выручка по направлениям за все месяцы года.""" return _collect_income(months, token, pay_items, *year_range_alfa()) def income_by_direction(month, token=None, pay_items=None): """Выручка по направлениям за ОДИН месяц (MM.YYYY) — тот же маппинг, что и за год.""" token = token or alfa_login() pay_items = pay_items if pay_items is not None else _alfa_pay_items(token) res = _collect_income([month], token, pay_items, *month_range_alfa(month))[month] return {d: res[d] for d in DIRECTIONS} def income_with_other(month, token=None, pay_items=None): """(направления, прочий_доход) за месяц MM.YYYY. income_by_direction отдаёт только DIRECTIONS и молча теряет «прочее» (мерч, внешние мероприятия). Канону доходных строк оно нужно — это строка «Прочее». """ token = token or alfa_login() pay_items = pay_items if pay_items is not None else _alfa_pay_items(token) res = _collect_income([month], token, pay_items, *month_range_alfa(month))[month] dirs = {d: res.get(d, 0.0) for d in DIRECTIONS} other = sum(res.get("__crm_other__", {}).values()) return dirs, other # --- Финтабло: расходы по направлениям --- _FT_TX_CACHE = {} # group -> все факт-операции года (тянем раз, фильтруем по месяцу в памяти) def _ft_tx(group, month): """факт-операции нужной группы за factMonth == month (MM.YYYY).""" if group not in _FT_TX_CACHE: _FT_TX_CACHE[group] = ft_transactions(group=group) return [t for t in _FT_TX_CACHE[group] if not t.get('isPlan') and t.get('factMonth') == month] def expenses_by_direction(month, cats=None): cats = cats or ft_categories() res = {d: 0.0 for d in DIRECTIONS} res["overhead"] = 0.0; res["__lump__"] = 0.0 detail = {} # постатейная разбивка overhead: имя статьи -> сумма for t in _ft_tx('outcome', month): name = cats.get(t.get('categoryId'), '') or "(без статьи)" d = expense_direction(name) val = float(t.get('value') or 0) res[d] = res.get(d, 0.0) + val if d == "overhead": detail[name] = detail.get(name, 0.0) + val res["__overhead_detail__"] = detail return res # «Прочие доходы» из Финтабло постатейно: всё, что НЕ b2c (даёт CRM по направлениям) # и НЕ b2b (уже учтено в направлении B2B). Сюда падают кредиты/займы, возвраты, # прочие поступления — чтобы в таблице сходились ВСЕ доходы с учётом. INCOME_OTHER_SKIP = {_norm("Выручка b2c"), _norm("Выручка b2b")} def ft_income_other(month, cats=None): cats = cats or ft_categories() detail = {} for t in _ft_tx('income', month): name = cats.get(t.get('categoryId'), '') or "(без статьи)" if _norm(name) in INCOME_OTHER_SKIP: continue detail[name] = detail.get(name, 0.0) + float(t.get('value') or 0) return detail # --- AlfaCRM: уроки по типам (для разбивки ЗП) --- def lesson_counts(month, token=None): """month = MM.YYYY. Считает проведённые уроки по типам.""" token = token or alfa_login() df, dt = month_range_alfa(month) counts = {} for br in BRANCHES: for x in alfa_all(f'/v2api/{br}/lesson/index', {'date_from': df, 'date_to': dt}, token): if x.get('status') != 3: continue name = x.get('lesson_type_name') or '' counts[name] = counts.get(name, 0) + 1 return counts def salary_by_direction(month, lump_amount, token=None): return split_lump(lump_amount, lesson_counts(month, token)) # --- сборка месячной матрицы --- def merge_direct(expenses, salary): direct = {d: expenses.get(d, 0.0) for d in DIRECTIONS} direct["1-1"] += salary.get("1-1", 0.0) direct["Групповые"] += salary.get("Групповые", 0.0) direct["B2B"] += salary.get("B2B", 0.0) return direct def month_pnl(month, income_m, cats, token): revenue = {d: income_m[d] for d in DIRECTIONS} # B2B-доход теперь из CRM постатейно (см. income_all_months), Финтабло «Выручка b2b» не используем. exp = expenses_by_direction(month, cats) salary = salary_by_direction(month, exp["__lump__"], token) direct = merge_direct(exp, salary) margin = {d: revenue[d] - direct[d] for d in DIRECTIONS} overhead = exp["overhead"] # лампу не разнесли (нет уроков в CRM) — кладём остаток в overhead, чтобы деньги не исчезли unsplit = exp["__lump__"] - sum(salary.values()) if unsplit > 1: print(f"⚠ {month}: ЗП педагогов {money(unsplit)} не разнесена по направлениям " f"(нет проведённых уроков в CRM) — учтена в overhead") overhead += unsplit income_other = ft_income_other(month, cats) # прочий CRM-доход (мерч, внешние ивенты) — операционный доход, но вне направлений for name, val in income_m.get("__crm_other__", {}).items(): income_other[name] = income_other.get(name, 0.0) + val return {"revenue": revenue, "direct": direct, "margin": margin, "overhead": overhead, "overhead_detail": exp["__overhead_detail__"], "income_other": income_other, "income_detail": income_m.get("__income_detail__", {})} def assemble(months): token = alfa_login() pay_items = _alfa_pay_items(token) cats = ft_categories() income = income_all_months(months, token, pay_items) return {m: month_pnl(m, income[m], cats, token) for m in months} def months_2026_to_now(): """Месяцы года до текущего включительно. Оставлено как псевдоним core.months_to_now.""" return months_to_now() if __name__ == "__main__": print("margins — библиотека. Пересчёт: python3 rebuild_matrix.py, заливка: python3 push_to_gsheet.py")