136 lines
4.1 KiB
Python
136 lines
4.1 KiB
Python
import datetime
|
|
import logging
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass
|
|
from typing import TYPE_CHECKING
|
|
|
|
from src import domain, dto
|
|
|
|
if TYPE_CHECKING:
|
|
from .. import Usecase
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _format_period(dt: 'datetime.datetime', grouping: dto.DateGrouping) -> str:
|
|
match grouping:
|
|
case dto.DateGrouping.DAY:
|
|
return str(dt.strftime('%Y-%m-%d'))
|
|
case dto.DateGrouping.WEEK:
|
|
# ISO week
|
|
return str(dt.strftime('%Y-W%W'))
|
|
case dto.DateGrouping.MONTH:
|
|
return str(dt.strftime('%Y-%m'))
|
|
case dto.DateGrouping.QUARTER:
|
|
quarter = (dt.month - 1) // 3 + 1
|
|
return f'{dt.year}-Q{quarter}'
|
|
case dto.DateGrouping.YEAR:
|
|
return str(dt.year)
|
|
|
|
raise ValueError('Invalid date grouping')
|
|
|
|
|
|
async def get_spending_analytics(
|
|
self: 'Usecase', input: dto.GetSpendingAnalyticsInput
|
|
) -> dto.GetSpendingAnalyticsOutput:
|
|
context = await self.ensure_workspace_permission(
|
|
input.workspace_id, input.user_id, domain.PermissionKey.ANALYTICS_READ
|
|
)
|
|
|
|
allowed_project_ids = context.allowed_project_ids(domain.PermissionKey.ANALYTICS_READ)
|
|
|
|
if input.project_id:
|
|
project = await self.database.get_project(input.workspace_id, input.project_id)
|
|
if not project:
|
|
raise domain.ProjectNotFound(input.project_id)
|
|
context.ensure_project_permission(domain.PermissionKey.ANALYTICS_READ, project.id)
|
|
allowed_project_ids = None
|
|
|
|
placements = await self.database.get_workspace_placements(
|
|
input.workspace_id,
|
|
input.project_id,
|
|
include_archived=False,
|
|
allowed_project_ids=allowed_project_ids,
|
|
)
|
|
|
|
filtered = [
|
|
p
|
|
for p in placements
|
|
if (not input.date_from or p.placement_date >= input.date_from)
|
|
and (not input.date_to or p.placement_date <= input.date_to)
|
|
]
|
|
|
|
total_cost = 0.0
|
|
total_subs = 0
|
|
total_views = 0
|
|
|
|
@dataclass
|
|
class PeriodData:
|
|
cost: float = 0.0
|
|
subscriptions: int = 0
|
|
views: int = 0
|
|
|
|
# Batch fetch views data for all posts
|
|
post_ids = [p.post.id for p in filtered if p.post]
|
|
views_map = await self.database.get_latest_views_data_batch(post_ids) if post_ids else {}
|
|
|
|
# Batch fetch subscriptions counts
|
|
placement_ids = [p.id for p in filtered]
|
|
subscriptions_counts = await self.database.count_subscriptions_by_placement_batch(placement_ids)
|
|
|
|
# Группировка по периодам
|
|
period_data: dict[str, PeriodData] = defaultdict(PeriodData)
|
|
|
|
for p in filtered:
|
|
period = _format_period(p.placement_date, input.grouping)
|
|
pd = period_data[period]
|
|
|
|
if p.cost is not None:
|
|
total_cost += p.cost
|
|
pd.cost += p.cost
|
|
|
|
subs_count = subscriptions_counts.get(p.id, 0)
|
|
total_subs += subs_count
|
|
pd.subscriptions += subs_count
|
|
|
|
# Get views from batch data
|
|
if p.post and p.post.id in views_map:
|
|
views_count = views_map[p.post.id][0]
|
|
total_views += views_count
|
|
pd.views += views_count
|
|
|
|
avg_cpf = total_cost / total_subs if total_subs > 0 and total_cost > 0 else None
|
|
avg_cpm = (total_cost / total_views * 1000) if total_views > 0 and total_cost > 0 else None
|
|
|
|
# Данные для графика
|
|
chart_data = []
|
|
for period in sorted(period_data.keys()):
|
|
d = period_data[period]
|
|
|
|
cost = d.cost
|
|
subs = d.subscriptions
|
|
views = d.views
|
|
|
|
cpf = cost / subs if subs > 0 and cost > 0 else None
|
|
cpm = (cost / views * 1000) if views > 0 and cost > 0 else None
|
|
|
|
chart_data.append(
|
|
dto.SpendingDataPoint(
|
|
period=period,
|
|
cost=cost,
|
|
subscriptions=subs,
|
|
views=views,
|
|
cpf=cpf,
|
|
cpm=cpm,
|
|
)
|
|
)
|
|
|
|
return dto.GetSpendingAnalyticsOutput(
|
|
total_cost=total_cost,
|
|
total_subscriptions=total_subs,
|
|
total_views=total_views,
|
|
avg_cpf=avg_cpf,
|
|
avg_cpm=avg_cpm,
|
|
chart_data=chart_data,
|
|
)
|