173 lines
8.0 KiB
Python
173 lines
8.0 KiB
Python
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()
|