434 lines
16 KiB
Python
434 lines
16 KiB
Python
import csv
|
||
from pathlib import Path
|
||
from sqlite3 import IntegrityError, Row
|
||
from typing import Iterable
|
||
|
||
from app.database import get_connection
|
||
from app.utils.formatters import current_month_key, format_money
|
||
|
||
|
||
class FinanceService:
|
||
"""Business layer for Finance Control.
|
||
|
||
UI classes call this object instead of duplicating SQL logic. It makes the
|
||
program easier to test, maintain and explain from the OOP point of view.
|
||
"""
|
||
|
||
def get_setting(self, key: str, default: str) -> str:
|
||
with get_connection() as connection:
|
||
row = connection.execute(
|
||
"SELECT value FROM settings WHERE key = ?",
|
||
(key,),
|
||
).fetchone()
|
||
return default if row is None else row["value"]
|
||
|
||
def set_setting(self, key: str, value: str) -> None:
|
||
with get_connection() as connection:
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO settings (key, value)
|
||
VALUES (?, ?)
|
||
ON CONFLICT(key) DO UPDATE SET value = excluded.value
|
||
""",
|
||
(key, value),
|
||
)
|
||
connection.commit()
|
||
|
||
def add_account(self, name: str, account_type: str, balance: float) -> None:
|
||
with get_connection() as connection:
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO accounts (name, account_type, balance)
|
||
VALUES (?, ?, ?)
|
||
""",
|
||
(name, account_type, balance),
|
||
)
|
||
connection.commit()
|
||
|
||
def delete_account(self, account_id: int) -> None:
|
||
with get_connection() as connection:
|
||
connection.execute("DELETE FROM accounts WHERE id = ?", (account_id,))
|
||
connection.commit()
|
||
|
||
def list_accounts(self) -> list[Row]:
|
||
with get_connection() as connection:
|
||
return connection.execute(
|
||
"SELECT * FROM accounts ORDER BY id DESC"
|
||
).fetchall()
|
||
|
||
def list_account_options(self) -> list[Row]:
|
||
with get_connection() as connection:
|
||
return connection.execute(
|
||
"SELECT id, name, account_type, balance FROM accounts ORDER BY name"
|
||
).fetchall()
|
||
|
||
def add_category(self, name: str, category_type: str) -> None:
|
||
with get_connection() as connection:
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO categories (name, category_type)
|
||
VALUES (?, ?)
|
||
""",
|
||
(name, category_type),
|
||
)
|
||
connection.commit()
|
||
|
||
def delete_category(self, category_id: int) -> None:
|
||
with get_connection() as connection:
|
||
connection.execute("DELETE FROM categories WHERE id = ?", (category_id,))
|
||
connection.commit()
|
||
|
||
def list_categories(self, category_type: str | None = None) -> list[Row]:
|
||
with get_connection() as connection:
|
||
if category_type is None:
|
||
return connection.execute(
|
||
"""
|
||
SELECT * FROM categories
|
||
ORDER BY category_type DESC, name
|
||
"""
|
||
).fetchall()
|
||
return connection.execute(
|
||
"""
|
||
SELECT * FROM categories
|
||
WHERE category_type = ?
|
||
ORDER BY name
|
||
""",
|
||
(category_type,),
|
||
).fetchall()
|
||
|
||
def add_transaction(
|
||
self,
|
||
title: str,
|
||
amount: float,
|
||
transaction_type: str,
|
||
account_id: int,
|
||
category_id: int,
|
||
created_at: str,
|
||
) -> None:
|
||
sign = 1 if transaction_type == "Доход" else -1
|
||
delta = amount * sign
|
||
with get_connection() as connection:
|
||
connection.execute("BEGIN")
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO transactions (
|
||
title, amount, transaction_type,
|
||
account_id, category_id, created_at
|
||
)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(title, amount, transaction_type, account_id, category_id, created_at),
|
||
)
|
||
connection.execute(
|
||
"UPDATE accounts SET balance = balance + ? WHERE id = ?",
|
||
(delta, account_id),
|
||
)
|
||
connection.commit()
|
||
|
||
def delete_transaction(self, transaction_id: int) -> None:
|
||
with get_connection() as connection:
|
||
row = connection.execute(
|
||
"SELECT * FROM transactions WHERE id = ?",
|
||
(transaction_id,),
|
||
).fetchone()
|
||
if row is None:
|
||
return
|
||
sign = -1 if row["transaction_type"] == "Доход" else 1
|
||
delta = row["amount"] * sign
|
||
connection.execute("BEGIN")
|
||
connection.execute(
|
||
"UPDATE accounts SET balance = balance + ? WHERE id = ?",
|
||
(delta, row["account_id"]),
|
||
)
|
||
connection.execute(
|
||
"DELETE FROM transactions WHERE id = ?",
|
||
(transaction_id,),
|
||
)
|
||
connection.commit()
|
||
|
||
def list_transactions(
|
||
self,
|
||
search: str = "",
|
||
transaction_type: str = "Все",
|
||
limit: int = 100,
|
||
) -> list[Row]:
|
||
clauses = []
|
||
params: list[object] = []
|
||
if search:
|
||
clauses.append("t.title LIKE ?")
|
||
params.append(f"%{search}%")
|
||
if transaction_type != "Все":
|
||
clauses.append("t.transaction_type = ?")
|
||
params.append(transaction_type)
|
||
|
||
where_sql = ""
|
||
if clauses:
|
||
where_sql = "WHERE " + " AND ".join(clauses)
|
||
|
||
params.append(limit)
|
||
with get_connection() as connection:
|
||
return connection.execute(
|
||
f"""
|
||
SELECT t.id,
|
||
t.title,
|
||
t.amount,
|
||
t.transaction_type,
|
||
t.created_at,
|
||
a.name AS account_name,
|
||
c.name AS category_name
|
||
FROM transactions AS t
|
||
JOIN accounts AS a ON a.id = t.account_id
|
||
JOIN categories AS c ON c.id = t.category_id
|
||
{where_sql}
|
||
ORDER BY t.created_at DESC, t.id DESC
|
||
LIMIT ?
|
||
""",
|
||
params,
|
||
).fetchall()
|
||
|
||
def add_subscription(
|
||
self,
|
||
name: str,
|
||
amount: float,
|
||
billing_day: int,
|
||
period: str,
|
||
status: str,
|
||
) -> None:
|
||
with get_connection() as connection:
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO subscriptions (name, amount, billing_day, period, status)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
""",
|
||
(name, amount, billing_day, period, status),
|
||
)
|
||
connection.commit()
|
||
|
||
def delete_subscription(self, subscription_id: int) -> None:
|
||
with get_connection() as connection:
|
||
connection.execute(
|
||
"DELETE FROM subscriptions WHERE id = ?",
|
||
(subscription_id,),
|
||
)
|
||
connection.commit()
|
||
|
||
def toggle_subscription_status(self, subscription_id: int) -> None:
|
||
with get_connection() as connection:
|
||
row = connection.execute(
|
||
"SELECT status FROM subscriptions WHERE id = ?",
|
||
(subscription_id,),
|
||
).fetchone()
|
||
if row is None:
|
||
return
|
||
new_status = "Пауза" if row["status"] == "Активна" else "Активна"
|
||
connection.execute(
|
||
"UPDATE subscriptions SET status = ? WHERE id = ?",
|
||
(new_status, subscription_id),
|
||
)
|
||
connection.commit()
|
||
|
||
def list_subscriptions(self) -> list[Row]:
|
||
with get_connection() as connection:
|
||
return connection.execute(
|
||
"SELECT * FROM subscriptions ORDER BY status, billing_day, name"
|
||
).fetchall()
|
||
|
||
def add_or_update_budget(self, category_id: int, limit_amount: float) -> None:
|
||
with get_connection() as connection:
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO budgets (category_id, limit_amount)
|
||
VALUES (?, ?)
|
||
ON CONFLICT(category_id)
|
||
DO UPDATE SET limit_amount = excluded.limit_amount
|
||
""",
|
||
(category_id, limit_amount),
|
||
)
|
||
connection.commit()
|
||
|
||
def delete_budget(self, budget_id: int) -> None:
|
||
with get_connection() as connection:
|
||
connection.execute("DELETE FROM budgets WHERE id = ?", (budget_id,))
|
||
connection.commit()
|
||
|
||
def list_budgets(self) -> list[Row]:
|
||
month_key = current_month_key()
|
||
with get_connection() as connection:
|
||
return connection.execute(
|
||
"""
|
||
SELECT b.id,
|
||
b.limit_amount,
|
||
c.name AS category_name,
|
||
COALESCE(SUM(t.amount), 0) AS spent
|
||
FROM budgets AS b
|
||
JOIN categories AS c ON c.id = b.category_id
|
||
LEFT JOIN transactions AS t
|
||
ON t.category_id = b.category_id
|
||
AND t.transaction_type = 'Расход'
|
||
AND strftime('%Y-%m', t.created_at) = ?
|
||
GROUP BY b.id, b.limit_amount, c.name
|
||
ORDER BY c.name
|
||
""",
|
||
(month_key,),
|
||
).fetchall()
|
||
|
||
def dashboard_summary(self) -> dict[str, object]:
|
||
month_key = current_month_key()
|
||
with get_connection() as connection:
|
||
balance = connection.execute(
|
||
"SELECT COALESCE(SUM(balance), 0) AS total FROM accounts"
|
||
).fetchone()["total"]
|
||
summary = connection.execute(
|
||
"""
|
||
SELECT
|
||
COALESCE(SUM(CASE WHEN transaction_type = 'Доход'
|
||
THEN amount END), 0) AS income,
|
||
COALESCE(SUM(CASE WHEN transaction_type = 'Расход'
|
||
THEN amount END), 0) AS expense
|
||
FROM transactions
|
||
WHERE strftime('%Y-%m', created_at) = ?
|
||
""",
|
||
(month_key,),
|
||
).fetchone()
|
||
subscriptions = connection.execute(
|
||
"""
|
||
SELECT COUNT(*) AS count,
|
||
COALESCE(SUM(
|
||
CASE WHEN period = 'Год'
|
||
THEN amount / 12.0
|
||
ELSE amount
|
||
END
|
||
), 0) AS monthly_total
|
||
FROM subscriptions
|
||
WHERE status = 'Активна'
|
||
"""
|
||
).fetchone()
|
||
upcoming = connection.execute(
|
||
"""
|
||
SELECT id, name, amount, billing_day, period
|
||
FROM subscriptions
|
||
WHERE status = 'Активна'
|
||
ORDER BY billing_day, name
|
||
LIMIT 8
|
||
"""
|
||
).fetchall()
|
||
recent = self.list_transactions(limit=8)
|
||
warnings = self.list_budget_warnings()
|
||
return {
|
||
"balance": balance,
|
||
"income": summary["income"],
|
||
"expense": summary["expense"],
|
||
"net": summary["income"] - summary["expense"],
|
||
"subscriptions_count": subscriptions["count"],
|
||
"subscriptions_total": subscriptions["monthly_total"],
|
||
"upcoming": upcoming,
|
||
"recent": recent,
|
||
"warnings": warnings,
|
||
}
|
||
|
||
def list_budget_warnings(self) -> list[Row]:
|
||
rows = self.list_budgets()
|
||
return [row for row in rows if row["limit_amount"] and row["spent"] >= row["limit_amount"]]
|
||
|
||
def report_data(self) -> dict[str, object]:
|
||
month_key = current_month_key()
|
||
with get_connection() as connection:
|
||
summary = connection.execute(
|
||
"""
|
||
SELECT
|
||
COALESCE(SUM(CASE WHEN transaction_type = 'Доход'
|
||
THEN amount END), 0) AS income,
|
||
COALESCE(SUM(CASE WHEN transaction_type = 'Расход'
|
||
THEN amount END), 0) AS expense
|
||
FROM transactions
|
||
WHERE strftime('%Y-%m', created_at) = ?
|
||
""",
|
||
(month_key,),
|
||
).fetchone()
|
||
balance = connection.execute(
|
||
"SELECT COALESCE(SUM(balance), 0) AS total FROM accounts"
|
||
).fetchone()["total"]
|
||
subscription_total = connection.execute(
|
||
"""
|
||
SELECT COALESCE(SUM(
|
||
CASE WHEN period = 'Год'
|
||
THEN amount / 12.0
|
||
ELSE amount
|
||
END
|
||
), 0) AS total
|
||
FROM subscriptions
|
||
WHERE status = 'Активна'
|
||
"""
|
||
).fetchone()["total"]
|
||
categories = connection.execute(
|
||
"""
|
||
SELECT c.name, SUM(t.amount) AS total
|
||
FROM transactions AS t
|
||
JOIN categories AS c ON c.id = t.category_id
|
||
WHERE t.transaction_type = 'Расход'
|
||
AND strftime('%Y-%m', t.created_at) = ?
|
||
GROUP BY c.name
|
||
HAVING total > 0
|
||
ORDER BY total DESC
|
||
""",
|
||
(month_key,),
|
||
).fetchall()
|
||
return {
|
||
"income": summary["income"],
|
||
"expense": summary["expense"],
|
||
"net": summary["income"] - summary["expense"],
|
||
"balance": balance,
|
||
"subscription_total": subscription_total,
|
||
"categories": categories,
|
||
}
|
||
|
||
def build_report_text(self) -> str:
|
||
data = self.report_data()
|
||
categories: Iterable[Row] = data["categories"]
|
||
top_category = "Нет данных"
|
||
top_category_sum = 0.0
|
||
category_lines = []
|
||
for row in categories:
|
||
if top_category == "Нет данных":
|
||
top_category = row["name"]
|
||
top_category_sum = row["total"]
|
||
category_lines.append(f"- {row['name']}: {format_money(row['total'])}")
|
||
|
||
details = "\n".join(category_lines) or "Нет расходов за текущий месяц."
|
||
return (
|
||
"ОТЧЁТ ЗА ТЕКУЩИЙ МЕСЯЦ\n\n"
|
||
f"Доходы: {format_money(data['income'])}\n"
|
||
f"Расходы: {format_money(data['expense'])}\n"
|
||
f"Финансовый результат: {format_money(data['net'])}\n"
|
||
f"Общий баланс по всем счетам: {format_money(data['balance'])}\n"
|
||
"Активные подписки в пересчёте на месяц: "
|
||
f"{format_money(data['subscription_total'])}\n"
|
||
"Самая затратная категория: "
|
||
f"{top_category} ({format_money(top_category_sum)})\n\n"
|
||
"Расходы по категориям:\n"
|
||
f"{details}\n"
|
||
)
|
||
|
||
def export_transactions_csv(self, path: str | Path) -> None:
|
||
rows = self.list_transactions(limit=10_000)
|
||
with open(path, "w", encoding="utf-8-sig", newline="") as file:
|
||
writer = csv.writer(file, delimiter=";")
|
||
writer.writerow(
|
||
["Дата", "Название", "Тип", "Сумма", "Счёт", "Категория"]
|
||
)
|
||
for row in rows:
|
||
writer.writerow(
|
||
[
|
||
row["created_at"],
|
||
row["title"],
|
||
row["transaction_type"],
|
||
row["amount"],
|
||
row["account_name"],
|
||
row["category_name"],
|
||
]
|
||
)
|
||
|
||
@staticmethod
|
||
def is_integrity_error(error: Exception) -> bool:
|
||
return isinstance(error, IntegrityError)
|