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-07 15:11:28 +03:00
parent f6669c3f5a
commit e4199fc9c9
30 changed files with 696 additions and 1201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/usecase/__init__.py
View File

@@ -175,6 +175,8 @@ class Database(typing.Protocol):
self, purchase_channel_id: UUID, purchase_id: UUID
) -> domain.PurchaseChannel | None: ...
async def get_purchase_channel_by_id(self, purchase_channel_id: UUID) -> domain.PurchaseChannel | None: ...
async def get_purchase_channels_by_project(self, project_id: UUID) -> list[domain.PurchaseChannel]: ...
async def add_channel_to_purchase(
@@ -225,6 +227,10 @@ class Database(typing.Protocol):
date_to: datetime.datetime | None = None,
) -> list[domain.Placement]: ...
async def create_placement(self, placement: domain.Placement) -> None: ...
async def update_placement(self, placement: domain.Placement) -> None: ...
async def delete_placement(self, placement_id: UUID) -> None: ...
async def get_placement_by_invite_link(self, invite_link: str) -> domain.Placement | None: ...
@@ -398,7 +404,7 @@ class Usecase:
if purchase:
return purchase
purchase = domain.Purchase(workspace_id=project.workspace_id, project_id=project.id, creative_id=creative.id)
purchase = domain.Purchase(project_id=project.id, creative_id=creative.id)
await self.database.create_purchase(purchase)
return purchase

10
src/usecase/analytics/get_channel_analytics.py
View File

@@ -47,8 +47,9 @@ async def get_channel_analytics(
if input.project_id is None and allowed_project_ids is not None:
channel_map: dict[UUID, domain.Channel] = {}
for placement in placements:
if placement.placement_channel:
channel_map[placement.placement_channel_id] = placement.placement_channel
purchase_channel = placement.purchase_channel
if purchase_channel and purchase_channel.channel:
channel_map[purchase_channel.channel_id] = purchase_channel.channel
channels = list(channel_map.values())
@dataclass
@@ -69,7 +70,10 @@ async def get_channel_analytics(
subscriptions_counts = await self.database.count_subscriptions_by_placement_batch(placement_ids)
for p in placements:
stats = channel_stats.get(p.placement_channel_id)
purchase_channel = p.purchase_channel
if not purchase_channel:
continue
stats = channel_stats.get(purchase_channel.channel_id)
if stats is None:
continue

16
src/usecase/analytics/get_overview_analytics.py
View File

@@ -131,19 +131,21 @@ async def get_overview_analytics(
project_meta: dict[UUID, domain.Project] = {}
for placement in current_placements:
placement_date = placement.wanted_placement_date.date()
placement_day = placement.wanted_placement_date.date()
project_meta[placement.project_id] = placement.project
cost_value = placement.cost or 0.0
cost_per_day[placement_date] += cost_value
cost_per_day[placement_day] += cost_value
project_spending_map[placement.project_id] += cost_value
subs = subs_per_placement_current.get(placement.id, 0)
channel_id = placement.placement_channel_id
if channel_id not in channel_stats:
channel_stats[channel_id] = ChannelAggregate(channel=placement.placement_channel)
channel_stats[channel_id].total_cost += cost_value
channel_stats[channel_id].total_subs += subs
purchase_channel = placement.purchase_channel
if purchase_channel:
channel_id = purchase_channel.channel_id
if channel_id not in channel_stats:
channel_stats[channel_id] = ChannelAggregate(channel=purchase_channel.channel)
channel_stats[channel_id].total_cost += cost_value
channel_stats[channel_id].total_subs += subs
start_date = input.date_from.date()
end_date = input.date_to.date()

4
src/usecase/analytics/get_placements_analytics.py
View File

@@ -57,8 +57,8 @@ async def get_placements_analytics(
id=p.id,
project_id=p.project_id,
project_channel_title=p.project.channel.title,
placement_channel_id=p.placement_channel_id,
placement_channel_title=p.placement_channel.title,
placement_channel_id=p.purchase_channel.channel_id,
placement_channel_title=p.purchase_channel.channel.title,
creative_id=p.creative_id,
creative_name=p.creative.name,
placement_date=p.wanted_placement_date,

1
src/usecase/channel/attach_channel_to_workspace.py
View File

@@ -47,4 +47,5 @@ async def attach_channel_to_workspace(
title=channel.title,
username=channel.username,
status=project.status,
purchase_invite_type_default=project.purchase_invite_type_default,
)

1
src/usecase/creative/create_creative.py
View File

@@ -43,7 +43,6 @@ async def create_creative(
buttons=[button.model_dump() for button in input.buttons],
status=domain.CreativeStatus.ACTIVE,
project_id=project.id,
workspace_id=workspace_id,
)
await self.database.create_creative(creative)

84
src/usecase/placement/create_placement.py
View File

@@ -1,84 +0,0 @@
import logging
import uuid
from typing import TYPE_CHECKING
import domain.purchase
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def create_placement(
self: 'Usecase', input: dto.CreatePlacementInput, user_id: uuid.UUID, workspace_id: uuid.UUID
) -> dto.PlacementOutput:
await self.ensure_workspace_permission(
workspace_id,
user_id,
domain.PermissionKey.PLACEMENTS_WRITE,
for_project_id=input.project_id,
)
project = await self.database.get_project(workspace_id, project_id=input.project_id)
if not project:
raise domain.ProjectNotFound(input.project_id)
placement_channel = await self.database.get_channel(channel_id=input.placement_channel_id)
if not placement_channel:
raise domain.ChannelNotFound(input.placement_channel_id)
creative = await self.database.get_creative(workspace_id, input.creative_id)
if not creative:
raise domain.CreativeNotFound(input.creative_id)
if creative.project_id != project.id:
raise domain.CreativeNotFound(input.creative_id)
# Берем тип ссылки из настроек проекта
requires_approval = project.purchase_invite_type_default == domain.purchase.InviteLinkType.APPROVAL
if project.channel.telegram_id is None:
raise domain.ChannelNotFound(project.channel.id)
invite_link = await self.telegram_bot.create_chat_invite_link(project.channel.telegram_id, requires_approval)
log.info(
'Created invite link for project=%s channel_telegram_id=%s requires_approval=%s invite_link=%s',
project.id,
project.channel.telegram_id,
requires_approval,
invite_link,
)
placement = domain.Placement(
project_id=input.project_id,
placement_channel_id=input.placement_channel_id,
creative_id=input.creative_id,
workspace_id=workspace_id,
placement_date=input.placement_date,
cost=input.cost,
comment=input.comment,
invite_link=invite_link,
status=domain.PlacementStatus.ACTIVE,
)
await self.database.create_placement(placement)
return dto.PlacementOutput(
id=placement.id,
project_id=placement.project_id,
project_channel_title=project.channel.title,
placement_channel_id=placement.placement_channel_id,
placement_channel_title=placement_channel.title,
creative_id=placement.creative_id,
creative_name=creative.name,
placement_date=placement.wanted_placement_date,
cost=placement.cost,
comment=placement.comment,
ad_post_url=None,
invite_link_type=project.purchase_invite_type_default,
invite_link=placement.invite_link,
status=placement.status,
subscriptions_count=0,
created_at=placement.created_at,
)

24
src/usecase/placement/delete_placement.py
View File

@@ -1,24 +0,0 @@
import logging
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def delete_placement(self: 'Usecase', input: dto.DeletePlacementInput) -> None:
context = await self.ensure_workspace_permission(
input.workspace_id, input.user_id, domain.PermissionKey.PLACEMENTS_WRITE
)
placement = await self.database.get_placement(input.workspace_id, input.placement_id)
if not placement:
log.warning('Placement %s not found for user %s', input.placement_id, input.user_id)
raise domain.PlacementNotFound(input.placement_id)
context.ensure_project_permission(domain.PermissionKey.PLACEMENTS_WRITE, placement.project_id)
await self.database.delete_placement(input.placement_id)

4
src/usecase/placement/fetch_placement_post_cycle.py
View File

@@ -44,12 +44,8 @@ async def fetch_placement_post_cycle(self: 'Usecase', interval_seconds: int) ->
wanted_placement_date=post.created_at,
cost=None,
comment=None,
invite_link=purchase_channel.invite_link,
workspace_id=purchase_channel.purchase.workspace_id,
project_id=purchase_channel.purchase.project_id,
creative_id=purchase_channel.purchase.creative_id,
placement_channel_id=purchase_channel.channel_id,
purchase_id=purchase_channel.purchase_id,
purchase_channel_id=purchase_channel.id,
post=post,
status=domain.PlacementStatus.ACTIVE,

14
src/usecase/placement/get_placement.py
View File

@@ -22,6 +22,10 @@ async def get_placement(self: 'Usecase', input: dto.GetPlacementInput) -> dto.Pl
context.ensure_project_permission(domain.PermissionKey.PLACEMENTS_READ, placement.project_id)
ad_post_url = placement.post.url if placement.post else None
purchase_channel = placement.purchase_channel
if not purchase_channel or not purchase_channel.channel:
log.warning('Placement %s missing purchase channel data', placement.id)
raise domain.PurchaseChannelNotFound()
subscriptions_count = await self.database.count_subscriptions_by_placement(placement.id)
@@ -29,17 +33,19 @@ async def get_placement(self: 'Usecase', input: dto.GetPlacementInput) -> dto.Pl
id=placement.id,
project_id=placement.project_id,
project_channel_title=placement.project.channel.title,
placement_channel_id=placement.placement_channel_id,
placement_channel_title=placement.placement_channel.title,
placement_channel_id=purchase_channel.channel_id,
placement_channel_title=purchase_channel.channel.title,
creative_id=placement.creative_id,
creative_name=placement.creative.name,
placement_date=placement.wanted_placement_date,
cost=placement.cost,
comment=placement.comment,
ad_post_url=ad_post_url,
invite_link_type=placement.project.purchase_invite_type_default,
invite_link=placement.invite_link,
invite_link_type=purchase_channel.invite_link_type,
invite_link=purchase_channel.invite_link,
status=placement.status,
subscriptions_count=subscriptions_count,
purchase_id=purchase_channel.purchase_id,
purchase_channel_id=purchase_channel.id,
created_at=placement.created_at,
)

17
src/usecase/placement/get_placements.py
View File

@@ -1,3 +1,4 @@
import logging
from typing import TYPE_CHECKING
from src import domain, dto
@@ -5,6 +6,8 @@ from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def get_placements(self: 'Usecase', input: dto.GetPlacementsInput) -> list[dto.PlacementOutput]:
context = await self.ensure_workspace_permission(
@@ -32,24 +35,30 @@ async def get_placements(self: 'Usecase', input: dto.GetPlacementsInput) -> list
placement_outputs = []
for placement in placements:
ad_post_url = placement.post.url if placement.post else None
purchase_channel = placement.purchase_channel
if not purchase_channel or not purchase_channel.channel:
log.warning('Placement %s missing purchase channel data', placement.id)
continue
placement_outputs.append(
dto.PlacementOutput(
id=placement.id,
project_id=placement.project_id,
project_channel_title=placement.project.channel.title,
placement_channel_id=placement.placement_channel_id,
placement_channel_title=placement.placement_channel.title,
placement_channel_id=purchase_channel.channel_id,
placement_channel_title=purchase_channel.channel.title,
creative_id=placement.creative_id,
creative_name=placement.creative.name,
placement_date=placement.wanted_placement_date,
cost=placement.cost,
comment=placement.comment,
ad_post_url=ad_post_url,
invite_link_type=placement.project.purchase_invite_type_default,
invite_link=placement.invite_link,
invite_link_type=purchase_channel.invite_link_type,
invite_link=purchase_channel.invite_link,
status=placement.status,
subscriptions_count=subscriptions_counts.get(placement.id, 0),
purchase_id=purchase_channel.purchase_id,
purchase_channel_id=purchase_channel.id,
created_at=placement.created_at,
)
)

61
src/usecase/placement/update_placement.py
View File

@@ -1,61 +0,0 @@
import logging
import uuid
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def update_placement(
self: 'Usecase',
placement_id: uuid.UUID,
input: dto.UpdatePlacementInput,
user_id: uuid.UUID,
workspace_id: uuid.UUID,
) -> dto.PlacementOutput:
context = await self.ensure_workspace_permission(workspace_id, user_id, domain.PermissionKey.PLACEMENTS_WRITE)
placement = await self.database.get_placement(workspace_id, placement_id)
if not placement:
log.warning('Placement %s not found for user %s', placement_id, user_id)
raise domain.PlacementNotFound(placement_id)
context.ensure_project_permission(domain.PermissionKey.PLACEMENTS_WRITE, placement.project_id)
if input.placement_date is not None:
placement.wanted_placement_date = input.placement_date
if input.cost is not None:
placement.cost = input.cost
if input.comment is not None:
placement.comment = input.comment
if input.status is not None:
placement.status = input.status
await self.database.update_placement(placement)
ad_post_url = placement.post.url if placement.post else None
subscriptions_count = await self.database.count_subscriptions_by_placement(placement.id)
return dto.PlacementOutput(
id=placement.id,
project_id=placement.project_id,
project_channel_title=placement.project.channel.title,
placement_channel_id=placement.placement_channel_id,
placement_channel_title=placement.placement_channel.title,
creative_id=placement.creative_id,
creative_name=placement.creative.name,
placement_date=placement.wanted_placement_date,
cost=placement.cost,
comment=placement.comment,
ad_post_url=ad_post_url,
invite_link_type=placement.project.purchase_invite_type_default,
invite_link=placement.invite_link,
status=placement.status,
subscriptions_count=subscriptions_count,
created_at=placement.created_at,
)

1
src/usecase/purchase/create_purchase.py
View File

@@ -99,7 +99,6 @@ async def create_purchase(
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,