root/tgex-backend
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1 from TelegramExchange/feat/add-analytics-overview-endpoint

Browse Source 
feat: add analytics overview endpoint
This commit is contained in:
Artem
2026-01-07 10:53:34 +03:00
committed by GitHub
parent 7ee1edc4fc 1a5ce621bc
commit 5b46737536
6 changed files with 332 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/usecase/__init__.py
View File

@@ -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
@@ -210,6 +211,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: ...
@@ -224,6 +227,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]: ...
@@ -408,6 +419,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
View 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,
)