from typing import Any from anydi import singleton from cachelib import BaseCache from fansifter_common.auth.identity import Identity from fansifter_common.httpclient.exceptions import HTTPClientError from fansifter_common.utils import timezone from fansifter_common.utils.cache import cached from dmp.audiences.dtos import AudienceCriteria from dmp.audiences.enums import AudienceTarget from dmp.audiences.exceptions import ( AudienceCityFilterError, AudienceNotFoundError, ) from dmp.audiences.filters import AudienceFilters, CityFilter from dmp.audiences.models import ( Audience, AudienceExport, AudienceShare, AudienceSnapshot, ) from dmp.audiences.repositories import ( AudienceEmailFanRepository, AudienceFanRepository, AudienceRepository, AudienceTextFanRepository, ) from dmp.audiences.validators import AudienceValidator from dmp.config import Settings from dmp.fandata.enums import FanSegment from dmp.locations.exceptions import CityNotFoundError from dmp.locations.services import LocationService from dmp.locations.types import BoundingBox, Point from dmp.locations.utils import get_bounding_center_and_radius_in_km from dmp.rosters.exceptions import MainRepArtistOnlyAllowedError @singleton class AudienceService: def __init__( self, audience_repository: AudienceRepository, audience_fan_repository: AudienceFanRepository, audience_email_fan_repository: AudienceEmailFanRepository, audience_text_fan_repository: AudienceTextFanRepository, audience_validator: AudienceValidator, location_service: LocationService, cache: BaseCache, settings: Settings, ) -> None: self.audience_repository = audience_repository self.audience_fan_repository = audience_fan_repository self.audience_email_fan_repository = audience_email_fan_repository self.audience_text_fan_repository = audience_text_fan_repository self.audience_validator = audience_validator self.location_service = location_service self.cache = cache self.settings = settings def get_audience(self, audience_id: str) -> Audience: audience = self.audience_repository.get(audience_id) if not audience or audience.is_archived: raise AudienceNotFoundError return audience def get_audience_size( self, criteria: AudienceCriteria, *, ignore_city_error: bool = False, ignore_main_rep_error: bool = False, ) -> int: """Get audience size.""" # Validates main rep only for when global fandata list is enabled if criteria.is_global: try: self.audience_validator.validate_main_rep_only_artists( fandata_list_ids=criteria.fandata_list_ids, account=criteria.account, target=criteria.target, ) except MainRepArtistOnlyAllowedError as exc: if ignore_main_rep_error: return 0 raise exc return self.audience_fan_repository.get_size( self.clean_criteria(criteria, ignore_city_error=ignore_city_error) ) def save_audience(self, audience: Audience) -> None: """Save audience.""" self.audience_repository.save(audience) def refresh_audience( self, audience: Audience, *, identity: Identity, is_global: bool ) -> Audience: """Refresh audience.""" criteria = self.clean_criteria( AudienceCriteria.from_audience( audience, identity=identity, is_global=is_global ) ) # Default fan count value fan_count = 0 # Calculate recipients count by email provider for EMAIL targets if audience.target == AudienceTarget.EMAIL: audience.recipients_count_by_email_provider = self.audience_email_fan_repository.get_recipients_count_by_email_provider( criteria=criteria ) else: audience.recipients_count_by_email_provider = None # Create snapshot if audience.target in (AudienceTarget.ADS, AudienceTarget.EMAIL): fan_count = self.get_audience_size( criteria, ignore_city_error=False, ignore_main_rep_error=False, ) snapshot = AudienceSnapshot( audience=audience, created_by=identity.id, fan_count=fan_count, ) audience.snapshots.append(snapshot) self.audience_fan_repository.insert_bulk( snapshot_id=snapshot.id, criteria=criteria ) elif audience.target == AudienceTarget.TEXT: fan_count = self.upsert_audience_fans( audience, identity=identity, is_global=is_global ) audience.fan_count = fan_count audience.refreshed_at = timezone.now() audience.refreshed_by = identity.id self.audience_repository.save(audience) return audience def upsert_audience_fans( self, audience: Audience, *, identity: Identity, is_global: bool ) -> int: criteria = self.clean_criteria( AudienceCriteria.from_audience( audience, identity=identity, is_global=is_global ) ) if audience.target == AudienceTarget.EMAIL: self.audience_email_fan_repository.delete_by_audience_id(audience.id) row_count = self.audience_email_fan_repository.insert_bulk( audience_id=audience.id, criteria=criteria ) # Calculate recipients count by email provider audience.recipients_count_by_email_provider = self.audience_email_fan_repository.get_recipients_count_by_email_provider( criteria=criteria ) self.audience_repository.save(audience) return row_count elif audience.target == AudienceTarget.TEXT: self.audience_text_fan_repository.delete_by_audience_id(audience.id) return self.audience_text_fan_repository.insert_bulk( audience_id=audience.id, criteria=criteria ) return 0 def export_audience_to_csv(self, export: AudienceExport) -> None: """Export audience to CSV.""" self.audience_fan_repository.export_to_csv( snapshot_id=export.snapshot_id, reason=export.reason, key=export.key, stage=self.settings.audience_export_stage, identity_id=export.created_by, ) def export_audience_share_to_csv( self, share: AudienceShare, context: dict[str, Any] | None = None ) -> None: """Export audience share to CSV.""" self.audience_fan_repository.export_share_to_csv( snapshot_id=share.snapshot_id, platform=share.platform, key=share.key, stage=self.settings.audience_export_stage, context=context, ) def clean_criteria( self, criteria: AudienceCriteria, *, ignore_city_error: bool = False ) -> AudienceCriteria: # Get excluded countries excluded_countries = [*self.settings.audience_global_excluded_countries] # Exception by https://theorchard.atlassian.net/browse/FS-4695 # Always exclude FR from audience criteria if not explicitly included if "FR" not in excluded_countries and ( not criteria.filters.countries and not criteria.filters.city_countries ): excluded_countries.append("FR") if criteria.target == AudienceTarget.ADS: excluded_countries.extend(self.settings.audience_ads_excluded_countries) # Always set include_secondary_fans to None or remove SECONDARY_FANS # from fan_segments for non-ads targets and external users if criteria.target != AudienceTarget.ADS or ( not criteria.identity.is_internal_employee ): criteria.filters.include_secondary_fans = False if ( criteria.filters.fan_segments and criteria.filters.fan_segments.value and FanSegment.SECONDARY_FANS in criteria.filters.fan_segments.value ): criteria.filters.fan_segments.value.remove(FanSegment.SECONDARY_FANS) criteria.excluded_countries = excluded_countries if criteria.target == AudienceTarget.TEXT: allowed_countries = set(self.settings.audience_text_allowed_countries) if criteria.filters.countries: criteria.filters.countries = list( set(criteria.filters.countries) & allowed_countries ) or ["XY"] else: # Only allowed countries criteria.filters.countries = list(allowed_countries) # Clean cities filter self.clean_cities_filter(criteria.filters, ignore_error=ignore_city_error) return criteria def clean_cities_filter( self, filters: AudienceFilters, *, ignore_error: bool = False ) -> None: """Clean cities filter.""" if not filters.cities: return # Copy the list to avoid modifying the original list cities: list[CityFilter] = [*filters.cities] updated_cities = [] for city_filter in cities: # Skip clean city filters if city_filter.is_clean: updated_cities.append(city_filter) continue # Retrieve the full city data using the place_id try: cleaned_city_filter = self._get_city_filter(city_filter.place_id) # Update the title if it's not set if city_filter.title: cleaned_city_filter.title = city_filter.title updated_cities.append(cleaned_city_filter) except AudienceCityFilterError as exc: if exc.location is not None: ignore_error = True if exc.location.address: city_filter.country_iso2 = exc.location.address.country.code2 if not ignore_error: raise exc city_filter.city_id = "-1" updated_cities.append(city_filter) continue # Update the original list filters.cities = updated_cities @cached("city_filter", timeout_param="cache_timeout") def _get_city_filter(self, place_id: str) -> CityFilter: """Get cached city filter.""" # Get location by place_id try: location = self.location_service.get_location(place_id) except HTTPClientError as exc: raise AudienceCityFilterError("Cannot find location") from exc if not location.address: raise AudienceCityFilterError("Location has no address") # Get the closest city by country code and name try: city = self.location_service.get_closest_city( country_code2=location.address.country.code2, name=location.address.locality, position=Point( latitude=location.position[1], longitude=location.position[0], ), ) except CityNotFoundError as exc: raise AudienceCityFilterError( "Cannot find city", location=location ) from exc # Unpack map view as bounding box min_longitude, min_latitude, max_longitude, max_latitude = location.map_view # Get the bounding center and radius (latitude, longitude), radius = get_bounding_center_and_radius_in_km( BoundingBox( min_latitude=min_latitude, min_longitude=min_longitude, max_latitude=max_latitude, max_longitude=max_longitude, ) ) return CityFilter.model_construct( place_id=place_id, title=location.title, country_iso2=city.country_code2, city_id=city.id, postal_code=location.address.postal_code, latitude=latitude, longitude=longitude, bounding_radius_in_km=radius, )