from dataclasses import dataclass from anydi import singleton from fansifter_common.auth.requests import AuthAccountFilterRequest from fansifter_common.auth.services import AuthService from fansifter_common.auth.types import Permission from ows_text_campaigns.campaigns.dtos import CampaignLinkAnalytics from ows_text_campaigns.campaigns.services import CampaignService from ows_text_campaigns.campaigns.types import CampaignAnalyticsCriteria from ows_text_campaigns.campaigns.utils import ( extract_non_shortened_urls_from_content, extract_shortened_urls_from_markup, ) @dataclass(kw_only=True) class GetCampaignLinkAnalyticsRequest(AuthAccountFilterRequest): campaign_id: str @singleton class GetCampaignLinkAnalyticsHandler: permission = Permission("text_campaign", "view") def __init__( self, auth_service: AuthService, campaign_service: CampaignService, ) -> None: self.auth_service = auth_service self.campaign_service = campaign_service def handle( self, request: GetCampaignLinkAnalyticsRequest ) -> list[CampaignLinkAnalytics]: campaign = self.campaign_service.get_campaign(request.campaign_id) long_urls = extract_non_shortened_urls_from_content(campaign.get_content()) # Pass short URLs to build response even without clicks short_urls = extract_shortened_urls_from_markup(campaign.get_content()) self.auth_service.authorize_account( request.identity_id, account=campaign.account, permission=self.permission, ) return self.campaign_service.get_campaign_link_analytics( criteria=CampaignAnalyticsCriteria( campaign_id=campaign.id, ), long_urls=long_urls, short_urls=short_urls, )