root/tgex-backend
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

размещения ручки

Browse Source
This commit is contained in:
Artem Tsyrulnikov
2026-01-20 20:15:18 +03:00
parent 33c9121571
commit e8ff317566
35 changed files with 949 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
src/usecase/purchase/get_placement.py
View File

@@ -7,39 +7,40 @@ from src.usecase.purchase.create_placements import _build_placement_output
log = logging.getLogger(__name__)
def _build_post_output(post: domain.Post) -> dto.PostOutput:
channel = post.channel
if not channel:
raise ValueError(f'Post {post.id} has no channel')
return dto.PostOutput(
id=post.id,
message_id=post.message_id,
text=post.text,
url=post.url,
deleted_from_channel_at=post.deleted_from_channel_at,
created_at=post.created_at,
updated_at=post.updated_at,
)
def _build_placement_post_output(
placement_post: domain.PlacementPost, subscriptions_count: int
placement_post: domain.PlacementPost, subscriptions_count: int, views_count: int | None
) -> dto.PlacementPostOutput | None:
placement = placement_post.placement
if not placement:
log.warning('PlacementPost %s missing placement data', placement_post.id)
if not placement_post.post:
log.warning('PlacementPost %s missing post', 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)
try:
post_output = _build_post_output(placement_post.post)
except ValueError:
log.warning('PlacementPost %s post missing channel 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),
views_count=views_count,
created_at=placement_post.created_at,
post=post_output,
)
@@ -75,17 +76,27 @@ async def get_placement_user(self: 'Usecase', input: dto.GetPlacementInput) -> d
)
placement_post_ids = [post.id for post in placement_posts]
subscriptions_counts = await self.database.count_subscriptions_by_placement_post_batch(placement_post_ids)
post_ids = [placement_post.post.id for placement_post in placement_posts if placement_post.post]
views_map = await self.database.get_latest_views_data_batch(post_ids) if post_ids else {}
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_post_output = None
if placement_posts:
if len(placement_posts) > 1:
log.warning('Placement %s has %s placement_posts, returning latest', placement.id, len(placement_posts))
for placement_post in placement_posts:
views_count = None
if placement_post.post:
views_count = views_map.get(placement_post.post.id, (None,))[0]
placement_post_output = _build_placement_post_output(
placement_post,
subscriptions_counts.get(placement_post.id, 0),
views_count,
)
if placement_post_output is not None:
break
placement_output = _build_placement_output(placement)
return dto.PlacementWithPostsOutput(
**placement_output.model_dump(),
placement_posts=placement_post_outputs,
placement_post=placement_post_output,
)