feat: add analytics overview endpoint
This commit is contained in:
@@ -10,6 +10,7 @@ from src import domain
|
||||
|
||||
from .analytics.get_channel_analytics import get_channel_analytics
|
||||
from .analytics.get_creatives_analytics import get_creatives_analytics
|
||||
from .analytics.get_overview_analytics import get_overview_analytics
|
||||
from .analytics.get_placements_analytics import get_placements_analytics
|
||||
from .analytics.get_spending_analytics import get_spending_analytics
|
||||
from .auth.create_telegram_login_token import create_telegram_login_token
|
||||
@@ -206,6 +207,8 @@ class Database(typing.Protocol):
|
||||
creative_id: UUID | None = None,
|
||||
include_archived: bool = False,
|
||||
allowed_project_ids: set[UUID] | None = None,
|
||||
date_from: datetime.datetime | None = None,
|
||||
date_to: datetime.datetime | None = None,
|
||||
) -> list[domain.Placement]: ...
|
||||
|
||||
async def delete_placement(self, placement_id: UUID) -> None: ...
|
||||
@@ -220,6 +223,14 @@ class Database(typing.Protocol):
|
||||
|
||||
async def update_subscription(self, subscription: domain.Subscription) -> None: ...
|
||||
|
||||
async def get_subscriptions_for_placements(
|
||||
self,
|
||||
placement_ids: list[UUID],
|
||||
*,
|
||||
date_from: datetime.datetime | None = None,
|
||||
date_to: datetime.datetime | None = None,
|
||||
) -> list[domain.Subscription]: ...
|
||||
|
||||
async def get_active_subscriptions_by_subscriber_and_project(
|
||||
self, subscriber_id: UUID, project_id: UUID
|
||||
) -> list[domain.Subscription]: ...
|
||||
@@ -391,6 +402,7 @@ class Usecase:
|
||||
get_creatives_analytics = get_creatives_analytics
|
||||
get_channel_analytics = get_channel_analytics
|
||||
get_spending_analytics = get_spending_analytics
|
||||
get_overview_analytics = get_overview_analytics
|
||||
get_workspaces = get_workspaces
|
||||
create_workspace = create_workspace
|
||||
update_workspace = update_workspace
|
||||
|
||||
213
src/usecase/analytics/get_overview_analytics.py
Normal file
213
src/usecase/analytics/get_overview_analytics.py
Normal file
@@ -0,0 +1,213 @@
|
||||
import datetime
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import HTTPException, status
|
||||
|
||||
from src import domain, dto
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .. import Usecase
|
||||
|
||||
|
||||
def _calc_delta(current: float | None, previous: float | None) -> float | None:
|
||||
if current is None or previous is None or previous == 0:
|
||||
return None
|
||||
return (current - previous) / previous * 100
|
||||
|
||||
|
||||
def _as_number_with_delta(value: float | None, previous: float | None) -> dto.NumberWithDelta:
|
||||
return dto.NumberWithDelta(value=value, delta_percent=_calc_delta(value, previous))
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelAggregate:
|
||||
channel: domain.Channel | None
|
||||
total_cost: float = 0.0
|
||||
total_subs: int = 0
|
||||
|
||||
|
||||
async def get_overview_analytics(
|
||||
self: 'Usecase', input: dto.GetOverviewAnalyticsInput
|
||||
) -> dto.GetOverviewAnalyticsOutput:
|
||||
if input.date_from > input.date_to:
|
||||
raise HTTPException(status.HTTP_400_BAD_REQUEST, 'date_from must be before date_to')
|
||||
|
||||
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
|
||||
|
||||
raw_duration = input.date_to - input.date_from
|
||||
if raw_duration.total_seconds() < 0:
|
||||
raise HTTPException(status.HTTP_400_BAD_REQUEST, 'date_from must be before date_to')
|
||||
|
||||
period_length = raw_duration if raw_duration.total_seconds() > 0 else datetime.timedelta(days=1)
|
||||
previous_period_start = input.date_from - period_length
|
||||
|
||||
placements = await self.database.get_workspace_placements(
|
||||
input.workspace_id,
|
||||
project_id=input.project_id,
|
||||
include_archived=False,
|
||||
allowed_project_ids=allowed_project_ids,
|
||||
date_from=previous_period_start,
|
||||
date_to=input.date_to,
|
||||
)
|
||||
|
||||
current_placements: list[domain.Placement] = []
|
||||
previous_placements: list[domain.Placement] = []
|
||||
|
||||
for placement in placements:
|
||||
placement_date = placement.wanted_placement_date
|
||||
if placement_date >= input.date_from and placement_date <= input.date_to:
|
||||
current_placements.append(placement)
|
||||
elif placement_date >= previous_period_start and placement_date < input.date_from:
|
||||
previous_placements.append(placement)
|
||||
|
||||
placement_ids = [p.id for p in placements]
|
||||
subscriptions = await self.database.get_subscriptions_for_placements(
|
||||
placement_ids, date_from=previous_period_start, date_to=input.date_to
|
||||
)
|
||||
|
||||
subs_per_placement_current: dict[UUID, int] = defaultdict(int)
|
||||
subs_per_placement_previous: dict[UUID, int] = defaultdict(int)
|
||||
subs_per_day_current: dict[datetime.date, int] = defaultdict(int)
|
||||
|
||||
for sub in subscriptions:
|
||||
created_at = sub.created_at
|
||||
if created_at is None:
|
||||
continue
|
||||
|
||||
if created_at >= input.date_from and created_at <= input.date_to:
|
||||
subs_per_placement_current[sub.placement_id] += 1
|
||||
subs_per_day_current[created_at.date()] += 1
|
||||
elif created_at >= previous_period_start and created_at < input.date_from:
|
||||
subs_per_placement_previous[sub.placement_id] += 1
|
||||
|
||||
post_ids = [p.post.id for p in placements if p.post]
|
||||
views_map = await self.database.get_latest_views_data_batch(post_ids) if post_ids else {}
|
||||
|
||||
def _aggregate_totals(
|
||||
placement_list: list[domain.Placement], subs_per_placement: dict[UUID, int]
|
||||
) -> tuple[float, int, int]:
|
||||
total_cost = 0.0
|
||||
total_subscriptions = 0
|
||||
total_views = 0
|
||||
|
||||
for placement in placement_list:
|
||||
if placement.cost is not None:
|
||||
total_cost += placement.cost
|
||||
|
||||
subs = subs_per_placement.get(placement.id, 0)
|
||||
total_subscriptions += subs
|
||||
|
||||
if placement.post and placement.post.id in views_map:
|
||||
total_views += views_map[placement.post.id][0]
|
||||
|
||||
return total_cost, total_subscriptions, total_views
|
||||
|
||||
current_cost, current_subs, current_views = _aggregate_totals(current_placements, subs_per_placement_current)
|
||||
previous_cost, previous_subs, previous_views = _aggregate_totals(previous_placements, subs_per_placement_previous)
|
||||
|
||||
current_avg_cpf = current_cost / current_subs if current_cost > 0 and current_subs > 0 else None
|
||||
previous_avg_cpf = previous_cost / previous_subs if previous_cost > 0 and previous_subs > 0 else None
|
||||
|
||||
current_avg_cpm = (current_cost / current_views * 1000) if current_cost > 0 and current_views > 0 else None
|
||||
previous_avg_cpm = (previous_cost / previous_views * 1000) if previous_cost > 0 and previous_views > 0 else None
|
||||
|
||||
cost_per_day: dict[datetime.date, float] = defaultdict(float)
|
||||
channel_stats: dict[UUID, ChannelAggregate] = {}
|
||||
project_spending_map: dict[UUID, float] = defaultdict(float)
|
||||
project_meta: dict[UUID, domain.Project] = {}
|
||||
|
||||
for placement in current_placements:
|
||||
placement_date = placement.wanted_placement_date.date()
|
||||
project_meta[placement.project_id] = placement.project
|
||||
|
||||
cost_value = placement.cost or 0.0
|
||||
cost_per_day[placement_date] += cost_value
|
||||
project_spending_map[placement.project_id] += cost_value
|
||||
subs = subs_per_placement_current.get(placement.id, 0)
|
||||
|
||||
channel_id = placement.placement_channel_id
|
||||
if channel_id not in channel_stats:
|
||||
channel_stats[channel_id] = ChannelAggregate(channel=placement.placement_channel)
|
||||
channel_stats[channel_id].total_cost += cost_value
|
||||
channel_stats[channel_id].total_subs += subs
|
||||
|
||||
start_date = input.date_from.date()
|
||||
end_date = input.date_to.date()
|
||||
days_span = (end_date - start_date).days
|
||||
|
||||
daily_stats: list[dto.OverviewDailyPoint] = []
|
||||
previous_day_subs: int | None = None
|
||||
for day_offset in range(days_span + 1):
|
||||
day = start_date + datetime.timedelta(days=day_offset)
|
||||
cost = cost_per_day.get(day, 0.0)
|
||||
subs = subs_per_day_current.get(day, 0)
|
||||
delta = (subs - previous_day_subs) if previous_day_subs is not None else subs
|
||||
cpf = cost / subs if subs > 0 and cost > 0 else None
|
||||
daily_stats.append(
|
||||
dto.OverviewDailyPoint(date=day, cost=cost, subscriptions=subs, subscriptions_delta=delta, cpf=cpf)
|
||||
)
|
||||
previous_day_subs = subs
|
||||
|
||||
channel_performance = []
|
||||
for stats in channel_stats.values():
|
||||
subs = stats.total_subs
|
||||
cost_value = stats.total_cost
|
||||
cpf = cost_value / subs if subs > 0 else None
|
||||
channel = stats.channel
|
||||
|
||||
if not channel or cpf is None:
|
||||
continue
|
||||
|
||||
channel_performance.append(
|
||||
dto.OverviewChannelPerformance(
|
||||
channel_id=channel.id,
|
||||
title=channel.title,
|
||||
username=channel.username,
|
||||
cpf=cpf,
|
||||
total_cost=cost_value,
|
||||
subscriptions=subs,
|
||||
)
|
||||
)
|
||||
|
||||
top_channels = sorted(channel_performance, key=lambda c: c.cpf or float('inf'))[:5]
|
||||
worst_channels = sorted(channel_performance, key=lambda c: c.cpf or float('inf'), reverse=True)[:5]
|
||||
|
||||
project_spending = []
|
||||
for project_id, total_cost in project_spending_map.items():
|
||||
project = project_meta.get(project_id)
|
||||
project_spending.append(
|
||||
dto.OverviewProjectSpending(
|
||||
project_id=project_id,
|
||||
project_title=project.channel.title if project and project.channel else 'Unnamed project',
|
||||
project_username=project.channel.username if project and project.channel else None,
|
||||
total_cost=total_cost,
|
||||
)
|
||||
)
|
||||
project_spending = sorted(project_spending, key=lambda p: p.total_cost, reverse=True)
|
||||
|
||||
return dto.GetOverviewAnalyticsOutput(
|
||||
total_cost=_as_number_with_delta(current_cost, previous_cost),
|
||||
total_reach=_as_number_with_delta(current_views, previous_views),
|
||||
placements_count=_as_number_with_delta(len(current_placements), len(previous_placements)),
|
||||
subscriptions_count=_as_number_with_delta(current_subs, previous_subs),
|
||||
avg_cpm=_as_number_with_delta(current_avg_cpm, previous_avg_cpm),
|
||||
avg_cpf=_as_number_with_delta(current_avg_cpf, previous_avg_cpf),
|
||||
daily_stats=daily_stats,
|
||||
top_channels_by_cpf=top_channels,
|
||||
worst_channels_by_cpf=worst_channels,
|
||||
project_spending=project_spending,
|
||||
)
|
||||
Reference in New Issue
Block a user