66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
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',
|
|
)
|