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

feat: /new_creative tg_bot

Browse Source
This commit is contained in:
Artem Tsyrulnikov
2025-12-01 13:20:08 +03:00
parent 3508a522ac
commit f7754b707d
21 changed files with 688 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

176
CLAUDE.md
View File

@@ -84,10 +84,15 @@ The codebase follows a clean architecture pattern with clear separation of conce
2. **Use Case Layer** (`src/usecase/`)
- Business logic functions organized by feature
- Each use case is a standalone function
- Depends on Protocol interfaces (Database, TelegramWriter, JWTEncoder)
- Each use case is a standalone function with signature: `async def use_case_name(self: 'Usecase', ...) -> ReturnType`
- Depends on Protocol interfaces (Database, TelegramBotWriter, JWTEncoder)
- Use cases are assembled in the `Usecase` dataclass for dependency injection
- Один use case отвечает за полноценный сценарий «от триггера до завершения» и не должен вызывать другие use case'ы напрямую (общую логику выносить в хелперы/сервисы внутри того же модуля)
- **IMPORTANT RULES:**
- Один use case отвечает за полноценный сценарий «от триггера до завершения»
- Use case НЕ должен вызывать другие use cases (общую логику выносить в хелперы/сервисы)
- Use case содержит ВСЮ бизнес-логику, включая управление состояниями диалога
- Use case может отправлять сообщения пользователям через `self.telegram_bot` protocol
- Всегда оборачивать операции БД в `async with self.database.transaction()`
3. **Adapter Layer** (`src/adapter/`)
- Concrete implementations of protocol interfaces:
@@ -98,7 +103,15 @@ The codebase follows a clean architecture pattern with clear separation of conce
4. **Controller Layer** (`src/controller/`)
- **HTTP controllers** (`src/controller/http/`): FastAPI route handlers
- **Telegram callbacks** (`src/controller/telegram_callback/`): Telegram event handlers
- Controllers call use cases to perform business logic
- **ВАЖНО: Controllers - это тонкий слой роутинга**
- Контроллер только извлекает данные из событий (HTTP request / Telegram update)
- Валидирует базовые параметры (user exists, message not empty)
- Вызывает соответствующий use case
- НЕ содержит бизнес-логику
- НЕ обращается к базе данных напрямую
- НЕ отправляет сообщения пользователям
- Все handlers регистрируются через декораторы на `telegram_callback_router`
- Handlers импортируются в `__init__.py` для auto-регистрации
5. **DTO Layer** (`src/dto/`)
- Pydantic models for request/response validation
@@ -168,3 +181,158 @@ The codebase follows a clean architecture pattern with clear separation of conce
- Line length: 120 characters
- Python 3.13+ syntax
- Strict mypy typing enabled
### Telegram Bot Patterns
**State Management:**
- Состояния хранятся в БД через модель `TelegramState`
- Используется `TelegramStateEnum` для типизации состояний (не строки!)
- Database protocol методы: `get_telegram_state()`, `set_telegram_state()`, `clear_telegram_state()`
**Controller Pattern (Telegram):**
```python
from aiogram import F
from aiogram.filters import Command
from aiogram.types import Message
@telegram_callback_router.message(Command('command_name'))
async def cmd_handler(message: Message) -> None:
if not message.from_user:
log.error('Failed to get user data')
return
usecase = dependencies.get_usecase()
await usecase.some_use_case(
telegram_id=message.from_user.id,
chat_id=message.chat.id
)
```
**Use Case Pattern (Telegram):**
```python
from src import domain
async def some_use_case(self: 'Usecase', telegram_id: int, chat_id: int) -> None:
async with self.database.transaction():
# 1. Получить/проверить пользователя
user = await self.database.get_user(telegram_id=telegram_id)
if not user:
await self.telegram_bot.send_message('❌ Не авторизован', chat_id=chat_id)
return
# 2. Получить состояние (если нужно)
state = await self.database.get_telegram_state(telegram_id)
# 3. Бизнес-логика
# ...
# 4. Обновить состояние
await self.database.set_telegram_state(
telegram_id=telegram_id,
state=domain.TelegramStateEnum.SOME_STATE,
context={'key': 'value'}
)
# 5. Отправить сообщение
await self.telegram_bot.send_message('✅ Успешно', chat_id=chat_id)
```
**Inline Keyboard Pattern:**
```python
from aiogram.types import InlineKeyboardButton
# В use case:
buttons = [
[InlineKeyboardButton(text='Кнопка', callback_data='callback_id:value')]
for item in items
]
await self.telegram_bot.send_message_with_inline_keyboard(
'Выберите опцию:', chat_id=chat_id, buttons=buttons
)
```
**Callback Query Handler:**
```python
@telegram_callback_router.callback_query(F.data.startswith('prefix:'))
async def callback_handler(callback: CallbackQuery) -> None:
if not callback.from_user or not callback.data:
return
usecase = dependencies.get_usecase()
await usecase.handle_callback(
telegram_id=callback.from_user.id,
chat_id=callback.message.chat.id,
callback_data=callback.data,
message_id=callback.message.message_id
)
await callback.answer() # Убрать "loading" на кнопке
```
**Text Message Filters:**
```python
# Обрабатывать только не-команды
@telegram_callback_router.message(F.text & ~F.text.startswith('/'))
async def text_handler(message: Message) -> None:
# Этот handler НЕ сработает для команд типа /start
...
```
**Важные правила для Telegram handlers:**
- Специфичные фильтры (Command) регистрировать РАНЬШЕ общих (F.text)
- Использовать `~F.text.startswith('/')` для исключения команд в текстовых handlers
- Всегда проверять `message.from_user` перед использованием
- Callback handlers должны вызывать `await callback.answer()` в конце
- НЕ импортировать aiogram типы в usecase layer (только через TYPE_CHECKING в Protocol)
### Adding New Protocol Methods
When adding new functionality that requires protocol methods:
**1. Define Protocol in usecase layer (`src/usecase/__init__.py`):**
```python
if typing.TYPE_CHECKING:
from aiogram.types import InlineKeyboardButton # Импорт только для типов
class TelegramBotWriter(typing.Protocol):
async def send_message_with_inline_keyboard(
self, text: str, chat_id: int, buttons: list[list['InlineKeyboardButton']]
) -> None: ...
```
**2. Implement in adapter (`src/adapter/telegram_bot.py`):**
```python
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
class TelegramBot(TelegramBase):
async def send_message_with_inline_keyboard(
self, text: str, chat_id: int, buttons: list[list[InlineKeyboardButton]]
) -> None:
keyboard = InlineKeyboardMarkup(inline_keyboard=buttons)
await self.bot.send_message(chat_id=chat_id, text=text, reply_markup=keyboard)
```
**3. Use in use case:**
```python
from aiogram.types import InlineKeyboardButton # Можно импортировать в use case
buttons = [[InlineKeyboardButton(text='Text', callback_data='data')]]
await self.telegram_bot.send_message_with_inline_keyboard('Message', chat_id, buttons)
```
**Типизация Enum в Domain:**
- Используйте `enum.StrEnum` для строковых enum (Python 3.11+)
- Enum значения в lowercase с underscores
- Передавайте enum в SQLAlchemy через `Enum(EnumClass)`
```python
import enum
from sqlalchemy import Enum
from sqlalchemy.orm import Mapped, mapped_column
class TelegramStateEnum(enum.StrEnum):
CREATIVE_WAITING_CHANNEL = 'creative_waiting_channel'
CREATIVE_WAITING_NAME = 'creative_waiting_name'
class TelegramState(Base):
state: Mapped[TelegramStateEnum] = mapped_column(Enum(TelegramStateEnum), nullable=False)
```

42
migration/versions/3b8259700b61_add_subscription_status_and_.py
View File

@@ -1,42 +0,0 @@
"""add subscription status and unsubscribed_at
Revision ID: 3b8259700b61
Revises: f05f85d10837
Create Date: 2025-11-25 18:45:04.642105
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = '3b8259700b61'
down_revision: str | None = 'f05f85d10837'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
# Создаём enum тип явно
subscription_status = sa.Enum('ACTIVE', 'UNSUBSCRIBED', name='subscriptionstatus')
subscription_status.create(op.get_bind(), checkfirst=True)
op.add_column(
'subscription',
sa.Column('status', subscription_status, nullable=False, server_default='ACTIVE'),
)
op.add_column('subscription', sa.Column('unsubscribed_at', sa.DateTime(timezone=True), nullable=True))
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('subscription', 'unsubscribed_at')
op.drop_column('subscription', 'status')
# Удаляем enum тип
sa.Enum(name='subscriptionstatus').drop(op.get_bind(), checkfirst=True)
# ### end Alembic commands ###

63
migration/versions/52ad7cc9bf1e_add_unique_constraint_for_active_.py
View File

@@ -1,63 +0,0 @@
"""add unique constraint for active subscriptions
Revision ID: 52ad7cc9bf1e
Revises: 3b8259700b61
Create Date: 2025-11-26 14:14:57.814576
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = '52ad7cc9bf1e'
down_revision: str | None = '3b8259700b61'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# Добавляем поле target_channel_id (сначала nullable)
op.add_column('subscription', sa.Column('target_channel_id', sa.Uuid(), nullable=True))
# Заполняем target_channel_id из placement
op.execute("""
UPDATE subscription
SET target_channel_id = placement.target_channel_id
FROM placement
WHERE subscription.placement_id = placement.id
""")
# Делаем поле NOT NULL после заполнения
op.alter_column('subscription', 'target_channel_id', nullable=False)
# Добавляем FK constraint
op.create_foreign_key(
'subscription_target_channel_id_fkey', 'subscription', 'target_channel', ['target_channel_id'], ['id']
)
# Создаем индекс на target_channel_id
op.create_index(op.f('ix_subscription_target_channel_id'), 'subscription', ['target_channel_id'])
# Создаём UNIQUE constraint для активных подписок
op.create_index(
'unique_active_subscription_per_channel',
'subscription',
['subscriber_id', 'target_channel_id'],
unique=True,
postgresql_where=sa.text("status = 'ACTIVE'::subscriptionstatus"),
)
def downgrade() -> None:
# Удаляем UNIQUE constraint
op.drop_index('unique_active_subscription_per_channel', table_name='subscription')
# Удаляем индекс и FK constraint
op.drop_index(op.f('ix_subscription_target_channel_id'), table_name='subscription')
op.drop_constraint('subscription_target_channel_id_fkey', 'subscription', type_='foreignkey')
# Удаляем колонку
op.drop_column('subscription', 'target_channel_id')

65
migration/versions/b1d510ad3751_.py Normal file
View File

@@ -0,0 +1,65 @@
"""empty message
Revision ID: b1d510ad3751
Revises: f05f85d10837
Create Date: 2025-12-01 13:05:17.215374
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
subscription_status_enum = sa.Enum('ACTIVE', 'UNSUBSCRIBED', name='subscriptionstatus')
# revision identifiers, used by Alembic.
revision: str = 'b1d510ad3751'
down_revision: str | None = 'f05f85d10837'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
'telegram_state',
sa.Column('telegram_id', sa.BigInteger(), nullable=False),
sa.Column(
'state',
sa.Enum(
'CREATIVE_WAITING_CHANNEL', 'CREATIVE_WAITING_NAME', 'CREATIVE_WAITING_TEXT', name='telegramstateenum'
),
nullable=False,
),
sa.Column('context', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_telegram_state_telegram_id'), 'telegram_state', ['telegram_id'], unique=True)
op.add_column('subscription', sa.Column('target_channel_id', sa.Uuid(), nullable=False))
subscription_status_enum.create(op.get_bind(), checkfirst=True)
op.add_column('subscription', sa.Column('status', subscription_status_enum, nullable=False))
op.add_column('subscription', sa.Column('unsubscribed_at', sa.DateTime(timezone=True), nullable=True))
op.create_index(op.f('ix_subscription_target_channel_id'), 'subscription', ['target_channel_id'], unique=False)
fk_name = op.f('fk_subscription_target_channel_id')
op.create_foreign_key(fk_name, 'subscription', 'target_channel', ['target_channel_id'], ['id'])
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
fk_name = op.f('fk_subscription_target_channel_id')
op.drop_constraint(fk_name, 'subscription', type_='foreignkey')
op.drop_index(op.f('ix_subscription_target_channel_id'), table_name='subscription')
op.drop_column('subscription', 'unsubscribed_at')
op.drop_column('subscription', 'status')
subscription_status_enum.drop(op.get_bind(), checkfirst=True)
op.drop_column('subscription', 'target_channel_id')
op.drop_index(op.f('ix_telegram_state_telegram_id'), table_name='telegram_state')
op.drop_table('telegram_state')
# ### end Alembic commands ###

29
src/adapter/postgres.py
View File

@@ -1,4 +1,5 @@
import datetime
import typing
import uuid
from sqlalchemy import delete, select
@@ -441,3 +442,31 @@ class Postgres(DatabaseBase):
result = await self.session.execute(q)
return list(result.scalars().all())
async def get_telegram_state(self, telegram_id: int) -> domain.TelegramState | None:
q = select(domain.TelegramState).where(domain.TelegramState.telegram_id == telegram_id)
result = await self.session.execute(q)
return result.scalar_one_or_none()
async def set_telegram_state(
self, telegram_id: int, state: domain.TelegramStateEnum, context: dict[str, typing.Any]
) -> domain.TelegramState:
existing = await self.get_telegram_state(telegram_id)
if existing:
existing.state = state
existing.context = context
existing.updated_at = datetime.datetime.now(datetime.UTC)
await self.session.flush()
await self.session.refresh(existing)
return existing
new_state = domain.TelegramState(telegram_id=telegram_id, state=state, context=context)
self.session.add(new_state)
await self.session.flush()
await self.session.refresh(new_state)
return new_state
async def clear_telegram_state(self, telegram_id: int) -> None:
q = delete(domain.TelegramState).where(domain.TelegramState.telegram_id == telegram_id)
await self.session.execute(q)

19
src/adapter/telegram_bot.py
View File

@@ -1,7 +1,7 @@
import logging
from typing import Any
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
from aiogram.types import BotCommand, InlineKeyboardButton, InlineKeyboardMarkup
from shared.telegram_base import TelegramBase
@@ -23,3 +23,20 @@ class TelegramBot(TelegramBase):
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)
return invite_link.invite_link
async def send_message_with_inline_keyboard(
self, text: str, chat_id: int, buttons: list[list[InlineKeyboardButton]]
) -> None:
keyboard = InlineKeyboardMarkup(inline_keyboard=buttons)
await self.bot.send_message(chat_id=chat_id, text=text, reply_markup=keyboard)
async def edit_message_reply_markup(self, chat_id: int, message_id: int) -> None:
await self.bot.edit_message_reply_markup(chat_id=chat_id, message_id=message_id, reply_markup=None)
async def set_commands(self) -> None:
commands = [
BotCommand(command='start', description='Начать работу с ботом'),
BotCommand(command='new_creative', description='Создать новый креатив'),
]
await self.bot.set_my_commands(commands)

8
src/controller/telegram_callback/__init__.py
View File

@@ -3,4 +3,10 @@ from aiogram import Router
telegram_callback_router = Router()
# Import handlers to register them
from . import chat_join_request, chat_member_updated, my_chat_member, start_with_login # noqa: E402, F401
from . import ( # noqa: E402, F401
chat_join_request,
chat_member_updated,
creative_commands,
my_chat_member,
start_with_login,
)

51
src/controller/telegram_callback/creative_commands.py Normal file
View File

@@ -0,0 +1,51 @@
import logging
from aiogram import F
from aiogram.filters import Command
from aiogram.types import CallbackQuery, Message
from src import dependencies
from src.controller.telegram_callback import telegram_callback_router
log = logging.getLogger(__name__)
@telegram_callback_router.message(Command('new_creative'))
async def cmd_new_creative(message: Message) -> None:
if not message.from_user:
log.error('Failed to get user data from message')
return
usecase = dependencies.get_usecase()
await usecase.start_creative_creation(telegram_id=message.from_user.id, chat_id=message.chat.id)
@telegram_callback_router.callback_query(F.data.startswith('cc_ch:'))
async def callback_creative_channel(callback: CallbackQuery) -> None:
if not callback.from_user or not callback.message or not callback.data:
log.error('Failed to get required data from callback query')
return
usecase = dependencies.get_usecase()
await usecase.handle_creative_callback(
telegram_id=callback.from_user.id,
chat_id=callback.message.chat.id,
callback_data=callback.data,
message_id=callback.message.message_id,
)
await callback.answer()
@telegram_callback_router.message(F.text & ~F.text.startswith('/'))
async def handle_text_message(message: Message) -> None:
if not message.from_user or not message.text:
return
usecase = dependencies.get_usecase()
await usecase.handle_creative_text_input(
telegram_id=message.from_user.id,
chat_id=message.chat.id,
text=message.text,
text_html=message.html_text,
)

11
src/controller/telegram_callback/start_with_login.py
View File

@@ -28,3 +28,14 @@ async def cmd_start_with_deeplink(message: Message) -> None:
)
await dependencies.get_usecase().telegram_login(input=input)
@telegram_callback_router.message(CommandStart())
async def cmd_start(message: Message) -> None:
"""Handle /start command without deep link."""
if not message.from_user:
log.error('Failed to get user data from message')
return
usecase = dependencies.get_usecase()
await usecase.telegram_start(telegram_id=message.from_user.id, chat_id=message.chat.id)

3
src/domain/__init__.py
View File

@@ -8,6 +8,8 @@ __all__ = (
'Subscription',
'Subscriber',
'PlacementViewsHistory',
'TelegramState',
'TelegramStateEnum',
'TargetChannelStatus',
'CreativeStatus',
'PlacementStatus',
@@ -52,4 +54,5 @@ from .placement_views_history import PlacementViewsHistory
from .subscriber import Subscriber
from .subscription import Subscription, SubscriptionStatus
from .target_channel import TargetChannel, TargetChannelStatus
from .telegram_state import TelegramState, TelegramStateEnum
from .user import User

24
src/domain/telegram_state.py Normal file
View File

@@ -0,0 +1,24 @@
import enum
from typing import Any
from sqlalchemy import BigInteger, Enum
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from . import Base
class TelegramStateEnum(str, enum.Enum):
"""Telegram bot conversation states."""
CREATIVE_WAITING_CHANNEL = 'creative_waiting_channel'
CREATIVE_WAITING_NAME = 'creative_waiting_name'
CREATIVE_WAITING_TEXT = 'creative_waiting_text'
class TelegramState(Base):
"""Хранит текущее состояние разговора пользователя с ботом для multi-step flows."""
telegram_id: Mapped[int] = mapped_column(BigInteger, unique=True, nullable=False, index=True)
state: Mapped[TelegramStateEnum] = mapped_column(Enum(TelegramStateEnum), nullable=False)
context: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict)

1
src/domain/user.py
View File

@@ -12,5 +12,4 @@ class User(domain.Base):
telegram_id: Mapped[int] = mapped_column(unique=True, index=True)
username: Mapped[str | None]
# Relationship with channels
target_channels: Mapped[list['TargetChannel']] = relationship(back_populates='user')

19
src/domain/user_conversation_state.py Normal file
View File

@@ -0,0 +1,19 @@
import enum
from typing import Any
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from src import domain
class TelegramStateEnum(enum.StrEnum):
CREATIVE_WAITING_CHANNEL = 'creative_waiting_channel'
CREATIVE_WAITING_NAME = 'creative_waiting_name'
CREATIVE_WAITING_TEXT = 'creative_waiting_text'
class TelegramState(domain.Base):
telegram_id: Mapped[int] = mapped_column(unique=True, index=True)
state: Mapped[TelegramStateEnum]
context: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict)

1
src/main.py
View File

@@ -21,6 +21,7 @@ from src.usecase import Usecase
async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
await postgres.connect()
await telegram.start_polling()
await telegram.set_commands()
await views_worker.start()
yield

27
src/usecase/__init__.py
View File

@@ -3,6 +3,9 @@ import typing
from dataclasses import dataclass
from uuid import UUID
if typing.TYPE_CHECKING:
from aiogram.types import InlineKeyboardButton
from src import domain
from .analytics.get_creatives_analytics import get_creatives_analytics
@@ -13,6 +16,9 @@ from .creative.create_creative import create_creative
from .creative.delete_creative import delete_creative
from .creative.get_creative import get_creative
from .creative.get_creatives import get_creatives
from .creative.handle_creative_callback import handle_creative_callback
from .creative.handle_creative_text_input import handle_creative_text_input
from .creative.start_creative_creation import start_creative_creation
from .creative.update_creative import update_creative
from .external_channel.create_external_channel import create_external_channel
from .external_channel.delete_external_channel import delete_external_channel
@@ -33,6 +39,7 @@ from .target_channel.disconnect_target_chan_by_tg_id import disconnect_target_ch
from .target_channel.get_user_target_chans import get_user_target_chans
from .target_channel.update_target_chan_permissions import update_target_chan_permissions
from .telegram_login import telegram_login
from .telegram_start import telegram_start
from .validate_login_token import validate_login_token
from .views.fetch_views_manually import fetch_views_manually
from .views.get_views_history import get_views_history
@@ -43,7 +50,7 @@ from .views.update_views_manually import update_views_manually
class Database(typing.Protocol):
def transaction(self) -> typing.AsyncContextManager[None]: ...
async def get_user(self, *, user_id: UUID | None = None, telegram_id: int | None = None) -> domain.User | None: ...
async def get_user(self, user_id: UUID | None = None, telegram_id: int | None = None) -> domain.User | None: ...
async def create_user(self, user: domain.User) -> domain.User: ...
@@ -156,12 +163,26 @@ class Database(typing.Protocol):
async def get_active_placements_with_urls(self) -> list[domain.Placement]: ... # Для worker
async def get_telegram_state(self, telegram_id: int) -> domain.TelegramState | None: ...
async def set_telegram_state(
self, telegram_id: int, state: domain.TelegramStateEnum, context: dict[str, typing.Any]
) -> domain.TelegramState: ...
async def clear_telegram_state(self, telegram_id: int) -> None: ...
class TelegramBotWriter(typing.Protocol):
async def send_message(self, text: str, chat_id: int) -> None: ...
async def send_message_with_button(self, text: str, chat_id: int, button_text: str, button_url: str) -> None: ...
async def send_message_with_inline_keyboard(
self, text: str, chat_id: int, buttons: list[list['InlineKeyboardButton']]
) -> None: ...
async def edit_message_reply_markup(self, chat_id: int, message_id: int) -> None: ...
async def get_chat_member(self, chat_id: int, user_id: int) -> dict[str, typing.Any]: ...
async def create_chat_invite_link(self, chat_id: int, requires_approval: bool = False) -> str: ...
@@ -184,6 +205,7 @@ class Usecase:
validate_login_token = validate_login_token
telegram_login = telegram_login
telegram_start = telegram_start
get_me = get_me
connect_target_chan = connect_target_chan
get_user_target_chans = get_user_target_chans
@@ -200,6 +222,9 @@ class Usecase:
create_creative = create_creative
update_creative = update_creative
delete_creative = delete_creative
start_creative_creation = start_creative_creation
handle_creative_callback = handle_creative_callback
handle_creative_text_input = handle_creative_text_input
get_placements = get_placements
get_placement = get_placement
create_placement = create_placement

55
src/usecase/creative/handle_creative_callback.py Normal file
View File

@@ -0,0 +1,55 @@
import logging
import uuid
from typing import TYPE_CHECKING
from src import domain
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def handle_creative_callback(
self: 'Usecase', telegram_id: int, chat_id: int, callback_data: str, message_id: int
) -> None:
async with self.database.transaction():
state = await self.database.get_telegram_state(telegram_id)
if not state or state.state != domain.TelegramStateEnum.CREATIVE_WAITING_CHANNEL:
log.debug('User %s has no active creative creation flow or wrong state', telegram_id)
return
if not callback_data.startswith('cc_ch:'):
log.warning('Invalid callback data format: %s', callback_data)
return
channel_id_str = callback_data.split(':', 1)[1]
try:
channel_id = uuid.UUID(channel_id_str)
except ValueError:
log.error('Invalid channel UUID in callback data: %s', channel_id_str)
return
user = await self.database.get_user(telegram_id=telegram_id)
if not user:
await self.telegram_bot.send_message('❌ Ошибка: пользователь не найден.', chat_id=chat_id)
await self.database.clear_telegram_state(telegram_id)
return
channel = await self.database.get_target_channel(user.id, channel_id=channel_id)
if not channel:
await self.telegram_bot.send_message('❌ Канал не найден или не принадлежит вам.', chat_id=chat_id)
await self.database.clear_telegram_state(telegram_id)
return
await self.database.set_telegram_state(
telegram_id=telegram_id,
state=domain.TelegramStateEnum.CREATIVE_WAITING_NAME,
context={'target_channel_id': str(channel_id), 'channel_title': channel.title},
)
await self.telegram_bot.edit_message_reply_markup(chat_id, message_id)
await self.telegram_bot.send_message(
f'✅ Канал выбран: {channel.title}\n\n📝 Введите название креатива:', chat_id=chat_id
)

98
src/usecase/creative/handle_creative_text_input.py Normal file
View File

@@ -0,0 +1,98 @@
import logging
import uuid
from typing import TYPE_CHECKING
from src import domain
from .text_formatting import prepare_creative_text_html
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def handle_creative_text_input(
self: 'Usecase',
telegram_id: int,
chat_id: int,
text: str,
text_html: str | None = None,
) -> None:
async with self.database.transaction():
state = await self.database.get_telegram_state(telegram_id)
if not state:
return
user = await self.database.get_user(telegram_id=telegram_id)
if not user:
await self.telegram_bot.send_message('❌ Ошибка: пользователь не найден.', chat_id=chat_id)
await self.database.clear_telegram_state(telegram_id)
return
if state.state == domain.TelegramStateEnum.CREATIVE_WAITING_NAME:
await self.database.set_telegram_state(
telegram_id=telegram_id,
state=domain.TelegramStateEnum.CREATIVE_WAITING_TEXT,
context={**state.context, 'creative_name': text},
)
await self.telegram_bot.send_message(
'✅ Название сохранено!\n\n'
'💬 Теперь отправьте или перешлите текст креатива.\n'
'Используйте {link} для обозначения места вставки ссылки на ваш канал.\n'
'Вы также можете указать текст ссылки в формате [текст]{link}.',
chat_id=chat_id,
)
elif state.state == domain.TelegramStateEnum.CREATIVE_WAITING_TEXT:
target_channel_id_str = state.context.get('target_channel_id')
creative_name = state.context.get('creative_name')
if not target_channel_id_str or not creative_name:
log.error('Missing context data for creative creation: %s', state.context)
await self.telegram_bot.send_message('❌ Ошибка: данные не найдены. Начните заново.', chat_id=chat_id)
await self.database.clear_telegram_state(telegram_id)
return
try:
target_channel_id = uuid.UUID(target_channel_id_str)
except ValueError:
log.error('Invalid target_channel_id in context: %s', target_channel_id_str)
await self.telegram_bot.send_message('❌ Ошибка: неверный формат данных.', chat_id=chat_id)
await self.database.clear_telegram_state(telegram_id)
return
channel = await self.database.get_target_channel(user.id, channel_id=target_channel_id)
if not channel:
await self.telegram_bot.send_message('❌ Канал не найден.', chat_id=chat_id)
await self.database.clear_telegram_state(telegram_id)
return
message_html = text_html or text
creative_text = prepare_creative_text_html(message_html)
creative = domain.Creative(
name=creative_name,
text=creative_text,
status=domain.CreativeStatus.ACTIVE,
user_id=user.id,
target_channel_id=target_channel_id,
)
created = await self.database.create_creative(creative)
await self.database.clear_telegram_state(telegram_id)
text_preview_source = text
text_preview = text_preview_source[:100] + '...' if len(text_preview_source) > 100 else text_preview_source
await self.telegram_bot.send_message(
f'✅ Креатив успешно создан!\n\n'
f'📝 Название: {created.name}\n'
f'📢 Канал: {channel.title}\n\n'
f'💬 Текст:\n{text_preview}',
chat_id=chat_id,
)
else:
log.warning('Unknown state: %s', state.state)
await self.database.clear_telegram_state(telegram_id)

43
src/usecase/creative/start_creative_creation.py Normal file
View File

@@ -0,0 +1,43 @@
import logging
from typing import TYPE_CHECKING
from aiogram.types import InlineKeyboardButton
from src import domain
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def start_creative_creation(self: 'Usecase', telegram_id: int, chat_id: int) -> None:
async with self.database.transaction():
user = await self.database.get_user(telegram_id=telegram_id)
if not user:
await self.telegram_bot.send_message(
'❌ Вы не авторизованы. Используйте /start для входа.', chat_id=chat_id
)
return
channels = await self.database.get_user_target_channels(user.id)
if not channels:
await self.telegram_bot.send_message(
'❌ У вас нет подключенных каналов. Добавьте бота в свой канал для начала работы.', chat_id=chat_id
)
return
buttons = [
[InlineKeyboardButton(text=channel.title, callback_data=f'cc_ch:{channel.id}')] for channel in channels
]
await self.telegram_bot.send_message_with_inline_keyboard(
'✨ Создание нового креатива\n\nВыберите канал, для которого создается креатив:',
chat_id=chat_id,
buttons=buttons,
)
await self.database.set_telegram_state(
telegram_id=telegram_id, state=domain.TelegramStateEnum.CREATIVE_WAITING_CHANNEL, context={}
)

22
src/usecase/creative/text_formatting.py Normal file
View File

@@ -0,0 +1,22 @@
import re
_LINK_WITH_LABEL_PATTERN: re.Pattern[str] = re.compile(r'\[(?P<label>[^\[\]]+?)\]\s*\{link\}', re.IGNORECASE)
_LINK_PLACEHOLDER_PATTERN: re.Pattern[str] = re.compile(r'\{link\}', re.IGNORECASE)
_LINK_CLASS = 'tg-link'
def _replace_label_placeholder(match: re.Match[str]) -> str:
label = match.group('label')
return f'<a class="{_LINK_CLASS}">{label}</a>'
def prepare_creative_text_html(source: str) -> str:
"""Normalize Telegram HTML and replace {link} placeholder with custom elements.
Supported patterns:
- ``{link}`` → ``<a class="tg-link"></a>``
- ``[Label]{link}`` → ``<a class="tg-link">Label</a>``
"""
with_labels = _LINK_WITH_LABEL_PATTERN.sub(_replace_label_placeholder, source)
return _LINK_PLACEHOLDER_PATTERN.sub(f'<a class="{_LINK_CLASS}"></a>', with_labels)

20
src/usecase/telegram_start.py Normal file
View File

@@ -0,0 +1,20 @@
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from . import Usecase
log = logging.getLogger(__name__)
async def telegram_start(self: 'Usecase', telegram_id: int, chat_id: int) -> None:
async with self.database.transaction():
user = await self.database.get_user(telegram_id=telegram_id)
if user:
username_display = f'@{user.username}' if user.username else 'пользователь'
message = f'Привет, {username_display}! 👋\n\nДоступные команды:\n/new_creative - Создать новый креатив'
else:
message = 'Привет! 👋\n\nДля работы с ботом необходимо авторизоваться через веб-интерфейс.'
await self.telegram_bot.send_message(message, chat_id=chat_id)

24
tests/test_creative_text_formatting.py Normal file
View File

@@ -0,0 +1,24 @@
"""Tests for creative text formatting helpers."""
from src.usecase.creative.text_formatting import prepare_creative_text_html
def test_placeholder_is_replaced_with_custom_tag() -> None:
source = 'Подписывайтесь: {link}!'
assert prepare_creative_text_html(source) == 'Подписывайтесь: <a class="tg-link"></a>!'
def test_label_placeholder_is_wrapped() -> None:
source = 'Жми [сюда]{link} скорее'
assert prepare_creative_text_html(source) == 'Жми <a class="tg-link">сюда</a> скорее'
def test_placeholder_replacement_is_case_insensitive() -> None:
source = 'Первый {LINK}, второй {Link}.'
expected = 'Первый <a class="tg-link"></a>, второй <a class="tg-link"></a>.'
assert prepare_creative_text_html(source) == expected
def test_plain_html_left_untouched() -> None:
source = '<b>Жми</b> и <i>делись</i>'
assert prepare_creative_text_html(source) == source