from dataclasses import dataclass from anydi import singleton from fansifter_common.auth.requests import AuthRequest from fansifter_common.auth.services import AuthService from fansifter_common.auth.types import Permission from dmp.adapters.db import DefaultDB, transactional from dmp.audiences.enums import AudienceExportStatus from dmp.tiktok.exceptions import TikTokAudienceNotFoundError from dmp.tiktok.repositories import TikTokAudienceRepository @dataclass class GetTikTokAudienceShareStatusRequest(AuthRequest): tiktok_audience_id: str @dataclass class GetTikTokAudienceShareStatusResponse: audience_id: str audience_name: str ad_account_name: str status: AudienceExportStatus fans_count: int processed_fans: int progress: float seconds_remaining: int | None url: str @singleton class GetTikTokAudienceShareStatusHandler: permission = Permission("ad_account", "view") def __init__( self, db: DefaultDB, auth_service: AuthService, tiktok_audience_repository: TikTokAudienceRepository, ) -> None: self.db = db self.auth_service = auth_service self.tiktok_audience_repository = tiktok_audience_repository @transactional def handle( self, request: GetTikTokAudienceShareStatusRequest ) -> GetTikTokAudienceShareStatusResponse: tiktok_audience = self.tiktok_audience_repository.get( request.tiktok_audience_id ) if tiktok_audience is None: raise TikTokAudienceNotFoundError self.auth_service.authorize_account( request.identity_id, permission=self.permission, account=tiktok_audience.share.audience.account, ) return GetTikTokAudienceShareStatusResponse( audience_id=tiktok_audience.id, audience_name=tiktok_audience.name, ad_account_name=tiktok_audience.ad_account.name or "", status=tiktok_audience.share.status, fans_count=tiktok_audience.fans_count, processed_fans=tiktok_audience.share.processed_fans, progress=tiktok_audience.share.progress, seconds_remaining=tiktok_audience.share.seconds_remaining, url=tiktok_audience.url, )