import logging from dataclasses import dataclass from anydi import singleton from fansifter_common.adapters.twilio import TwilioClient from fansifter_common.auth.requests import AuthRequest from fansifter_common.auth.types import Permission from ows_text_campaigns.adapters.db import DB from ows_text_campaigns.adapters.scanner import ContentScanner from ows_text_campaigns.artist.dtos import ArtistSettings from ows_text_campaigns.artist.repositories import ArtistSettingsRepository from ows_text_campaigns.artist.services import ArtistSettingsService from ows_text_campaigns.artist.validators import ArtistSettingsValidator from ows_text_campaigns.assets.exceptions import AssetNotFoundError, InvalidAssetIdError from ows_text_campaigns.config import Settings logger = logging.getLogger(__name__) @dataclass(kw_only=True) class UpdateArtistSettingsRequest(AuthRequest): global_participant_id: str welcome_message: str | None = None opt_in_message: str | None = None asset_id: str | None = None @singleton class UpdateArtistSettingsHandler: permission = Permission("text_campaign", "edit") def __init__( self, db: DB, artist_settings_repository: ArtistSettingsRepository, artist_settings_service: ArtistSettingsService, artist_settings_validator: ArtistSettingsValidator, twilio_client: TwilioClient, settings: Settings, scanner: ContentScanner, ) -> None: self.db = db self.artist_settings_repository = artist_settings_repository self.artist_settings_service = artist_settings_service self.artist_settings_validator = artist_settings_validator self.twilio_client = twilio_client self.settings = settings self.scanner = scanner def handle(self, request: UpdateArtistSettingsRequest) -> ArtistSettings: self.artist_settings_validator.validate_roster_access( identity_id=request.identity_id, global_participant_id=request.global_participant_id, permission=self.permission, ) with self.db.transaction(): artist_settings = self.artist_settings_service.get_settings( request.global_participant_id ) # Update welcome message if request.welcome_message is not None: artist_settings.welcome_message = request.welcome_message artist_settings.welcome_message_scan_result = self.scanner.scan( request.welcome_message, artist_settings.artist_name ) # Update opt in message opt_in_message_changed = False if request.opt_in_message is not None: artist_settings.opt_in_message = request.opt_in_message artist_settings.opt_in_message_scan_result = self.scanner.scan( request.opt_in_message, artist_settings.artist_name ) opt_in_message_changed = True # Update asset if request.asset_id is not None: try: asset = self.artist_settings_service.get_artist_image( global_participant_id=artist_settings.global_participant_id, asset_id=request.asset_id, ) except AssetNotFoundError: raise InvalidAssetIdError from None artist_settings.asset_id = asset.id artist_settings.asset = asset artist_settings.updated_by = request.identity_id self.artist_settings_repository.save(artist_settings) if opt_in_message_changed: pass return ArtistSettings( 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, asset_id=artist_settings.asset_id, )