47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
from src import domain, dto
|
|
|
|
if TYPE_CHECKING:
|
|
from .. import Usecase
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
async def get_views_history(self: 'Usecase', input: dto.GetViewsHistoryInput) -> list[dto.PostViewsHistoryOutput]:
|
|
context = await self.ensure_workspace_permission(
|
|
input.workspace_id, input.user_id, domain.PermissionKey.PLACEMENTS_READ
|
|
)
|
|
|
|
publication = await self.database.get_publication(input.workspace_id, input.publication_id)
|
|
if not publication:
|
|
log.warning('Publication %s not found for user %s', input.publication_id, input.user_id)
|
|
raise domain.PublicationNotFound(input.publication_id)
|
|
|
|
if not publication.placement:
|
|
log.error('Publication %s has no placement', publication.id)
|
|
raise domain.PlacementNotFound(publication.placement_id)
|
|
|
|
context.ensure_project_permission(domain.PermissionKey.PLACEMENTS_READ, publication.placement.project_id)
|
|
|
|
if not publication.post:
|
|
return []
|
|
|
|
histories = await self.database.get_views_history(
|
|
publication.post.id,
|
|
from_date=input.from_date,
|
|
to_date=input.to_date,
|
|
)
|
|
|
|
return [
|
|
dto.PostViewsHistoryOutput(
|
|
id=history.id,
|
|
post_id=history.post_id,
|
|
views_count=history.views_count,
|
|
fetched_at=history.fetched_at,
|
|
created_at=history.created_at,
|
|
)
|
|
for history in histories
|
|
]
|