64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
from datetime import date
|
|
|
|
from sqlalchemy import select, func, Subquery, cast, CTE
|
|
from sqlalchemy.dialects.postgresql import TIMESTAMP
|
|
|
|
from models import PaymentRecord
|
|
from schemas.statistics import CommonProfitFilters
|
|
from services.base import BaseService
|
|
from services.statistics.common import generate_date_range
|
|
|
|
|
|
class PaymentStatisticsService(BaseService):
|
|
date_from: date
|
|
date_to: date
|
|
|
|
@staticmethod
|
|
def _fill_date_gaps(payments: Subquery, all_dates: CTE) -> Subquery:
|
|
return (
|
|
select(
|
|
all_dates.c.date,
|
|
(all_dates.c.expenses + func.coalesce(payments.c.expenses, 0)).label("expenses"),
|
|
)
|
|
.join(payments, all_dates.c.date == payments.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
|
|
|
|
@staticmethod
|
|
def _apply_payments(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_payments(self, filters: CommonProfitFilters, deals_by_dates: Subquery):
|
|
self.date_from, self.date_to = filters.date_range
|
|
|
|
salary_expenses = self._get_payment_records_sub()
|
|
deals_by_dates = self._apply_payments(deals_by_dates, salary_expenses)
|
|
|
|
return deals_by_dates
|