175 lines
6.7 KiB
Python
175 lines
6.7 KiB
Python
import logging
|
|
import uuid
|
|
from typing import TYPE_CHECKING
|
|
|
|
from src import domain, dto
|
|
|
|
if TYPE_CHECKING:
|
|
from .. import Usecase
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _build_cost_info(cost_type: domain.purchase.CostType | None, cost_value: float | None) -> dto.CostInfo | None:
|
|
if cost_type is None or cost_value is None:
|
|
return None
|
|
return dto.CostInfo(type=cost_type, value=cost_value)
|
|
|
|
|
|
def _build_purchase_details(purchase: domain.Purchase) -> dto.PurchaseDetails | None:
|
|
details = dto.PurchaseDetails(
|
|
placement_at=purchase.placement_at,
|
|
payment_at=purchase.payment_at,
|
|
cost=_build_cost_info(purchase.cost_type, purchase.cost_value),
|
|
cost_before_bargain=purchase.cost_before_bargain,
|
|
purchase_type=purchase.purchase_type,
|
|
format=purchase.format,
|
|
comment=purchase.comment,
|
|
)
|
|
if details.model_dump(exclude_none=True):
|
|
return details
|
|
return None
|
|
|
|
|
|
def _build_purchase_channel_details(purchase_channel: domain.PurchaseChannel) -> dto.PurchaseChannelDetails | None:
|
|
details = dto.PurchaseChannelDetails(
|
|
placement_at=purchase_channel.placement_at,
|
|
cost=_build_cost_info(purchase_channel.cost_type, purchase_channel.cost_value),
|
|
cost_before_bargain=purchase_channel.cost_before_bargain,
|
|
format=purchase_channel.format,
|
|
)
|
|
if details.model_dump(exclude_none=True):
|
|
return details
|
|
return None
|
|
|
|
|
|
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,
|
|
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,
|
|
),
|
|
details=_build_purchase_channel_details(purchase_channel),
|
|
)
|
|
)
|
|
|
|
return dto.PurchaseOutput(
|
|
id=purchase.id,
|
|
status=purchase.status,
|
|
creative_id=purchase.creative_id,
|
|
channels=channels_output,
|
|
details=_build_purchase_details(purchase),
|
|
)
|
|
|
|
|
|
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)
|
|
|
|
details = input.details
|
|
purchase = domain.Purchase(
|
|
workspace_id=workspace_id,
|
|
project_id=project.id,
|
|
creative_id=creative.id,
|
|
placement_at=details.placement_at if details else None,
|
|
payment_at=details.payment_at if details else None,
|
|
cost_type=details.cost.type if details and details.cost else None,
|
|
cost_value=details.cost.value if details and details.cost else None,
|
|
cost_before_bargain=details.cost_before_bargain if details else None,
|
|
purchase_type=details.purchase_type if details else None,
|
|
format=details.format if details else None,
|
|
comment=details.comment if details else None,
|
|
)
|
|
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:
|
|
parser_response = await self.parser.fetch_telegram_channel(channel_input.username)
|
|
if not parser_response:
|
|
raise domain.TelegramChannelNotFound(channel_input.username)
|
|
|
|
channel = domain.Channel(
|
|
username=parser_response.username,
|
|
telegram_id=parser_response.telegram_id,
|
|
title=parser_response.title,
|
|
)
|
|
await self.database.create_channel(channel)
|
|
log.info('Created channel @%s with telegram_id=%s', channel.username, channel.telegram_id)
|
|
|
|
invite_link = await self.telegram_bot.create_chat_invite_link(project.channel.telegram_id, requires_approval)
|
|
log.info(
|
|
'Created invite link for purchase %s channel_telegram_id=%s requires_approval=%s invite_link=%s',
|
|
purchase.id,
|
|
project.channel.telegram_id,
|
|
requires_approval,
|
|
invite_link,
|
|
)
|
|
|
|
channel_details = channel_input.details
|
|
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,
|
|
comment=channel_input.comment,
|
|
placement_at=channel_details.placement_at if channel_details else None,
|
|
cost_type=channel_details.cost.type if channel_details and channel_details.cost else None,
|
|
cost_value=channel_details.cost.value if channel_details and channel_details.cost else None,
|
|
cost_before_bargain=channel_details.cost_before_bargain if channel_details else None,
|
|
format=channel_details.format if channel_details else None,
|
|
)
|
|
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)
|