102 lines
3.2 KiB
Python
102 lines
3.2 KiB
Python
import logging
|
|
from dataclasses import dataclass
|
|
from typing import TYPE_CHECKING
|
|
from uuid import UUID
|
|
|
|
from src import domain, dto
|
|
|
|
if TYPE_CHECKING:
|
|
from .. import Usecase
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
async def get_creatives_analytics(
|
|
self: 'Usecase', input: dto.GetCreativesAnalyticsInput
|
|
) -> list[dto.CreativeAnalyticsOutput]:
|
|
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
|
|
|
|
creatives = await self.database.get_workspace_creatives(
|
|
input.workspace_id,
|
|
input.project_id,
|
|
include_archived=False,
|
|
allowed_project_ids=allowed_project_ids,
|
|
)
|
|
placements = await self.database.get_workspace_placement_posts(
|
|
input.workspace_id,
|
|
input.project_id,
|
|
include_archived=False,
|
|
allowed_project_ids=allowed_project_ids,
|
|
)
|
|
|
|
@dataclass
|
|
class CreativeStats:
|
|
total_cost: float = 0.0
|
|
total_subscriptions: int = 0
|
|
total_views: int = 0
|
|
placements_count: int = 0
|
|
|
|
creative_stats: dict[UUID, CreativeStats] = {cr.id: CreativeStats() for cr in creatives}
|
|
|
|
# Batch fetch views data for all posts
|
|
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 {}
|
|
|
|
# Batch fetch subscriptions counts
|
|
placement_ids = [p.id for p in placements]
|
|
subscriptions_counts = await self.database.count_subscriptions_by_placement_post_batch(placement_ids)
|
|
|
|
for p in placements:
|
|
stats = creative_stats.get(p.creative_id)
|
|
if stats is None:
|
|
continue
|
|
|
|
stats.total_cost += p.cost if p.cost is not None else 0
|
|
stats.total_subscriptions += subscriptions_counts.get(p.id, 0)
|
|
|
|
# Get views from batch data
|
|
if p.post and p.post.id in views_map:
|
|
views_count = views_map[p.post.id][0]
|
|
stats.total_views += views_count
|
|
|
|
stats.placements_count += 1
|
|
|
|
result = []
|
|
for cr in creatives:
|
|
stats = creative_stats[cr.id]
|
|
|
|
avg_cpf = (
|
|
(stats.total_cost / stats.total_subscriptions)
|
|
if stats.total_subscriptions > 0 and stats.total_cost > 0
|
|
else None
|
|
)
|
|
avg_cpm = (
|
|
(stats.total_cost / stats.total_views * 1000) if stats.total_views > 0 and stats.total_cost > 0 else None
|
|
)
|
|
|
|
result.append(
|
|
dto.CreativeAnalyticsOutput(
|
|
id=cr.id,
|
|
name=cr.name,
|
|
placements_count=stats.placements_count,
|
|
total_cost=stats.total_cost,
|
|
total_subscriptions=stats.total_subscriptions,
|
|
total_views=stats.total_views,
|
|
avg_cpf=avg_cpf,
|
|
avg_cpm=avg_cpm,
|
|
)
|
|
)
|
|
|
|
return result
|