feat: аналитика

src/usecase/analytics/get_creatives_analytics.py

@@ -0,0 +1,78 @@
+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
+) -> dto.GetCreativesAnalyticsOutput:
+    async with self.database.transaction():
+        if input.target_channel_id:
+            target_channel = await self.database.get_target_channel(input.user_id, input.target_channel_id)
+            if not target_channel:
+                raise domain.TargetChannelNotFound(input.target_channel_id)
+
+        creatives = await self.database.get_user_creatives(
+            input.user_id, input.target_channel_id, include_archived=False
+        )
+
+        placements = await self.database.get_user_placements(
+            input.user_id, input.target_channel_id, include_archived=False
+        )
+
+        @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}
+
+        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 += p.subscriptions_count
+            stats.total_views += p.views_count if p.views_count is not None else 0
+            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 dto.GetCreativesAnalyticsOutput(creatives=result)