change domen

src/usecase/purchase/create_purchase.py

@@ -0,0 +1,117 @@
+import logging
+import random
+import uuid
+from typing import TYPE_CHECKING
+
+from src import domain, dto
+
+if TYPE_CHECKING:
+    from .. import Usecase
+
+log = logging.getLogger(__name__)
+
+
+def _build_purchase_output(
+    purchase: domain.Purchase, purchase_channels: list[domain.PurchaseChannel]
+) -> dto.PurchaseOutput:
+    channels_output: list[dto.PurchaseChannelOutput] = []
+    for purchase_channel in purchase_channels:
+        channel = purchase_channel.channel
+        if channel is None:
+            log.warning('Purchase channel %s has no channel prefetched', purchase_channel.id)
+            continue
+        channels_output.append(
+            dto.PurchaseChannelOutput(
+                id=purchase_channel.id,
+                status=purchase_channel.status,
+                planned_cost=purchase_channel.planned_cost,
+                comment=purchase_channel.comment,
+                invite_link=purchase_channel.invite_link,
+                invite_link_type=purchase_channel.invite_link_type,
+                channel=dto.ChannelOutput(
+                    id=channel.id,
+                    telegram_id=channel.telegram_id,
+                    title=channel.title,
+                    username=channel.username,
+                ),
+            )
+        )
+
+    return dto.PurchaseOutput(
+        id=purchase.id,
+        status=purchase.status,
+        creative_id=purchase.creative_id,
+        channels=channels_output,
+    )
+
+
+async def create_purchase(
+    self: 'Usecase',
+    project_id: uuid.UUID,
+    workspace_id: uuid.UUID,
+    user_id: uuid.UUID,
+    input: dto.CreatePurchaseInput,
+) -> dto.PurchaseOutput:
+    context = await self.ensure_workspace_permission(workspace_id, user_id, domain.PermissionKey.PLACEMENTS_WRITE)
+
+    project = await self.database.get_project(workspace_id, project_id=project_id)
+    if not project:
+        raise domain.ProjectNotFound(project_id)
+
+    context.ensure_project_permission(domain.PermissionKey.PLACEMENTS_WRITE, project.id)
+
+    creative = await self.database.get_creative(workspace_id, input.creative_id)
+    if not creative or creative.project_id != project.id:
+        raise domain.CreativeNotFound(input.creative_id)
+
+    existing_purchase = await self.database.get_active_purchase(project.id, creative.id)
+    if existing_purchase:
+        channels = await self.database.get_purchase_channels(existing_purchase.id)
+        return _build_purchase_output(existing_purchase, channels)
+
+    purchase = domain.Purchase(workspace_id=workspace_id, project_id=project.id, creative_id=creative.id)
+    await self.database.create_purchase(purchase)
+
+    if project.channel.telegram_id is None:
+        raise domain.ChannelNotFound(project.channel.id)
+
+    invite_link_type = project.purchase_invite_type_default
+    requires_approval = invite_link_type == domain.purchase.InviteLinkType.APPROVAL
+
+    purchase_channels: list[domain.PurchaseChannel] = []
+
+    for channel_input in input.channels:
+        channel = await self.database.get_channel(username=channel_input.username)
+
+        if not channel:
+            channel = domain.Channel(
+                username=channel_input.username,
+                telegram_id=random.randint(1, 10000),
+                title='parser_response.title',
+            )
+            await self.database.create_channel(channel)
+            log.info('Created channel @%s with telegram_id=%s', channel_input.username, channel.telegram_id)
+
+        invite_link = await self.telegram_bot.create_chat_invite_link(project.channel.telegram_id, requires_approval)
+
+        purchase_channel = await self.database.add_channel_to_purchase(
+            purchase.id,
+            channel.id,
+            invite_link=invite_link,
+            invite_link_type=invite_link_type,
+            status=channel_input.status,
+            planned_cost=channel_input.planned_cost,
+            comment=channel_input.comment,
+        )
+        purchase_channel.channel = channel
+        purchase_channels.append(purchase_channel)
+
+    log.info(
+        'Purchase %s created for project %s (creative %s) with %s channels',
+        purchase.id,
+        project.id,
+        creative.id,
+        len(purchase_channels),
+    )
+
+    return _build_purchase_output(purchase, purchase_channels)

src/usecase/purchase/get_purchase.py

@@ -0,0 +1,30 @@
+import logging
+from typing import TYPE_CHECKING
+
+from src import domain, dto
+from .create_purchase import _build_purchase_output
+
+if TYPE_CHECKING:
+    from .. import Usecase
+
+log = logging.getLogger(__name__)
+
+
+async def get_purchase(self: 'Usecase', input: dto.GetPurchaseInput) -> dto.PurchaseOutput:
+    context = await self.ensure_workspace_permission(
+        input.workspace_id, input.user_id, domain.PermissionKey.PLACEMENTS_READ
+    )
+
+    project = await self.database.get_project(input.workspace_id, project_id=input.project_id)
+    if not project:
+        raise domain.ProjectNotFound(input.project_id)
+
+    context.ensure_project_permission(domain.PermissionKey.PLACEMENTS_READ, project.id)
+
+    purchase = await self.database.get_purchase(input.workspace_id, input.purchase_id)
+    if not purchase or purchase.project_id != project.id:
+        raise domain.PurchaseNotFound(input.purchase_id)
+
+    channels = await self.database.get_purchase_channels(purchase.id)
+
+    return _build_purchase_output(purchase, channels)

src/usecase/purchase/get_purchases.py

@@ -0,0 +1,32 @@
+import logging
+from typing import TYPE_CHECKING
+
+from src import domain, dto
+from .create_purchase import _build_purchase_output
+
+if TYPE_CHECKING:
+    from .. import Usecase
+
+log = logging.getLogger(__name__)
+
+
+async def get_purchases(self: 'Usecase', input: dto.GetPurchasesInput) -> list[dto.PurchaseOutput]:
+    context = await self.ensure_workspace_permission(
+        input.workspace_id, input.user_id, domain.PermissionKey.PLACEMENTS_READ
+    )
+
+    project = await self.database.get_project(input.workspace_id, project_id=input.project_id)
+    if not project:
+        raise domain.ProjectNotFound(input.project_id)
+
+    context.ensure_project_permission(domain.PermissionKey.PLACEMENTS_READ, project.id)
+
+    purchases = await self.database.get_project_purchases(input.workspace_id, project.id)
+    log.debug('Fetched %s purchases for project %s', len(purchases), project.id)
+
+    output: list[dto.PurchaseOutput] = []
+    for purchase in purchases:
+        channels = await self.database.get_purchase_channels(purchase.id)
+        output.append(_build_purchase_output(purchase, channels))
+
+    return output