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-02-01 17:57:46 +03:00
parent a92f33c22f
commit 4e35c73a66
15 changed files with 1573 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
src/adapter/telegram_bot.py
View File

@@ -17,11 +17,21 @@ log = logging.getLogger(__name__)
class TelegramBot(TelegramBase, TelegramBotWriter):
async def send_message(
self, text: str, chat_id: int, parse_mode: str | None = None, disable_preview: bool = False
) -> None:
await self.bot.send_message(
chat_id=chat_id, text=text, parse_mode=parse_mode, disable_web_page_preview=disable_preview
self,
text: str,
chat_id: int,
parse_mode: str | None = None,
disable_preview: bool = False,
reply_to_message_id: int | None = None,
) -> int:
message = await self.bot.send_message(
chat_id=chat_id,
text=text,
parse_mode=parse_mode,
disable_web_page_preview=disable_preview,
reply_to_message_id=reply_to_message_id,
)
return message.message_id
async def create_chat_invite_link(self, chat_id: int, requires_approval: bool = False) -> str:
invite_link = await self.bot.create_chat_invite_link(chat_id=chat_id, creates_join_request=requires_approval)
@@ -34,15 +44,18 @@ class TelegramBot(TelegramBase, TelegramBotWriter):
buttons: list[list[InlineKeyboardButton]],
parse_mode: str | None = None,
disable_preview: bool = False,
) -> None:
reply_to_message_id: int | None = None,
) -> int:
keyboard = InlineKeyboardMarkup(inline_keyboard=buttons)
await self.bot.send_message(
message = await self.bot.send_message(
chat_id=chat_id,
text=text,
reply_markup=keyboard,
parse_mode=parse_mode,
disable_web_page_preview=disable_preview,
reply_to_message_id=reply_to_message_id,
)
return message.message_id
async def send_media_with_inline_keyboard(
self,
@@ -52,41 +65,47 @@ class TelegramBot(TelegramBase, TelegramBotWriter):
media_file_id: str,
buttons: list[list[InlineKeyboardButton]],
parse_mode: str | None = None,
) -> None:
reply_to_message_id: int | None = None,
) -> int:
keyboard = InlineKeyboardMarkup(inline_keyboard=buttons) if buttons else None
if media_type == 'photo':
await self.bot.send_photo(
message = await self.bot.send_photo(
chat_id=chat_id,
photo=media_file_id,
caption=text,
parse_mode=parse_mode,
reply_markup=keyboard,
reply_to_message_id=reply_to_message_id,
)
return
return message.message_id
if media_type == 'video':
await self.bot.send_video(
message = await self.bot.send_video(
chat_id=chat_id,
video=media_file_id,
caption=text,
parse_mode=parse_mode,
reply_markup=keyboard,
reply_to_message_id=reply_to_message_id,
)
return
return message.message_id
if media_type == 'animation':
await self.bot.send_animation(
message = await self.bot.send_animation(
chat_id=chat_id,
animation=media_file_id,
caption=text,
parse_mode=parse_mode,
reply_markup=keyboard,
reply_to_message_id=reply_to_message_id,
)
return
await self.bot.send_message(
return message.message_id
message = await self.bot.send_message(
chat_id=chat_id,
text=text,
parse_mode=parse_mode,
reply_markup=keyboard,
reply_to_message_id=reply_to_message_id,
)
return message.message_id
async def send_media_group(
self,
@@ -94,9 +113,10 @@ class TelegramBot(TelegramBase, TelegramBotWriter):
media_items: Sequence[Any],
caption: str | None = None,
parse_mode: str | None = None,
) -> None:
reply_to_message_id: int | None = None,
) -> int:
if not media_items:
return
raise ValueError('No media items to send')
group: list[InputMediaPhoto | InputMediaVideo] = []
for index, item in enumerate(media_items):
@@ -113,7 +133,11 @@ class TelegramBot(TelegramBase, TelegramBotWriter):
raise ValueError(f'Unsupported media type for group: {item.media_type}')
# Type ignore: aiogram expects Union type but we only use Photo/Video
await self.bot.send_media_group(chat_id=chat_id, media=group) # type: ignore[arg-type]
messages = await self.bot.send_media_group(
chat_id=chat_id, media=group, reply_to_message_id=reply_to_message_id
) # type: ignore[arg-type]
# Return message_id of first message (with caption)
return messages[0].message_id
async def edit_message_text(self, text: str, chat_id: int, message_id: int) -> None:
await self.bot.edit_message_text(chat_id=chat_id, message_id=message_id, text=text, reply_markup=None)

1
src/domain/placement.py
View File

@@ -89,6 +89,7 @@ class Placement(TimestampedModel):
payment_at = fields.DatetimeField(null=True)
cost_type = fields.CharEnumField(CostType, null=True, max_length=8)
cost_value = fields.FloatField(null=True)
cost_before_bargain_type = fields.CharEnumField(CostType, null=True, max_length=8)
cost_before_bargain = fields.FloatField(null=True)
placement_type = fields.CharEnumField(PlacementType, null=True, max_length=16)
format = fields.TextField(null=True)

5
src/dto/purchase.py
View File

@@ -18,11 +18,12 @@ class PlacementDetails(pydantic.BaseModel):
placement_at: datetime.datetime | None = None
payment_at: datetime.datetime | None = None
cost: CostInfo | None = None
cost_before_bargain: float | None = None
cost_before_bargain: CostInfo | None = None
placement_type: PlacementType | None = None
format: str | None = None
comment: str | None = None
creative_id: uuid.UUID | None = None
invite_link_type: InviteLinkType | None = None
class PlacementOutput(pydantic.BaseModel):
@@ -102,7 +103,7 @@ class UpdatePlacementInput(pydantic.BaseModel):
placement_at: datetime.datetime | None = None
payment_at: datetime.datetime | None = None
cost: CostInfo | None = None
cost_before_bargain: float | None = None
cost_before_bargain: CostInfo | None = None
placement_type: PlacementType | None = None
format: str | None = None

18
src/usecase/__init__.py
View File

@@ -72,8 +72,13 @@ class MediaItem(typing.Protocol):
class TelegramBotWriter(typing.Protocol):
async def send_message(
self, text: str, chat_id: int, parse_mode: str | None = None, disable_preview: bool = False
) -> None: ...
self,
text: str,
chat_id: int,
parse_mode: str | None = None,
disable_preview: bool = False,
reply_to_message_id: int | None = None,
) -> int: ...
async def send_message_with_inline_keyboard(
self,
@@ -82,7 +87,8 @@ class TelegramBotWriter(typing.Protocol):
buttons: list[list['InlineKeyboardButton']],
parse_mode: str | None = None,
disable_preview: bool = False,
) -> None: ...
reply_to_message_id: int | None = None,
) -> int: ...
async def send_media_with_inline_keyboard(
self,
@@ -92,7 +98,8 @@ class TelegramBotWriter(typing.Protocol):
media_file_id: str,
buttons: list[list['InlineKeyboardButton']],
parse_mode: str | None = None,
) -> None: ...
reply_to_message_id: int | None = None,
) -> int: ...
async def send_media_group(
self,
@@ -100,7 +107,8 @@ class TelegramBotWriter(typing.Protocol):
media_items: Sequence[MediaItem],
caption: str | None = None,
parse_mode: str | None = None,
) -> None: ...
reply_to_message_id: int | None = None,
) -> int: ...
async def edit_message_text(self, text: str, chat_id: int, message_id: int) -> None: ...

82
src/usecase/purchase/build_placement_creative.py
View File

@@ -51,6 +51,73 @@ def _format_channel_name(channel: domain.Channel) -> str:
return 'Без названия'
def _format_channel_link(channel: domain.Channel) -> str:
"""Format channel as HTML link if possible, otherwise return plain text."""
name = _format_channel_name(channel)
# Try invite_link first
if channel.invite_link:
return f'<a href="{channel.invite_link}">{name}</a>'
# Try username
if channel.username:
return f'<a href="https://t.me/{channel.username}">{name}</a>'
# No link available, return plain name
return name
def _build_info_message(
placement: domain.Placement,
project: domain.Project,
) -> str:
"""Build informational message about placement."""
lines = []
# Header
lines.append('Ссылка зашита с помощью @smartpost_tg_bot')
lines.append('')
# Placement channel (always present)
placement_channel_link = _format_channel_link(placement.channel)
lines.append(f'Размещение в канале: {placement_channel_link}')
# Project channel (always present)
project_channel_link = _format_channel_link(project.channel)
lines.append(f'Рекламируемый проект: {project_channel_link}')
# Invite link type (always present)
link_type_text = 'Открытая' if placement.invite_link_type == domain.InviteLinkType.PUBLIC else 'С заявками'
lines.append(f'Тип ссылки: {link_type_text}')
# Cost / CPM (optional)
if placement.cost_value is not None and placement.cost_value > 0:
if placement.cost_type == domain.CostType.CPM:
lines.append(f'Ставка CPM: {placement.cost_value:.0f} ₽')
else:
lines.append(f'Стоимость: {placement.cost_value:.0f} ₽')
# Date and time (optional)
if placement.placement_at:
# Check if time is 00:00 (midnight) - then show only date
if placement.placement_at.hour == 0 and placement.placement_at.minute == 0:
date_str = placement.placement_at.strftime('%d.%m.%Y')
lines.append(f'Дата: {date_str}')
else:
datetime_str = placement.placement_at.strftime('%d.%m.%Y %H:%M')
lines.append(f'Дата и время: {datetime_str}')
# Format (optional)
if placement.format:
lines.append(f'Формат: {placement.format}')
# Add empty line before footer
lines.append('')
lines.append('Пост ниже')
return '\n'.join(lines)
def _build_keyboard_buttons(buttons: list[dto.CreativeButton]) -> list[list[InlineKeyboardButton]]:
return [[InlineKeyboardButton(text=btn.text, url=btn.url)] for btn in buttons]
@@ -120,10 +187,12 @@ async def build_placement_creative(
raise domain.UserNotFound(user_id)
chat_id = user.telegram_user.telegram_id
channel_name = _format_channel_name(placement.channel)
header = f'<b>Канал:</b> {channel_name}'
await self.telegram_bot.send_message(header, chat_id, parse_mode='HTML')
# Build and send informational message
info_message = _build_info_message(placement, project)
info_message_id = await self.telegram_bot.send_message(info_message, chat_id, parse_mode='HTML')
# Send creative as reply to informational message
keyboard_buttons = _build_keyboard_buttons(preview.buttons)
if preview.media_items:
if len(preview.media_items) == 1:
@@ -135,6 +204,7 @@ async def build_placement_creative(
media_file_id=media_item.media_file_id,
buttons=keyboard_buttons,
parse_mode='HTML',
reply_to_message_id=info_message_id,
)
else:
await self.telegram_bot.send_media_group(
@@ -142,6 +212,7 @@ async def build_placement_creative(
media_items=preview.media_items,
caption=preview.text,
parse_mode='HTML',
reply_to_message_id=info_message_id,
)
elif keyboard_buttons:
await self.telegram_bot.send_message_with_inline_keyboard(
@@ -150,8 +221,11 @@ async def build_placement_creative(
keyboard_buttons,
parse_mode='HTML',
disable_preview=True,
reply_to_message_id=info_message_id,
)
else:
await self.telegram_bot.send_message(preview.text, chat_id, parse_mode='HTML', disable_preview=True)
await self.telegram_bot.send_message(
preview.text, chat_id, parse_mode='HTML', disable_preview=True, reply_to_message_id=info_message_id
)
return preview

8
src/usecase/purchase/create_placements.py
View File

@@ -21,7 +21,7 @@ def _build_placement_details(placement: domain.Placement) -> dto.PlacementDetail
placement_at=placement.placement_at,
payment_at=placement.payment_at,
cost=_build_cost_info(placement.cost_type, placement.cost_value),
cost_before_bargain=placement.cost_before_bargain,
cost_before_bargain=_build_cost_info(placement.cost_before_bargain_type, placement.cost_before_bargain),
placement_type=placement.placement_type,
format=placement.format,
comment=placement.comment,
@@ -117,8 +117,10 @@ async def create_placements(
or (details.cost.type if details and details.cost else None),
cost_value=(channel_details.cost.value if channel_details and channel_details.cost else None)
or (details.cost.value if details and details.cost else None),
cost_before_bargain=(channel_details.cost_before_bargain if channel_details else None)
or (details.cost_before_bargain if details else None),
cost_before_bargain_type=(channel_details.cost_before_bargain.type if channel_details and channel_details.cost_before_bargain else None)
or (details.cost_before_bargain.type if details and details.cost_before_bargain else None),
cost_before_bargain=(channel_details.cost_before_bargain.value if channel_details and channel_details.cost_before_bargain else None)
or (details.cost_before_bargain.value if details and details.cost_before_bargain else None),
placement_type=details.placement_type if details else None,
format=(channel_details.format if channel_details else None) or (details.format if details else None),
)

7
src/usecase/purchase/update_placement.py
View File

@@ -61,7 +61,12 @@ async def update_placement(
placement.cost_type = input.cost.type
placement.cost_value = input.cost.value
if 'cost_before_bargain' in fields_set:
placement.cost_before_bargain = input.cost_before_bargain
if input.cost_before_bargain is None:
placement.cost_before_bargain_type = None
placement.cost_before_bargain = None
else:
placement.cost_before_bargain_type = input.cost_before_bargain.type
placement.cost_before_bargain = input.cost_before_bargain.value
if 'placement_type' in fields_set:
placement.placement_type = input.placement_type
if 'format' in fields_set: