Initial release of family finance Telegram bot

This commit is contained in:
Ochenstarik 2026-07-14 12:21:12 +07:00
commit 2ce385554f
25 changed files with 3254 additions and 0 deletions

10
.dockerignore Normal file
View file

@ -0,0 +1,10 @@
.git
.gitignore
.env
.venv
__pycache__
*.py[cod]
.pytest_cache
tests
data

11
.env.example Normal file
View file

@ -0,0 +1,11 @@
# Получите токен у @BotFather
BOT_TOKEN=123456789:replace_me
# IANA-таймзона и время автоматического дневного отчёта
BOT_TIMEZONE=Asia/Novosibirsk
DAILY_REPORT_TIME=21:00
# В Docker оставьте это значение без изменений
DATABASE_PATH=/app/data/finance.db
LOG_LEVEL=INFO

9
.gitignore vendored Normal file
View file

@ -0,0 +1,9 @@
.env
.venv/
__pycache__/
*.py[cod]
.pytest_cache/
.mypy_cache/
*.egg-info/
data/*.db
data/*.db-*

20
Dockerfile Normal file
View file

@ -0,0 +1,20 @@
FROM python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_NO_CACHE_DIR=1
WORKDIR /app
COPY pyproject.toml README.md ./
COPY app ./app
RUN pip install --upgrade pip && pip install . && \
groupadd --system financebot && \
useradd --system --gid financebot --home-dir /app financebot && \
mkdir -p /app/data && chown -R financebot:financebot /app/data
USER financebot
CMD ["python", "-m", "app"]

158
README.md Normal file
View file

@ -0,0 +1,158 @@
# Финансовый Telegram-бот
Аккуратный бот для повседневного учёта личных финансов. Расход записывается одной строкой — `467 ярче` или `3000 бензин`. Доход — через `+ 85000 зарплата` или `доход 5000 подработка`.
## Что уже работает
- автоматическое определение категорий: Продукты, Развлечения, Машина, Кредиты, Ремонт и Другое;
- доходы, расходы и точные суммы с копейками;
- изменение категории и удаление только что добавленной записи;
- итоги за текущий месяц и за всё время;
- аналитика по категориям с процентами и визуальными полосами — за месяц или всё время;
- последние 10 операций;
- ежедневный отчёт: траты и доходы за день плюс накопительный итог месяца;
- отдельные данные для каждого Telegram-пользователя;
- защищённый семейный бюджет для двух супругов с подтверждением владельца;
- мгновенные уведомления супругу и в подключённый семейный Telegram-чат;
- один общий вечерний отчёт в семейный чат;
- общий долг: доходы автоматически уменьшают его, расходы увеличивают;
- категории и подкатегории с возможностью добавлять собственные;
- SQLite в режиме WAL и защита от повторного дневного отчёта;
- готовый Docker-запуск на сервере.
## Автоматическая установка на сервер
Для Ubuntu или Debian используйте интерактивный установщик `install.sh`. Он скачивает актуальную копию проекта с публичной папки Яндекс.Диска, при необходимости устанавливает Docker, спрашивает токен и настройки отчётов, после чего запускает бота.
Скопируйте `install.sh` на сервер и выполните:
```bash
sudo bash install.sh
```
Или скачайте сам установщик из публичной папки прямо на сервере:
```bash
PUBLIC_URL='https://disk.yandex.ru/d/hhbDTO08CB030Q'
INSTALLER_URL="$(curl -fsSG 'https://cloud-api.yandex.net/v1/disk/public/resources/download' \
--data-urlencode "public_key=$PUBLIC_URL" --data-urlencode 'path=/install.sh' \
| python3 -c 'import json,sys; print(json.load(sys.stdin)["href"])')"
curl -fsSL "$INSTALLER_URL" -o install.sh
sudo bash install.sh
```
Установщик спросит:
- папку установки;
- токен, полученный у [@BotFather](https://t.me/BotFather);
- часовой пояс;
- время ежедневного отчёта.
Повторный запуск обновляет код, но сохраняет `.env` и базу данных в Docker-томе. Публичная ссылка проекта уже записана в скрипте. При необходимости её можно переопределить:
```bash
sudo FINANCE_BOT_PUBLIC_URL="https://disk.yandex.ru/d/другая-ссылка" bash install.sh
```
## Ручной запуск на сервере
Понадобятся Docker и Docker Compose.
1. Создайте бота через [@BotFather](https://t.me/BotFather) и скопируйте токен.
2. На сервере скопируйте `.env.example` в `.env`.
3. В `.env` замените `BOT_TOKEN`, при необходимости настройте таймзону и время отчёта.
4. Запустите контейнер:
```bash
docker compose up -d --build
```
Посмотреть состояние и журнал:
```bash
docker compose ps
docker compose logs -f finance-bot
```
Данные сохраняются в Docker-томе `finance-data` и не исчезают при пересборке контейнера.
## Настройки
| Переменная | Пример | Назначение |
|---|---|---|
| `BOT_TOKEN` | `123:ABC...` | токен от BotFather |
| `BOT_TIMEZONE` | `Asia/Novosibirsk` | таймзона операций и отчётов |
| `DAILY_REPORT_TIME` | `21:00` | локальное время дневного отчёта |
| `DATABASE_PATH` | `/app/data/finance.db` | путь к SQLite |
| `LOG_LEVEL` | `INFO` | уровень журналирования |
## Семейный бюджет и общий чат
1. Первый супруг открывает кнопку **«Семья»** и выбирает **«Создать общий бюджет»**.
2. Второй супруг запускает этого же бота, открывает **«Семья»**, нажимает **«Ввести код супруга»** и отправляет полученный код.
3. Код не даёт доступ автоматически: создатель бюджета получает запрос с кнопками **«Подтвердить»** и **«Отклонить»**.
4. После подтверждения итоги, аналитика и история становятся общими. Новые записи мгновенно отправляются второму супругу.
5. Для общего журнала создайте Telegram-группу, добавьте туда бота. Создатель семейного бюджета должен быть администратором этой группы и отправить в ней команду `/family_chat`.
После привязки каждая новая операция обоих супругов дублируется в группу. В заданное время туда также приходит один общий дневной отчёт. Посторонний человек не сможет подключиться только по коду: требуется ручное подтверждение владельца, семейный бюджет ограничен двумя участниками, а привязать групповой чат может только создатель бюджета, являющийся его администратором.
### Общий долг
После создания семейного бюджета откройте кнопку **«💳 Общий долг»** и укажите текущую сумму. Сделать это может только создатель бюджета, второй супруг видит результат.
Например, исходный долг составляет `530713`. После записи `+ 400000 зарплата` бот покажет долг `130713`. Следующая запись `5000 продукты` увеличит его до `135713`.
Расчёт ведётся по операциям после установки исходной суммы:
- каждый расход увеличивает долг;
- каждый доход уменьшает долг;
- удаление ошибочной операции автоматически пересчитывает сумму;
- если доходы превысят долг, бот покажет разницу как семейный резерв;
- текущий долг или резерв отображается в подтверждениях, семейном чате и дневном отчёте.
Повторная установка текущей суммы начинает новый расчёт с указанного значения, не меняя историю доходов и расходов.
### Категории и подкатегории
В семейном бюджете доступна кнопка **«🗂 Категории»**. Категории общие для обоих супругов: каждый участник может добавить новую основную категорию или подкатегорию.
При создании бюджета бот автоматически добавляет базовое дерево, например:
- Машина → Бензин, Ремонт, Запчасти, Мойка и парковка, Страховка;
- Продукты → Еда, Алкоголь, Кафе и рестораны, Бытовые товары;
- Развлечения → Кино и театр, Игры, Подписки, Хобби;
- Кредиты → Ипотека, Кредиты, Рассрочки;
- Ремонт → Материалы, Мебель, Работы, Инструменты.
Стандартные подкатегории определяются автоматически: запись `3000 бензин` попадёт в **Машина → Бензин**, а `500 пиво` — в **Продукты → Алкоголь**.
Чтобы создать свою структуру, нажмите **«🗂 Категории»**, затем **«Добавить категорию»** или **«Добавить подкатегорию»**. Новые пользовательские категории выбираются вручную при изменении записи.
После добавления расхода нажмите **«Изменить категорию»**. Можно выбрать как основную категорию, так и любую подкатегорию. Это работает и для записей, которые первоначально попали в **«Другое»**. Изменить категорию может только автор операции; супруг и семейный чат сразу получают уведомление о переносе.
## Локальная разработка
Нужен Python 3.11 или новее.
```bash
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install -e ".[dev]"
cp .env.example .env # Windows: copy .env.example .env
```
Переменные из `.env` нужно загрузить в окружение, затем запустить:
```bash
python -m app
```
Тесты:
```bash
pytest
```
## Как расширять категории
Названия, иконки и ключевые слова находятся в `app/categories.py`. Новые ключевые слова можно добавлять без изменения базы данных. Суммы хранятся целым числом копеек, поэтому арифметика не накапливает ошибки `float`.

2
app/__init__.py Normal file
View file

@ -0,0 +1,2 @@
"""Personal finance Telegram bot."""

68
app/__main__.py Normal file
View file

@ -0,0 +1,68 @@
from __future__ import annotations
import asyncio
import logging
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.types import (
BotCommand,
BotCommandScopeAllGroupChats,
BotCommandScopeAllPrivateChats,
)
from app.config import Settings
from app.database import Database
from app.handlers import create_router
from app.scheduler import daily_report_loop, stop_task
async def main() -> None:
settings = Settings.from_env()
logging.basicConfig(
level=getattr(logging, settings.log_level, logging.INFO),
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
database = Database(settings.database_path)
await database.initialize()
bot = Bot(
token=settings.bot_token,
default=DefaultBotProperties(parse_mode=ParseMode.HTML),
)
dispatcher = Dispatcher()
dispatcher.include_router(create_router(database, settings))
await bot.set_my_commands(
[
BotCommand(command="start", description="Открыть главное меню"),
BotCommand(command="month", description="Итог за месяц"),
BotCommand(command="total", description="Итог за всё время"),
BotCommand(command="analytics", description="Расходы по категориям"),
BotCommand(command="recent", description="Последние записи"),
BotCommand(command="family", description="Семейный бюджет"),
BotCommand(command="debt", description="Общий долг"),
BotCommand(command="categories", description="Категории и подкатегории"),
BotCommand(command="help", description="Как пользоваться ботом"),
],
scope=BotCommandScopeAllPrivateChats(),
)
await bot.set_my_commands(
[BotCommand(command="family_chat", description="Подключить семейный чат")],
scope=BotCommandScopeAllGroupChats(),
)
report_task = asyncio.create_task(daily_report_loop(bot, database))
try:
await dispatcher.start_polling(
bot,
allowed_updates=dispatcher.resolve_used_update_types(),
)
finally:
await stop_task(report_task)
await bot.session.close()
if __name__ == "__main__":
asyncio.run(main())

129
app/categories.py Normal file
View file

@ -0,0 +1,129 @@
from __future__ import annotations
import re
from app.domain import Category
CATEGORY_META: dict[Category, tuple[str, str]] = {
Category.PRODUCTS: ("🛒", "Продукты"),
Category.ENTERTAINMENT: ("🎉", "Развлечения"),
Category.CAR: ("🚗", "Машина"),
Category.CREDITS: ("🏦", "Кредиты"),
Category.RENOVATION: ("🛠", "Ремонт"),
Category.OTHER: ("📦", "Другое"),
Category.INCOME: ("💚", "Доход"),
}
# Сначала проверяются более специфичные слова. Список легко расширять без миграции БД.
CATEGORY_KEYWORDS: dict[Category, tuple[str, ...]] = {
Category.PRODUCTS: (
"ярче", "мария ра", "пятерочка", "пятёрочка", "магнит", "лента",
"ашан", "metro", "продукт", "еда", "молоко", "хлеб", "мясо",
"овощ", "фрукт", "кофе", "кафе", "ресторан", "доставка еды",
"алкоголь", "пиво", "вино", "супермаркет",
),
Category.ENTERTAINMENT: (
"кино", "театр", "концерт", "бар", "клуб", "игра", "steam",
"подписка", "музей", "развлеч", "боулинг", "караоке", "хобби",
),
Category.CAR: (
"бензин", "топливо", "азс", "газпромнефть", "лукойл", "g-drive",
"шиномонтаж", "автосервис", "мойка", "запчаст", "масло", "осаго",
"парковка", "штраф гибдд", "машин", "авто",
),
Category.CREDITS: (
"кредит", "ипотек", "долг", "рассроч", "процент банку", "заём", "займ",
),
Category.RENOVATION: (
"ремонт квартир", "стройматериал", "обои", "краска", "ламинат",
"сантехник", "электрик", "мебель", "инструмент", "стройка", "леруа", "ремонт",
),
}
def normalize(text: str) -> str:
text = text.casefold().replace("ё", "е")
return re.sub(r"[^a-zа-я0-9]+", " ", text).strip()
def classify(description: str) -> Category:
normalized_text = normalize(description)
normalized = f" {normalized_text} "
words = normalized_text.split()
for category, keywords in CATEGORY_KEYWORDS.items():
for keyword in keywords:
normalized_keyword = normalize(keyword)
is_phrase = " " in normalized_keyword
exact_phrase = is_phrase and f" {normalized_keyword} " in normalized
word_match = not is_phrase and any(
word == normalized_keyword
or (len(normalized_keyword) >= 4 and word.startswith(normalized_keyword))
for word in words
)
if exact_phrase or word_match:
return category
return Category.OTHER
def category_label(category: Category) -> str:
icon, name = CATEGORY_META[category]
return f"{icon} {name}"
EXPENSE_CATEGORIES = tuple(category for category in Category if category is not Category.INCOME)
# system_key, icon, visible name, keywords. These are seeded for every family budget.
DEFAULT_SUBCATEGORIES: dict[Category, tuple[tuple[str, str, str, tuple[str, ...]], ...]] = {
Category.PRODUCTS: (
("products_food", "🥦", "Еда", ("продукт", "еда", "ярче", "магнит", "пятер", "лента", "молоко", "хлеб", "мясо", "овощ", "фрукт")),
("products_alcohol", "🍷", "Алкоголь", ("алкоголь", "пиво", "вино", "водка")),
("products_cafe", "", "Кафе и рестораны", ("кафе", "ресторан", "кофе", "доставка еды")),
("products_household", "🧻", "Бытовые товары", ("бытовая химия", "хозтовар", "порошок", "салфетк")),
),
Category.ENTERTAINMENT: (
("entertainment_cinema", "🎬", "Кино и театр", ("кино", "театр", "концерт", "музей")),
("entertainment_games", "🎮", "Игры", ("игра", "steam", "playstation", "xbox")),
("entertainment_subscriptions", "📺", "Подписки", ("подписка", "кинопоиск", "ivi", "netflix")),
("entertainment_hobby", "🎨", "Хобби", ("хобби", "боулинг", "караоке")),
),
Category.CAR: (
("car_fuel", "", "Бензин", ("бензин", "топливо", "азс", "газпромнефть", "лукойл", "g drive")),
("car_repair", "🔧", "Ремонт", ("автосервис", "ремонт машины", "ремонт авто", "шиномонтаж")),
("car_parts", "⚙️", "Запчасти", ("запчаст", "масло", "аккумулятор", "шина", "резина")),
("car_service", "🅿️", "Мойка и парковка", ("мойка", "парковка")),
("car_insurance", "🛡", "Страховка", ("осаго", "каско", "страховка авто")),
),
Category.CREDITS: (
("credits_mortgage", "🏠", "Ипотека", ("ипотек",)),
("credits_bank", "💳", "Кредиты", ("кредит", "процент банку")),
("credits_installment", "📆", "Рассрочки", ("рассроч",)),
),
Category.RENOVATION: (
("renovation_materials", "🧱", "Материалы", ("стройматериал", "обои", "краска", "ламинат", "леруа")),
("renovation_furniture", "🪑", "Мебель", ("мебель", "диван", "шкаф", "стол")),
("renovation_work", "👷", "Работы", ("сантехник", "электрик", "ремонт квартир", "стройка")),
("renovation_tools", "🧰", "Инструменты", ("инструмент", "дрель", "шуруповерт")),
),
Category.OTHER: (),
}
def suggest_subcategory(category: Category, description: str) -> str | None:
normalized_text = normalize(description)
words = normalized_text.split()
padded = f" {normalized_text} "
for system_key, _icon, _name, keywords in DEFAULT_SUBCATEGORIES.get(category, ()):
for keyword in keywords:
normalized_keyword = normalize(keyword)
if " " in normalized_keyword:
if f" {normalized_keyword} " in padded:
return system_key
elif any(
word == normalized_keyword
or (len(normalized_keyword) >= 4 and word.startswith(normalized_keyword))
for word in words
):
return system_key
return None

44
app/config.py Normal file
View file

@ -0,0 +1,44 @@
from __future__ import annotations
import os
from dataclasses import dataclass
from datetime import time
from pathlib import Path
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
@dataclass(frozen=True, slots=True)
class Settings:
bot_token: str
database_path: Path
timezone: ZoneInfo
daily_report_time: time
log_level: str = "INFO"
@classmethod
def from_env(cls) -> "Settings":
token = os.getenv("BOT_TOKEN", "").strip()
if not token:
raise RuntimeError("Переменная BOT_TOKEN не задана")
timezone_name = os.getenv("BOT_TIMEZONE", "Asia/Novosibirsk").strip()
try:
timezone = ZoneInfo(timezone_name)
except ZoneInfoNotFoundError as exc:
raise RuntimeError(f"Неизвестная таймзона BOT_TIMEZONE: {timezone_name}") from exc
report_time_raw = os.getenv("DAILY_REPORT_TIME", "21:00").strip()
try:
hour, minute = (int(part) for part in report_time_raw.split(":", maxsplit=1))
report_time = time(hour=hour, minute=minute)
except (ValueError, TypeError) as exc:
raise RuntimeError("DAILY_REPORT_TIME должен иметь формат ЧЧ:ММ") from exc
return cls(
bot_token=token,
database_path=Path(os.getenv("DATABASE_PATH", "data/finance.db")),
timezone=timezone,
daily_report_time=report_time,
log_level=os.getenv("LOG_LEVEL", "INFO").upper(),
)

884
app/database.py Normal file
View file

@ -0,0 +1,884 @@
from __future__ import annotations
from datetime import datetime
from pathlib import Path
import secrets
import aiosqlite
from app.categories import CATEGORY_META, DEFAULT_SUBCATEGORIES, category_label
from app.domain import (
BudgetCategory,
Category,
FamilyInfo,
FamilyJoinRequest,
Totals,
Transaction,
TransactionKind,
)
class FamilyJoinError(ValueError):
pass
class Database:
def __init__(self, path: Path) -> None:
self.path = path
async def initialize(self) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
async with aiosqlite.connect(self.path) as db:
await db.execute("PRAGMA journal_mode=WAL")
await db.execute("PRAGMA foreign_keys=ON")
await db.executescript(
"""
CREATE TABLE IF NOT EXISTS users (
user_id INTEGER PRIMARY KEY,
chat_id INTEGER NOT NULL,
username TEXT,
full_name TEXT NOT NULL,
timezone TEXT NOT NULL,
report_time TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS households (
id INTEGER PRIMARY KEY AUTOINCREMENT,
invite_code TEXT NOT NULL UNIQUE,
created_by INTEGER NOT NULL REFERENCES users(user_id),
notification_chat_id INTEGER,
notification_chat_title TEXT,
debt_base_kopecks INTEGER,
debt_started_at TEXT,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS household_members (
household_id INTEGER NOT NULL REFERENCES households(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL UNIQUE REFERENCES users(user_id) ON DELETE CASCADE,
joined_at TEXT NOT NULL,
PRIMARY KEY (household_id, user_id)
);
CREATE TABLE IF NOT EXISTS budget_categories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
household_id INTEGER NOT NULL REFERENCES households(id) ON DELETE CASCADE,
name TEXT NOT NULL,
icon TEXT NOT NULL DEFAULT '📁',
parent_id INTEGER REFERENCES budget_categories(id) ON DELETE RESTRICT,
system_key TEXT,
created_by INTEGER REFERENCES users(user_id),
created_at TEXT NOT NULL,
UNIQUE (household_id, system_key)
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_budget_category_name
ON budget_categories(household_id, COALESCE(parent_id, 0), lower(name));
CREATE TABLE IF NOT EXISTS pending_family_joins (
household_id INTEGER NOT NULL REFERENCES households(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL UNIQUE REFERENCES users(user_id) ON DELETE CASCADE,
requested_at TEXT NOT NULL,
PRIMARY KEY (household_id, user_id)
);
CREATE TABLE IF NOT EXISTS transactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(user_id) ON DELETE CASCADE,
amount_kopecks INTEGER NOT NULL CHECK (amount_kopecks > 0),
description TEXT NOT NULL,
kind TEXT NOT NULL CHECK (kind IN ('expense', 'income')),
category TEXT NOT NULL,
household_id INTEGER REFERENCES households(id),
budget_category_id INTEGER REFERENCES budget_categories(id),
occurred_at TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_transactions_user_date
ON transactions(user_id, occurred_at);
CREATE TABLE IF NOT EXISTS daily_reports (
user_id INTEGER NOT NULL REFERENCES users(user_id) ON DELETE CASCADE,
local_date TEXT NOT NULL,
sent_at TEXT NOT NULL,
PRIMARY KEY (user_id, local_date)
);
CREATE TABLE IF NOT EXISTS family_daily_reports (
household_id INTEGER NOT NULL REFERENCES households(id) ON DELETE CASCADE,
local_date TEXT NOT NULL,
sent_at TEXT NOT NULL,
PRIMARY KEY (household_id, local_date)
);
"""
)
columns_cursor = await db.execute("PRAGMA table_info(transactions)")
columns = {row[1] for row in await columns_cursor.fetchall()}
if "household_id" not in columns:
await db.execute(
"ALTER TABLE transactions ADD COLUMN household_id INTEGER REFERENCES households(id)"
)
if "budget_category_id" not in columns:
await db.execute(
"ALTER TABLE transactions ADD COLUMN budget_category_id INTEGER "
"REFERENCES budget_categories(id)"
)
await db.execute(
"CREATE INDEX IF NOT EXISTS idx_transactions_household_date "
"ON transactions(household_id, occurred_at)"
)
household_columns_cursor = await db.execute("PRAGMA table_info(households)")
household_columns = {row[1] for row in await household_columns_cursor.fetchall()}
if "notification_chat_id" not in household_columns:
await db.execute("ALTER TABLE households ADD COLUMN notification_chat_id INTEGER")
if "notification_chat_title" not in household_columns:
await db.execute("ALTER TABLE households ADD COLUMN notification_chat_title TEXT")
if "debt_base_kopecks" not in household_columns:
await db.execute("ALTER TABLE households ADD COLUMN debt_base_kopecks INTEGER")
if "debt_started_at" not in household_columns:
await db.execute("ALTER TABLE households ADD COLUMN debt_started_at TEXT")
household_cursor = await db.execute("SELECT id FROM households")
for household_row in await household_cursor.fetchall():
await self._seed_default_categories(db, int(household_row[0]))
await db.execute(
"""
UPDATE transactions
SET budget_category_id=(
SELECT bc.id FROM budget_categories bc
WHERE bc.household_id=transactions.household_id
AND bc.system_key=transactions.category
)
WHERE household_id IS NOT NULL AND budget_category_id IS NULL
"""
)
await db.commit()
async def upsert_user(
self,
*,
user_id: int,
chat_id: int,
username: str | None,
full_name: str,
timezone: str,
report_time: str,
now: datetime,
) -> None:
timestamp = now.isoformat()
async with aiosqlite.connect(self.path) as db:
await db.execute(
"""
INSERT INTO users (
user_id, chat_id, username, full_name, timezone,
report_time, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(user_id) DO UPDATE SET
chat_id=excluded.chat_id,
username=excluded.username,
full_name=excluded.full_name,
timezone=excluded.timezone,
report_time=excluded.report_time,
updated_at=excluded.updated_at
""",
(user_id, chat_id, username, full_name, timezone, report_time, timestamp, timestamp),
)
await db.commit()
async def add_transaction(
self,
*,
user_id: int,
amount_kopecks: int,
description: str,
kind: TransactionKind,
category: Category,
budget_category_id: int | None = None,
occurred_at: datetime,
) -> Transaction:
now = datetime.now(tz=occurred_at.tzinfo)
async with aiosqlite.connect(self.path) as db:
user_cursor = await db.execute(
"""
SELECT hm.household_id, u.full_name
FROM users u
LEFT JOIN household_members hm ON hm.user_id=u.user_id
WHERE u.user_id=?
""",
(user_id,),
)
user_row = await user_cursor.fetchone()
household_id = user_row[0] if user_row else None
author_name = str(user_row[1]) if user_row else None
cursor = await db.execute(
"""
INSERT INTO transactions (
user_id, amount_kopecks, description, kind, category, household_id,
budget_category_id, occurred_at, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
user_id,
amount_kopecks,
description,
kind.value,
category.value,
household_id,
budget_category_id,
occurred_at.isoformat(),
now.isoformat(),
),
)
await db.commit()
transaction_id = int(cursor.lastrowid)
transaction = await self.get_transaction(transaction_id, user_id)
if transaction is None:
raise RuntimeError("Операция не сохранилась")
return transaction
async def get_transaction(self, transaction_id: int, user_id: int) -> Transaction | None:
async with aiosqlite.connect(self.path) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute(
"""
SELECT t.*, u.full_name AS author_name,
c.name AS budget_category_name, c.icon AS budget_category_icon,
p.name AS parent_category_name, p.icon AS parent_category_icon
FROM transactions t JOIN users u ON u.user_id=t.user_id
LEFT JOIN budget_categories c ON c.id=t.budget_category_id
LEFT JOIN budget_categories p ON p.id=c.parent_id
WHERE t.id=? AND t.user_id=?
""",
(transaction_id, user_id),
)
row = await cursor.fetchone()
return self._row_to_transaction(row) if row else None
async def update_category(
self, transaction_id: int, user_id: int, category: Category
) -> bool:
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(
"UPDATE transactions SET category=? WHERE id=? AND user_id=? AND kind='expense'",
(category.value, transaction_id, user_id),
)
await db.commit()
return cursor.rowcount > 0
async def delete_transaction(self, transaction_id: int, user_id: int) -> bool:
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(
"DELETE FROM transactions WHERE id=? AND user_id=?",
(transaction_id, user_id),
)
await db.commit()
return cursor.rowcount > 0
async def totals(
self, user_id: int, start: datetime | None = None, end: datetime | None = None
) -> Totals:
where = [self._scope_clause("t")]
params: list[object] = [user_id, user_id]
if start is not None:
where.append("occurred_at >= ?")
params.append(start.isoformat())
if end is not None:
where.append("occurred_at < ?")
params.append(end.isoformat())
query = f"""
SELECT
COALESCE(SUM(CASE WHEN kind='income' THEN amount_kopecks ELSE 0 END), 0),
COALESCE(SUM(CASE WHEN kind='expense' THEN amount_kopecks ELSE 0 END), 0)
FROM transactions t WHERE {' AND '.join(where)}
"""
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(query, params)
row = await cursor.fetchone()
return Totals(income=int(row[0]), expense=int(row[1]))
async def category_totals(
self, user_id: int, start: datetime | None = None, end: datetime | None = None
) -> dict[str, int]:
where = [self._scope_clause("t"), "t.kind='expense'"]
params: list[object] = [user_id, user_id]
if start is not None:
where.append("occurred_at>=?")
params.append(start.isoformat())
if end is not None:
where.append("occurred_at<?")
params.append(end.isoformat())
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(
f"""
SELECT t.category, c.name, c.icon, p.name, p.icon,
SUM(t.amount_kopecks) AS total
FROM transactions t
LEFT JOIN budget_categories c ON c.id=t.budget_category_id
LEFT JOIN budget_categories p ON p.id=c.parent_id
WHERE {' AND '.join(where)}
GROUP BY t.category, c.id, p.id ORDER BY total DESC
""",
params,
)
rows = await cursor.fetchall()
result: dict[str, int] = {}
for legacy_key, name, icon, parent_name, parent_icon, total in rows:
if name and parent_name:
label = f"{parent_icon or '📁'} {parent_name}{icon or '📁'} {name}"
elif name:
label = f"{icon or '📁'} {name}"
else:
label = category_label(Category(legacy_key))
result[label] = result.get(label, 0) + int(total)
return result
async def recent_transactions(self, user_id: int, limit: int = 10) -> list[Transaction]:
async with aiosqlite.connect(self.path) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute(
f"""
SELECT t.*, u.full_name AS author_name,
c.name AS budget_category_name, c.icon AS budget_category_icon,
p.name AS parent_category_name, p.icon AS parent_category_icon
FROM transactions t JOIN users u ON u.user_id=t.user_id
LEFT JOIN budget_categories c ON c.id=t.budget_category_id
LEFT JOIN budget_categories p ON p.id=c.parent_id
WHERE {self._scope_clause("t")}
ORDER BY t.occurred_at DESC, t.id DESC LIMIT ?
""",
(user_id, user_id, limit),
)
rows = await cursor.fetchall()
return [self._row_to_transaction(row) for row in rows]
async def create_family(self, user_id: int, now: datetime) -> FamilyInfo:
async with aiosqlite.connect(self.path) as db:
await db.execute("PRAGMA foreign_keys=ON")
await db.execute("BEGIN IMMEDIATE")
existing = await self._family_info_in_connection(db, user_id)
if existing is not None:
await db.rollback()
return existing
for _ in range(10):
code = "FIN-" + "".join(
secrets.choice("ABCDEFGHJKLMNPQRSTUVWXYZ23456789") for _ in range(6)
)
try:
cursor = await db.execute(
"INSERT INTO households(invite_code, created_by, created_at) VALUES (?, ?, ?)",
(code, user_id, now.isoformat()),
)
break
except aiosqlite.IntegrityError:
continue
else:
await db.rollback()
raise RuntimeError("Не удалось создать уникальный код семьи")
household_id = int(cursor.lastrowid)
await self._seed_default_categories(db, household_id)
await db.execute(
"INSERT INTO household_members(household_id, user_id, joined_at) VALUES (?, ?, ?)",
(household_id, user_id, now.isoformat()),
)
await db.execute(
"UPDATE transactions SET household_id=? WHERE user_id=? AND household_id IS NULL",
(household_id, user_id),
)
await db.execute(
"""
UPDATE transactions
SET budget_category_id=(
SELECT id FROM budget_categories
WHERE household_id=? AND system_key=transactions.category
)
WHERE user_id=? AND household_id=? AND budget_category_id IS NULL
""",
(household_id, user_id, household_id),
)
await db.commit()
info = await self.family_info(user_id)
if info is None:
raise RuntimeError("Семья не сохранилась")
return info
async def request_family_join(
self, user_id: int, invite_code: str, now: datetime
) -> FamilyJoinRequest:
code = invite_code.strip().upper()
async with aiosqlite.connect(self.path) as db:
await db.execute("PRAGMA foreign_keys=ON")
await db.execute("BEGIN IMMEDIATE")
if await self._family_info_in_connection(db, user_id) is not None:
await db.rollback()
raise FamilyJoinError("Вы уже состоите в семейном бюджете")
cursor = await db.execute(
"""
SELECT h.id, owner.chat_id, candidate.full_name,
(SELECT COUNT(*) FROM household_members WHERE household_id=h.id)
FROM households h
JOIN users owner ON owner.user_id=h.created_by
JOIN users candidate ON candidate.user_id=?
WHERE h.invite_code=?
""",
(user_id, code),
)
row = await cursor.fetchone()
if row is None:
await db.rollback()
raise FamilyJoinError("Код приглашения не найден")
if int(row[3]) >= 2:
await db.rollback()
raise FamilyJoinError("В этом семейном бюджете уже два участника")
await db.execute(
"""
INSERT INTO pending_family_joins(household_id, user_id, requested_at)
VALUES (?, ?, ?)
ON CONFLICT(user_id) DO UPDATE SET
household_id=excluded.household_id,
requested_at=excluded.requested_at
""",
(row[0], user_id, now.isoformat()),
)
await db.commit()
return FamilyJoinRequest(
candidate_user_id=user_id,
candidate_name=str(row[2]),
owner_chat_id=int(row[1]),
)
async def resolve_family_join(
self,
*,
owner_user_id: int,
candidate_user_id: int,
approve: bool,
now: datetime,
) -> tuple[bool, int, FamilyInfo | None]:
async with aiosqlite.connect(self.path) as db:
await db.execute("PRAGMA foreign_keys=ON")
await db.execute("BEGIN IMMEDIATE")
cursor = await db.execute(
"""
SELECT p.household_id, u.chat_id
FROM pending_family_joins p
JOIN households h ON h.id=p.household_id
JOIN users u ON u.user_id=p.user_id
WHERE h.created_by=? AND p.user_id=?
""",
(owner_user_id, candidate_user_id),
)
row = await cursor.fetchone()
if row is None:
await db.rollback()
raise FamilyJoinError("Запрос уже обработан или не найден")
household_id, candidate_chat_id = int(row[0]), int(row[1])
await db.execute(
"DELETE FROM pending_family_joins WHERE household_id=? AND user_id=?",
(household_id, candidate_user_id),
)
if not approve:
await db.commit()
return False, candidate_chat_id, None
count_cursor = await db.execute(
"SELECT COUNT(*) FROM household_members WHERE household_id=?",
(household_id,),
)
if int((await count_cursor.fetchone())[0]) >= 2:
await db.rollback()
raise FamilyJoinError("В семейном бюджете уже два участника")
await db.execute(
"INSERT INTO household_members(household_id, user_id, joined_at) VALUES (?, ?, ?)",
(household_id, candidate_user_id, now.isoformat()),
)
await db.execute(
"UPDATE transactions SET household_id=? WHERE user_id=? AND household_id IS NULL",
(household_id, candidate_user_id),
)
await db.execute(
"""
UPDATE transactions
SET budget_category_id=(
SELECT id FROM budget_categories
WHERE household_id=? AND system_key=transactions.category
)
WHERE user_id=? AND household_id=? AND budget_category_id IS NULL
""",
(household_id, candidate_user_id, household_id),
)
await db.commit()
return True, candidate_chat_id, await self.family_info(candidate_user_id)
async def bind_family_chat(
self, owner_user_id: int, chat_id: int, chat_title: str
) -> FamilyInfo:
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(
"""
UPDATE households
SET notification_chat_id=?, notification_chat_title=?
WHERE created_by=?
""",
(chat_id, chat_title, owner_user_id),
)
await db.commit()
if cursor.rowcount == 0:
raise FamilyJoinError("Привязать чат может только создатель семейного бюджета")
info = await self.family_info(owner_user_id)
if info is None:
raise FamilyJoinError("Семейный бюджет не найден")
return info
async def set_family_debt(
self, owner_user_id: int, amount_kopecks: int, now: datetime
) -> int:
if amount_kopecks < 0:
raise ValueError("Долг не может быть отрицательным")
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(
"""
UPDATE households
SET debt_base_kopecks=?, debt_started_at=?
WHERE created_by=?
""",
(amount_kopecks, now.isoformat(), owner_user_id),
)
await db.commit()
if cursor.rowcount == 0:
raise FamilyJoinError("Изменить общий долг может только создатель бюджета")
return amount_kopecks
async def family_debt(self, user_id: int) -> int | None:
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(
"""
SELECT h.debt_base_kopecks, h.debt_started_at,
COALESCE(SUM(
CASE
WHEN t.kind='expense' THEN t.amount_kopecks
WHEN t.kind='income' THEN -t.amount_kopecks
ELSE 0
END
), 0)
FROM households h
JOIN household_members hm ON hm.household_id=h.id
LEFT JOIN transactions t
ON t.household_id=h.id AND t.occurred_at>h.debt_started_at
WHERE hm.user_id=?
GROUP BY h.id
""",
(user_id,),
)
row = await cursor.fetchone()
if row is None or row[0] is None or row[1] is None:
return None
return int(row[0]) + int(row[2])
async def budget_categories(self, user_id: int) -> list[BudgetCategory]:
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(
"""
SELECT c.id, c.name, c.icon, c.parent_id, p.name, p.icon, c.system_key
FROM household_members hm
JOIN budget_categories c ON c.household_id=hm.household_id
LEFT JOIN budget_categories p ON p.id=c.parent_id
WHERE hm.user_id=? AND (c.system_key IS NULL OR c.system_key<>'income')
ORDER BY COALESCE(p.id, c.id), c.parent_id IS NOT NULL, c.name
""",
(user_id,),
)
rows = await cursor.fetchall()
return [
BudgetCategory(
id=int(row[0]),
name=str(row[1]),
icon=str(row[2]),
parent_id=int(row[3]) if row[3] is not None else None,
parent_name=str(row[4]) if row[4] is not None else None,
parent_icon=str(row[5]) if row[5] is not None else None,
system_key=str(row[6]) if row[6] is not None else None,
)
for row in rows
]
async def resolve_budget_category(
self, user_id: int, main_category: Category, subcategory_key: str | None
) -> int | None:
key = subcategory_key or main_category.value
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(
"""
SELECT c.id
FROM household_members hm
JOIN budget_categories c ON c.household_id=hm.household_id
WHERE hm.user_id=? AND c.system_key=?
""",
(user_id, key),
)
row = await cursor.fetchone()
return int(row[0]) if row else None
async def create_budget_category(
self,
*,
user_id: int,
name: str,
parent_id: int | None,
now: datetime,
) -> BudgetCategory:
clean_name = " ".join(name.split()).strip()
if not 2 <= len(clean_name) <= 40:
raise ValueError("Название должно содержать от 2 до 40 символов")
async with aiosqlite.connect(self.path) as db:
await db.execute("PRAGMA foreign_keys=ON")
household_cursor = await db.execute(
"SELECT household_id FROM household_members WHERE user_id=?", (user_id,)
)
household_row = await household_cursor.fetchone()
if household_row is None:
raise FamilyJoinError("Категории настраиваются внутри семейного бюджета")
household_id = int(household_row[0])
count_cursor = await db.execute(
"SELECT COUNT(*) FROM budget_categories WHERE household_id=?", (household_id,)
)
if int((await count_cursor.fetchone())[0]) >= 60:
raise ValueError("Достигнут лимит в 60 категорий и подкатегорий")
if parent_id is not None:
parent_cursor = await db.execute(
"""
SELECT 1 FROM budget_categories
WHERE id=? AND household_id=? AND parent_id IS NULL
AND (system_key IS NULL OR system_key<>'income')
""",
(parent_id, household_id),
)
if await parent_cursor.fetchone() is None:
raise ValueError("Основная категория не найдена")
try:
cursor = await db.execute(
"""
INSERT INTO budget_categories(
household_id, name, icon, parent_id, created_by, created_at
) VALUES (?, ?, '📁', ?, ?, ?)
""",
(household_id, clean_name, parent_id, user_id, now.isoformat()),
)
await db.commit()
except aiosqlite.IntegrityError as exc:
raise ValueError("Такая категория уже существует") from exc
category_id = int(cursor.lastrowid)
return next(
category for category in await self.budget_categories(user_id)
if category.id == category_id
)
async def update_transaction_budget_category(
self, transaction_id: int, user_id: int, category_id: int
) -> bool:
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(
"""
UPDATE transactions
SET budget_category_id=?
WHERE id=? AND user_id=? AND kind='expense'
AND household_id=(
SELECT household_id FROM budget_categories WHERE id=?
)
""",
(category_id, transaction_id, user_id, category_id),
)
await db.commit()
return cursor.rowcount > 0
async def family_info(self, user_id: int) -> FamilyInfo | None:
async with aiosqlite.connect(self.path) as db:
return await self._family_info_in_connection(db, user_id)
async def family_recipients(self, user_id: int) -> list[aiosqlite.Row]:
async with aiosqlite.connect(self.path) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute(
"""
SELECT u.user_id, u.chat_id, u.full_name
FROM household_members mine
JOIN household_members other ON other.household_id=mine.household_id
JOIN users u ON u.user_id=other.user_id
WHERE mine.user_id=? AND other.user_id<>?
""",
(user_id, user_id),
)
return await cursor.fetchall()
async def family_notification_chats(self, user_id: int) -> list[int]:
recipients = [int(row["chat_id"]) for row in await self.family_recipients(user_id)]
info = await self.family_info(user_id)
if info and info.notification_chat_id is not None:
recipients.append(info.notification_chat_id)
return list(dict.fromkeys(recipients))
async def _family_info_in_connection(
self, db: aiosqlite.Connection, user_id: int
) -> FamilyInfo | None:
cursor = await db.execute(
"""
SELECT h.id, h.invite_code, h.created_by,
h.notification_chat_id, h.notification_chat_title
FROM households h
JOIN household_members hm ON hm.household_id=h.id
WHERE hm.user_id=?
""",
(user_id,),
)
row = await cursor.fetchone()
if row is None:
return None
members_cursor = await db.execute(
"""
SELECT u.full_name
FROM household_members hm JOIN users u ON u.user_id=hm.user_id
WHERE hm.household_id=? ORDER BY hm.joined_at
""",
(row[0],),
)
members = tuple(str(member[0]) for member in await members_cursor.fetchall())
return FamilyInfo(
household_id=int(row[0]),
invite_code=str(row[1]),
members=members,
owner_user_id=int(row[2]),
notification_chat_id=int(row[3]) if row[3] is not None else None,
notification_chat_title=str(row[4]) if row[4] is not None else None,
)
async def family_report_target(self, user_id: int) -> tuple[int, int] | None:
info = await self.family_info(user_id)
if info is None or info.notification_chat_id is None:
return None
return info.household_id, info.notification_chat_id
async def family_report_was_sent(self, household_id: int, local_date: str) -> bool:
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(
"SELECT 1 FROM family_daily_reports WHERE household_id=? AND local_date=?",
(household_id, local_date),
)
return await cursor.fetchone() is not None
async def mark_family_report_sent(
self, household_id: int, local_date: str, sent_at: datetime
) -> None:
async with aiosqlite.connect(self.path) as db:
await db.execute(
"""
INSERT OR IGNORE INTO family_daily_reports(household_id, local_date, sent_at)
VALUES (?, ?, ?)
""",
(household_id, local_date, sent_at.isoformat()),
)
await db.commit()
async def users_for_reports(self) -> list[aiosqlite.Row]:
async with aiosqlite.connect(self.path) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute(
"SELECT user_id, chat_id, timezone, report_time FROM users"
)
return await cursor.fetchall()
async def report_was_sent(self, user_id: int, local_date: str) -> bool:
async with aiosqlite.connect(self.path) as db:
cursor = await db.execute(
"SELECT 1 FROM daily_reports WHERE user_id=? AND local_date=?",
(user_id, local_date),
)
return await cursor.fetchone() is not None
async def mark_report_sent(self, user_id: int, local_date: str, sent_at: datetime) -> None:
async with aiosqlite.connect(self.path) as db:
await db.execute(
"INSERT OR IGNORE INTO daily_reports(user_id, local_date, sent_at) VALUES (?, ?, ?)",
(user_id, local_date, sent_at.isoformat()),
)
await db.commit()
@staticmethod
def _row_to_transaction(row: aiosqlite.Row) -> Transaction:
return Transaction(
id=int(row["id"]),
user_id=int(row["user_id"]),
amount_kopecks=int(row["amount_kopecks"]),
description=str(row["description"]),
kind=TransactionKind(row["kind"]),
category=Category(row["category"]),
occurred_at=datetime.fromisoformat(row["occurred_at"]),
author_name=str(row["author_name"]) if "author_name" in row.keys() else None,
budget_category_id=(
int(row["budget_category_id"])
if "budget_category_id" in row.keys() and row["budget_category_id"] is not None
else None
),
budget_category_name=(
str(row["budget_category_name"])
if "budget_category_name" in row.keys() and row["budget_category_name"] is not None
else None
),
budget_category_icon=(
str(row["budget_category_icon"])
if "budget_category_icon" in row.keys() and row["budget_category_icon"] is not None
else None
),
parent_category_name=(
str(row["parent_category_name"])
if "parent_category_name" in row.keys() and row["parent_category_name"] is not None
else None
),
parent_category_icon=(
str(row["parent_category_icon"])
if "parent_category_icon" in row.keys() and row["parent_category_icon"] is not None
else None
),
)
@staticmethod
async def _seed_default_categories(
db: aiosqlite.Connection, household_id: int
) -> None:
created_at = datetime.now().astimezone().isoformat()
for category in Category:
icon, name = CATEGORY_META[category]
await db.execute(
"""
INSERT OR IGNORE INTO budget_categories(
household_id, name, icon, parent_id, system_key, created_at
) VALUES (?, ?, ?, NULL, ?, ?)
""",
(household_id, name, icon, category.value, created_at),
)
parent_cursor = await db.execute(
"SELECT id FROM budget_categories WHERE household_id=? AND system_key=?",
(household_id, category.value),
)
parent_id = int((await parent_cursor.fetchone())[0])
for system_key, sub_icon, sub_name, _keywords in DEFAULT_SUBCATEGORIES.get(
category, ()
):
await db.execute(
"""
INSERT OR IGNORE INTO budget_categories(
household_id, name, icon, parent_id, system_key, created_at
) VALUES (?, ?, ?, ?, ?, ?)
""",
(household_id, sub_name, sub_icon, parent_id, system_key, created_at),
)
@staticmethod
def _scope_clause(alias: str) -> str:
return f"""(
({alias}.household_id IS NOT NULL AND {alias}.household_id=(
SELECT household_id FROM household_members WHERE user_id=?
))
OR ({alias}.household_id IS NULL AND {alias}.user_id=?)
)"""

88
app/domain.py Normal file
View file

@ -0,0 +1,88 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
class TransactionKind(str, Enum):
EXPENSE = "expense"
INCOME = "income"
class Category(str, Enum):
PRODUCTS = "products"
ENTERTAINMENT = "entertainment"
CAR = "car"
CREDITS = "credits"
RENOVATION = "renovation"
OTHER = "other"
INCOME = "income"
@dataclass(frozen=True, slots=True)
class ParsedTransaction:
amount_kopecks: int
description: str
kind: TransactionKind
@dataclass(frozen=True, slots=True)
class Transaction:
id: int
user_id: int
amount_kopecks: int
description: str
kind: TransactionKind
category: Category
occurred_at: datetime
author_name: str | None = None
budget_category_id: int | None = None
budget_category_name: str | None = None
budget_category_icon: str | None = None
parent_category_name: str | None = None
parent_category_icon: str | None = None
@dataclass(frozen=True, slots=True)
class BudgetCategory:
id: int
name: str
icon: str
parent_id: int | None
parent_name: str | None = None
parent_icon: str | None = None
system_key: str | None = None
@property
def display_name(self) -> str:
if self.parent_name:
return f"{self.parent_icon or '📁'} {self.parent_name}{self.icon} {self.name}"
return f"{self.icon} {self.name}"
@dataclass(frozen=True, slots=True)
class FamilyInfo:
household_id: int
invite_code: str
members: tuple[str, ...]
owner_user_id: int
notification_chat_id: int | None = None
notification_chat_title: str | None = None
@dataclass(frozen=True, slots=True)
class FamilyJoinRequest:
candidate_user_id: int
candidate_name: str
owner_chat_id: int
@dataclass(frozen=True, slots=True)
class Totals:
income: int = 0
expense: int = 0
@property
def balance(self) -> int:
return self.income - self.expense

592
app/handlers.py Normal file
View file

@ -0,0 +1,592 @@
from __future__ import annotations
from datetime import datetime
from html import escape
from aiogram import Bot, F, Router
from aiogram.exceptions import TelegramForbiddenError
from aiogram.enums import ChatMemberStatus, ChatType
from aiogram.filters import Command, CommandStart
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import CallbackQuery, Message
from app.categories import Category, category_label, classify, suggest_subcategory
from app.config import Settings
from app.database import Database, FamilyJoinError
from app.domain import TransactionKind
from app.keyboards import (
ALL_TOTAL,
ANALYTICS,
CATEGORIES,
DEBT,
FAMILY,
HELP,
MONTH_TOTAL,
RECENT,
analytics_period_keyboard,
category_management_keyboard,
category_tree_keyboard,
category_keyboard,
debt_keyboard,
family_setup_keyboard,
family_approval_keyboard,
main_keyboard,
parent_category_keyboard,
transaction_actions,
)
from app.messages import (
HELP_TEXT,
MONTHS,
WELCOME,
analytics_text,
debt_short_text,
debt_status_text,
family_status_text,
family_transaction_notice,
recent_text,
summary_text,
transaction_confirmation,
transaction_category_text,
)
from app.parser import ParseError, parse_money_amount, parse_transaction
from app.periods import month_bounds
class FamilySetup(StatesGroup):
waiting_for_code = State()
class DebtSetup(StatesGroup):
waiting_for_amount = State()
class CategorySetup(StatesGroup):
waiting_for_name = State()
def create_router(database: Database, settings: Settings) -> Router:
root_router = Router(name=__name__)
group_router = Router(name=f"{__name__}.groups")
router = Router(name=f"{__name__}.private")
router.message.filter(F.chat.type == ChatType.PRIVATE)
@group_router.message(
Command("family_chat"),
F.chat.type.in_({ChatType.GROUP, ChatType.SUPERGROUP}),
)
async def bind_family_chat(message: Message, bot: Bot) -> None:
if message.from_user is None:
return
member = await bot.get_chat_member(message.chat.id, message.from_user.id)
if member.status not in {ChatMemberStatus.CREATOR, ChatMemberStatus.ADMINISTRATOR}:
await message.reply("Привязать семейный чат может только его администратор.")
return
try:
info = await database.bind_family_chat(
message.from_user.id,
message.chat.id,
message.chat.title or "Семейный чат",
)
except FamilyJoinError as exc:
await message.reply(f"Не удалось привязать чат: {escape(str(exc))}")
return
await message.reply(
"✅ <b>Чат подключён к семейному бюджету.</b>\n\n"
"Сюда будут приходить новые записи обоих супругов и один общий вечерний отчёт."
)
async def register(message: Message) -> None:
if message.from_user is None:
return
await database.upsert_user(
user_id=message.from_user.id,
chat_id=message.chat.id,
username=message.from_user.username,
full_name=message.from_user.full_name,
timezone=settings.timezone.key,
report_time=settings.daily_report_time.strftime("%H:%M"),
now=datetime.now(settings.timezone),
)
@router.message(CommandStart())
async def start(message: Message) -> None:
await register(message)
await message.answer(WELCOME, reply_markup=main_keyboard())
@router.message(Command("help"))
@router.message(F.text == HELP)
async def help_message(message: Message) -> None:
await register(message)
await message.answer(HELP_TEXT, reply_markup=main_keyboard())
@router.message(Command("month"))
@router.message(F.text == MONTH_TOTAL)
async def month_total(message: Message) -> None:
await register(message)
now = datetime.now(settings.timezone)
start_at, end_at = month_bounds(now)
totals = await database.totals(message.from_user.id, start_at, end_at)
await message.answer(summary_text(f"Итог за {MONTHS[now.month]}", totals))
@router.message(Command("total"))
@router.message(F.text == ALL_TOTAL)
async def all_total(message: Message) -> None:
await register(message)
totals = await database.totals(message.from_user.id)
await message.answer(summary_text("Итог за всё время", totals))
@router.message(Command("analytics"))
@router.message(F.text == ANALYTICS)
async def analytics(message: Message) -> None:
await register(message)
now = datetime.now(settings.timezone)
start_at, end_at = month_bounds(now)
categories = await database.category_totals(message.from_user.id, start_at, end_at)
await message.answer(
analytics_text(f"за {MONTHS[now.month]}", categories, sum(categories.values())),
reply_markup=analytics_period_keyboard("month"),
)
@router.callback_query(F.data.in_({"analytics:month", "analytics:all"}))
async def switch_analytics_period(callback: CallbackQuery) -> None:
period = (callback.data or "").split(":", maxsplit=1)[1]
now = datetime.now(settings.timezone)
if period == "month":
start_at, end_at = month_bounds(now)
categories = await database.category_totals(
callback.from_user.id, start_at, end_at
)
period_name = f"за {MONTHS[now.month]}"
else:
categories = await database.category_totals(callback.from_user.id)
period_name = "за всё время"
if callback.message:
await callback.message.edit_text(
analytics_text(period_name, categories, sum(categories.values())),
reply_markup=analytics_period_keyboard(period),
)
await callback.answer()
@router.message(Command("recent"))
@router.message(F.text == RECENT)
async def recent(message: Message) -> None:
await register(message)
transactions = await database.recent_transactions(message.from_user.id)
await message.answer(recent_text(transactions))
@router.message(Command("family"))
@router.message(F.text == FAMILY)
async def family_menu(message: Message) -> None:
await register(message)
info = await database.family_info(message.from_user.id)
await message.answer(
family_status_text(info),
reply_markup=family_setup_keyboard() if info is None else None,
)
@router.message(Command("debt"))
@router.message(F.text == DEBT)
async def debt_menu(message: Message) -> None:
await register(message)
info = await database.family_info(message.from_user.id)
if info is None:
await message.answer(
"Сначала создайте семейный бюджет через кнопку «👨‍👩‍👧 Семья»."
)
return
debt = await database.family_debt(message.from_user.id)
can_edit = info.owner_user_id == message.from_user.id
await message.answer(
debt_status_text(debt, can_edit),
reply_markup=debt_keyboard(can_edit),
)
@router.message(Command("categories"))
@router.message(F.text == CATEGORIES)
async def categories_menu(message: Message) -> None:
await register(message)
categories = await database.budget_categories(message.from_user.id)
if not categories:
await message.answer(
"Настраиваемые категории доступны после подключения семейного бюджета."
)
return
lines = ["🗂 <b>Категории семейного бюджета</b>", ""]
for category in categories:
prefix = "" if category.parent_id is not None else ""
lines.append(f"{prefix}{escape(category.icon)} {escape(category.name)}")
await message.answer(
"\n".join(lines), reply_markup=category_management_keyboard()
)
@router.callback_query(F.data == "catmanage:add-main")
async def ask_main_category_name(callback: CallbackQuery, state: FSMContext) -> None:
await state.set_state(CategorySetup.waiting_for_name)
await state.set_data({"parent_id": None})
if callback.message:
await callback.message.answer(
"Введите название новой категории, например <code>Здоровье</code>.\n\n"
"Для отмены отправьте /cancel"
)
await callback.answer()
@router.callback_query(F.data == "catmanage:add-sub")
async def choose_parent_category(callback: CallbackQuery) -> None:
categories = await database.budget_categories(callback.from_user.id)
if callback.message:
await callback.message.answer(
"К какой основной категории добавить подкатегорию?",
reply_markup=parent_category_keyboard(categories),
)
await callback.answer()
@router.callback_query(F.data.startswith("catmanage:parent:"))
async def ask_subcategory_name(callback: CallbackQuery, state: FSMContext) -> None:
try:
parent_id = int((callback.data or "").rsplit(":", maxsplit=1)[1])
except (ValueError, IndexError):
await callback.answer("Категория не найдена", show_alert=True)
return
await state.set_state(CategorySetup.waiting_for_name)
await state.set_data({"parent_id": parent_id})
if callback.message:
await callback.message.answer(
"Введите название подкатегории, например <code>Врачи</code>.\n\n"
"Для отмены отправьте /cancel"
)
await callback.answer()
@router.callback_query(F.data == "catmanage:cancel")
async def cancel_category_callback(callback: CallbackQuery, state: FSMContext) -> None:
await state.clear()
if callback.message:
await callback.message.edit_reply_markup(reply_markup=None)
await callback.answer("Отменено")
@router.message(CategorySetup.waiting_for_name, Command("cancel"))
async def cancel_new_category(message: Message, state: FSMContext) -> None:
await state.clear()
await message.answer("Добавление категории отменено.", reply_markup=main_keyboard())
@router.message(CategorySetup.waiting_for_name, F.text)
async def save_new_category(message: Message, state: FSMContext) -> None:
data = await state.get_data()
try:
category = await database.create_budget_category(
user_id=message.from_user.id,
name=message.text or "",
parent_id=data.get("parent_id"),
now=datetime.now(settings.timezone),
)
except (ValueError, FamilyJoinError) as exc:
await message.answer(f"Не получилось добавить категорию: {escape(str(exc))}")
return
await state.clear()
await message.answer(
f"✅ Добавлено: <b>{escape(category.display_name)}</b>",
reply_markup=main_keyboard(),
)
@router.callback_query(F.data == "debt:set")
async def ask_debt_amount(callback: CallbackQuery, state: FSMContext) -> None:
info = await database.family_info(callback.from_user.id)
if info is None or info.owner_user_id != callback.from_user.id:
await callback.answer("Изменить долг может только создатель бюджета", show_alert=True)
return
await state.set_state(DebtSetup.waiting_for_amount)
if callback.message:
await callback.message.answer(
"Введите текущую сумму общего долга.\n"
"Например: <code>530713</code> или <code>530 713 ₽</code>\n\n"
"Для отмены отправьте /cancel"
)
await callback.answer()
@router.message(DebtSetup.waiting_for_amount, Command("cancel"))
async def cancel_debt(message: Message, state: FSMContext) -> None:
await state.clear()
await message.answer("Изменение долга отменено.", reply_markup=main_keyboard())
@router.message(DebtSetup.waiting_for_amount, F.text)
async def save_debt(message: Message, state: FSMContext, bot: Bot) -> None:
try:
amount = parse_money_amount(message.text or "")
await database.set_family_debt(
message.from_user.id, amount, datetime.now(settings.timezone)
)
except (ParseError, FamilyJoinError) as exc:
await message.answer(f"Не получилось сохранить сумму: {escape(str(exc))}")
return
await state.clear()
await message.answer(
debt_status_text(amount, True),
reply_markup=debt_keyboard(True),
)
for chat_id in await database.family_notification_chats(message.from_user.id):
try:
await bot.send_message(
chat_id,
f"💳 <b>{escape(message.from_user.full_name)} обновил(а) общий долг.</b>\n\n"
f"{debt_short_text(amount)}",
)
except TelegramForbiddenError:
pass
@router.callback_query(F.data == "family:create")
async def create_family(callback: CallbackQuery) -> None:
info = await database.create_family(
callback.from_user.id, datetime.now(settings.timezone)
)
if callback.message:
await callback.message.answer(family_status_text(info))
await callback.message.edit_reply_markup(reply_markup=None)
await callback.answer("Общий бюджет создан")
@router.callback_query(F.data == "family:join")
async def ask_family_code(callback: CallbackQuery, state: FSMContext) -> None:
await state.set_state(FamilySetup.waiting_for_code)
if callback.message:
await callback.message.answer(
"Введите код приглашения, который показан у супруга.\n"
"Например: <code>FIN-ABC123</code>\n\n"
"Для отмены отправьте /cancel"
)
await callback.answer()
@router.message(FamilySetup.waiting_for_code, Command("cancel"))
async def cancel_family_join(message: Message, state: FSMContext) -> None:
await state.clear()
await message.answer("Подключение отменено.", reply_markup=main_keyboard())
@router.message(FamilySetup.waiting_for_code, F.text)
async def join_family(message: Message, state: FSMContext, bot: Bot) -> None:
await register(message)
try:
request = await database.request_family_join(
message.from_user.id,
message.text or "",
datetime.now(settings.timezone),
)
except FamilyJoinError as exc:
await message.answer(
f"Не получилось подключиться: {exc}\n\n"
"Проверьте код или отправьте /cancel."
)
return
await state.clear()
await message.answer(
"🔐 <b>Запрос отправлен владельцу бюджета.</b>\n\n"
"Доступ появится только после его подтверждения.",
reply_markup=main_keyboard(),
)
try:
await bot.send_message(
request.owner_chat_id,
f"🔐 <b>{escape(request.candidate_name)} просит доступ к семейному бюджету.</b>\n\n"
"Подтвердите только знакомого вам человека.",
reply_markup=family_approval_keyboard(request.candidate_user_id),
)
except TelegramForbiddenError:
await message.answer("Владелец бюджета заблокировал бота, запрос нельзя подтвердить.")
@router.callback_query(F.data.startswith("family:approve:"))
@router.callback_query(F.data.startswith("family:reject:"))
async def resolve_family_request(callback: CallbackQuery, bot: Bot) -> None:
parts = (callback.data or "").split(":")
try:
approve = parts[1] == "approve"
candidate_user_id = int(parts[2])
approved, candidate_chat_id, info = await database.resolve_family_join(
owner_user_id=callback.from_user.id,
candidate_user_id=candidate_user_id,
approve=approve,
now=datetime.now(settings.timezone),
)
except (IndexError, ValueError, FamilyJoinError) as exc:
await callback.answer(str(exc), show_alert=True)
return
if callback.message:
await callback.message.edit_reply_markup(reply_markup=None)
await callback.message.answer(
"✅ Доступ подтверждён." if approved else "🚫 Запрос отклонён."
)
if approved and info is not None:
await bot.send_message(
candidate_chat_id,
family_status_text(info),
reply_markup=main_keyboard(),
)
else:
await bot.send_message(
candidate_chat_id,
"🚫 Владелец семейного бюджета отклонил запрос на подключение.",
)
await callback.answer()
@router.callback_query(F.data.startswith("choose:"))
async def choose_category(callback: CallbackQuery) -> None:
transaction_id = _callback_id(callback.data)
transaction = await database.get_transaction(transaction_id, callback.from_user.id)
if transaction is None or transaction.kind is TransactionKind.INCOME:
await callback.answer("Запись не найдена", show_alert=True)
return
categories = await database.budget_categories(callback.from_user.id)
if callback.message:
await callback.message.edit_reply_markup(
reply_markup=(
category_tree_keyboard(transaction_id, categories)
if categories
else category_keyboard(transaction_id)
)
)
await callback.answer("Выберите категорию")
@router.callback_query(F.data.startswith("bcat:"))
async def set_budget_category(callback: CallbackQuery, bot: Bot) -> None:
try:
_, transaction_id_raw, category_id_raw = (callback.data or "").split(":")
transaction_id = int(transaction_id_raw)
category_id = int(category_id_raw)
except (ValueError, TypeError):
await callback.answer("Некорректная категория", show_alert=True)
return
changed = await database.update_transaction_budget_category(
transaction_id, callback.from_user.id, category_id
)
transaction = await database.get_transaction(transaction_id, callback.from_user.id)
if not changed or transaction is None:
await callback.answer("Запись или категория не найдена", show_alert=True)
return
debt = await database.family_debt(callback.from_user.id)
if callback.message:
await callback.message.edit_text(
transaction_confirmation(transaction, debt),
reply_markup=transaction_actions(transaction.id, is_expense=True),
)
for chat_id in await database.family_notification_chats(callback.from_user.id):
try:
await bot.send_message(
chat_id,
f"🗂 <b>{escape(callback.from_user.full_name)} изменил(а) категорию записи.</b>\n\n"
f"{escape(transaction.description)}{transaction_category_text(transaction)}",
)
except TelegramForbiddenError:
pass
await callback.answer("Категория изменена")
@router.callback_query(F.data.startswith("category:"))
async def set_category(callback: CallbackQuery) -> None:
try:
_, transaction_id_raw, category_raw = (callback.data or "").split(":", maxsplit=2)
transaction_id = int(transaction_id_raw)
category = Category(category_raw)
if category is Category.INCOME:
raise ValueError
except (ValueError, TypeError):
await callback.answer("Некорректная категория", show_alert=True)
return
changed = await database.update_category(transaction_id, callback.from_user.id, category)
transaction = await database.get_transaction(transaction_id, callback.from_user.id)
if not changed or transaction is None:
await callback.answer("Запись не найдена", show_alert=True)
return
if callback.message:
debt = await database.family_debt(callback.from_user.id)
await callback.message.edit_text(
transaction_confirmation(transaction, debt),
reply_markup=transaction_actions(transaction.id, is_expense=True),
)
await callback.answer(f"Категория: {category_label(category)}")
@router.callback_query(F.data.startswith("cancel:"))
async def cancel_category(callback: CallbackQuery) -> None:
transaction_id = _callback_id(callback.data)
transaction = await database.get_transaction(transaction_id, callback.from_user.id)
if transaction is None:
await callback.answer("Запись не найдена", show_alert=True)
return
if callback.message:
await callback.message.edit_reply_markup(
reply_markup=transaction_actions(
transaction.id, is_expense=transaction.kind is TransactionKind.EXPENSE
)
)
await callback.answer()
@router.callback_query(F.data.startswith("delete:"))
async def delete_transaction(callback: CallbackQuery) -> None:
transaction_id = _callback_id(callback.data)
deleted = await database.delete_transaction(transaction_id, callback.from_user.id)
if not deleted:
await callback.answer("Запись уже удалена", show_alert=True)
return
if callback.message:
debt = await database.family_debt(callback.from_user.id)
debt_line = f"\n\n{debt_short_text(debt)}" if debt is not None else ""
await callback.message.edit_text(f"🗑 <i>Запись удалена</i>{debt_line}")
await callback.answer("Удалено")
@router.message(F.text)
async def add_transaction(message: Message, bot: Bot) -> None:
await register(message)
try:
parsed = parse_transaction(message.text or "")
except ParseError:
await message.answer(
"Не получилось распознать запись 🤔\n\n"
"Напишите сумму и описание, например: <code>467 ярче</code>\n"
"Доход: <code>+ 50000 зарплата</code>"
)
return
category = (
Category.INCOME
if parsed.kind is TransactionKind.INCOME
else classify(parsed.description)
)
subcategory_key = (
None
if parsed.kind is TransactionKind.INCOME
else suggest_subcategory(category, parsed.description)
)
budget_category_id = await database.resolve_budget_category(
message.from_user.id, category, subcategory_key
)
transaction = await database.add_transaction(
user_id=message.from_user.id,
amount_kopecks=parsed.amount_kopecks,
description=parsed.description,
kind=parsed.kind,
category=category,
budget_category_id=budget_category_id,
occurred_at=datetime.now(settings.timezone),
)
debt = await database.family_debt(message.from_user.id)
await message.answer(
transaction_confirmation(transaction, debt),
reply_markup=transaction_actions(
transaction.id, is_expense=transaction.kind is TransactionKind.EXPENSE
),
)
for chat_id in await database.family_notification_chats(message.from_user.id):
try:
await bot.send_message(
chat_id, family_transaction_notice(transaction, debt)
)
except TelegramForbiddenError:
pass
@router.message()
async def unsupported(message: Message) -> None:
await message.answer("Пока я понимаю только текстовые записи — например, <code>3000 бензин</code>.")
root_router.include_routers(group_router, router)
return root_router
def _callback_id(data: str | None) -> int:
try:
return int((data or "").split(":", maxsplit=1)[1])
except (ValueError, IndexError) as exc:
raise ValueError("Некорректный идентификатор операции") from exc

152
app/keyboards.py Normal file
View file

@ -0,0 +1,152 @@
from __future__ import annotations
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, KeyboardButton, ReplyKeyboardMarkup
from app.categories import EXPENSE_CATEGORIES, category_label
from app.domain import BudgetCategory
MONTH_TOTAL = "📅 Итог за месяц"
ALL_TOTAL = "💰 Итог общий"
ANALYTICS = "📊 Аналитика"
RECENT = "🧾 Последние записи"
FAMILY = "👨‍👩‍👧 Семья"
DEBT = "💳 Общий долг"
CATEGORIES = "🗂 Категории"
HELP = "❓ Помощь"
def main_keyboard() -> ReplyKeyboardMarkup:
return ReplyKeyboardMarkup(
keyboard=[
[KeyboardButton(text=MONTH_TOTAL), KeyboardButton(text=ALL_TOTAL)],
[KeyboardButton(text=ANALYTICS), KeyboardButton(text=RECENT)],
[KeyboardButton(text=FAMILY), KeyboardButton(text=DEBT)],
[KeyboardButton(text=CATEGORIES), KeyboardButton(text=HELP)],
],
resize_keyboard=True,
input_field_placeholder="Например: 467 ярче",
is_persistent=True,
)
def family_setup_keyboard() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[
[InlineKeyboardButton(text="Создать общий бюджет", callback_data="family:create")],
[InlineKeyboardButton(text="Ввести код супруга", callback_data="family:join")],
]
)
def family_approval_keyboard(candidate_user_id: int) -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(
text="✅ Подтвердить",
callback_data=f"family:approve:{candidate_user_id}",
),
InlineKeyboardButton(
text="🚫 Отклонить",
callback_data=f"family:reject:{candidate_user_id}",
),
]
]
)
def debt_keyboard(can_edit: bool) -> InlineKeyboardMarkup | None:
if not can_edit:
return None
return InlineKeyboardMarkup(
inline_keyboard=[
[InlineKeyboardButton(text="✏️ Указать текущий долг", callback_data="debt:set")]
]
)
def category_tree_keyboard(
transaction_id: int, categories: list[BudgetCategory]
) -> InlineKeyboardMarkup:
rows: list[list[InlineKeyboardButton]] = []
for category in categories:
if category.parent_name:
text = f"{category.icon} {category.name}"
else:
text = f"{category.icon} {category.name}"
rows.append(
[
InlineKeyboardButton(
text=text,
callback_data=f"bcat:{transaction_id}:{category.id}",
)
]
)
rows.append([InlineKeyboardButton(text="Отмена", callback_data=f"cancel:{transaction_id}")])
return InlineKeyboardMarkup(inline_keyboard=rows)
def category_management_keyboard() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[
[InlineKeyboardButton(text=" Добавить категорию", callback_data="catmanage:add-main")],
[InlineKeyboardButton(text="↳ Добавить подкатегорию", callback_data="catmanage:add-sub")],
]
)
def parent_category_keyboard(categories: list[BudgetCategory]) -> InlineKeyboardMarkup:
parents = [category for category in categories if category.parent_id is None]
return InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(
text=f"{category.icon} {category.name}",
callback_data=f"catmanage:parent:{category.id}",
)
]
for category in parents
]
+ [[InlineKeyboardButton(text="Отмена", callback_data="catmanage:cancel")]]
)
def transaction_actions(transaction_id: int, *, is_expense: bool) -> InlineKeyboardMarkup:
buttons: list[InlineKeyboardButton] = []
if is_expense:
buttons.append(
InlineKeyboardButton(text="Изменить категорию", callback_data=f"choose:{transaction_id}")
)
buttons.append(InlineKeyboardButton(text="Удалить", callback_data=f"delete:{transaction_id}"))
return InlineKeyboardMarkup(inline_keyboard=[buttons])
def category_keyboard(transaction_id: int) -> InlineKeyboardMarkup:
rows = []
categories = list(EXPENSE_CATEGORIES)
for index in range(0, len(categories), 2):
rows.append(
[
InlineKeyboardButton(
text=category_label(category),
callback_data=f"category:{transaction_id}:{category.value}",
)
for category in categories[index : index + 2]
]
)
rows.append([InlineKeyboardButton(text="Отмена", callback_data=f"cancel:{transaction_id}")])
return InlineKeyboardMarkup(inline_keyboard=rows)
def analytics_period_keyboard(active: str = "month") -> InlineKeyboardMarkup:
month_label = "✓ Этот месяц" if active == "month" else "Этот месяц"
all_label = "✓ Всё время" if active == "all" else "Всё время"
return InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(text=month_label, callback_data="analytics:month"),
InlineKeyboardButton(text=all_label, callback_data="analytics:all"),
]
]
)

210
app/messages.py Normal file
View file

@ -0,0 +1,210 @@
from __future__ import annotations
from html import escape
from app.categories import category_label
from app.domain import FamilyInfo, Totals, Transaction, TransactionKind
from app.parser import format_money
MONTHS = (
"",
"январь",
"февраль",
"март",
"апрель",
"май",
"июнь",
"июль",
"август",
"сентябрь",
"октябрь",
"ноябрь",
"декабрь",
)
WELCOME = """<b>Привет! Я помогу держать деньги под контролем 👋</b>
Просто присылай сумму и описание я сам определю категорию:
<code>467 ярче</code>
<code>3000 бензин</code>
<code>+ 85000 зарплата</code>
Обычная запись расход. Знак <b>+</b> или слово <b>доход</b> поступление.
Каждый вечер я пришлю короткий итог дня и месяца. А в разделе «Семья» можно подключить общий бюджет с супругом. Начнём?"""
HELP_TEXT = """<b>Как записывать операции</b>
Расход:
<code>1250 продукты</code>
<code>3 500 ремонт машины</code>
Доход:
<code>+ 75000 зарплата</code>
<code>доход 5000 подработка</code>
Можно вводить копейки: <code>199,90 кофе</code>.
После записи категорию можно изменить, а ошибочную операцию удалить. Кнопки внизу показывают итоги и аналитику.
В семейном режиме кнопка <b>«Общий долг»</b> включает автоматический расчёт: расходы увеличивают долг, доходы уменьшают.
Через <b>«Категории»</b> можно добавить свои категории и подкатегории. Кнопка «Изменить категорию» переносит любую запись, в том числе из «Другое»."""
def transaction_confirmation(transaction: Transaction, debt: int | None = None) -> str:
is_income = transaction.kind is TransactionKind.INCOME
heading = "Доход записан" if is_income else "Расход записан"
icon = "🟢" if is_income else "🔴"
signed_amount = transaction.amount_kopecks if is_income else -transaction.amount_kopecks
debt_line = f"\n\n{debt_short_text(debt)}" if debt is not None else ""
return (
f"{icon} <b>{heading}</b>\n\n"
f"<b>{format_money(signed_amount, signed=True)}</b> · {escape(transaction.description)}\n"
f"{transaction_category_text(transaction)}{debt_line}"
)
def summary_text(title: str, totals: Totals) -> str:
balance_icon = "🟢" if totals.balance >= 0 else "🔴"
return (
f"<b>{escape(title)}</b>\n\n"
f"↗️ Доходы <b>{format_money(totals.income)}</b>\n"
f"↘️ Расходы <b>{format_money(totals.expense)}</b>\n"
f"━━━━━━━━━━━━\n"
f"{balance_icon} Баланс <b>{format_money(totals.balance, signed=True)}</b>"
)
def analytics_text(period_name: str, category_totals: dict, total: int) -> str:
title = f"📊 <b>Расходы {escape(period_name)}</b>"
if total == 0:
return f"{title}\n\nПока расходов нет. Самое время записать первый 👇"
lines = [title, f"Всего: <b>{format_money(total)}</b>", ""]
for label, amount in category_totals.items():
percent = round(amount / total * 100)
filled = max(1, round(percent / 10))
bar = "" * filled + "" * (10 - filled)
lines.append(escape(label))
lines.append(f"<code>{bar}</code> {percent}% · <b>{format_money(amount)}</b>")
lines.append("")
return "\n".join(lines).rstrip()
def recent_text(transactions: list[Transaction]) -> str:
if not transactions:
return "🧾 <b>Последние записи</b>\n\nЗдесь пока пусто."
lines = ["🧾 <b>Последние записи</b>", ""]
for transaction in transactions:
is_income = transaction.kind is TransactionKind.INCOME
amount = transaction.amount_kopecks if is_income else -transaction.amount_kopecks
author = f" · {escape(transaction.author_name)}" if transaction.author_name else ""
lines.append(
f"{transaction.occurred_at:%d.%m} <b>{format_money(amount, signed=True)}</b> "
f"{escape(transaction.description)}{author}"
)
lines.append("")
lines.append("Удалить или исправить запись можно сразу после её добавления.")
return "\n".join(lines)
def family_status_text(info: FamilyInfo | None) -> str:
if info is None:
return (
"👨‍👩‍👧 <b>Семейный бюджет</b>\n\n"
"Объедините два Telegram-аккаунта. После подключения вы оба будете видеть "
"общие итоги, аналитику и новые записи в реальном времени."
)
members = "\n".join(f"{escape(name)}" for name in info.members)
waiting = (
"\n\nОтправьте код супругу. Подключение произойдёт только после вашего подтверждения."
if len(info.members) < 2
else ""
)
chat = (
f"\n\n💬 Общий чат: <b>{escape(info.notification_chat_title or 'Подключён')}</b>"
if info.notification_chat_id is not None
else (
"\n\n💬 <b>Общий чат пока не подключён.</b> Создатель бюджета должен добавить "
"бота в семейную группу, стать администратором группы и отправить там /family_chat."
)
)
return (
"👨‍👩‍👧 <b>Семейный бюджет подключён</b>\n\n"
f"<b>Участники:</b>\n{members}\n\n"
f"Код приглашения: <code>{escape(info.invite_code)}</code>"
f"{waiting}{chat}"
)
def family_transaction_notice(transaction: Transaction, debt: int | None = None) -> str:
is_income = transaction.kind is TransactionKind.INCOME
action = "добавил(а) доход" if is_income else "добавил(а) расход"
icon = "🟢" if is_income else "🔴"
amount = transaction.amount_kopecks if is_income else -transaction.amount_kopecks
author = escape(transaction.author_name or "Супруг")
debt_line = f"\n\n{debt_short_text(debt)}" if debt is not None else ""
return (
f"👨‍👩‍👧 <b>{author} {action}</b>\n\n"
f"{icon} <b>{format_money(amount, signed=True)}</b> · {escape(transaction.description)}\n"
f"{transaction_category_text(transaction)}{debt_line}"
)
def daily_report_text(day: Totals, month: Totals, debt: int | None = None) -> str:
balance_icon = "" if day.balance >= 0 else "📌"
debt_section = f"\n\n{debt_short_text(debt)}" if debt is not None else ""
return (
"🌙 <b>Итог дня</b>\n\n"
f"Потрачено сегодня: <b>{format_money(day.expense)}</b>\n"
f"Получено сегодня: <b>{format_money(day.income)}</b>\n"
f"{balance_icon} Баланс дня: <b>{format_money(day.balance, signed=True)}</b>\n\n"
"<b>С начала месяца</b>\n"
f"Расходы: <b>{format_money(month.expense)}</b>\n"
f"Доходы: <b>{format_money(month.income)}</b>\n"
f"Баланс: <b>{format_money(month.balance, signed=True)}</b>"
f"{debt_section}"
)
def debt_short_text(debt: int) -> str:
if debt > 0:
return f"💳 Общий долг: <b>{format_money(debt)}</b>"
if debt < 0:
return f"🌿 Семейный резерв: <b>{format_money(abs(debt))}</b>"
return "🎉 Общий долг: <b>погашен</b>"
def debt_status_text(debt: int | None, can_edit: bool) -> str:
if debt is None:
status = "Общий долг пока не указан."
else:
status = debt_short_text(debt)
edit_hint = (
"\n\nУкажите текущую сумму долга. С этого момента каждый расход будет её увеличивать, "
"а каждый доход — уменьшать."
if can_edit
else "\n\nИзменить исходную сумму может только создатель семейного бюджета."
)
return f"<b>💳 Общий долг</b>\n\n{status}{edit_hint}"
def transaction_category_text(transaction: Transaction) -> str:
if transaction.budget_category_name and transaction.parent_category_name:
return (
f"{escape(transaction.parent_category_icon or '📁')} "
f"{escape(transaction.parent_category_name)}"
f"{escape(transaction.budget_category_icon or '📁')} "
f"{escape(transaction.budget_category_name)}"
)
if transaction.budget_category_name:
return (
f"{escape(transaction.budget_category_icon or '📁')} "
f"{escape(transaction.budget_category_name)}"
)
return category_label(transaction.category)

84
app/parser.py Normal file
View file

@ -0,0 +1,84 @@
from __future__ import annotations
import re
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from app.domain import ParsedTransaction, TransactionKind
class ParseError(ValueError):
pass
_TRANSACTION_RE = re.compile(
r"^\s*(?P<sign>[+-])?\s*"
r"(?:(?P<kind>доход|приход|расход|трата)\s+)?"
r"(?P<amount>\d[\d ]*(?:[,.]\d{1,2})?)\s*"
r"(?:₽|р\.?|руб(?:лей|ля|ль)?\.?)?\s+"
r"(?P<description>\S.*)\s*$",
flags=re.IGNORECASE,
)
def parse_transaction(text: str) -> ParsedTransaction:
match = _TRANSACTION_RE.match(text)
if not match:
raise ParseError("Не удалось распознать сумму и описание")
raw_amount = match.group("amount").replace(" ", "").replace(",", ".")
try:
amount = Decimal(raw_amount)
except InvalidOperation as exc:
raise ParseError("Некорректная сумма") from exc
if amount <= 0:
raise ParseError("Сумма должна быть больше нуля")
amount_kopecks = int((amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
description = " ".join(match.group("description").split())
kind_word = (match.group("kind") or "").casefold()
sign = match.group("sign")
kind = (
TransactionKind.INCOME
if sign == "+" or kind_word in {"доход", "приход"}
else TransactionKind.EXPENSE
)
if sign == "-" and kind is TransactionKind.INCOME:
raise ParseError("Доход не может быть отрицательным")
return ParsedTransaction(
amount_kopecks=amount_kopecks,
description=description,
kind=kind,
)
def format_money(kopecks: int, *, signed: bool = False) -> str:
sign = ""
if signed:
sign = "+" if kopecks > 0 else ("" if kopecks < 0 else "")
absolute = abs(kopecks)
rubles, cents = divmod(absolute, 100)
grouped = f"{rubles:,}".replace(",", " ")
value = f"{grouped},{cents:02d}" if cents else grouped
return f"{sign}{value}"
_MONEY_ONLY_RE = re.compile(
r"^\s*(?P<amount>\d[\d ]*(?:[,.]\d{1,2})?)\s*"
r"(?:₽|р\.?|руб(?:лей|ля|ль)?\.?)?\s*$",
flags=re.IGNORECASE,
)
def parse_money_amount(text: str) -> int:
match = _MONEY_ONLY_RE.match(text)
if not match:
raise ParseError("Некорректная сумма")
raw_amount = match.group("amount").replace(" ", "").replace(",", ".")
try:
amount = Decimal(raw_amount)
except InvalidOperation as exc:
raise ParseError("Некорректная сумма") from exc
if amount < 0:
raise ParseError("Сумма не может быть отрицательной")
return int((amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))

18
app/periods.py Normal file
View file

@ -0,0 +1,18 @@
from __future__ import annotations
from datetime import datetime, timedelta
def day_bounds(now: datetime) -> tuple[datetime, datetime]:
start = now.replace(hour=0, minute=0, second=0, microsecond=0)
return start, start + timedelta(days=1)
def month_bounds(now: datetime) -> tuple[datetime, datetime]:
start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
if start.month == 12:
end = start.replace(year=start.year + 1, month=1)
else:
end = start.replace(month=start.month + 1)
return start, end

80
app/scheduler.py Normal file
View file

@ -0,0 +1,80 @@
from __future__ import annotations
import asyncio
import logging
from contextlib import suppress
from datetime import datetime, time
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from aiogram import Bot
from aiogram.exceptions import TelegramForbiddenError
from app.database import Database
from app.keyboards import main_keyboard
from app.messages import daily_report_text
from app.periods import day_bounds, month_bounds
logger = logging.getLogger(__name__)
async def daily_report_loop(bot: Bot, database: Database) -> None:
"""Check report schedules periodically; the database makes delivery idempotent."""
while True:
try:
await _send_due_reports(bot, database)
except asyncio.CancelledError:
raise
except Exception:
logger.exception("Ошибка цикла ежедневных отчётов")
await asyncio.sleep(30)
async def _send_due_reports(bot: Bot, database: Database) -> None:
users = await database.users_for_reports()
for user in users:
try:
timezone = ZoneInfo(user["timezone"])
report_at = time.fromisoformat(user["report_time"])
except (ZoneInfoNotFoundError, ValueError):
logger.error("Некорректные настройки времени у пользователя %s", user["user_id"])
continue
now = datetime.now(timezone)
if now.time().replace(tzinfo=None) < report_at:
continue
local_date = now.date().isoformat()
day_start, day_end = day_bounds(now)
month_start, month_end = month_bounds(now)
day = await database.totals(user["user_id"], day_start, day_end)
month = await database.totals(user["user_id"], month_start, month_end)
debt = await database.family_debt(user["user_id"])
report_text = daily_report_text(day, month, debt)
if not await database.report_was_sent(user["user_id"], local_date):
try:
await bot.send_message(
chat_id=user["chat_id"],
text=report_text,
reply_markup=main_keyboard(),
)
except TelegramForbiddenError:
logger.info("Пользователь %s заблокировал бота", user["user_id"])
await database.mark_report_sent(user["user_id"], local_date, now)
family_target = await database.family_report_target(user["user_id"])
if family_target is None:
continue
household_id, family_chat_id = family_target
if await database.family_report_was_sent(household_id, local_date):
continue
try:
await bot.send_message(chat_id=family_chat_id, text=report_text)
except TelegramForbiddenError:
logger.info("Бот не может писать в семейный чат %s", family_chat_id)
await database.mark_family_report_sent(household_id, local_date, now)
async def stop_task(task: asyncio.Task) -> None:
task.cancel()
with suppress(asyncio.CancelledError):
await task

12
compose.yaml Normal file
View file

@ -0,0 +1,12 @@
services:
finance-bot:
build: .
container_name: finance-telegram-bot
restart: unless-stopped
env_file: .env
volumes:
- finance-data:/app/data
volumes:
finance-data:

251
install.sh Normal file
View file

@ -0,0 +1,251 @@
#!/usr/bin/env bash
set -Eeuo pipefail
PUBLIC_URL="${FINANCE_BOT_PUBLIC_URL:-https://disk.yandex.ru/d/hhbDTO08CB030Q}"
DEFAULT_INSTALL_DIR="/opt/finance-bot"
YANDEX_API="https://cloud-api.yandex.net/v1/disk/public/resources/download"
TMP_DIR=""
if [[ -t 1 ]]; then
BOLD='\033[1m'
GREEN='\033[0;32m'
YELLOW='\033[0;33m'
RED='\033[0;31m'
RESET='\033[0m'
else
BOLD=''
GREEN=''
YELLOW=''
RED=''
RESET=''
fi
info() { printf '%b\n' "${GREEN}${RESET} $*"; }
warn() { printf '%b\n' "${YELLOW}!${RESET} $*"; }
fail() { printf '%b\n' "${RED}Ошибка:${RESET} $*" >&2; exit 1; }
cleanup() {
if [[ -n "$TMP_DIR" && -d "$TMP_DIR" ]]; then
rm -rf -- "$TMP_DIR"
fi
}
trap cleanup EXIT
require_root() {
if [[ "${EUID}" -ne 0 ]]; then
fail "запустите установщик с правами администратора: sudo bash install.sh"
fi
}
install_base_tools() {
local missing=()
local command_name
for command_name in curl unzip python3; do
command -v "$command_name" >/dev/null 2>&1 || missing+=("$command_name")
done
[[ -e /usr/share/zoneinfo/UTC ]] || missing+=("tzdata")
if (( ${#missing[@]} == 0 )); then
return
fi
command -v apt-get >/dev/null 2>&1 \
|| fail "установите curl, unzip и python3, затем запустите скрипт повторно"
warn "Устанавливаю системные утилиты: ${missing[*]}"
apt-get update
DEBIAN_FRONTEND=noninteractive apt-get install -y ca-certificates curl unzip python3 tzdata
}
install_docker() {
if command -v docker >/dev/null 2>&1 && docker compose version >/dev/null 2>&1; then
info "Docker и Docker Compose уже установлены"
return
fi
local answer
read -r -p "Docker Compose не найден. Установить Docker из официального репозитория? [Y/n]: " answer
answer="${answer:-Y}"
[[ "$answer" =~ ^[YyДд]$ ]] || fail "для запуска бота необходимы Docker Engine и Docker Compose"
[[ -r /etc/os-release ]] || fail "не удалось определить дистрибутив Linux"
# shellcheck disable=SC1091
. /etc/os-release
case "${ID:-}" in
ubuntu|debian) ;;
*) fail "автоматическая установка Docker поддерживает Ubuntu и Debian; установите Docker вручную" ;;
esac
info "Подключаю официальный репозиторий Docker"
apt-get update
DEBIAN_FRONTEND=noninteractive apt-get install -y ca-certificates curl
install -m 0755 -d /etc/apt/keyrings
curl -fsSL "https://download.docker.com/linux/${ID}/gpg" -o /etc/apt/keyrings/docker.asc
chmod a+r /etc/apt/keyrings/docker.asc
local codename architecture
codename="${UBUNTU_CODENAME:-${VERSION_CODENAME:-}}"
[[ -n "$codename" ]] || fail "не удалось определить кодовое имя дистрибутива"
architecture="$(dpkg --print-architecture)"
cat > /etc/apt/sources.list.d/docker.sources <<EOF
Types: deb
URIs: https://download.docker.com/linux/${ID}
Suites: ${codename}
Components: stable
Architectures: ${architecture}
Signed-By: /etc/apt/keyrings/docker.asc
EOF
apt-get update
DEBIAN_FRONTEND=noninteractive apt-get install -y \
docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin
systemctl enable --now docker
docker compose version >/dev/null 2>&1 || fail "Docker Compose не запустился после установки"
info "Docker установлен"
}
prompt_install_dir() {
local entered
read -r -p "Папка установки [${DEFAULT_INSTALL_DIR}]: " entered
INSTALL_DIR="${entered:-$DEFAULT_INSTALL_DIR}"
[[ "$INSTALL_DIR" == /* ]] || fail "путь установки должен быть абсолютным"
case "$INSTALL_DIR" in
/|/etc|/usr|/var|/opt) fail "выберите отдельную папку, например ${DEFAULT_INSTALL_DIR}" ;;
esac
}
read_existing_value() {
local key="$1"
local env_file="$2"
[[ -f "$env_file" ]] || return 0
sed -n "s/^${key}=//p" "$env_file" | tail -n 1
}
prompt_bot_settings() {
local env_file="${INSTALL_DIR}/.env"
local existing_token existing_timezone existing_report_time keep answer
existing_token="$(read_existing_value BOT_TOKEN "$env_file")"
existing_timezone="$(read_existing_value BOT_TIMEZONE "$env_file")"
existing_report_time="$(read_existing_value DAILY_REPORT_TIME "$env_file")"
BOT_TOKEN=""
if [[ -n "$existing_token" ]]; then
read -r -p "Найдены прежние настройки. Сохранить текущий токен бота? [Y/n]: " keep
keep="${keep:-Y}"
if [[ "$keep" =~ ^[YyДд]$ ]]; then
BOT_TOKEN="$existing_token"
fi
fi
while [[ ! "$BOT_TOKEN" =~ ^[0-9]{6,}:[A-Za-z0-9_-]{20,}$ ]]; do
read -r -s -p "Токен от @BotFather: " BOT_TOKEN
printf '\n'
[[ "$BOT_TOKEN" =~ ^[0-9]{6,}:[A-Za-z0-9_-]{20,}$ ]] \
|| warn "Токен выглядит некорректно. Попробуйте ещё раз."
done
local default_timezone="${existing_timezone:-Asia/Novosibirsk}"
while true; do
read -r -p "Часовой пояс [${default_timezone}]: " answer
BOT_TIMEZONE="${answer:-$default_timezone}"
if [[ "$BOT_TIMEZONE" =~ ^[A-Za-z0-9_+.-]+(/[A-Za-z0-9_+.-]+)+$ ]] \
&& [[ -e "/usr/share/zoneinfo/${BOT_TIMEZONE}" ]]; then
break
fi
warn "Неизвестный часовой пояс. Пример: Asia/Novosibirsk"
done
local default_report_time="${existing_report_time:-21:00}"
while true; do
read -r -p "Время ежедневного отчёта [${default_report_time}]: " answer
DAILY_REPORT_TIME="${answer:-$default_report_time}"
if [[ "$DAILY_REPORT_TIME" =~ ^([01][0-9]|2[0-3]):[0-5][0-9]$ ]]; then
break
fi
warn "Введите время в формате ЧЧ:ММ, например 21:00"
done
}
download_project() {
TMP_DIR="$(mktemp -d)"
local api_response download_url archive project_root
archive="${TMP_DIR}/finance-bot.zip"
info "Получаю проект с Яндекс.Диска"
api_response="$(curl -fsS --get --data-urlencode "public_key=${PUBLIC_URL}" "$YANDEX_API")" \
|| fail "Яндекс.Диск не вернул ссылку на скачивание"
download_url="$(printf '%s' "$api_response" | python3 -c '
import json, sys
try:
payload = json.load(sys.stdin)
print(payload["href"])
except (KeyError, ValueError):
raise SystemExit(1)
')" || fail "не удалось разобрать ответ Яндекс.Диска; проверьте публичную ссылку"
curl -fL --retry 3 --connect-timeout 20 "$download_url" -o "$archive" \
|| fail "не удалось скачать архив проекта"
[[ -s "$archive" ]] || fail "Яндекс.Диск вернул пустой архив"
if (( $(stat -c '%s' "$archive") > 209715200 )); then
fail "архив больше 200 МБ; установка остановлена"
fi
unzip -tq "$archive" >/dev/null || fail "скачанный файл не является исправным ZIP-архивом"
if unzip -Z1 "$archive" | grep -Eq '(^/|(^|/)\.\.(/|$))'; then
fail "в архиве обнаружены небезопасные пути"
fi
mkdir -p "${TMP_DIR}/extracted"
unzip -q "$archive" -d "${TMP_DIR}/extracted"
COMPOSE_FILE="$(find "${TMP_DIR}/extracted" -maxdepth 4 -type f -name compose.yaml -print -quit)"
[[ -n "$COMPOSE_FILE" ]] || fail "в архиве не найден compose.yaml"
project_root="$(dirname "$COMPOSE_FILE")"
[[ -f "${project_root}/Dockerfile" && -f "${project_root}/pyproject.toml" && -d "${project_root}/app" ]] \
|| fail "структура проекта в архиве неполная"
DOWNLOADED_PROJECT_DIR="$project_root"
info "Проект скачан и проверен"
}
deploy_project() {
mkdir -p "$INSTALL_DIR"
cp -a "${DOWNLOADED_PROJECT_DIR}/." "$INSTALL_DIR/"
umask 077
cat > "${INSTALL_DIR}/.env" <<EOF
BOT_TOKEN=${BOT_TOKEN}
BOT_TIMEZONE=${BOT_TIMEZONE}
DAILY_REPORT_TIME=${DAILY_REPORT_TIME}
DATABASE_PATH=/app/data/finance.db
LOG_LEVEL=INFO
EOF
chmod 600 "${INSTALL_DIR}/.env"
info "Собираю и запускаю контейнер"
docker compose --project-directory "$INSTALL_DIR" -f "${INSTALL_DIR}/compose.yaml" \
up -d --build --remove-orphans
sleep 5
local status
status="$(docker inspect --format '{{.State.Status}}' finance-telegram-bot 2>/dev/null || true)"
if [[ "$status" != "running" ]]; then
docker compose --project-directory "$INSTALL_DIR" -f "${INSTALL_DIR}/compose.yaml" \
logs --tail=40 finance-bot || true
fail "контейнер не запустился; выше показаны последние строки журнала"
fi
info "Бот установлен и запущен"
printf '\n%b\n' "${BOLD}Папка:${RESET} ${INSTALL_DIR}"
printf '%b\n' "${BOLD}Статус:${RESET} docker compose -f ${INSTALL_DIR}/compose.yaml ps"
printf '%b\n' "${BOLD}Журнал:${RESET} docker compose -f ${INSTALL_DIR}/compose.yaml logs -f"
printf '%b\n' "Повторный запуск этого установщика обновит код и сохранит базу данных."
}
main() {
printf '%b\n\n' "${BOLD}Установка финансового Telegram-бота${RESET}"
require_root
install_base_tools
prompt_install_dir
prompt_bot_settings
install_docker
download_project
deploy_project
}
main "$@"

25
pyproject.toml Normal file
View file

@ -0,0 +1,25 @@
[build-system]
requires = ["setuptools>=69"]
build-backend = "setuptools.build_meta"
[project]
name = "finance-telegram-bot"
version = "1.0.0"
description = "Telegram bot for effortless personal income and expense tracking"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"aiogram>=3.29,<4",
"aiosqlite>=0.20,<1",
"tzdata>=2025.2",
]
[project.optional-dependencies]
dev = ["pytest>=8,<9", "pytest-asyncio>=0.24,<1"]
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
[tool.setuptools.packages.find]
include = ["app*"]

39
tests/test_categories.py Normal file
View file

@ -0,0 +1,39 @@
import pytest
from app.categories import classify, suggest_subcategory
from app.domain import Category
@pytest.mark.parametrize(
("description", "category"),
[
("ярче", Category.PRODUCTS),
("Пиво и снеки", Category.PRODUCTS),
("бензин газпромнефть", Category.CAR),
("платёж по ипотеке", Category.CREDITS),
("билеты в кино", Category.ENTERTAINMENT),
("обои в спальню", Category.RENOVATION),
("5000 ремонт", Category.RENOVATION),
("ремонт машины", Category.CAR),
("подарок коллеге", Category.OTHER),
],
)
def test_classification(description, category):
assert classify(description) is category
def test_keyword_does_not_match_inside_another_word():
assert classify("корм для барсика") is Category.OTHER
@pytest.mark.parametrize(
("description", "category", "subcategory"),
[
("3000 бензин", Category.CAR, "car_fuel"),
("запчасти на машину", Category.CAR, "car_parts"),
("пиво", Category.PRODUCTS, "products_alcohol"),
("кофе в кафе", Category.PRODUCTS, "products_cafe"),
],
)
def test_subcategory_suggestion(description, category, subcategory):
assert suggest_subcategory(category, description) == subcategory

257
tests/test_database.py Normal file
View file

@ -0,0 +1,257 @@
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
import pytest
from app.database import Database
from app.database import FamilyJoinError
from app.categories import category_label
from app.domain import Category, TransactionKind
@pytest.mark.asyncio
async def test_transaction_lifecycle_and_totals(tmp_path):
database = Database(tmp_path / "finance.db")
await database.initialize()
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
await database.upsert_user(
user_id=10,
chat_id=10,
username="tester",
full_name="Test User",
timezone="UTC",
report_time="21:00",
now=now,
)
await database.upsert_user(
user_id=20,
chat_id=20,
username=None,
full_name="Other User",
timezone="UTC",
report_time="21:00",
now=now,
)
expense = await database.add_transaction(
user_id=10,
amount_kopecks=30_000,
description="бензин",
kind=TransactionKind.EXPENSE,
category=Category.CAR,
occurred_at=now,
)
await database.add_transaction(
user_id=10,
amount_kopecks=100_000,
description="подработка",
kind=TransactionKind.INCOME,
category=Category.INCOME,
occurred_at=now,
)
totals = await database.totals(10, now - timedelta(hours=1), now + timedelta(hours=1))
assert totals.income == 100_000
assert totals.expense == 30_000
assert totals.balance == 70_000
assert await database.category_totals(
10, now - timedelta(hours=1), now + timedelta(hours=1)
) == {category_label(Category.CAR): 30_000}
assert await database.category_totals(10) == {category_label(Category.CAR): 30_000}
# Другой пользователь не может изменить или удалить чужую запись.
assert not await database.update_category(expense.id, 20, Category.OTHER)
assert not await database.delete_transaction(expense.id, 20)
assert await database.update_category(expense.id, 10, Category.OTHER)
assert (await database.get_transaction(expense.id, 10)).category is Category.OTHER
assert await database.delete_transaction(expense.id, 10)
@pytest.mark.asyncio
async def test_daily_report_idempotency(tmp_path):
database = Database(tmp_path / "finance.db")
await database.initialize()
now = datetime(2026, 7, 13, 21, tzinfo=ZoneInfo("UTC"))
await database.upsert_user(
user_id=10,
chat_id=10,
username=None,
full_name="Test User",
timezone="UTC",
report_time="21:00",
now=now,
)
assert not await database.report_was_sent(10, "2026-07-13")
await database.mark_report_sent(10, "2026-07-13", now)
await database.mark_report_sent(10, "2026-07-13", now)
assert await database.report_was_sent(10, "2026-07-13")
@pytest.mark.asyncio
async def test_family_budget_shares_history_and_new_transactions(tmp_path):
database = Database(tmp_path / "finance.db")
await database.initialize()
now = datetime(2026, 7, 13, 20, tzinfo=ZoneInfo("UTC"))
for user_id, name in ((10, "Алексей"), (20, "Мария"), (30, "Посторонний")):
await database.upsert_user(
user_id=user_id,
chat_id=user_id,
username=None,
full_name=name,
timezone="UTC",
report_time="21:00",
now=now,
)
await database.add_transaction(
user_id=10,
amount_kopecks=30_000,
description="бензин",
kind=TransactionKind.EXPENSE,
category=Category.CAR,
occurred_at=now,
)
await database.add_transaction(
user_id=20,
amount_kopecks=10_000,
description="ярче",
kind=TransactionKind.EXPENSE,
category=Category.PRODUCTS,
occurred_at=now,
)
family = await database.create_family(10, now)
request = await database.request_family_join(20, family.invite_code.lower(), now)
assert request.owner_chat_id == 10
with pytest.raises(FamilyJoinError, match="обработан"):
await database.resolve_family_join(
owner_user_id=30,
candidate_user_id=20,
approve=True,
now=now,
)
approved, candidate_chat_id, joined = await database.resolve_family_join(
owner_user_id=10,
candidate_user_id=20,
approve=True,
now=now,
)
assert approved
assert candidate_chat_id == 20
assert joined is not None
assert joined.household_id == family.household_id
assert joined.members == ("Алексей", "Мария")
for user_id in (10, 20):
totals = await database.totals(user_id)
assert totals.expense == 40_000
assert await database.category_totals(user_id) == {
category_label(Category.CAR): 30_000,
category_label(Category.PRODUCTS): 10_000,
}
recent = await database.recent_transactions(user_id)
assert {item.author_name for item in recent} == {"Алексей", "Мария"}
recipients = await database.family_recipients(10)
assert [(row["user_id"], row["chat_id"]) for row in recipients] == [(20, 20)]
with pytest.raises(FamilyJoinError, match="создатель"):
await database.bind_family_chat(20, -100500, "Наш бюджет")
bound = await database.bind_family_chat(10, -100500, "Наш бюджет")
assert bound.notification_chat_id == -100500
assert await database.family_notification_chats(10) == [20, -100500]
assert await database.family_report_target(20) == (family.household_id, -100500)
assert not await database.family_report_was_sent(family.household_id, "2026-07-13")
await database.mark_family_report_sent(family.household_id, "2026-07-13", now)
assert await database.family_report_was_sent(family.household_id, "2026-07-13")
with pytest.raises(FamilyJoinError, match="создатель"):
await database.set_family_debt(20, 53_071_300, now)
await database.set_family_debt(10, 53_071_300, now)
salary = await database.add_transaction(
user_id=20,
amount_kopecks=40_000_000,
description="зарплата",
kind=TransactionKind.INCOME,
category=Category.INCOME,
occurred_at=now + timedelta(seconds=1),
)
assert await database.family_debt(10) == 13_071_300
purchase = await database.add_transaction(
user_id=10,
amount_kopecks=2_000_000,
description="покупка",
kind=TransactionKind.EXPENSE,
category=Category.OTHER,
occurred_at=now + timedelta(seconds=2),
)
assert await database.family_debt(20) == 15_071_300
assert await database.delete_transaction(purchase.id, 10)
assert await database.family_debt(10) == 13_071_300
assert salary.author_name == "Мария"
categories = await database.budget_categories(10)
fuel = next(category for category in categories if category.system_key == "car_fuel")
assert fuel.parent_name == "Машина"
health = await database.create_budget_category(
user_id=20,
name="Здоровье",
parent_id=None,
now=now,
)
doctors = await database.create_budget_category(
user_id=10,
name="Врачи",
parent_id=health.id,
now=now,
)
other = await database.add_transaction(
user_id=10,
amount_kopecks=50_000,
description="консультация",
kind=TransactionKind.EXPENSE,
category=Category.OTHER,
occurred_at=now + timedelta(seconds=3),
)
assert await database.update_transaction_budget_category(other.id, 10, doctors.id)
moved = await database.get_transaction(other.id, 10)
assert moved.parent_category_name == "Здоровье"
assert moved.budget_category_name == "Врачи"
analytics = await database.category_totals(20)
assert analytics["📁 Здоровье → 📁 Врачи"] == 50_000
with pytest.raises(FamilyJoinError, match="два участника"):
await database.request_family_join(30, family.invite_code, now)
@pytest.mark.asyncio
async def test_existing_database_is_migrated_for_family_budget(tmp_path):
import aiosqlite
path = tmp_path / "old.db"
async with aiosqlite.connect(path) as db:
await db.executescript(
"""
CREATE TABLE users (
user_id INTEGER PRIMARY KEY, chat_id INTEGER NOT NULL, username TEXT,
full_name TEXT NOT NULL, timezone TEXT NOT NULL, report_time TEXT NOT NULL,
created_at TEXT NOT NULL, updated_at TEXT NOT NULL
);
CREATE TABLE transactions (
id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL,
amount_kopecks INTEGER NOT NULL, description TEXT NOT NULL,
kind TEXT NOT NULL, category TEXT NOT NULL,
occurred_at TEXT NOT NULL, created_at TEXT NOT NULL
);
"""
)
await db.commit()
database = Database(path)
await database.initialize()
async with aiosqlite.connect(path) as db:
cursor = await db.execute("PRAGMA table_info(transactions)")
columns = {row[1] for row in await cursor.fetchall()}
assert "household_id" in columns
async with aiosqlite.connect(path) as db:
cursor = await db.execute("PRAGMA table_info(households)")
household_columns = {row[1] for row in await cursor.fetchall()}
assert {"debt_base_kopecks", "debt_started_at"} <= household_columns

41
tests/test_parser.py Normal file
View file

@ -0,0 +1,41 @@
import pytest
from app.domain import TransactionKind
from app.parser import ParseError, format_money, parse_money_amount, parse_transaction
@pytest.mark.parametrize(
("text", "kopecks", "description", "kind"),
[
("467 ярче", 46_700, "ярче", TransactionKind.EXPENSE),
("3 000 бензин", 300_000, "бензин", TransactionKind.EXPENSE),
("199,90 кофе", 19_990, "кофе", TransactionKind.EXPENSE),
("+ 85 000 зарплата", 8_500_000, "зарплата", TransactionKind.INCOME),
("доход 5000 подработка", 500_000, "подработка", TransactionKind.INCOME),
("расход 1200 кино", 120_000, "кино", TransactionKind.EXPENSE),
],
)
def test_parse_transaction(text, kopecks, description, kind):
parsed = parse_transaction(text)
assert parsed.amount_kopecks == kopecks
assert parsed.description == description
assert parsed.kind is kind
@pytest.mark.parametrize("text", ["просто текст", "0 кофе", "500", "- доход 500 премия"])
def test_rejects_invalid_input(text):
with pytest.raises(ParseError):
parse_transaction(text)
def test_format_money():
assert format_money(123_456_00) == "123 456 ₽"
assert format_money(-199_90, signed=True) == "199,90 ₽"
@pytest.mark.parametrize(
("text", "kopecks"),
[("530713", 53_071_300), ("530 713 ₽", 53_071_300), ("1250,50 руб", 125_050)],
)
def test_parse_money_amount(text, kopecks):
assert parse_money_amount(text) == kopecks

19
tests/test_periods.py Normal file
View file

@ -0,0 +1,19 @@
from datetime import datetime
from zoneinfo import ZoneInfo
from app.periods import day_bounds, month_bounds
def test_month_bounds_in_december():
now = datetime(2026, 12, 15, 10, tzinfo=ZoneInfo("UTC"))
start, end = month_bounds(now)
assert start == datetime(2026, 12, 1, tzinfo=ZoneInfo("UTC"))
assert end == datetime(2027, 1, 1, tzinfo=ZoneInfo("UTC"))
def test_day_bounds():
now = datetime(2026, 7, 13, 22, 10, tzinfo=ZoneInfo("UTC"))
start, end = day_bounds(now)
assert start.day == 13
assert end.day == 14

51
tests/test_scheduler.py Normal file
View file

@ -0,0 +1,51 @@
from datetime import datetime
from zoneinfo import ZoneInfo
import pytest
from app.database import Database
from app.scheduler import _send_due_reports
class FakeBot:
def __init__(self):
self.messages = []
async def send_message(self, **kwargs):
self.messages.append(kwargs)
@pytest.mark.asyncio
async def test_family_daily_report_is_sent_to_group_only_once(tmp_path):
database = Database(tmp_path / "finance.db")
await database.initialize()
now = datetime.now(ZoneInfo("UTC"))
for user_id, name in ((10, "Алексей"), (20, "Мария")):
await database.upsert_user(
user_id=user_id,
chat_id=user_id,
username=None,
full_name=name,
timezone="UTC",
report_time="00:00",
now=now,
)
family = await database.create_family(10, now)
await database.request_family_join(20, family.invite_code, now)
await database.resolve_family_join(
owner_user_id=10,
candidate_user_id=20,
approve=True,
now=now,
)
await database.bind_family_chat(10, -100500, "Наш бюджет")
bot = FakeBot()
await _send_due_reports(bot, database)
assert [message["chat_id"] for message in bot.messages].count(10) == 1
assert [message["chat_id"] for message in bot.messages].count(20) == 1
assert [message["chat_id"] for message in bot.messages].count(-100500) == 1
await _send_due_reports(bot, database)
assert len(bot.messages) == 3