from typing import Any from backend.db.database import get_connection from backend.services.settings_service import get_income_settings, get_exclude_transfers def get_summary( period_start: str | None = None, period_end: str | None = None, bank_ids: list[int] | None = None, ) -> dict[str, Any]: """ Итог по всем счетам (или по выбранным банкам): суммарный баланс, доход, дебет, кредит. period_start/period_end в формате YYYY-MM-DD; bank_ids — список id банков для фильтра (пусто = все). """ conn = get_connection() try: salary_bank_id = conn.execute( "SELECT id FROM banks WHERE is_salary = 1" ).fetchone() salary_bank_id = salary_bank_id["id"] if salary_bank_id else None only_salary_card, salary_account_id = get_income_settings() income_account_filter = salary_account_id if (only_salary_card and salary_account_id) else None accounts_sql = "SELECT a.id, a.bank_id, a.external_id, a.name, b.code FROM accounts a JOIN banks b ON a.bank_id = b.id" params_acc = [] if bank_ids: placeholders = ",".join("?" * len(bank_ids)) accounts_sql += f" WHERE a.bank_id IN ({placeholders})" params_acc = list(bank_ids) accounts = conn.execute(accounts_sql, params_acc).fetchall() total_balance = 0.0 by_account = [] income = 0.0 debit = 0.0 credit = 0.0 exclude_filter = " AND (t.excluded_from_balance = 0 OR t.excluded_from_balance IS NULL)" if get_exclude_transfers(): exclude_filter += " AND t.description NOT LIKE '%Перевод%'" for acc in accounts: # Начальный остаток на начало периода (последний установленный на дату <= period_start) ob = conn.execute( "SELECT amount FROM opening_balances WHERE account_id = ? AND (? IS NULL OR period_start <= ?) ORDER BY period_start DESC LIMIT 1", (acc["id"], period_start, period_start or "0000-01-01"), ).fetchone() opening = float(ob["amount"]) if ob else 0.0 sql = "SELECT COALESCE(SUM(t.amount), 0) FROM transactions t WHERE t.account_id = ?" + exclude_filter params = [acc["id"]] if period_start: sql += " AND t.operation_date >= ?" params.append(period_start) if period_end: sql += " AND t.operation_date <= ?" params.append(period_end) row = conn.execute(sql, params).fetchone() tx_sum = float(row[0]) balance = opening + tx_sum total_balance += balance # Доход: приходы по зарплатному банку или только по выбранной зарплатной карте count_income = False if income_account_filter is not None: count_income = acc["id"] == income_account_filter elif acc["bank_id"] == salary_bank_id: count_income = True if count_income: inc_sql = "SELECT COALESCE(SUM(t.amount), 0) FROM transactions t WHERE t.account_id = ? AND t.amount > 0" + exclude_filter inc_params = [acc["id"]] if period_start: inc_sql += " AND t.operation_date >= ?" inc_params.append(period_start) if period_end: inc_sql += " AND t.operation_date <= ?" inc_params.append(period_end) income += float(conn.execute(inc_sql, inc_params).fetchone()[0]) # Дебет = приход (увеличение счёта), Кредит = расход (уменьшение счёта) d_sql = "SELECT COALESCE(SUM(CASE WHEN t.amount > 0 THEN t.amount ELSE 0 END), 0), COALESCE(SUM(CASE WHEN t.amount < 0 THEN -t.amount ELSE 0 END), 0) FROM transactions t WHERE t.account_id = ?" + exclude_filter d_params = [acc["id"]] if period_start: d_sql += " AND t.operation_date >= ?" d_params.append(period_start) if period_end: d_sql += " AND t.operation_date <= ?" d_params.append(period_end) d_row = conn.execute(d_sql, d_params).fetchone() debit += float(d_row[0]) # приход credit += float(d_row[1]) # расход by_account.append({ "account_id": acc["id"], "bank_code": acc["code"], "external_id": acc["external_id"], "name": acc["name"], "opening_balance": opening, "transactions_sum": tx_sum, "balance": balance, }) # Копилка: только если в фильтр входит Я-банк (или фильтра нет) piggy_sql = "SELECT COALESCE(SUM(-t.amount), 0) FROM transactions t WHERE t.description LIKE ? " + exclude_filter piggy_params = ["%Перевод между счетами одного клиента%"] if bank_ids: placeholders = ",".join("?" * len(bank_ids)) piggy_sql += f" AND t.account_id IN (SELECT id FROM accounts WHERE bank_id IN ({placeholders}))" piggy_params.extend(bank_ids) piggy_row = conn.execute(piggy_sql, piggy_params).fetchone() savings_balance = float(piggy_row[0]) if piggy_row else 0.0 net_flow = debit - credit # за период: положительный = накопление, отрицательный = траты return { "total_balance": round(total_balance, 2), "income": round(income, 2), "debit": round(debit, 2), "credit": round(credit, 2), "net_flow": round(net_flow, 2), "savings_balance": round(savings_balance, 2), "by_account": by_account, "salary_bank_id": salary_bank_id, } finally: conn.close() def get_transactions( account_id: int | None = None, bank_ids: list[int] | None = None, period_start: str | None = None, period_end: str | None = None, limit: int = 50, offset: int = 0, ) -> tuple[list[dict], int]: """Возвращает (список транзакций, общее количество). Фильтры: bank_ids, period_start, period_end.""" conn = get_connection() try: conditions = [] params = [] if account_id is not None: conditions.append("t.account_id = ?") params.append(account_id) if bank_ids: placeholders = ",".join("?" * len(bank_ids)) conditions.append(f"a.bank_id IN ({placeholders})") params.extend(bank_ids) if period_start: conditions.append("t.operation_date >= ?") params.append(period_start) if period_end: conditions.append("t.operation_date <= ?") params.append(period_end) where = " WHERE " + " AND ".join(conditions) if conditions else "" count_sql = "SELECT COUNT(*) FROM transactions t JOIN accounts a ON t.account_id = a.id JOIN banks b ON a.bank_id = b.id" + where total = conn.execute(count_sql, params).fetchone()[0] sql = """ SELECT t.id, t.account_id, t.operation_date, t.amount, t.description, t.source_file, t.excluded_from_balance, a.external_id, b.code as bank_code, b.name as bank_name FROM transactions t JOIN accounts a ON t.account_id = a.id JOIN banks b ON a.bank_id = b.id """ + where + " ORDER BY t.operation_date DESC, t.id DESC LIMIT ? OFFSET ?" params.extend([limit, offset]) rows = conn.execute(sql, params).fetchall() items = [dict(r) for r in rows] return items, total finally: conn.close()