112 lines
4.4 KiB
Python
112 lines
4.4 KiB
Python
"""Report service for analytics"""
|
|
|
|
from typing import List, Dict
|
|
from datetime import datetime, timedelta
|
|
from sqlalchemy.orm import Session
|
|
from app.db.repositories import TransactionRepository, CategoryRepository
|
|
from app.db.models import TransactionType
|
|
|
|
|
|
class ReportService:
|
|
"""Service for generating financial reports"""
|
|
|
|
def __init__(self, session: Session):
|
|
self.session = session
|
|
self.transaction_repo = TransactionRepository(session)
|
|
self.category_repo = CategoryRepository(session)
|
|
|
|
def get_expenses_by_category(
|
|
self, family_id: int, start_date: datetime, end_date: datetime
|
|
) -> Dict[str, float]:
|
|
"""Get expense breakdown by category"""
|
|
transactions = self.transaction_repo.get_transactions_by_period(
|
|
family_id, start_date, end_date
|
|
)
|
|
|
|
expenses_by_category = {}
|
|
for transaction in transactions:
|
|
if transaction.transaction_type == TransactionType.EXPENSE:
|
|
category_name = transaction.category.name if transaction.category else "Без категории"
|
|
if category_name not in expenses_by_category:
|
|
expenses_by_category[category_name] = 0
|
|
expenses_by_category[category_name] += transaction.amount
|
|
|
|
# Sort by amount descending
|
|
return dict(sorted(expenses_by_category.items(), key=lambda x: x[1], reverse=True))
|
|
|
|
def get_expenses_by_user(
|
|
self, family_id: int, start_date: datetime, end_date: datetime
|
|
) -> Dict[str, float]:
|
|
"""Get expense breakdown by user"""
|
|
transactions = self.transaction_repo.get_transactions_by_period(
|
|
family_id, start_date, end_date
|
|
)
|
|
|
|
expenses_by_user = {}
|
|
for transaction in transactions:
|
|
if transaction.transaction_type == TransactionType.EXPENSE:
|
|
user_name = f"{transaction.user.first_name or ''} {transaction.user.last_name or ''}".strip()
|
|
if not user_name:
|
|
user_name = transaction.user.username or f"User {transaction.user.id}"
|
|
if user_name not in expenses_by_user:
|
|
expenses_by_user[user_name] = 0
|
|
expenses_by_user[user_name] += transaction.amount
|
|
|
|
return dict(sorted(expenses_by_user.items(), key=lambda x: x[1], reverse=True))
|
|
|
|
def get_daily_expenses(
|
|
self, family_id: int, days: int = 30
|
|
) -> Dict[str, float]:
|
|
"""Get daily expenses for period"""
|
|
end_date = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
|
|
start_date = end_date - timedelta(days=days)
|
|
|
|
transactions = self.transaction_repo.get_transactions_by_period(
|
|
family_id, start_date, end_date
|
|
)
|
|
|
|
daily_expenses = {}
|
|
for transaction in transactions:
|
|
if transaction.transaction_type == TransactionType.EXPENSE:
|
|
date_key = transaction.transaction_date.date().isoformat()
|
|
if date_key not in daily_expenses:
|
|
daily_expenses[date_key] = 0
|
|
daily_expenses[date_key] += transaction.amount
|
|
|
|
return dict(sorted(daily_expenses.items()))
|
|
|
|
def get_month_comparison(self, family_id: int) -> Dict[str, float]:
|
|
"""Compare expenses: current month vs last month"""
|
|
today = datetime.utcnow()
|
|
current_month_start = today.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
|
|
|
|
# Last month
|
|
last_month_end = current_month_start - timedelta(days=1)
|
|
last_month_start = last_month_end.replace(day=1)
|
|
|
|
current_transactions = self.transaction_repo.get_transactions_by_period(
|
|
family_id, current_month_start, today
|
|
)
|
|
last_transactions = self.transaction_repo.get_transactions_by_period(
|
|
family_id, last_month_start, last_month_end
|
|
)
|
|
|
|
current_expenses = sum(
|
|
t.amount for t in current_transactions
|
|
if t.transaction_type == TransactionType.EXPENSE
|
|
)
|
|
last_expenses = sum(
|
|
t.amount for t in last_transactions
|
|
if t.transaction_type == TransactionType.EXPENSE
|
|
)
|
|
|
|
difference = current_expenses - last_expenses
|
|
percent_change = ((difference / last_expenses * 100) if last_expenses > 0 else 0)
|
|
|
|
return {
|
|
"current_month": current_expenses,
|
|
"last_month": last_expenses,
|
|
"difference": difference,
|
|
"percent_change": percent_change,
|
|
}
|