98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
from datetime import date
|
|
|
|
from sqlalchemy import select, func, Subquery, cast, CTE
|
|
from sqlalchemy.dialects.postgresql import TIMESTAMP
|
|
|
|
from models import PaymentRecord, Expense, expenses_expense_tags
|
|
from schemas.statistics import CommonProfitFilters
|
|
from services.base import BaseService
|
|
from services.statistics.common import generate_date_range
|
|
|
|
|
|
class ExpensesStatisticsService(BaseService):
|
|
date_from: date
|
|
date_to: date
|
|
|
|
@staticmethod
|
|
def _fill_date_gaps(expenses: Subquery, all_dates: CTE) -> Subquery:
|
|
return (
|
|
select(
|
|
all_dates.c.date,
|
|
(all_dates.c.expenses + func.coalesce(expenses.c.expenses, 0)).label("expenses"),
|
|
)
|
|
.join(expenses, all_dates.c.date == expenses.c.date, isouter=True)
|
|
.order_by(all_dates.c.date)
|
|
.subquery()
|
|
)
|
|
|
|
def _get_payment_records_sub(self) -> Subquery:
|
|
all_dates = generate_date_range(self.date_from, self.date_to, ["expenses"])
|
|
|
|
expenses = (
|
|
select(
|
|
func.sum(PaymentRecord.amount).label("expenses"),
|
|
cast(PaymentRecord.start_date, TIMESTAMP(timezone=False)).label("date"),
|
|
)
|
|
.group_by("date")
|
|
.subquery()
|
|
)
|
|
|
|
expenses_with_filled_gaps = self._fill_date_gaps(expenses, all_dates)
|
|
return expenses_with_filled_gaps
|
|
|
|
def _get_additional_expenses_sub(self, tag_id: int) -> Subquery:
|
|
all_dates = generate_date_range(self.date_from, self.date_to, ["expenses"])
|
|
|
|
expenses = (
|
|
select(Expense)
|
|
)
|
|
|
|
if tag_id != -1:
|
|
expenses = (
|
|
expenses
|
|
.join(expenses_expense_tags)
|
|
.where(expenses_expense_tags.c.expense_tag_id == tag_id)
|
|
)
|
|
|
|
expenses = expenses.subquery()
|
|
|
|
expenses = (
|
|
select(
|
|
func.sum(expenses.c.amount).label("expenses"),
|
|
cast(expenses.c.spent_date, TIMESTAMP(timezone=False)).label("date"),
|
|
)
|
|
.where(expenses.c.spent_date.between(self.date_from, self.date_to))
|
|
.group_by("date")
|
|
.subquery()
|
|
)
|
|
|
|
expenses_with_filled_gaps = self._fill_date_gaps(expenses, all_dates)
|
|
return expenses_with_filled_gaps
|
|
|
|
@staticmethod
|
|
def _apply_expenses(deals_by_dates: Subquery, expenses_subquery: Subquery):
|
|
return (
|
|
select(
|
|
deals_by_dates.c.date,
|
|
deals_by_dates.c.deals_count,
|
|
deals_by_dates.c.revenue,
|
|
(func.coalesce(deals_by_dates.c.profit, 0) - func.coalesce(expenses_subquery.c.expenses, 0)).label(
|
|
"profit"),
|
|
(deals_by_dates.c.expenses + expenses_subquery.c.expenses).label("expenses"),
|
|
)
|
|
.join(expenses_subquery, expenses_subquery.c.date == deals_by_dates.c.date)
|
|
)
|
|
|
|
def apply_expenses(self, filters: CommonProfitFilters, deals_by_dates: Subquery):
|
|
self.date_from, self.date_to = filters.date_range
|
|
|
|
# Apply salary expenses
|
|
salary_expenses = self._get_payment_records_sub()
|
|
deals_by_dates = self._apply_expenses(deals_by_dates, salary_expenses)
|
|
|
|
# Apply additional expenses
|
|
additional_expenses = self._get_additional_expenses_sub(filters.tag_id)
|
|
deals_by_dates = self._apply_expenses(deals_by_dates, additional_expenses)
|
|
|
|
return deals_by_dates
|