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

feat: parser, refactoring

Browse Source
This commit is contained in:
Artem Tsyrulnikov
2025-12-07 22:17:24 +03:00
parent e90c6bfb75
commit 2af23f84fe
39 changed files with 503 additions and 822 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
src/adapter/postgres.py
View File

@@ -67,12 +67,6 @@ class Postgres(DatabaseBase):
result = await self.session.execute(q)
return result.scalar_one_or_none()
async def create_target_channel(self, channel: domain.TargetChannel) -> domain.TargetChannel:
self.session.add(channel)
await self.session.flush()
await self.session.refresh(channel)
return channel
async def upsert_target_channel(self, channel: domain.TargetChannel) -> domain.TargetChannel:
existing = await self.get_target_channel(user_id=channel.user_id, telegram_id=channel.telegram_id)
@@ -302,14 +296,6 @@ class Postgres(DatabaseBase):
await self.session.execute(q)
await self.session.flush()
async def check_creative_in_use(self, creative_id: uuid.UUID) -> bool:
q = select(domain.Placement).where(
domain.Placement.creative_id == creative_id,
domain.Placement.status == domain.PlacementStatus.ACTIVE,
)
result = await self.session.execute(q)
return result.scalar_one_or_none() is not None
async def get_placement_by_invite_link(self, invite_link: str) -> domain.Placement | None:
q = (
select(domain.Placement)
@@ -323,6 +309,50 @@ class Postgres(DatabaseBase):
result = await self.session.execute(q)
return result.scalar_one_or_none()
async def get_placement_by_id(self, placement_id: uuid.UUID) -> domain.Placement | None:
return await self.session.get(domain.Placement, placement_id)
async def update_placement_from_post_found(
self, placement_id: uuid.UUID, ad_post_url: str, message_id: int, views_count: int
) -> domain.Placement:
placement = await self.session.get(domain.Placement, placement_id)
if not placement:
raise ValueError(f'Placement {placement_id} not found')
placement.status = domain.PlacementStatus.ACTIVE
placement.ad_post_url = ad_post_url
placement.external_channel_message_id = message_id
placement.views_count = views_count
placement.views_availability = domain.PostViewsAvailability.AVAILABLE
placement.last_views_fetch_at = datetime.datetime.now(datetime.UTC)
await self.session.flush()
await self.session.refresh(placement)
return placement
async def update_placement_from_post_deleted(self, placement_id: uuid.UUID) -> domain.Placement:
placement = await self.session.get(domain.Placement, placement_id)
if not placement:
raise ValueError(f'Placement {placement_id} not found')
placement.status = domain.PlacementStatus.DELETED
await self.session.flush()
await self.session.refresh(placement)
return placement
async def update_placement_views(self, placement_id: uuid.UUID, views_count: int) -> domain.Placement:
placement = await self.session.get(domain.Placement, placement_id)
if not placement:
raise ValueError(f'Placement {placement_id} not found')
placement.views_count = views_count
placement.last_views_fetch_at = datetime.datetime.now(datetime.UTC)
await self.session.flush()
await self.session.refresh(placement)
return placement
async def get_subscriber(self, telegram_id: int) -> domain.Subscriber | None:
q = select(domain.Subscriber).where(domain.Subscriber.telegram_id == telegram_id)
result = await self.session.execute(q)
@@ -429,20 +459,6 @@ class Postgres(DatabaseBase):
result = await self.session.execute(q)
return list(result.scalars().all())
async def get_active_placements_with_urls(self) -> list[domain.Placement]:
q = (
select(domain.Placement)
.where(
domain.Placement.status == domain.PlacementStatus.ACTIVE,
domain.Placement.ad_post_url.isnot(None),
domain.Placement.views_availability != domain.PostViewsAvailability.MANUAL,
)
.order_by(domain.Placement.created_at.asc())
)
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)

99
src/adapter/telegram_parser.py
View File

@@ -1,99 +0,0 @@
import logging
import re
from bs4 import BeautifulSoup
from shared.http_parser_base import HttpParserBase, HttpParserConfig
log = logging.getLogger(__name__)
class TelegramParser(HttpParserBase):
def __init__(self, config: HttpParserConfig) -> None:
super().__init__(config)
log.info('TelegramParser initialized (timeout=%s)', config.TIMEOUT)
def _parse_url(self, url: str) -> tuple[str, int] | None:
"""Parse Telegram post URL into username and message_id."""
match = re.match(r'(?:https?://)?t\.me/([\w]+)/(\d+)', url)
if not match:
return None
username, message_id = match.groups()
return username, int(message_id)
def _parse_view_count(self, text: str) -> int | None:
"""Convert view counter text (1.2K, 5.3M) to integer."""
if not text:
return None
text = text.strip().replace(',', '').replace(' ', '')
multipliers = {'K': 1_000, 'M': 1_000_000, 'B': 1_000_000_000}
try:
if text[-1].upper() in multipliers:
return int(float(text[:-1]) * multipliers[text[-1].upper()])
return int(text)
except (ValueError, IndexError):
return None
async def _fetch_from_preview(self, username: str, message_id: int) -> int | None:
url = f'https://t.me/s/{username}/{message_id}'
status_code, html = await self.fetch_html(url)
if status_code != 200:
return None
soup = BeautifulSoup(html, 'lxml')
messages = soup.find_all('div', class_='tgme_widget_message')
for message_div in messages:
data_post = message_div.get('data-post', '')
if isinstance(data_post, str) and data_post.endswith(f'/{message_id}'):
views_span = message_div.find('span', class_='tgme_widget_message_views')
if views_span:
views_text = views_span.get_text(strip=True)
return self._parse_view_count(views_text)
return None
async def _fetch_from_embed(self, username: str, message_id: int) -> int | None:
url = f'https://t.me/{username}/{message_id}?embed=1'
status_code, html = await self.fetch_html(url)
if status_code != 200:
return None
soup = BeautifulSoup(html, 'lxml')
views_span = soup.find('span', class_='tgme_widget_message_views')
if not views_span:
return None
views_text = views_span.get_text(strip=True)
return self._parse_view_count(views_text)
async def get_post_views(self, post_url: str) -> int | None:
parsed = self._parse_url(post_url)
if not parsed:
log.error('Invalid Telegram URL format: %s', post_url)
return None
username, message_id = parsed
try:
views = await self._fetch_from_preview(username, message_id)
if views is not None:
log.info('Fetched views for %s: %d', post_url, views)
return views
views = await self._fetch_from_embed(username, message_id)
if views is not None:
log.info('Fetched views for %s: %d', post_url, views)
return views
log.warning('Unable to extract views from %s', post_url)
return None
except Exception as exc:
log.exception('Failed to fetch %s: %s', post_url, exc.__class__.__name__)
return None

5
src/config.py
View File

@@ -3,10 +3,8 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
from shared.config_helper import load_settings
from shared.datebase_base import DatabaseConfig
from shared.http_parser_base import HttpParserConfig
from shared.logger import LoggerConfig
from shared.telegram_base import TelegramConfig
from shared.worker_base import WorkerConfig
from src.adapter.jwt import JWTConfig
@@ -21,10 +19,7 @@ class Settings(BaseSettings):
db: DatabaseConfig
logger: LoggerConfig
telegram: TelegramConfig
telegram_parser: HttpParserConfig
jwt: JWTConfig
fetch_views_worker: WorkerConfig
settings: Settings = load_settings(Settings)

18
src/controller/http/analytics.py
View File

@@ -5,7 +5,7 @@ from typing import Annotated
from fastapi import Depends
from fastapi.routing import APIRouter
from src import dependencies, dto
from src import deps, dto
from src.adapter.jwt import JWTPayload
analytics_router = APIRouter(prefix='/analytics', tags=['analytics'])
@@ -13,43 +13,43 @@ analytics_router = APIRouter(prefix='/analytics', tags=['analytics'])
@analytics_router.get('/placements')
async def get_placements_analytics(
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
target_channel_id: uuid.UUID | None = None,
) -> dto.GetPlacementsAnalyticsOutput:
input = dto.GetPlacementsAnalyticsInput(
user_id=current_user.user_id,
target_channel_id=target_channel_id,
)
return await dependencies.get_usecase().get_placements_analytics(input)
return await deps.get_usecase().get_placements_analytics(input)
@analytics_router.get('/creatives')
async def get_creatives_analytics(
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
target_channel_id: uuid.UUID | None = None,
) -> dto.GetCreativesAnalyticsOutput:
input = dto.GetCreativesAnalyticsInput(
user_id=current_user.user_id,
target_channel_id=target_channel_id,
)
return await dependencies.get_usecase().get_creatives_analytics(input)
return await deps.get_usecase().get_creatives_analytics(input)
@analytics_router.get('/external-channels')
async def get_external_channels_analytics(
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
target_channel_id: uuid.UUID | None = None,
) -> dto.GetExternalChannelsAnalyticsOutput:
input = dto.GetExternalChannelsAnalyticsInput(
user_id=current_user.user_id,
target_channel_id=target_channel_id,
)
return await dependencies.get_usecase().get_external_channels_analytics(input)
return await deps.get_usecase().get_external_channels_analytics(input)
@analytics_router.get('/spending')
async def get_spending_analytics(
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
target_channel_id: uuid.UUID | None = None,
date_from: datetime.datetime | None = None,
date_to: datetime.datetime | None = None,
@@ -62,4 +62,4 @@ async def get_spending_analytics(
date_to=date_to,
grouping=grouping,
)
return await dependencies.get_usecase().get_spending_analytics(input)
return await deps.get_usecase().get_spending_analytics(input)

8
src/controller/http/auth.py
View File

@@ -3,7 +3,7 @@ from typing import Annotated
from fastapi import Depends
from fastapi.routing import APIRouter
from src import dependencies, dto
from src import deps, dto
from src.adapter.jwt import JWTPayload
auth_router = APIRouter(prefix='/auth', tags=['auth'])
@@ -11,7 +11,7 @@ auth_router = APIRouter(prefix='/auth', tags=['auth'])
@auth_router.get('/complete')
async def complete_auth(token: str) -> dto.ValidateLoginTokenOutput:
return await dependencies.get_usecase().validate_login_token(
return await deps.get_usecase().validate_login_token(
input=dto.ValidateLoginTokenInput(
token=token,
)
@@ -20,6 +20,6 @@ async def complete_auth(token: str) -> dto.ValidateLoginTokenOutput:
@auth_router.get('/me')
async def get_me(
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.UserOutput:
return await dependencies.get_usecase().get_me(user_id=current_user.user_id)
return await deps.get_usecase().get_me(user_id=current_user.user_id)

22
src/controller/http/creatives.py
View File

@@ -4,7 +4,7 @@ from typing import Annotated
from fastapi import Depends
from fastapi.routing import APIRouter
from src import dependencies, dto
from src import deps, dto
from src.adapter.jwt import JWTPayload
creatives_router = APIRouter(prefix='/creatives', tags=['creatives'])
@@ -12,7 +12,7 @@ creatives_router = APIRouter(prefix='/creatives', tags=['creatives'])
@creatives_router.get('')
async def list_creatives(
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
target_channel_id: uuid.UUID | None = None,
include_archived: bool = False,
) -> dto.GetCreativesOutput:
@@ -22,37 +22,37 @@ async def list_creatives(
include_archived=include_archived,
)
return await dependencies.get_usecase().get_creatives(input=input)
return await deps.get_usecase().get_creatives(input=input)
@creatives_router.get('/{creative_id}')
async def get_creative(
creative_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.CreativeOutput:
input = dto.GetCreativeInput(
creative_id=creative_id,
user_id=current_user.user_id,
)
return await dependencies.get_usecase().get_creative(input=input)
return await deps.get_usecase().get_creative(input=input)
@creatives_router.post('')
async def create_creative(
request: dto.CreateCreativeInput,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.CreativeOutput:
return await dependencies.get_usecase().create_creative(request, current_user.user_id)
return await deps.get_usecase().create_creative(request, current_user.user_id)
@creatives_router.patch('/{creative_id}')
async def update_creative(
creative_id: uuid.UUID,
request: dto.UpdateCreativeInput,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.CreativeOutput:
return await dependencies.get_usecase().update_creative(
return await deps.get_usecase().update_creative(
creative_id=creative_id,
input=request,
user_id=current_user.user_id,
@@ -62,11 +62,11 @@ async def update_creative(
@creatives_router.delete('/{creative_id}')
async def delete_creative(
creative_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> None:
input = dto.DeleteCreativeInput(
creative_id=creative_id,
user_id=current_user.user_id,
)
await dependencies.get_usecase().delete_creative(input)
await deps.get_usecase().delete_creative(input)

22
src/controller/http/external_channels.py
View File

@@ -4,7 +4,7 @@ from typing import Annotated
from fastapi import Depends
from fastapi.routing import APIRouter
from src import dependencies, dto
from src import deps, dto
from src.adapter.jwt import JWTPayload
external_channels_router = APIRouter(prefix='/external_channels', tags=['external_channels'])
@@ -13,7 +13,7 @@ external_channels_router = APIRouter(prefix='/external_channels', tags=['externa
@external_channels_router.get('/target/{target_channel_id}')
async def get_external_channels(
target_channel_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.GetExternalChannelsOutput:
"""Get all external channels for a specific target channel."""
input = dto.GetExternalChannelsInput(
@@ -21,16 +21,16 @@ async def get_external_channels(
user_id=current_user.user_id,
)
return await dependencies.get_usecase().get_external_channels(input=input)
return await deps.get_usecase().get_external_channels(input=input)
@external_channels_router.post('')
async def create_external_channel(
request: dto.CreateExternalChannelInput,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.ExternalChannelOutput:
"""Create a new external channel and link it to target channels."""
return await dependencies.get_usecase().create_external_channel(
return await deps.get_usecase().create_external_channel(
input=request,
user_id=current_user.user_id,
)
@@ -40,10 +40,10 @@ async def create_external_channel(
async def update_external_channel(
channel_id: uuid.UUID,
request: dto.UpdateExternalChannelInput,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.ExternalChannelOutput:
"""Update external channel fields (title, username, description, subscribers_count)."""
return await dependencies.get_usecase().update_external_channel(
return await deps.get_usecase().update_external_channel(
channel_id=channel_id,
input=request,
user_id=current_user.user_id,
@@ -54,14 +54,14 @@ async def update_external_channel(
async def update_external_channel_links(
channel_id: uuid.UUID,
request: dto.UpdateExternalChannelLinksInput,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.ExternalChannelOutput:
"""
Update target channel links for an existing external channel.
You can add new target channels, remove existing ones, or both in a single request.
"""
return await dependencies.get_usecase().update_external_channel_links(
return await deps.get_usecase().update_external_channel_links(
channel_id=channel_id,
input=request,
user_id=current_user.user_id,
@@ -71,7 +71,7 @@ async def update_external_channel_links(
@external_channels_router.delete('/{channel_id}')
async def delete_external_channel(
channel_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> None:
"""Delete an external channel."""
input = dto.DeleteExternalChannelInput(
@@ -79,4 +79,4 @@ async def delete_external_channel(
user_id=current_user.user_id,
)
await dependencies.get_usecase().delete_external_channel(input=input)
await deps.get_usecase().delete_external_channel(input=input)

22
src/controller/http/placements.py
View File

@@ -4,7 +4,7 @@ from typing import Annotated
from fastapi import Depends
from fastapi.routing import APIRouter
from src import dependencies, dto
from src import deps, dto
from src.adapter.jwt import JWTPayload
placements_router = APIRouter(prefix='/placements', tags=['placements'])
@@ -12,7 +12,7 @@ placements_router = APIRouter(prefix='/placements', tags=['placements'])
@placements_router.get('')
async def list_placements(
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
target_channel_id: uuid.UUID | None = None,
external_channel_id: uuid.UUID | None = None,
creative_id: uuid.UUID | None = None,
@@ -26,37 +26,37 @@ async def list_placements(
include_archived=include_archived,
)
return await dependencies.get_usecase().get_placements(input=input)
return await deps.get_usecase().get_placements(input=input)
@placements_router.get('/{placement_id}')
async def get_placement(
placement_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.PlacementOutput:
input = dto.GetPlacementInput(
placement_id=placement_id,
user_id=current_user.user_id,
)
return await dependencies.get_usecase().get_placement(input=input)
return await deps.get_usecase().get_placement(input=input)
@placements_router.post('')
async def create_placement(
request: dto.CreatePlacementInput,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.PlacementOutput:
return await dependencies.get_usecase().create_placement(request, current_user.user_id)
return await deps.get_usecase().create_placement(request, current_user.user_id)
@placements_router.patch('/{placement_id}')
async def update_placement(
placement_id: uuid.UUID,
request: dto.UpdatePlacementInput,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.PlacementOutput:
return await dependencies.get_usecase().update_placement(
return await deps.get_usecase().update_placement(
placement_id=placement_id,
input=request,
user_id=current_user.user_id,
@@ -66,11 +66,11 @@ async def update_placement(
@placements_router.delete('/{placement_id}')
async def delete_placement(
placement_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> None:
input = dto.DeletePlacementInput(
placement_id=placement_id,
user_id=current_user.user_id,
)
await dependencies.get_usecase().delete_placement(input)
await deps.get_usecase().delete_placement(input)

10
src/controller/http/target_channels.py
View File

@@ -4,7 +4,7 @@ from typing import Annotated
from fastapi import Depends
from fastapi.routing import APIRouter
from src import dependencies, dto
from src import deps, dto
from src.adapter.jwt import JWTPayload
target_channels_router = APIRouter(prefix='/target_channels', tags=['target_channels'])
@@ -12,23 +12,23 @@ target_channels_router = APIRouter(prefix='/target_channels', tags=['target_chan
@target_channels_router.get('')
async def get_user_target_chans(
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.GetUserTargetChansOutput:
input = dto.GetUserTargetChansInput(
user_id=current_user.user_id,
)
return await dependencies.get_usecase().get_user_target_chans(input=input)
return await deps.get_usecase().get_user_target_chans(input=input)
@target_channels_router.delete('/{channel_id}')
async def disconnect_target_chan(
channel_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> None:
input = dto.DisconnectTargetChanInput(
channel_id=channel_id,
user_id=current_user.user_id,
)
await dependencies.get_usecase().disconnect_target_chan(input=input)
await deps.get_usecase().disconnect_target_chan(input=input)

23
src/controller/http/views.py
View File

@@ -5,7 +5,7 @@ from typing import Annotated
from fastapi import Depends
from fastapi.routing import APIRouter
from src import dependencies, dto
from src import deps, dto
from src.adapter.jwt import JWTPayload
views_router = APIRouter(prefix='/placements/{placement_id}/views', tags=['views'])
@@ -14,7 +14,7 @@ views_router = APIRouter(prefix='/placements/{placement_id}/views', tags=['views
@views_router.get('/history')
async def get_views_history(
placement_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
from_date: datetime.datetime | None = None,
to_date: datetime.datetime | None = None,
) -> dto.GetViewsHistoryOutput:
@@ -25,27 +25,14 @@ async def get_views_history(
to_date=to_date,
)
return await dependencies.get_usecase().get_views_history(input=input_data)
@views_router.post('/fetch')
async def fetch_views(
placement_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
) -> dto.FetchViewsManuallyOutput:
input_data = dto.FetchViewsInputManually(
placement_id=placement_id,
user_id=current_user.user_id,
)
return await dependencies.get_usecase().fetch_views(input=input_data)
return await deps.get_usecase().get_views_history(input=input_data)
@views_router.post('/manual')
async def update_views_manually(
placement_id: uuid.UUID,
views_count: int,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.PlacementOutput:
input_data = dto.UpdateViewsManuallyInput(
placement_id=placement_id,
@@ -53,4 +40,4 @@ async def update_views_manually(
views_count=views_count,
)
return await dependencies.get_usecase().update_views_manually(input=input_data)
return await deps.get_usecase().update_views_manually(input=input_data)

4
src/controller/telegram_callback/chat_join_request.py
View File

@@ -2,7 +2,7 @@ import logging
from aiogram.types import ChatJoinRequest
from src import dependencies
from src import deps
from src.controller.telegram_callback import telegram_callback_router
log = logging.getLogger(__name__)
@@ -23,7 +23,7 @@ async def on_chat_join_request(event: ChatJoinRequest) -> None:
log.debug('No invite_link in chat join request event')
return
usecase = dependencies.get_usecase()
usecase = deps.get_usecase()
await usecase.handle_subscription(
user_telegram_id=event.from_user.id,

4
src/controller/telegram_callback/chat_member_updated.py
View File

@@ -2,7 +2,7 @@ import logging
from aiogram.types import ChatMemberUpdated
from src import dependencies
from src import deps
from src.controller.telegram_callback import telegram_callback_router
log = logging.getLogger(__name__)
@@ -30,7 +30,7 @@ async def on_chat_member_updated(event: ChatMemberUpdated) -> None:
is_now_not_member = new_status in ('left', 'kicked')
user_left = was_member and is_now_not_member
usecase = dependencies.get_usecase()
usecase = deps.get_usecase()
if user_joined:
invite_link = event.invite_link.invite_link if event.invite_link else None

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

@@ -4,7 +4,7 @@ from aiogram import F
from aiogram.filters import Command
from aiogram.types import CallbackQuery, Message
from src import dependencies
from src import deps
from src.controller.telegram_callback import telegram_callback_router
log = logging.getLogger(__name__)
@@ -16,7 +16,7 @@ async def cmd_new_creative(message: Message) -> None:
log.error('Failed to get user data from message')
return
usecase = dependencies.get_usecase()
usecase = deps.get_usecase()
await usecase.start_creative_creation(telegram_id=message.from_user.id, chat_id=message.chat.id)
@@ -26,7 +26,7 @@ async def callback_creative_channel(callback: CallbackQuery) -> None:
log.error('Failed to get required data from callback query')
return
usecase = dependencies.get_usecase()
usecase = deps.get_usecase()
await usecase.handle_creative_callback(
telegram_id=callback.from_user.id,
chat_id=callback.message.chat.id,
@@ -42,7 +42,7 @@ async def handle_text_message(message: Message) -> None:
if not message.from_user or not message.text:
return
usecase = dependencies.get_usecase()
usecase = deps.get_usecase()
await usecase.handle_creative_text_input(
telegram_id=message.from_user.id,
chat_id=message.chat.id,

4
src/controller/telegram_callback/my_chat_member.py
View File

@@ -3,7 +3,7 @@ import logging
from aiogram.types import ChatMemberAdministrator, ChatMemberUpdated
from fastapi import HTTPException
from src import dependencies, dto
from src import deps, dto
from src.controller.telegram_callback import telegram_callback_router
log = logging.getLogger(__name__)
@@ -18,7 +18,7 @@ async def on_bot_added_to_chat(event: ChatMemberUpdated) -> None:
log.error('Failed to get user data from chat member update')
return
usecase = dependencies.get_usecase()
usecase = deps.get_usecase()
old_status = event.old_chat_member.status
new_status = event.new_chat_member.status

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

@@ -3,7 +3,7 @@ import logging
from aiogram.filters import CommandStart
from aiogram.types import Message
from src import dependencies, dto
from src import deps, dto
from src.controller.telegram_callback import telegram_callback_router
log = logging.getLogger(__name__)
@@ -27,7 +27,7 @@ async def cmd_start_with_deeplink(message: Message) -> None:
chat_id=message.chat.id,
)
await dependencies.get_usecase().telegram_login(input=input)
await deps.get_usecase().telegram_login(input=input)
@telegram_callback_router.message(CommandStart())
@@ -37,5 +37,5 @@ async def cmd_start(message: Message) -> None:
log.error('Failed to get user data from message')
return
usecase = dependencies.get_usecase()
usecase = deps.get_usecase()
await usecase.telegram_start(telegram_id=message.from_user.id, chat_id=message.chat.id)

15
src/controller/worker/views_worker.py
View File

@@ -1,15 +0,0 @@
import logging
from typing import TYPE_CHECKING
from shared.worker_base import WorkerBase
from src import dependencies
if TYPE_CHECKING:
pass
log = logging.getLogger(__name__)
class FetchViewsWorker(WorkerBase):
async def _cycle_func(self) -> None:
await dependencies.get_usecase().run_views_worker_cycle(self.config.INTERVAL_SECONDS)

0
src/dependencies.py → src/deps.py
View File

9
src/domain/placement.py
View File

@@ -17,8 +17,10 @@ if TYPE_CHECKING:
class PlacementStatus(StrEnum):
ACTIVE = 'active'
ARCHIVED = 'archived'
PENDING = 'pending' # Ожидается публикация поста
ACTIVE = 'active' # Пост найден и отслеживается
DELETED = 'deleted' # Пост удален из канала
ARCHIVED = 'archived' # Вручную архивировано
class InviteLinkType(StrEnum):
@@ -27,8 +29,6 @@ class InviteLinkType(StrEnum):
class PostViewsAvailability(StrEnum):
"""Статус доступности получения просмотров поста."""
UNKNOWN = 'unknown' # Ещё не проверялось
AVAILABLE = 'available' # Просмотры доступны
UNAVAILABLE = 'unavailable' # Нет доступа / битая ссылка / приватный канал
@@ -49,6 +49,7 @@ class Placement(domain.Base):
ad_post_url: Mapped[str | None] # ссылка на рекламное сообщение во внешнем канале
invite_link_type: Mapped[InviteLinkType] # тип пригласительной ссылки
invite_link: Mapped[str] # уникальная пригласительная ссылка
external_channel_message_id: Mapped[int | None] # message_id в Telegram после нахождения поста
# Статус
status: Mapped[PlacementStatus] = mapped_column(default=PlacementStatus.ACTIVE)

16
src/dto/__init__.py
View File

@@ -34,8 +34,6 @@ __all__ = (
'PlacementViewsHistoryOutput',
'GetViewsHistoryInput',
'GetViewsHistoryOutput',
'FetchViewsInputManually',
'FetchViewsManuallyOutput',
'UpdateViewsManuallyInput',
'UserOutput',
'DateGrouping',
@@ -51,6 +49,11 @@ __all__ = (
'SpendingDataPoint',
'GetSpendingAnalyticsInput',
'GetSpendingAnalyticsOutput',
'ParserTrackTaskRequest',
'ParserGetViewsRequest',
'ParserPostFoundEvent',
'ParserPostDeletedEvent',
'ParserViewsUpdatedEvent',
)
from .analytics import (
@@ -86,6 +89,13 @@ from .external_channel import (
UpdateExternalChannelInput,
UpdateExternalChannelLinksInput,
)
from .nats_messages import (
ParserGetViewsRequest,
ParserPostDeletedEvent,
ParserPostFoundEvent,
ParserTrackTaskRequest,
ParserViewsUpdatedEvent,
)
from .placement import (
CreatePlacementInput,
DeletePlacementInput,
@@ -109,8 +119,6 @@ from .telegram_login import TelegramLoginInput
from .user import UserOutput
from .validate_login_token import ValidateLoginTokenInput, ValidateLoginTokenOutput
from .views import (
FetchViewsInputManually,
FetchViewsManuallyOutput,
GetViewsHistoryInput,
GetViewsHistoryOutput,
PlacementViewsHistoryOutput,

48
src/dto/nats_messages.py Normal file
View File

@@ -0,0 +1,48 @@
import pydantic
# Исходящие сообщения (к tgex-parser)
class ParserTrackTaskRequest(pydantic.BaseModel):
"""Сообщение для parser.track - добавить задачу на отслеживание канала"""
task_id: str # UUID placement
channel: str # username внешнего канала
target_text: str # invite_link для поиска в постах
class ParserGetViewsRequest(pydantic.BaseModel):
"""Сообщение для parser.get_views - синхронно получить просмотры поста"""
channel: str
message_id: int
# Входящие события (от tgex-parser)
class ParserPostFoundEvent(pydantic.BaseModel):
"""Событие parser.event.post_found - найден новый пост с target_text"""
task_id: str # UUID placement
channel: str
message_id: int
link: str # полная ссылка https://t.me/channel/message_id
text: str # текст поста
views: int
class ParserPostDeletedEvent(pydantic.BaseModel):
"""Событие parser.event.post_deleted - отслеживаемый пост удален"""
task_id: str
channel: str
message_id: int
link: str
views: int
class ParserViewsUpdatedEvent(pydantic.BaseModel):
"""Событие parser.event.views_updated - обновилось количество просмотров"""
task_id: str
channel: str
message_id: int
views: int

1
src/dto/placement.py
View File

@@ -52,7 +52,6 @@ class CreatePlacementInput(pydantic.BaseModel):
placement_date: datetime.datetime
cost: float | None = None
comment: str | None = None
ad_post_url: str | None = None
invite_link_type: domain.InviteLinkType = domain.InviteLinkType.PUBLIC

17
src/dto/views.py
View File

@@ -3,8 +3,6 @@ import uuid
import pydantic
from src import domain
class PlacementViewsHistoryOutput(pydantic.BaseModel):
"""История просмотров placement."""
@@ -31,21 +29,6 @@ class GetViewsHistoryOutput(pydantic.BaseModel):
histories: list[PlacementViewsHistoryOutput]
class FetchViewsInputManually(pydantic.BaseModel):
"""Запросить обновление просмотров для закупа."""
placement_id: uuid.UUID
user_id: uuid.UUID
class FetchViewsManuallyOutput(pydantic.BaseModel):
placement_id: uuid.UUID
views_count: int | None # None если не удалось получить
views_availability: domain.PostViewsAvailability
fetched_at: datetime.datetime
error_message: str | None = None # Сообщение об ошибке если не удалось
class UpdateViewsManuallyInput(pydantic.BaseModel):
"""Ручное обновление просмотров."""

13
src/main.py
View File

@@ -5,15 +5,13 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from shared import logger
from src import dependencies
from src import deps
from src.adapter.jwt import JWT
from src.adapter.postgres import Postgres
from src.adapter.telegram_bot import TelegramBot
from src.adapter.telegram_parser import TelegramParser
from src.config import settings
from src.controller.http import api_router
from src.controller.telegram_callback import telegram_callback_router
from src.controller.worker.views_worker import FetchViewsWorker
from src.usecase import Usecase
@@ -22,32 +20,25 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
await postgres.connect()
await telegram.start_polling()
await telegram.set_commands()
await views_worker.start()
yield
await views_worker.stop()
await telegram.stop_polling()
await postgres.close()
logger.init(settings.logger)
# Dependencies
postgres = Postgres(settings.db)
telegram = TelegramBot(settings.telegram, telegram_callback_router)
telegram_views_fetcher = TelegramParser(settings.telegram_parser)
jwt_encoder = JWT(settings.jwt)
usecase = Usecase(
database=postgres,
telegram_bot=telegram,
telegram_parser=telegram_views_fetcher,
jwt_encoder=jwt_encoder,
)
dependencies.set_usecase(usecase)
views_worker = FetchViewsWorker(config=settings.fetch_views_worker)
deps.set_usecase(usecase)
app = FastAPI(lifespan=lifespan, version=settings.logger.APP_VERSION)

26
src/usecase/__init__.py
View File

@@ -41,9 +41,7 @@ from .target_channel.disconnect_target_chan import disconnect_target_chan
from .target_channel.disconnect_target_chan_by_tg_id import disconnect_target_chan_by_tg_id
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 .views.fetch_views_manually import fetch_views_manually
from .views.get_views_history import get_views_history
from .views.run_views_worker_cycle import run_views_worker_cycle
from .views.update_views_manually import update_views_manually
@@ -66,8 +64,6 @@ class Database(typing.Protocol):
self, user_id: UUID, channel_id: UUID | None = None, telegram_id: int | None = None
) -> domain.TargetChannel | None: ...
async def create_target_channel(self, channel: domain.TargetChannel) -> domain.TargetChannel: ...
async def upsert_target_channel(self, channel: domain.TargetChannel) -> domain.TargetChannel: ...
async def get_user_target_channels(self, user_id: UUID) -> list[domain.TargetChannel]: ...
@@ -123,8 +119,6 @@ class Database(typing.Protocol):
async def delete_placement(self, placement_id: UUID) -> None: ...
async def check_creative_in_use(self, creative_id: UUID) -> bool: ...
async def get_placement_by_invite_link(self, invite_link: str) -> domain.Placement | None: ...
async def get_subscriber(self, telegram_id: int) -> domain.Subscriber | None: ...
@@ -161,8 +155,6 @@ class Database(typing.Protocol):
to_date: datetime.datetime | None = None,
) -> list[domain.PlacementViewsHistory]: ...
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(
@@ -171,6 +163,17 @@ class Database(typing.Protocol):
async def clear_telegram_state(self, telegram_id: int) -> None: ...
# NATS-related methods for placement tracking
async def get_placement_by_id(self, placement_id: UUID) -> domain.Placement | None: ...
async def update_placement_from_post_found(
self, placement_id: UUID, ad_post_url: str, message_id: int, views_count: int
) -> domain.Placement: ...
async def update_placement_from_post_deleted(self, placement_id: UUID) -> domain.Placement: ...
async def update_placement_views(self, placement_id: UUID, views_count: int) -> domain.Placement: ...
class TelegramBotWriter(typing.Protocol):
async def send_message(self, text: str, chat_id: int) -> None: ...
@@ -188,10 +191,6 @@ class TelegramBotWriter(typing.Protocol):
async def create_chat_invite_link(self, chat_id: int, requires_approval: bool = False) -> str: ...
class TelegramParser(typing.Protocol):
async def get_post_views(self, post_url: str) -> int | None: ...
class JWTEncoder(typing.Protocol):
def encode_access_token(self, user_id: UUID, telegram_id: int, username: str | None = None) -> str: ...
@@ -200,7 +199,6 @@ class JWTEncoder(typing.Protocol):
class Usecase:
database: Database
telegram_bot: TelegramBotWriter
telegram_parser: TelegramParser
jwt_encoder: JWTEncoder
validate_login_token = validate_login_token
@@ -232,8 +230,6 @@ class Usecase:
delete_placement = delete_placement
handle_subscription = handle_subscription
handle_unsubscription = handle_unsubscription
fetch_views = fetch_views_manually
run_views_worker_cycle = run_views_worker_cycle
get_views_history = get_views_history
update_views_manually = update_views_manually
get_placements_analytics = get_placements_analytics

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

@@ -41,7 +41,7 @@ async def create_placement(self: 'Usecase', input: dto.CreatePlacementInput, use
requires_approval=requires_approval,
)
# Создаём закуп
# Создаём закуп со статусом PENDING (ожидаем публикацию поста)
placement = domain.Placement(
target_channel_id=input.target_channel_id,
external_channel_id=input.external_channel_id,
@@ -50,10 +50,9 @@ async def create_placement(self: 'Usecase', input: dto.CreatePlacementInput, use
placement_date=input.placement_date,
cost=input.cost,
comment=input.comment,
ad_post_url=input.ad_post_url,
invite_link_type=input.invite_link_type,
invite_link=invite_link,
status=domain.PlacementStatus.ACTIVE,
status=domain.PlacementStatus.PENDING,
)
created = await self.database.create_placement(placement)
@@ -62,7 +61,6 @@ async def create_placement(self: 'Usecase', input: dto.CreatePlacementInput, use
creative.placements_count += 1
await self.database.update_creative(creative)
# Формируем ответ внутри транзакции, пока объекты в сессии
return dto.PlacementOutput(
id=created.id,
target_channel_id=created.target_channel_id,

65
src/usecase/views/fetch_views_manually.py
View File

@@ -1,65 +0,0 @@
import datetime
import logging
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def fetch_views_manually(self: 'Usecase', input: dto.FetchViewsInputManually) -> dto.FetchViewsManuallyOutput:
async with self.database.transaction():
placement = await self.database.get_placement(input.user_id, input.placement_id)
if not placement:
raise domain.PlacementNotFound(input.placement_id)
if not placement.ad_post_url:
return dto.FetchViewsManuallyOutput(
placement_id=placement.id,
views_count=None,
views_availability=domain.PostViewsAvailability.UNAVAILABLE,
fetched_at=datetime.datetime.now(datetime.UTC),
error_message='No ad_post_url specified',
)
views_count = await self.telegram_parser.get_post_views(placement.ad_post_url)
fetched_at = datetime.datetime.now(datetime.UTC)
if views_count:
placement.views_count = views_count
placement.views_availability = domain.PostViewsAvailability.AVAILABLE
placement.last_views_fetch_at = fetched_at
await self.database.update_placement(placement)
history = domain.PlacementViewsHistory(
placement_id=placement.id,
views_count=views_count,
fetched_at=fetched_at,
)
await self.database.create_placement_views_history(history)
log.info('Views fetched for placement %s: %d', placement.id, views_count)
return dto.FetchViewsManuallyOutput(
placement_id=placement.id,
views_count=views_count,
views_availability=domain.PostViewsAvailability.AVAILABLE,
fetched_at=fetched_at,
)
placement.views_availability = domain.PostViewsAvailability.UNAVAILABLE
placement.last_views_fetch_at = fetched_at
await self.database.update_placement(placement)
log.warning('Failed to fetch views for placement %s', placement.id)
return dto.FetchViewsManuallyOutput(
placement_id=placement.id,
views_count=None,
views_availability=domain.PostViewsAvailability.UNAVAILABLE,
fetched_at=fetched_at,
error_message='Failed to fetch views from Telegram API',
)

66
src/usecase/views/run_views_worker_cycle.py
View File

@@ -1,66 +0,0 @@
import asyncio
import datetime
import logging
import uuid
from typing import TYPE_CHECKING, cast
from aiolimiter import AsyncLimiter
from src import domain
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def _process_placement(self: 'Usecase', placement_id: uuid.UUID, user_id: uuid.UUID, ad_post_url: str) -> None:
views_count = await self.telegram_parser.get_post_views(ad_post_url)
fetched_at = datetime.datetime.now(datetime.UTC)
async with self.database.transaction():
p = await self.database.get_placement(user_id=user_id, placement_id=placement_id)
if not p:
log.error('Placement %s not found', placement_id)
return
p.last_views_fetch_at = fetched_at
if not views_count:
p.views_availability = domain.PostViewsAvailability.UNAVAILABLE
await self.database.update_placement(p)
log.warning('Failed to fetch views for %s', placement_id)
return
p.views_count = views_count
p.views_availability = domain.PostViewsAvailability.AVAILABLE
await self.database.update_placement(p)
history = domain.PlacementViewsHistory(
placement_id=placement_id,
views_count=views_count,
fetched_at=fetched_at,
)
await self.database.create_placement_views_history(history)
log.info('Updated placement %s: %d', placement_id, views_count)
async def run_views_worker_cycle(self: 'Usecase', interval_seconds: int) -> None:
async with self.database.transaction():
placements = await self.database.get_active_placements_with_urls()
if not placements:
log.debug('ViewsWorkerCycle: No placements to process')
return
limiter = AsyncLimiter(max_rate=len(placements), time_period=interval_seconds)
log.info('Processing %d placements for %d seconds', len(placements), interval_seconds)
async def process(p: domain.Placement) -> None:
async with limiter:
try:
await _process_placement(self, p.id, p.user_id, cast(str, p.ad_post_url))
except Exception:
log.exception('Error updating views for %s', p.id)
await asyncio.gather(*(process(p) for p in placements))