import logging from dataclasses import dataclass from typing import NamedTuple from anydi import singleton from fansifter_common.adapters.graphql_router import GlobalParticipant from fansifter_common.adapters.twilio import TwilioClient, TwilioClientError from fansifter_common.auth.requests import AuthRequest from fansifter_common.auth.types import Permission from pydantic import SecretStr from ows_text_campaigns.adapters.db.decorators import transactional from ows_text_campaigns.adapters.scanner import ContentScanner from ows_text_campaigns.artist.dtos import ArtistSettings as ArtistSettingsDtos from ows_text_campaigns.artist.exceptions import ArtistTwilioAccountError from ows_text_campaigns.artist.models import ArtistPhoneNumber, ArtistSettings from ows_text_campaigns.artist.repositories import ArtistSettingsRepository from ows_text_campaigns.artist.services import ArtistSettingsService from ows_text_campaigns.artist.types import PhoneNumberStr from ows_text_campaigns.artist.validators import ArtistSettingsValidator from ows_text_campaigns.assets.models import Asset from ows_text_campaigns.assets.services import AssetService from ows_text_campaigns.assets.types import BinaryAsset from ows_text_campaigns.config import Settings from ows_text_campaigns.twilio.services import TwilioAccountService logger = logging.getLogger(__name__) class ArtistImage(NamedTuple): asset: Asset | None binary: BinaryAsset | None @dataclass(kw_only=True) class CreateOrUpdateArtistSettingsRequest(AuthRequest): global_participant_id: str artist_name: str | None welcome_message: str | None opt_in_message: str | None phone_numbers: list[PhoneNumberStr] twilio_account_sid: str twilio_auth_token: SecretStr twilio_messaging_service_sid: str refresh_api_key: bool = False @singleton class CreateOrUpdateArtistSettingsHandler: permission = Permission("text_campaign", "create") def __init__( self, asset_service: AssetService, artist_settings_repository: ArtistSettingsRepository, artist_settings_service: ArtistSettingsService, artist_settings_validator: ArtistSettingsValidator, twilio_account_service: TwilioAccountService, twilio_client: TwilioClient, settings: Settings, scanner: ContentScanner, ) -> None: self.asset_service = asset_service self.artist_settings_repository = artist_settings_repository self.artist_settings_service = artist_settings_service self.artist_settings_validator = artist_settings_validator self.twilio_account_service = twilio_account_service self.twilio_client = twilio_client self.settings = settings self.scanner = scanner @transactional def handle( self, request: CreateOrUpdateArtistSettingsRequest ) -> ArtistSettingsDtos: self.artist_settings_validator.validate_roster_access( identity_id=request.identity_id, global_participant_id=request.global_participant_id, permission=self.permission, ) artist_settings = self.artist_settings_repository.get_by_global_participant_id( request.global_participant_id ) global_participant = self.artist_settings_service.get_global_participant( request.global_participant_id ) # Validate Twilio account self.validate_twilio_account( account_sid=request.twilio_account_sid, auth_token=request.twilio_auth_token, messaging_service_sid=request.twilio_messaging_service_sid, phone_numbers=request.phone_numbers, ) # Create artist settings if not exist if artist_settings is None: artist_settings = ArtistSettings( global_participant_id=global_participant.id, artist_name=global_participant.name, ) # Overwrite artist name with if provided if request.artist_name is not None: artist_settings.artist_name = request.artist_name # Overwrite artist welcome message if provided if request.welcome_message is not None: artist_settings.welcome_message = request.welcome_message welcome_message_scan_result = self.scanner.scan( request.welcome_message, artist_settings.artist_name ) artist_settings.welcome_message_scan_result = welcome_message_scan_result # Overwrite artist opt in message if provided if request.opt_in_message is not None: artist_settings.opt_in_message = request.opt_in_message opt_in_message_scan_result = self.scanner.scan( request.opt_in_message, artist_settings.artist_name ) artist_settings.opt_in_message_scan_result = opt_in_message_scan_result # Get or create default artist image if artist_settings.asset is None: artist_image = ArtistImage(asset=None, binary=None) try: artist_image = self._create_and_upload_default_image(global_participant) except Exception as exc: logger.exception( "Failed to save default image for artist `%s`", global_participant.id, extra={"global_participant_id": global_participant.id}, exc_info=exc, ) if artist_image.asset is not None: artist_settings.asset_id = artist_image.asset.id artist_settings.asset = artist_image.asset else: binary = self.asset_service.get_asset_from_s3(artist_settings.asset.key) artist_image = ArtistImage(asset=artist_settings.asset, binary=binary) # Set twilio settings artist_settings.twilio_account_sid = request.twilio_account_sid artist_settings.twilio_messaging_service_sid = ( request.twilio_messaging_service_sid ) # Set artist phone numbers phone_numbers: list[ArtistPhoneNumber] = [] for phone_number in request.phone_numbers: phone_numbers.append( self._get_or_create_phone_number( global_participant_id=global_participant.id, artist_name=global_participant.name, phone_number=phone_number, photo=artist_image.binary, ) ) artist_settings.phone_numbers = phone_numbers # Set user artist_settings.updated_by = request.identity_id # Save artist settings self.artist_settings_repository.save(artist_settings) # Save Twilio account credentials self.twilio_account_service.save_account( account_sid=request.twilio_account_sid, auth_token=request.twilio_auth_token.get_secret_value(), refresh_api_key=request.refresh_api_key, ) return ArtistSettingsDtos( welcome_message=artist_settings.welcome_message, welcome_message_instructions=self.settings.artist_welcome_message_instructions, opt_in_message=artist_settings.opt_in_message, opt_in_message_terms=self.settings.artist_opt_in_message_terms, image_url=artist_settings.image_url, phone_numbers=[ { "phone_number": phone_number.phone_number, "country_code": phone_number.country_code, } for phone_number in artist_settings.phone_numbers ], welcome_message_scan_result=artist_settings.welcome_message_scan_result, opt_in_message_scan_result=artist_settings.opt_in_message_scan_result, ) def _get_or_create_phone_number( self, *, global_participant_id: str, artist_name: str, phone_number: str, photo: BinaryAsset | None, ) -> ArtistPhoneNumber: # Create and upload vCard vcard_asset = ( self.artist_settings_service.create_and_upload_vcard_for_phone_number( global_participant_id=global_participant_id, artist_name=artist_name, phone_number=phone_number, photo=photo, ) ) return ArtistPhoneNumber( global_participant_id=global_participant_id, phone_number=phone_number, vcard_asset_id=vcard_asset.id, ) def _create_and_upload_default_image( self, global_participant: GlobalParticipant, ) -> ArtistImage: if not global_participant.image_url: return ArtistImage(asset=None, binary=None) binary_asset = self.asset_service.get_asset_from_url( global_participant.image_url ) image_key = self.artist_settings_service.create_asset_key( global_participant.id, filename=f"avatar.{binary_asset.extension}", ) asset = self.artist_settings_service.create_and_upload_artist_image( global_participant_id=global_participant.id, asset=binary_asset, key=image_key, ) return ArtistImage(asset=asset, binary=binary_asset) def validate_twilio_account( self, *, account_sid: str, auth_token: SecretStr, messaging_service_sid: str, phone_numbers: list[str], ) -> None: try: messaging_service = self.twilio_client.get_messaging_service( messaging_service_sid=messaging_service_sid, auth=( account_sid, auth_token.get_secret_value(), ), ) except TwilioClientError as exc: if exc.status_code == 404: raise ArtistTwilioAccountError( "Invalid Messaging Service SID" ) from None raise if messaging_service.account_sid != account_sid: raise ArtistTwilioAccountError("Invalid Account SID") messaging_phone_numbers = [ messaging_phone_number.phone_number for messaging_phone_number in self.twilio_client.get_messaging_phone_numbers( messaging_service_sid=messaging_service.sid, auth=(account_sid, auth_token.get_secret_value()), ) ] for phone_number in phone_numbers: if phone_number not in messaging_phone_numbers: raise ArtistTwilioAccountError( f"`{phone_number}` is not supported by " f"Messaging Service {messaging_service_sid}" )