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

new telegram bot

Browse Source
This commit is contained in:
Artem Tsyrulnikov
2025-12-28 15:50:06 +03:00
parent 69179c64be
commit 2fb8c83041
75 changed files with 3719 additions and 1046 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/usecase/__init__.py
View File

@@ -3,6 +3,7 @@ import typing
from dataclasses import dataclass
from uuid import UUID
if typing.TYPE_CHECKING:
from aiogram.types import InlineKeyboardButton
@@ -12,19 +13,17 @@ from .analytics.get_channel_analytics import get_channel_analytics
from .analytics.get_creatives_analytics import get_creatives_analytics
from .analytics.get_placements_analytics import get_placements_analytics
from .analytics.get_spending_analytics import get_spending_analytics
from .auth.get_jwt_by_telegram_id import get_jwt_by_telegram_id
from .auth.create_telegram_login_token import create_telegram_login_token
from .auth.get_me import get_me
from .auth.telegram_login import telegram_login
from .auth.telegram_start import telegram_start
from .auth.validate_login_token import validate_login_token
from .channel.attach_channel_to_workspace import attach_channel_to_workspace
from .channel.get_channel import get_channel
from .channel.get_channels import get_channels
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.tg_create_creative_1 import tg_create_creative_1
from .creative.tg_create_creative_2 import tg_create_creative_2
from .creative.tg_create_creative_3 import tg_create_creative_3
from .creative.tg_create_creative_4 import tg_create_creative_4
from .creative.update_creative import update_creative
from .placement.create_placement import create_placement
from .placement.delete_placement import delete_placement
@@ -33,9 +32,10 @@ from .placement.get_placement import get_placement
from .placement.get_placements import get_placements
from .placement.update_placement import update_placement
from .project.disconnect_project_by_tg_id import disconnect_project_by_tg_id
from .project.get_project import get_project
from .project.get_workspace_projects import get_workspace_projects
from .project.tg_add_project_1 import tg_add_project_1
from .project.tg_add_project_2 import tg_add_project_2
from .project.tg_add_project import tg_add_project
from .project.update_project_invite_link_type import update_project_invite_link_type
from .project.update_project_permissions import update_project_permissions
from .purchase_plan.attach_channel_to_purchase_plan import attach_channel_to_purchase_plan
from .purchase_plan.get_purchase_plan_channels import get_purchase_plan_channels
@@ -44,6 +44,7 @@ from .purchase_plan.update_purchase_plan_channel import update_purchase_plan_cha
from .subscription.handle_subscription import handle_subscription
from .subscription.handle_unsubscription import handle_unsubscription
from .views.get_views_history import get_views_history
from .workspace.accept_workspace_invite import accept_workspace_invite
from .workspace.create_workspace import create_workspace
from .workspace.create_workspace_invite import create_workspace_invite
from .workspace.delete_workspace import delete_workspace
@@ -371,15 +372,18 @@ class Usecase:
return plan
validate_login_token = validate_login_token
telegram_login = telegram_login
telegram_start = telegram_start
create_telegram_login_token = create_telegram_login_token
get_jwt_by_telegram_id = get_jwt_by_telegram_id
get_me = get_me
tg_add_project_1 = tg_add_project_1
tg_add_project_2 = tg_add_project_2
tg_add_project = tg_add_project
get_workspace_projects = get_workspace_projects
get_project = get_project
disconnect_project_by_tg_id = disconnect_project_by_tg_id
update_project_permissions = update_project_permissions
update_project_invite_link_type = update_project_invite_link_type
get_channels = get_channels
get_channel = get_channel
attach_channel_to_workspace = attach_channel_to_workspace
attach_channel_to_purchase_plan = attach_channel_to_purchase_plan
get_purchase_plan_channels = get_purchase_plan_channels
update_purchase_plan_channel = update_purchase_plan_channel
@@ -389,10 +393,6 @@ class Usecase:
create_creative = create_creative
update_creative = update_creative
delete_creative = delete_creative
tg_create_creative_1 = tg_create_creative_1
tg_create_creative_2 = tg_create_creative_2
tg_create_creative_3 = tg_create_creative_3
tg_create_creative_4 = tg_create_creative_4
get_placements = get_placements
get_placement = get_placement
create_placement = create_placement
@@ -414,4 +414,5 @@ class Usecase:
update_workspace_member_permissions = update_workspace_member_permissions
create_workspace_invite = create_workspace_invite
get_workspace_invites = get_workspace_invites
accept_workspace_invite = accept_workspace_invite
tg_accept_workspace_invite = tg_accept_workspace_invite

32
src/usecase/auth/create_telegram_login_token.py Normal file
View File

@@ -0,0 +1,32 @@
import datetime
import secrets
import typing
from tortoise import timezone
from src import domain
if typing.TYPE_CHECKING:
from .. import Usecase
async def create_telegram_login_token(self: 'Usecase', telegram_id: int, username: str | None) -> str:
user = await self.database.get_user(telegram_id=telegram_id)
if not user:
user = domain.User(
telegram_id=telegram_id,
username=username,
)
await self.database.create_user(user)
token = secrets.token_urlsafe(32)
expires_at = timezone.now() + datetime.timedelta(minutes=10)
login_token = domain.LoginToken(
token=token,
user=user,
expires_at=expires_at,
)
await self.database.create_login_token(login_token)
return token

23
src/usecase/auth/get_jwt_by_telegram_id.py Normal file
View File

@@ -0,0 +1,23 @@
import typing
from src import domain, dto
if typing.TYPE_CHECKING:
from .. import Usecase
async def get_jwt_by_telegram_id(self: 'Usecase', telegram_id: int) -> dto.ValidateLoginTokenOutput:
user = await self.database.get_user(telegram_id=telegram_id)
if not user:
raise domain.UserNotFound()
access_token = self.jwt_encoder.encode_access_token(
user_id=user.id,
telegram_id=user.telegram_id,
username=user.username,
)
return dto.ValidateLoginTokenOutput(
access_token=access_token,
)

40
src/usecase/auth/telegram_login.py
View File

@@ -1,40 +0,0 @@
import datetime
import secrets
import typing
from tortoise import timezone
from src import domain, dto
from src.config import settings
if typing.TYPE_CHECKING:
from .. import Usecase
async def telegram_login(self: 'Usecase', input: dto.TelegramLoginInput) -> None:
user = await self.database.get_user(telegram_id=input.telegram_id)
if not user:
user = domain.User(
telegram_id=input.telegram_id,
username=input.username,
)
await self.database.create_user(user)
token = secrets.token_urlsafe(32)
expires_at = timezone.now() + datetime.timedelta(minutes=10)
login_token = domain.LoginToken(
token=token,
user=user,
expires_at=expires_at,
)
await self.database.create_login_token(login_token)
login_url = f'{settings.app.URL}{login_token.token}'
await self.telegram_bot.send_message_with_button(
f'Привет, {user.username or "аноним"}!\n\nНажми кнопку ниже для входа на сайт.',
input.chat_id,
button_text='Войти на сайт',
button_url=login_url,
)

19
src/usecase/auth/telegram_start.py
View File

@@ -1,19 +0,0 @@
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:
user = await self.database.get_user(telegram_id=telegram_id)
if not user:
message = 'Привет! 👋\n\nДля работы с ботом необходимо авторизоваться через веб-интерфейс.'
await self.telegram_bot.send_message(message, chat_id)
return
message = f'Привет, {user.username or "аноним"}! 👋\n\nДоступные команды:\n/new_creative - Создать новый креатив'
await self.telegram_bot.send_message(message, chat_id)

52
src/usecase/channel/attach_channel_to_workspace.py Normal file
View File

@@ -0,0 +1,52 @@
import logging
from typing import TYPE_CHECKING
from fastapi import HTTPException
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def attach_channel_to_workspace(
self: 'Usecase', input: dto.AttachChannelToWorkspaceInput
) -> dto.ProjectOutput:
"""Привязать канал к workspace (вызывается из Golang бота после выбора workspace пользователем)"""
channel = await self.database.get_channel(channel_id=input.channel_id)
if not channel:
raise HTTPException(status_code=404, detail='Channel not found')
workspace = await self.database.get_workspace(input.workspace_id)
if not workspace:
raise HTTPException(status_code=404, detail='Workspace not found')
# Проверяем что проект еще не существует
project = await self.database.get_project(workspace.id, channel_id=channel.id)
if project:
# Проект уже существует - просто активируем
project.status = domain.ProjectStatus.ACTIVE
await self.database.update_project(project)
log.info('Project %s reactivated in workspace %s', project.id, workspace.id)
else:
# Создаем новый проект
project = domain.Project(
workspace_id=workspace.id,
channel_id=channel.id,
status=domain.ProjectStatus.ACTIVE,
)
await self.database.create_project(project)
log.info('Project created for channel %s in workspace %s', channel.id, workspace.id)
await self.ensure_active_purchase_plan(project)
return dto.ProjectOutput(
id=project.id,
telegram_id=channel.telegram_id,
title=channel.title,
username=channel.username,
status=project.status,
)

22
src/usecase/channel/get_channel.py Normal file
View File

@@ -0,0 +1,22 @@
from typing import TYPE_CHECKING
from fastapi import HTTPException
from src import dto
if TYPE_CHECKING:
from .. import Usecase
async def get_channel(self: 'Usecase', input: dto.GetChannelInput) -> dto.ChannelOutput:
channel = await self.database.get_channel(channel_id=input.channel_id)
if not channel:
raise HTTPException(status_code=404, detail='Channel not found')
return dto.ChannelOutput(
id=channel.id,
telegram_id=channel.telegram_id,
title=channel.title,
username=channel.username,
)

87
src/usecase/creative/tg_create_creative_1.py
View File

@@ -1,87 +0,0 @@
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 tg_create_creative_1(self: 'Usecase', telegram_id: int, chat_id: int) -> None:
user = await self.database.get_user(telegram_id=telegram_id)
if not user:
await self.telegram_bot.send_message('❌ Вы не авторизованы. Используйте /start для входа.', chat_id)
return
memberships = await self.database.get_user_workspaces(user.id)
if not memberships:
await self.telegram_bot.send_message(
'❌ У вас нет рабочих пространств. Создайте workspace в веб-панели, чтобы продолжить.', chat_id
)
await self.database.clear_telegram_state(telegram_id)
return
if len(memberships) == 1:
workspace: domain.Workspace | None = memberships[0].workspace
if not workspace:
workspace = await self.database.get_workspace(memberships[0].workspace_id)
if not workspace:
log.error('Workspace %s not found for user %s', memberships[0].workspace_id, user.id)
await self.telegram_bot.send_message('❌ Рабочее пространство не найдено. Попробуйте позже.', chat_id)
await self.database.clear_telegram_state(telegram_id)
return
projects = await self.database.get_workspace_projects(workspace.id)
if not projects:
await self.telegram_bot.send_message(
'❌ В выбранном рабочем пространстве нет подключенных проектов. Добавьте канал и попробуйте снова.',
chat_id,
)
await self.database.clear_telegram_state(telegram_id)
return
buttons = [
[
InlineKeyboardButton(
text=project.channel.title or project.channel.username or 'Unnamed Channel',
callback_data=f'tg_create_creative_3:{project.id}',
)
]
for project in projects
]
await self.telegram_bot.send_message_with_inline_keyboard(
'✨ Создание нового креатива\n\nВыберите канал, для которого создается креатив:', chat_id, buttons
)
await self.database.set_telegram_state(
telegram_id,
domain.TelegramStateEnum.CREATIVE_WAITING_CHANNEL,
context={'workspace_id': str(workspace.id)},
)
return
buttons = [
[
InlineKeyboardButton(
text=membership.workspace.name if membership.workspace else 'Workspace',
callback_data=f'tg_create_creative_2:{membership.workspace_id}',
)
]
for membership in memberships
]
await self.telegram_bot.send_message_with_inline_keyboard(
'✨ Создание нового креатива\n\nВыберите рабочее пространство:', chat_id, buttons
)
await self.database.set_telegram_state(
telegram_id,
domain.TelegramStateEnum.CREATIVE_WAITING_WORKSPACE,
context={},
)

77
src/usecase/creative/tg_create_creative_2.py
View File

@@ -1,77 +0,0 @@
import logging
import uuid
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 tg_create_creative_2(
self: 'Usecase', telegram_id: int, chat_id: int, callback_data: str, message_id: int
) -> None:
state = await self.database.get_telegram_state(telegram_id)
if not state or state.state != domain.TelegramStateEnum.CREATIVE_WAITING_WORKSPACE:
log.debug('User %s has no active creative workspace selection flow', telegram_id)
return
if not callback_data.startswith('tg_create_creative_2:'):
log.warning('Invalid workspace callback data format: %s', callback_data)
return
workspace_id_str = callback_data.split(':', 1)[1]
try:
workspace_id = uuid.UUID(workspace_id_str)
except ValueError:
log.error('Invalid workspace UUID in callback data: %s', workspace_id_str)
await self.telegram_bot.send_message('❌ Некорректное рабочее пространство.', chat_id)
await self.database.clear_telegram_state(telegram_id)
return
user = await self.database.get_user(telegram_id=telegram_id)
if not user:
await self.telegram_bot.send_message('Вы не зарегистрированы на платформе.', chat_id)
await self.database.clear_telegram_state(telegram_id)
return
workspace = await self.database.get_workspace_for_user(workspace_id, user.id)
if not workspace:
await self.telegram_bot.send_message('❌ Рабочее пространство недоступно.', chat_id)
await self.database.clear_telegram_state(telegram_id)
return
await self.telegram_bot.edit_message_reply_markup(chat_id, message_id)
projects = await self.database.get_workspace_projects(workspace.id)
if not projects:
await self.telegram_bot.send_message(
'❌ В выбранном рабочем пространстве нет подключенных проектов. Добавьте канал и попробуйте снова.',
chat_id,
)
await self.database.clear_telegram_state(telegram_id)
return
buttons = [
[
InlineKeyboardButton(
text=project.channel.title or project.channel.username or 'Unnamed Channel',
callback_data=f'tg_create_creative_3:{project.id}',
)
]
for project in projects
]
await self.telegram_bot.send_message_with_inline_keyboard(
'✨ Создание нового креатива\n\nВыберите канал, для которого создается креатив:', chat_id, buttons
)
await self.database.set_telegram_state(
telegram_id,
domain.TelegramStateEnum.CREATIVE_WAITING_CHANNEL,
context={'workspace_id': str(workspace.id)},
)

81
src/usecase/creative/tg_create_creative_3.py
View File

@@ -1,81 +0,0 @@
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 tg_create_creative_3(
self: 'Usecase', telegram_id: int, chat_id: int, callback_data: str, message_id: int
) -> None:
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('tg_create_creative_3:'):
log.warning('Invalid callback data format: %s', callback_data)
return
project_id_str = callback_data.split(':', 1)[1]
try:
project_id = uuid.UUID(project_id_str)
except ValueError:
log.error('Invalid project UUID in callback data: %s', project_id_str)
return
user = await self.database.get_user(telegram_id=telegram_id)
if not user:
await self.telegram_bot.send_message('Вы не зарегистрированы на платформе.', chat_id)
await self.database.clear_telegram_state(telegram_id)
return
workspace = None
workspace_id = None
workspace_id_str = state.context.get('workspace_id') if state.context else None
if workspace_id_str:
try:
workspace_id = uuid.UUID(workspace_id_str)
except ValueError:
workspace_id = None
if workspace_id:
workspace = await self.database.get_workspace_for_user(workspace_id, user.id)
if not workspace:
workspace = await self.get_or_create_personal_workspace(user)
workspace_id = workspace.id
if not workspace_id:
log.error('Workspace ID missing for user %s when selecting project %s', telegram_id, project_id)
await self.telegram_bot.send_message('❌ Ошибка: рабочее пространство не найдено. Начните заново.', chat_id)
await self.database.clear_telegram_state(telegram_id)
return
project = await self.database.get_project(workspace_id, project_id=project_id)
if not project:
await self.telegram_bot.send_message('❌ Канал не найден или не принадлежит вам.', chat_id)
await self.database.clear_telegram_state(telegram_id)
return
await self.database.set_telegram_state(
telegram_id,
domain.TelegramStateEnum.CREATIVE_WAITING_NAME,
{
'project_id': str(project_id),
'channel_title': project.channel.title,
'workspace_id': str(workspace_id),
},
)
await self.telegram_bot.edit_message_reply_markup(chat_id, message_id)
await self.telegram_bot.send_message(
f'✅ Канал выбран: {project.channel.title}\n\n📝 Введите название креатива:', chat_id
)

112
src/usecase/creative/tg_create_creative_4.py
View File

@@ -1,112 +0,0 @@
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 tg_create_creative_4(
self: 'Usecase', telegram_id: int, chat_id: int, text: str, text_html: str | None = None
) -> None:
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)
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, domain.TelegramStateEnum.CREATIVE_WAITING_TEXT, {**state.context, 'creative_name': text}
)
await self.telegram_bot.send_message(
'✅ Название сохранено!\n\n'
'💬 Теперь отправьте или перешлите текст креатива.\n'
'Пригласительная ссылка на ваш канал будет автоматически '
'заменена на уникальную ссылку для отслеживания.',
chat_id,
)
elif state.state == domain.TelegramStateEnum.CREATIVE_WAITING_TEXT:
project_id_str = state.context.get('project_id')
creative_name = state.context.get('creative_name')
workspace_id_str = state.context.get('workspace_id')
if not project_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)
await self.database.clear_telegram_state(telegram_id)
return
try:
project_id = uuid.UUID(project_id_str)
except ValueError:
log.error('Invalid project_id in context: %s', project_id_str)
await self.telegram_bot.send_message('❌ Ошибка: неверный формат данных.', chat_id)
await self.database.clear_telegram_state(telegram_id)
return
workspace = None
workspace_id = None
if workspace_id_str:
try:
workspace_id = uuid.UUID(workspace_id_str)
except ValueError:
workspace_id = None
if workspace_id:
workspace = await self.database.get_workspace_for_user(workspace_id, user.id)
if not workspace:
workspace = await self.get_or_create_personal_workspace(user)
workspace_id = workspace.id
if not workspace_id:
log.error('Workspace ID missing while finalizing creative for user %s', telegram_id)
await self.telegram_bot.send_message('❌ Ошибка: рабочее пространство не найдено. Начните заново.', chat_id)
await self.database.clear_telegram_state(telegram_id)
return
project = await self.database.get_project(workspace_id, project_id=project_id)
if not project:
await self.telegram_bot.send_message('❌ Канал не найден.', chat_id)
await self.database.clear_telegram_state(telegram_id)
return
message_html = text_html or text
creative_text = domain.replace_invite_link_with_tag(message_html)
creative = domain.Creative(
name=creative_name,
text=creative_text,
status=domain.CreativeStatus.ACTIVE,
workspace_id=workspace_id,
project_id=project_id,
)
async with self.database.transaction():
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'📝 Название: {creative.name}\n'
f'📢 Канал: {project.channel.title}\n\n'
f'💬 Текст:\n{text_preview}',
chat_id,
)
else:
log.warning('Unknown state: %s', state.state)
await self.database.clear_telegram_state(telegram_id)

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

@@ -34,7 +34,8 @@ async def create_placement(
if creative.project_id != project.id:
raise domain.CreativeNotFound(input.creative_id)
requires_approval = input.invite_link_type == domain.InviteLinkType.APPROVAL
# Берем тип ссылки из настроек проекта
requires_approval = project.invite_link_type == domain.InviteLinkType.APPROVAL
if project.channel.telegram_id is None:
raise domain.ChannelNotFound(project.channel.id)
@@ -49,7 +50,6 @@ async def create_placement(
placement_date=input.placement_date,
cost=input.cost,
comment=input.comment,
invite_link_type=input.invite_link_type,
invite_link=invite_link,
status=domain.PlacementStatus.ACTIVE,
)
@@ -68,7 +68,7 @@ async def create_placement(
cost=placement.cost,
comment=placement.comment,
ad_post_url=None,
invite_link_type=placement.invite_link_type,
invite_link_type=project.invite_link_type,
invite_link=placement.invite_link,
status=placement.status,
subscriptions_count=0,

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

@@ -37,7 +37,7 @@ async def get_placement(self: 'Usecase', input: dto.GetPlacementInput) -> dto.Pl
cost=placement.cost,
comment=placement.comment,
ad_post_url=ad_post_url,
invite_link_type=placement.invite_link_type,
invite_link_type=placement.project.invite_link_type,
invite_link=placement.invite_link,
status=placement.status,
subscriptions_count=subscriptions_count,

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

@@ -46,7 +46,7 @@ async def get_placements(self: 'Usecase', input: dto.GetPlacementsInput) -> list
cost=placement.cost,
comment=placement.comment,
ad_post_url=ad_post_url,
invite_link_type=placement.invite_link_type,
invite_link_type=placement.project.invite_link_type,
invite_link=placement.invite_link,
status=placement.status,
subscriptions_count=subscriptions_counts.get(placement.id, 0),

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

@@ -53,7 +53,7 @@ async def update_placement(
cost=placement.cost,
comment=placement.comment,
ad_post_url=ad_post_url,
invite_link_type=placement.invite_link_type,
invite_link_type=placement.project.invite_link_type,
invite_link=placement.invite_link,
status=placement.status,
subscriptions_count=subscriptions_count,

24
src/usecase/project/get_project.py Normal file
View File

@@ -0,0 +1,24 @@
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
async def get_project(self: 'Usecase', input: dto.GetProjectInput) -> dto.ProjectOutput:
project = await self.database.get_project(input.workspace_id, project_id=input.project_id)
if not project:
raise domain.ProjectNotFound()
await project.fetch_related('channel')
return dto.ProjectOutput(
id=project.id,
telegram_id=project.channel.telegram_id,
title=project.channel.title,
username=project.channel.username,
status=project.status,
invite_link_type=project.invite_link_type,
)

1
src/usecase/project/get_workspace_projects.py
View File

@@ -24,6 +24,7 @@ async def get_workspace_projects(
title=project.channel.title,
username=project.channel.username,
status=project.status,
invite_link_type=project.invite_link_type,
)
for project in projects
]

59
src/usecase/project/tg_add_project_1.py → src/usecase/project/tg_add_project.py
View File

@@ -11,7 +11,7 @@ if TYPE_CHECKING:
log = logging.getLogger(__name__)
async def tg_add_project_1(self: 'Usecase', input: dto.ConnectProjectInput) -> dto.ProjectOutput | None:
async def tg_add_project(self: 'Usecase', input: dto.ConnectProjectInput) -> dto.ProjectOutput | None:
permissions = input.bot_permissions
if not permissions.is_admin:
@@ -125,41 +125,40 @@ async def tg_add_project_1(self: 'Usecase', input: dto.ConnectProjectInput) -> d
status=project.status,
)
current_state = await self.database.get_telegram_state(user.telegram_id)
if current_state and current_state.state == domain.TelegramStateEnum.PROJECT_WAITING_WORKSPACE:
await self.telegram_bot.send_message(
'⚠️ Сначала завершите выбор рабочего пространства для предыдущего канала.', user.telegram_id
# Если >1 workspace - создаем/обновляем канал и отправляем уведомление
channel = await self.database.get_channel(telegram_id=input.telegram_id)
if channel:
channel.title = input.title
if input.username is not None:
channel.username = input.username
await self.database.update_channel(channel)
else:
if input.username is None:
log.warning('Cannot create channel %s without username', input.telegram_id)
await self.telegram_bot.send_message(
f'⚠️ Канал "{input.title}" не может быть подключен.\n\nУ канала отсутствует публичный username.',
input.user_telegram_id,
)
return None
channel = domain.Channel(
telegram_id=input.telegram_id,
title=input.title,
username=input.username,
)
return None
await self.database.create_channel(channel)
context: dict[str, object] = {
'channel_data': {
'telegram_id': input.telegram_id,
'title': input.title,
'username': input.username,
'permissions': input.bot_permissions.model_dump(),
}
}
if current_state:
context['previous_state'] = current_state.state.value
context['previous_context'] = dict(current_state.context or {})
await self.database.set_telegram_state(
user.telegram_id,
domain.TelegramStateEnum.PROJECT_WAITING_WORKSPACE,
context,
)
buttons = [
[InlineKeyboardButton(text=workspace.name, callback_data=f'tg_add_project_2:{workspace.id}')]
for workspace in workspaces
]
# Callback будет обработан Golang ботом который покажет экран выбора workspace
buttons = [[InlineKeyboardButton(text='Выбрать workspace', callback_data=f'pending_channel:{channel.id}')]]
await self.telegram_bot.send_message_with_inline_keyboard(
(f'✨ Подключение канала\n\nВыберите рабочее пространство, в которое добавить канал "{input.title}":'),
f'Бот добавлен в канал "{input.title}".\n\n'
f'У вас {len(workspaces)} рабочих пространств. '
'Нажмите кнопку ниже чтобы выбрать, где создать проект.',
user.telegram_id,
buttons,
)
log.info('Pending channel notification sent to user %s for channel %s', user.telegram_id, channel.id)
return None

108
src/usecase/project/tg_add_project_2.py
View File

@@ -1,108 +0,0 @@
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 tg_add_project_2(
self: 'Usecase', telegram_id: int, chat_id: int, callback_data: str, message_id: int
) -> None:
state = await self.database.get_telegram_state(telegram_id)
if not state or state.state != domain.TelegramStateEnum.PROJECT_WAITING_WORKSPACE:
log.debug('User %s has no active project workspace selection flow', telegram_id)
return
context = state.context or {}
previous_state_value = context.get('previous_state')
previous_context = context.get('previous_context') or {}
previous_state = None
if previous_state_value:
try:
previous_state = domain.TelegramStateEnum(previous_state_value)
except ValueError:
previous_state = None
if not callback_data.startswith('tg_add_project_2:'):
log.warning('Invalid project workspace callback data: %s', callback_data)
return
workspace_id_str = callback_data.split(':', 1)[1]
try:
workspace_id = uuid.UUID(workspace_id_str)
except ValueError:
log.error('Invalid workspace UUID in callback data: %s', workspace_id_str)
await self.telegram_bot.send_message('❌ Некорректное рабочее пространство.', chat_id)
if previous_state:
await self.database.set_telegram_state(telegram_id, previous_state, previous_context)
else:
await self.database.clear_telegram_state(telegram_id)
return
channel_data = context.get('channel_data')
if not channel_data:
log.error('Missing channel data for user %s in project flow', telegram_id)
await self.telegram_bot.send_message('❌ Не удалось подключить канал. Попробуйте снова.', chat_id)
if previous_state:
await self.database.set_telegram_state(telegram_id, previous_state, previous_context)
else:
await self.database.clear_telegram_state(telegram_id)
return
user = await self.database.get_user(telegram_id=telegram_id)
if not user:
await self.telegram_bot.send_message('Вы не зарегистрированы на платформе.', chat_id)
await self.database.clear_telegram_state(telegram_id)
return
workspace = await self.database.get_workspace_for_user(workspace_id, user.id)
if not workspace:
await self.telegram_bot.send_message('❌ У вас нет доступа к выбранному рабочему пространству.', chat_id)
return
telegram_channel_id = int(channel_data['telegram_id'])
channel = await self.database.get_channel(telegram_id=telegram_channel_id)
if channel:
channel.title = str(channel_data['title'])
channel.username = channel_data.get('username')
await self.database.update_channel(channel)
else:
channel = domain.Channel(
telegram_id=telegram_channel_id,
title=str(channel_data['title']),
username=channel_data.get('username'),
)
await self.database.create_channel(channel)
project = await self.database.get_project(workspace.id, channel_id=channel.id)
if project:
project.status = domain.ProjectStatus.ACTIVE
await self.database.update_project(project)
else:
project = domain.Project(workspace_id=workspace.id, channel_id=channel.id, status=domain.ProjectStatus.ACTIVE)
await self.database.create_project(project)
await self.ensure_active_purchase_plan(project)
log.info(
'Channel %s connected/updated successfully in workspace %s by user %s',
channel.telegram_id,
workspace.id,
telegram_id,
)
await self.telegram_bot.send_message(
f'✅ Канал "{channel.title}" добавлен в рабочее пространство "{workspace.name}".', telegram_id
)
await self.telegram_bot.edit_message_reply_markup(chat_id, message_id)
if previous_state:
await self.database.set_telegram_state(telegram_id, previous_state, previous_context)
else:
await self.database.clear_telegram_state(telegram_id)

42
src/usecase/project/update_project_invite_link_type.py Normal file
View File

@@ -0,0 +1,42 @@
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 update_project_invite_link_type(
self: 'Usecase',
workspace_id: uuid.UUID,
project_id: uuid.UUID,
invite_link_type: domain.InviteLinkType,
user_id: uuid.UUID,
) -> domain.Project:
await self.ensure_workspace_permission(
workspace_id,
user_id,
domain.PermissionKey.PROJECTS_WRITE,
for_project_id=project_id,
)
async with self.database.transaction():
project = await self.database.get_project(workspace_id, project_id=project_id)
if not project:
raise domain.ProjectNotFound(project_id)
project.invite_link_type = invite_link_type
await project.save()
log.info(
'Project %s invite_link_type updated to %s by user %s',
project_id,
invite_link_type,
user_id,
)
return project

36
src/usecase/workspace/accept_workspace_invite.py Normal file
View File

@@ -0,0 +1,36 @@
from __future__ import annotations
import uuid
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
async def accept_workspace_invite(
self: Usecase,
invite_id: uuid.UUID,
user_id: uuid.UUID,
) -> dto.WorkspaceInviteOutput:
invite = await self.database.get_workspace_invite(invite_id)
if not invite or invite.user_id != user_id:
raise domain.WorkspaceInviteNotFound()
if invite.status != domain.WorkspaceInviteStatus.PENDING:
raise domain.WorkspaceInviteAlreadyProcessed()
async with self.database.transaction():
invite.status = domain.WorkspaceInviteStatus.ACCEPTED
await self.database.update_workspace_invite(invite)
membership = await self.database.get_workspace_membership(invite.workspace_id, user_id)
if membership:
if membership.status != domain.WorkspaceUserStatus.ACTIVE:
membership.status = domain.WorkspaceUserStatus.ACTIVE
await self.database.update_workspace_user(membership)
else:
await self.database.add_user_to_workspace(invite.workspace_id, user_id)
return dto.WorkspaceInviteOutput.from_domain(invite)