ref

src/domain/__init__.py

@@ -20,6 +20,7 @@ __all__ = (
     'PurchasePlanStatus',
     'PurchasePlanChannelStatus',
     'Creative',
+    'replace_invite_link_with_tag',
     'Placement',
     'Post',
     'Subscription',
@@ -52,17 +53,21 @@ __all__ = (
     'TelegramChannelNotFound',
     'CreativeNotFound',
     'CreativeInUse',
+    'CreativeInviteLinkNotFound',
+    'CreativeMultipleInviteLinks',
     'PlacementNotFound',
     'UserByUsernameNotFound',
 )
 
 from .channel import Channel
-from .creative import Creative, CreativeStatus
+from .creative import Creative, CreativeStatus, replace_invite_link_with_tag
 from .error import (
     ChannelAlreadyExists,
     ChannelNoAdminRights,
     ChannelNotFound,
     CreativeInUse,
+    CreativeInviteLinkNotFound,
+    CreativeMultipleInviteLinks,
     CreativeNotFound,
     LoginTokenAlreadyUsed,
     LoginTokenExpired,

src/domain/creative.py

@@ -1,10 +1,12 @@
 import enum
+import re
 from typing import TYPE_CHECKING
 from uuid import UUID
 
 from tortoise import fields
 
 from .base import TimestampedModel
+from .error import CreativeInviteLinkNotFound, CreativeMultipleInviteLinks
 
 if TYPE_CHECKING:
     from .project import Project
@@ -34,3 +36,24 @@ class Creative(TimestampedModel):
 
     class Meta:
         table = 'creative'
+
+
+_INVITE_LINK_HTML = re.compile(r'<a\s+href=["\']https?://t\.me/\+[a-zA-Z0-9_-]+["\']>([^<]*)</a>', re.IGNORECASE)
+_INVITE_LINK_PLAIN = re.compile(r'https?://t\.me/\+[a-zA-Z0-9_-]+', re.IGNORECASE)
+
+
+def replace_invite_link_with_tag(text: str) -> str:
+    """Replace Telegram invite link (t.me/+xxx) with tracking tag."""
+
+    html_count = len(_INVITE_LINK_HTML.findall(text))
+    plain_count = len(_INVITE_LINK_PLAIN.findall(text))
+    total = html_count + plain_count
+
+    if total == 0:
+        raise CreativeInviteLinkNotFound()
+    if total > 1:
+        raise CreativeMultipleInviteLinks()
+
+    if html_count == 1:
+        return _INVITE_LINK_HTML.sub(r'<a class="tg-link">\1</a>', text)
+    return _INVITE_LINK_PLAIN.sub('<a class="tg-link"></a>', text)

src/domain/error.py

@@ -71,6 +71,14 @@ def CreativeNotFound(creative_id: uuid.UUID | None = None) -> HTTPException:
     return HTTPException(status.HTTP_404_NOT_FOUND, f'Creative {creative_id} not found')
 
 
+def CreativeInviteLinkNotFound() -> HTTPException:
+    return HTTPException(status.HTTP_400_BAD_REQUEST, 'Creative text must contain one invite link (t.me/+xxx)')
+
+
+def CreativeMultipleInviteLinks() -> HTTPException:
+    return HTTPException(status.HTTP_400_BAD_REQUEST, 'Creative text must contain only one invite link')
+
+
 def CreativeInUse(creative_id: uuid.UUID) -> HTTPException:
     return HTTPException(
         status.HTTP_400_BAD_REQUEST,

src/domain/post.py

@@ -24,3 +24,9 @@ class Post(TimestampedModel):
     class Meta:
         table = 'post'
         unique_together = (('channel_id', 'message_id'),)
+
+    @property
+    def url(self) -> str | None:
+        if not self.channel.username:
+            return None
+        return f'https://t.me/{self.channel.username}/{self.message_id}'

src/domain/workspace.py

@@ -101,10 +101,14 @@ class PermissionScopeType(enum.StrEnum):
 
 
 class WorkspaceUserPermission(TimestampedModel):
+    permission = fields.CharEnumField(PermissionKey)
+
     workspace_user: fields.ForeignKeyRelation[WorkspaceUser] = fields.ForeignKeyField(
         'models.WorkspaceUser', related_name='permissions', on_delete=fields.CASCADE, index=True
     )
-    permission = fields.CharEnumField(PermissionKey)
+
+    if TYPE_CHECKING:
+        workspace_user_id: uuid.UUID
 
     class Meta:
         table = 'workspace_user_permission'
@@ -120,6 +124,9 @@ class WorkspaceUserPermissionScope(TimestampedModel):
         'models.WorkspaceUser', related_name='permission_scopes', on_delete=fields.CASCADE, index=True
     )
 
+    if TYPE_CHECKING:
+        workspace_user_id: uuid.UUID
+
     class Meta:
         table = 'workspace_user_permission_scope'
         unique_together = (('workspace_user_id', 'permission', 'scope_type', 'scope_id'),)

src/dto/workspace.py

@@ -47,6 +47,35 @@ class WorkspaceMemberOutput(pydantic.BaseModel):
     user: WorkspaceMemberUserOutput
     permissions: list[WorkspacePermissionOutput]
 
+    @classmethod
+    def from_domain(cls, member: domain.WorkspaceUser) -> 'WorkspaceMemberOutput':
+        if member.user is None:
+            raise ValueError('Workspace member user relation is not loaded')
+
+        permissions_map: dict[domain.PermissionKey, list[WorkspacePermissionScopeOutput]] = {}
+
+        for permission in getattr(member, 'permissions', []) or []:
+            permissions_map.setdefault(permission.permission, [])
+
+        for scope in getattr(member, 'permission_scopes', []) or []:
+            permissions_map.setdefault(scope.permission, []).append(
+                WorkspacePermissionScopeOutput(type=scope.scope_type, id=scope.scope_id)
+            )
+
+        return cls(
+            id=member.id,
+            status=member.status,
+            user=WorkspaceMemberUserOutput(
+                id=member.user.id,
+                telegram_id=member.user.telegram_id,
+                username=member.user.username,
+            ),
+            permissions=[
+                WorkspacePermissionOutput(key=key, scopes=scopes)
+                for key, scopes in sorted(permissions_map.items(), key=lambda item: item[0].value)
+            ],
+        )
+
 
 class GetWorkspaceMembersOutput(pydantic.BaseModel):
     members: list[WorkspaceMemberOutput]
@@ -119,6 +148,26 @@ class WorkspaceInviteOutput(pydantic.BaseModel):
     user: WorkspaceMemberUserOutput
     invited_by: WorkspaceMemberUserOutput
 
+    @classmethod
+    def from_domain(cls, invite: domain.WorkspaceInvite) -> 'WorkspaceInviteOutput':
+        if invite.user is None or invite.invited_by is None:
+            raise ValueError('Workspace invite relations are not loaded')
+
+        return cls(
+            id=invite.id,
+            status=invite.status,
+            user=WorkspaceMemberUserOutput(
+                id=invite.user.id,
+                telegram_id=invite.user.telegram_id,
+                username=invite.user.username,
+            ),
+            invited_by=WorkspaceMemberUserOutput(
+                id=invite.invited_by.id,
+                telegram_id=invite.invited_by.telegram_id,
+                username=invite.invited_by.username,
+            ),
+        )
+
 
 class GetWorkspaceInvitesOutput(pydantic.BaseModel):
     invites: list[WorkspaceInviteOutput]

src/usecase/creative/create_creative.py

@@ -14,10 +14,7 @@ async def create_creative(
     self: 'Usecase', input: dto.CreateCreativeInput, project_id: uuid.UUID, user_id: uuid.UUID, workspace_id: uuid.UUID
 ) -> dto.CreativeOutput:
     await self.ensure_workspace_permission(
-        workspace_id,
+        workspace_id, user_id, domain.PermissionKey.PROJECTS_WRITE, for_project_id=project_id
-        user_id,
-        domain.PermissionKey.PROJECTS_WRITE,
-        for_project_id=project_id,
     )
 
     project = await self.database.get_project(workspace_id, project_id=project_id)
@@ -25,9 +22,11 @@ async def create_creative(
         log.warning('User %s attempted to create creative for unavailable project %s', user_id, project_id)
         raise domain.ProjectNotFound(project_id)
 
+    creative_text = domain.replace_invite_link_with_tag(input.text)
+
     creative = domain.Creative(
         name=input.name,
-        text=input.text,
+        text=creative_text,
         status=domain.CreativeStatus.ACTIVE,
         project_id=project.id,
         workspace_id=workspace_id,

src/usecase/creative/text_formatting.py

@@ -1,22 +0,0 @@
-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)

src/usecase/creative/tg_create_creative_4.py

@@ -4,8 +4,6 @@ from typing import TYPE_CHECKING
 
 from src import domain
 
-from .text_formatting import prepare_creative_text_html
-
 if TYPE_CHECKING:
     from .. import Usecase
 
@@ -32,8 +30,8 @@ async def tg_create_creative_4(
         await self.telegram_bot.send_message(
             '✅ Название сохранено!\n\n'
             '💬 Теперь отправьте или перешлите текст креатива.\n'
-            'Используйте {link} для обозначения места вставки ссылки на ваш канал.\n'
+            'Пригласительная ссылка на ваш канал будет автоматически '
-            'Вы также можете указать текст ссылки в формате [текст]{link}.',
+            'заменена на уникальную ссылку для отслеживания.',
             chat_id,
         )
 
@@ -85,7 +83,7 @@ async def tg_create_creative_4(
             return
 
         message_html = text_html or text
-        creative_text = prepare_creative_text_html(message_html)
+        creative_text = domain.replace_invite_link_with_tag(message_html)
 
         creative = domain.Creative(
             name=creative_name,

src/usecase/creative/update_creative.py

@@ -29,7 +29,7 @@ async def update_creative(
     if input.name:
         creative.name = input.name
     if input.text:
-        creative.text = input.text
+        creative.text = domain.replace_invite_link_with_tag(input.text)
     if input.status:
         creative.status = input.status
 

src/usecase/placement/get_placement.py

@@ -3,8 +3,6 @@ from typing import TYPE_CHECKING
 
 from src import domain, dto
 
-from .helpers import generate_post_url
-
 if TYPE_CHECKING:
     from .. import Usecase
 
@@ -23,11 +21,7 @@ async def get_placement(self: 'Usecase', input: dto.GetPlacementInput) -> dto.Pl
 
     context.ensure_project_permission(domain.PermissionKey.PLACEMENTS_READ, placement.project_id)
 
-    # Compute post-related fields
+    ad_post_url = placement.post.url if placement.post else None
-    ad_post_url = None
-
-    if placement.post:
-        ad_post_url = generate_post_url(placement.post.channel.username, placement.post.message_id)
 
     subscriptions_count = await self.database.count_subscriptions_by_placement(placement.id)
 

src/usecase/placement/get_placements.py

@@ -2,8 +2,6 @@ from typing import TYPE_CHECKING
 
 from src import domain, dto
 
-from .helpers import generate_post_url
-
 if TYPE_CHECKING:
     from .. import Usecase
 
@@ -33,10 +31,7 @@ async def get_placements(self: 'Usecase', input: dto.GetPlacementsInput) -> list
 
     placement_outputs = []
     for placement in placements:
-        ad_post_url = None
+        ad_post_url = placement.post.url if placement.post else None
-
-        if placement.post:
-            ad_post_url = generate_post_url(placement.post.channel.username, placement.post.message_id)
 
         placement_outputs.append(
             dto.PlacementOutput(

src/usecase/placement/helpers.py

@@ -1,7 +0,0 @@
-def generate_post_url(channel_username: str | None, message_id: int) -> str | None:
-    """Generate Telegram post URL from channel username and message ID."""
-    if not channel_username:
-        return None
-    # Remove @ prefix if present
-    username = channel_username.lstrip('@')
-    return f'https://t.me/{username}/{message_id}'

src/usecase/placement/update_placement.py

@@ -4,8 +4,6 @@ from typing import TYPE_CHECKING
 
 from src import domain, dto
 
-from .helpers import generate_post_url
-
 if TYPE_CHECKING:
     from .. import Usecase
 
@@ -39,10 +37,7 @@ async def update_placement(
 
     await self.database.update_placement(placement)
 
-    ad_post_url = None
+    ad_post_url = placement.post.url if placement.post else None
-
-    if placement.post:
-        ad_post_url = generate_post_url(placement.post.channel.username, placement.post.message_id)
 
     subscriptions_count = await self.database.count_subscriptions_by_placement(placement.id)
 

src/usecase/workspace/create_workspace_invite.py

@@ -6,7 +6,6 @@ from typing import TYPE_CHECKING
 from aiogram.types import InlineKeyboardButton
 
 from src import domain, dto
-from src.usecase.workspace.mappers import workspace_invite_to_dto
 
 if TYPE_CHECKING:
     from .. import Usecase
@@ -73,4 +72,4 @@ async def create_workspace_invite(
         buttons=[[accept_button]],
     )
 
-    return workspace_invite_to_dto(invite)
+    return dto.WorkspaceInviteOutput.from_domain(invite)

src/usecase/workspace/get_workspace_invites.py

@@ -4,7 +4,6 @@ import uuid
 from typing import TYPE_CHECKING
 
 from src import domain, dto
-from src.usecase.workspace.mappers import workspace_invite_to_dto
 
 if TYPE_CHECKING:
     from .. import Usecase
@@ -17,4 +16,4 @@ async def get_workspace_invites(
 
     invites = await self.database.get_workspace_invites(workspace_id)
 
-    return [workspace_invite_to_dto(invite) for invite in invites]
+    return [dto.WorkspaceInviteOutput.from_domain(invite) for invite in invites]

src/usecase/workspace/get_workspace_members.py

@@ -4,7 +4,6 @@ import uuid
 from typing import TYPE_CHECKING
 
 from src import domain, dto
-from src.usecase.workspace.mappers import workspace_member_to_dto
 
 if TYPE_CHECKING:
     from .. import Usecase
@@ -17,4 +16,4 @@ async def get_workspace_members(
 
     members = await self.database.get_workspace_members(workspace_id)
 
-    return [workspace_member_to_dto(member) for member in members]
+    return [dto.WorkspaceMemberOutput.from_domain(member) for member in members]

src/usecase/workspace/mappers.py

@@ -1,57 +0,0 @@
-from __future__ import annotations
-
-from src import domain, dto
-
-
-def workspace_member_to_dto(member: domain.WorkspaceUser) -> dto.WorkspaceMemberOutput:
-    if member.user is None:
-        raise ValueError('Workspace member user relation is not loaded')
-
-    permissions_map: dict[domain.PermissionKey, list[dto.WorkspacePermissionScopeOutput]] = {}
-
-    for permission in getattr(member, 'permissions', []) or []:
-        permissions_map.setdefault(permission.permission, [])
-
-    for scope in getattr(member, 'permission_scopes', []) or []:
-        permissions_map.setdefault(scope.permission, []).append(
-            dto.WorkspacePermissionScopeOutput(
-                type=scope.scope_type,
-                id=scope.scope_id,
-            )
-        )
-
-    permissions_output = [
-        dto.WorkspacePermissionOutput(key=key, scopes=scopes)
-        for key, scopes in sorted(permissions_map.items(), key=lambda item: item[0].value)
-    ]
-
-    return dto.WorkspaceMemberOutput(
-        id=member.id,
-        status=member.status,
-        user=dto.WorkspaceMemberUserOutput(
-            id=member.user.id,
-            telegram_id=member.user.telegram_id,
-            username=member.user.username,
-        ),
-        permissions=permissions_output,
-    )
-
-
-def workspace_invite_to_dto(invite: domain.WorkspaceInvite) -> dto.WorkspaceInviteOutput:
-    if invite.user is None or invite.invited_by is None:
-        raise ValueError('Workspace invite relations are not loaded')
-
-    return dto.WorkspaceInviteOutput(
-        id=invite.id,
-        status=invite.status,
-        user=dto.WorkspaceMemberUserOutput(
-            id=invite.user.id,
-            telegram_id=invite.user.telegram_id,
-            username=invite.user.username,
-        ),
-        invited_by=dto.WorkspaceMemberUserOutput(
-            id=invite.invited_by.id,
-            telegram_id=invite.invited_by.telegram_id,
-            username=invite.invited_by.username,
-        ),
-    )

src/usecase/workspace/update_workspace_member_permissions.py

@@ -4,7 +4,6 @@ import uuid
 from typing import TYPE_CHECKING
 
 from src import domain, dto
-from src.usecase.workspace.mappers import workspace_member_to_dto
 
 if TYPE_CHECKING:
     from .. import Usecase
@@ -43,4 +42,4 @@ async def update_workspace_member_permissions(
     if not updated_member:
         raise domain.WorkspaceAccessDenied(workspace_id)
 
-    return workspace_member_to_dto(updated_member)
+    return dto.WorkspaceMemberOutput.from_domain(updated_member)