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 ows_text_campaigns.adapters.db.decorators import transactional from ows_text_campaigns.adapters.ows_url_shortener import OwsUrlShortenerClient from ows_text_campaigns.adapters.scanner import ContentRecommendations, ContentScanner from ows_text_campaigns.artist.services import ArtistSettingsService from ows_text_campaigns.assets.exceptions import InvalidAssetIdError from ows_text_campaigns.assets.services import AssetService from ows_text_campaigns.campaigns.exceptions import ( CampaignAttachmentLimitExceededError, CampaignMustBeDraftError, ) from ows_text_campaigns.campaigns.models import Campaign from ows_text_campaigns.campaigns.services import CampaignService from ows_text_campaigns.campaigns.utils import extract_shortened_urls_from_markup from ows_text_campaigns.campaigns.validators import ShortenedUrlValidator from ows_text_campaigns.config import Settings @dataclass(kw_only=True) class UpdateCampaignContentRequest(AuthRequest): campaign_id: str content: str | None asset_ids: list[str] | None @singleton class UpdateCampaignContentHandler: permission = Permission("text_campaign", "edit") audience_permission = Permission("audience", "create_text_target") def __init__( self, auth_service: AuthService, campaign_service: CampaignService, asset_service: AssetService, shortened_url_validator: ShortenedUrlValidator, ows_url_shortener_client: OwsUrlShortenerClient, settings: Settings, artist_settings_service: ArtistSettingsService, scanner: ContentScanner, ) -> None: self.auth_service = auth_service self.campaign_service = campaign_service self.asset_service = asset_service self.shortened_url_validator = shortened_url_validator self.ows_url_shortener_client = ows_url_shortener_client self.settings = settings self.artist_settings_service = artist_settings_service self.scanner = scanner @transactional def handle(self, request: UpdateCampaignContentRequest) -> Campaign: # noqa: C901 campaign = self.campaign_service.get_campaign(request.campaign_id) # Check if user has permission to edit email campaign self.auth_service.check_account_resource( request.identity_id, account=campaign.account, permission=self.permission, resource_id=campaign.id, ) # Check if text campaign is not draft if not campaign.is_draft: raise CampaignMustBeDraftError # Update shortened urls if request.content is not None: # run one more content scanning artist_name = self.artist_settings_service.get_configured_settings( campaign.global_participant_id ).artist_name if ( campaign.content != request.content or campaign.content_scan_result is None ): scan_result = self.scanner.scan(request.content, artist_name) else: scan_result = campaign.content_scan_result self._update_content( campaign, content=request.content, identity_id=request.identity_id, content_scan_result=scan_result, ) # Update assets if request.asset_ids is not None: self._update_assets(campaign, asset_ids=request.asset_ids) campaign.updated_by = request.identity_id self.campaign_service.save_campaign(campaign) return campaign def _update_content( self, campaign: Campaign, *, content: str, identity_id: str, content_scan_result: ContentRecommendations, ) -> None: # Build a mapping of existing shortened URLs by ID existing_standard_urls = { shortened_url.id: shortened_url for shortened_url in campaign.shortened_urls if shortened_url.is_standard } # Extract shortened URLs from the new content requested_shortened_urls = extract_shortened_urls_from_markup( content, raise_on_error=True ) requested_standard_urls = { shortened_url.id: shortened_url for shortened_url in requested_shortened_urls if shortened_url.is_standard } # Determine which standard URLs to delete (missing or with method changed) deleted_shortened_paths: list[str] = [] for shortened_url_id, existing_url in existing_standard_urls.items(): requested_url = requested_standard_urls.get(shortened_url_id) if requested_url is None: deleted_shortened_paths.append(existing_url.path) elif requested_url != existing_url: # Method has changed, treat as deleted and re-added deleted_shortened_paths.append(existing_url.path) # Delete removed URLs if deleted_shortened_paths: self.ows_url_shortener_client.delete_paths(deleted_shortened_paths) # Shorten new or updated URLs for shortened_url in requested_shortened_urls: if not shortened_url.is_standard: continue existing_url = existing_standard_urls.get(shortened_url.id) if shortened_url == existing_url: continue # No changes needed # Validate allowed domain url_domain = self.shortened_url_validator.validate_allowed_domain( identity_id=identity_id, account=campaign.account, domain=shortened_url.domain, ) # Shorten the new URL self.ows_url_shortener_client.shorten_url( url=shortened_url.url, domain=url_domain.domain, path=shortened_url.path, additional_attributes={ "campaign_id": campaign.id, "shortened_url_id": shortened_url.id, }, ) # Update campaign content campaign.content = content campaign.content_scan_result = content_scan_result def _update_assets(self, campaign: Campaign, *, asset_ids: list[str]) -> None: # Update assets current_assets = campaign.assets if asset_ids: if len(asset_ids) > self.settings.campaign_max_attachments: raise CampaignAttachmentLimitExceededError( max_allowed_limit=self.settings.campaign_max_attachments ) updated_assets = self.asset_service.get_assets_for( asset_ids=asset_ids, object_id=campaign.id, object_type="text_campaign", ) updated_asset_ids = {asset.id for asset in updated_assets} # Check for invalid asset IDs invalid_ids = set(asset_ids) - updated_asset_ids if invalid_ids: raise InvalidAssetIdError( f"Invalid asset IDs: {', '.join(invalid_ids)}" ) # Delete removed assets deleted_assets = [ asset for asset in current_assets if asset.id not in updated_asset_ids ] # Final updated assets campaign.assets = list(updated_assets) else: # If asset_ids is None or empty, remove all assets deleted_assets = current_assets campaign.assets = [] for deleted_asset in deleted_assets: self.asset_service.delete_asset(deleted_asset)