diff --git a/src/adapter/postgres.py b/src/adapter/postgres.py index fb1761e..fa934fb 100644 --- a/src/adapter/postgres.py +++ b/src/adapter/postgres.py @@ -388,6 +388,8 @@ class Postgres(DatabaseBase, Database): creative_id: uuid.UUID | None = None, include_archived: bool = False, allowed_project_ids: set[uuid.UUID] | None = None, + date_from: datetime.datetime | None = None, + date_to: datetime.datetime | None = None, ) -> list[domain.Placement]: query = domain.Placement.filter(workspace_id=workspace_id) @@ -402,6 +404,11 @@ class Postgres(DatabaseBase, Database): if creative_id: query = query.filter(creative_id=creative_id) + if date_from: + query = query.filter(wanted_placement_date__gte=date_from) + if date_to: + query = query.filter(wanted_placement_date__lte=date_to) + if not include_archived: query = query.filter(status=domain.PlacementStatus.ACTIVE) @@ -409,7 +416,7 @@ class Postgres(DatabaseBase, Database): await query.prefetch_related( 'project', 'project__channel', 'placement_channel', 'creative', 'post', 'post__channel' ) - .order_by('-placement_date') + .order_by('-wanted_placement_date') .all() ) @@ -432,6 +439,25 @@ class Postgres(DatabaseBase, Database): async def update_subscription(self, subscription: domain.Subscription) -> None: await subscription.save() + async def get_subscriptions_for_placements( + self, + placement_ids: list[uuid.UUID], + *, + date_from: datetime.datetime | None = None, + date_to: datetime.datetime | None = None, + ) -> list[domain.Subscription]: + if not placement_ids: + return [] + + query = domain.Subscription.filter(placement_id__in=placement_ids) + + if date_from: + query = query.filter(created_at__gte=date_from) + if date_to: + query = query.filter(created_at__lte=date_to) + + return await query.all() + async def get_active_subscriptions_by_subscriber_and_project( self, subscriber_id: uuid.UUID, project_id: uuid.UUID ) -> list[domain.Subscription]: diff --git a/src/controller/http_v1/analytics.py b/src/controller/http_v1/analytics.py index 81bb42c..c039bec 100644 --- a/src/controller/http_v1/analytics.py +++ b/src/controller/http_v1/analytics.py @@ -75,3 +75,21 @@ async def get_spending_analytics( grouping=grouping, ) return await deps.get_usecase().get_spending_analytics(input) + + +@analytics_router.get('/overview') +async def get_overview_analytics( + workspace_id: uuid.UUID, + current_user: Annotated[JWTPayload, Depends(deps.get_current_user)], + date_from: datetime.datetime, + date_to: datetime.datetime, + project_id: uuid.UUID | None = None, +) -> dto.GetOverviewAnalyticsOutput: + input = dto.GetOverviewAnalyticsInput( + user_id=current_user.user_id, + workspace_id=workspace_id, + date_from=date_from, + date_to=date_to, + project_id=project_id, + ) + return await deps.get_usecase().get_overview_analytics(input) diff --git a/src/dto/__init__.py b/src/dto/__init__.py index b6805dd..a804cdd 100644 --- a/src/dto/__init__.py +++ b/src/dto/__init__.py @@ -50,6 +50,12 @@ __all__ = ( 'GetChannelAnalyticsInput', 'GetChannelAnalyticsOutput', 'SpendingDataPoint', + 'NumberWithDelta', + 'OverviewDailyPoint', + 'OverviewChannelPerformance', + 'OverviewProjectSpending', + 'GetOverviewAnalyticsInput', + 'GetOverviewAnalyticsOutput', 'GetSpendingAnalyticsInput', 'GetSpendingAnalyticsOutput', 'WorkspaceMembershipOutput', @@ -82,9 +88,15 @@ from .analytics import ( GetCreativesAnalyticsOutput, GetPlacementsAnalyticsInput, GetPlacementsAnalyticsOutput, + GetOverviewAnalyticsInput, GetSpendingAnalyticsInput, GetSpendingAnalyticsOutput, + GetOverviewAnalyticsOutput, + NumberWithDelta, PlacementAnalyticsOutput, + OverviewDailyPoint, + OverviewChannelPerformance, + OverviewProjectSpending, SpendingDataPoint, ) from .channel import ( diff --git a/src/dto/analytics.py b/src/dto/analytics.py index ed3b452..7f76263 100644 --- a/src/dto/analytics.py +++ b/src/dto/analytics.py @@ -107,3 +107,53 @@ class GetSpendingAnalyticsOutput(pydantic.BaseModel): avg_cpf: float | None avg_cpm: float | None chart_data: list[SpendingDataPoint] + + +class NumberWithDelta(pydantic.BaseModel): + value: float | int | None + delta_percent: float | None + + +class OverviewDailyPoint(pydantic.BaseModel): + date: datetime.date + cost: float + subscriptions: int + subscriptions_delta: int | None = None + cpf: float | None + + +class OverviewChannelPerformance(pydantic.BaseModel): + channel_id: uuid.UUID + title: str + username: str | None + cpf: float | None + total_cost: float + subscriptions: int + + +class OverviewProjectSpending(pydantic.BaseModel): + project_id: uuid.UUID + project_title: str + project_username: str | None + total_cost: float + + +class GetOverviewAnalyticsInput(pydantic.BaseModel): + user_id: uuid.UUID + workspace_id: uuid.UUID + date_from: datetime.datetime + date_to: datetime.datetime + project_id: uuid.UUID | None = None + + +class GetOverviewAnalyticsOutput(pydantic.BaseModel): + total_cost: NumberWithDelta + total_reach: NumberWithDelta + placements_count: NumberWithDelta + subscriptions_count: NumberWithDelta + avg_cpm: NumberWithDelta + avg_cpf: NumberWithDelta + daily_stats: list[OverviewDailyPoint] + top_channels_by_cpf: list[OverviewChannelPerformance] + worst_channels_by_cpf: list[OverviewChannelPerformance] + project_spending: list[OverviewProjectSpending] diff --git a/src/usecase/__init__.py b/src/usecase/__init__.py index 3a4bd4b..86806f1 100644 --- a/src/usecase/__init__.py +++ b/src/usecase/__init__.py @@ -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 diff --git a/src/usecase/analytics/get_overview_analytics.py b/src/usecase/analytics/get_overview_analytics.py new file mode 100644 index 0000000..3a6dfa5 --- /dev/null +++ b/src/usecase/analytics/get_overview_analytics.py @@ -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, + )