92 lines
3.5 KiB
Python
92 lines
3.5 KiB
Python
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
from src import domain, dto
|
|
from src.usecase.purchase.create_placements import _build_placement_output
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _build_placement_post_output(
|
|
placement_post: domain.PlacementPost, subscriptions_count: int
|
|
) -> dto.PlacementPostOutput | None:
|
|
placement = placement_post.placement
|
|
if not placement:
|
|
log.warning('PlacementPost %s missing placement data', placement_post.id)
|
|
return None
|
|
|
|
if not placement.channel or not placement.project or not placement.creative:
|
|
log.warning('PlacementPost %s placement missing related data', placement_post.id)
|
|
return None
|
|
|
|
ad_post_url = placement_post.post.url if placement_post.post else None
|
|
|
|
return dto.PlacementPostOutput(
|
|
id=placement_post.id,
|
|
project_id=placement.project_id,
|
|
project_channel_title=placement.project.channel.title if placement.project.channel else None,
|
|
placement_channel_id=placement.channel_id,
|
|
placement_channel_title=placement.channel.title,
|
|
creative_id=placement.creative_id,
|
|
creative_name=placement.creative.name,
|
|
placement_date=placement.placement_at or placement_post.created_at,
|
|
cost=placement.cost_value,
|
|
comment=placement.comment,
|
|
ad_post_url=ad_post_url,
|
|
invite_link_type=placement.invite_link_type,
|
|
invite_link=placement.invite_link,
|
|
status=placement_post.status,
|
|
subscriptions_count=subscriptions_count,
|
|
placement_id=placement.id,
|
|
placement=_build_placement_output(placement),
|
|
created_at=placement_post.created_at,
|
|
)
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
from .. import Usecase
|
|
|
|
|
|
async def get_placement_user(self: 'Usecase', input: dto.GetPlacementInput) -> dto.PlacementWithPostsOutput:
|
|
"""Get single placement by ID (user-managed)"""
|
|
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)
|
|
|
|
placement = await self.database.get_placement(input.workspace_id, input.placement_id)
|
|
if not placement or placement.project_id != project.id:
|
|
raise domain.PlacementNotFound(input.placement_id)
|
|
|
|
# Prefetch channel for output building
|
|
if placement.channel is None:
|
|
channel = await self.database.get_channel(channel_id=placement.channel_id)
|
|
placement.channel = channel
|
|
|
|
placement_posts = await self.database.get_workspace_placement_posts(
|
|
input.workspace_id,
|
|
placement_id=placement.id,
|
|
include_archived=True,
|
|
)
|
|
placement_post_ids = [post.id for post in placement_posts]
|
|
subscriptions_counts = await self.database.count_subscriptions_by_placement_post_batch(placement_post_ids)
|
|
|
|
placement_post_outputs = []
|
|
for placement_post in placement_posts:
|
|
placement_post_output = _build_placement_post_output(
|
|
placement_post, subscriptions_counts.get(placement_post.id, 0)
|
|
)
|
|
if placement_post_output is not None:
|
|
placement_post_outputs.append(placement_post_output)
|
|
|
|
placement_output = _build_placement_output(placement)
|
|
return dto.PlacementWithPostsOutput(
|
|
**placement_output.model_dump(),
|
|
placement_posts=placement_post_outputs,
|
|
)
|