from typing import Annotated, Any from anydi import Provide from fansifter_common.api.schemas import response_errors from fansifter_common.artist.exceptions import InvalidGlobalParticipantIdError from fastapi import APIRouter, Body, Depends, Header, Path, Query from fastapi.security import HTTPBasic, HTTPBasicCredentials from starlette import status from starlette.responses import HTMLResponse, Response from email_campaigns.api import auth, schemas from email_campaigns.automated.exceptions import ( AutomatedEmailHasTriggerError, AutomatedEmailNotFoundError, AutomatedEmailSenderEmailNotValidError, AutomatedEmailUniqueNameError, InvalidSourceCampaignIdError, SourceCampaignConnectionsEmptyError, SourceCampaignNotAttachedError, SourceCampaignSubscriptionOnlyRequiredError, SourceCampaignVendorMismatchError, ) from email_campaigns.automated.handlers import ( ApplyAutomatedEmailChangesHandler, ApplyAutomatedEmailChangesRequest, AttachSourceCampaignsHandler, AttachSourceCampaignsRequest, AutosaveAutomatedEmailHandler, AutosaveAutomatedEmailV2Handler, CreateAutomatedEmailHandler, CreateAutomatedEmailRequest, DeleteAutomatedEmailHandler, DeleteAutomatedEmailRequest, DetachSourceCampaignsHandler, DetachSourceCampaignsRequest, DuplicateAutomatedEmailHandler, DuplicateAutomatedEmailRequest, GenerateAutomatedEmailPublicPreviewUrlHandler, GetAutomatedEmailAnalyticsHandler, GetAutomatedEmailHandler, GetAutomatedEmailLinkAnalyticsHandler, GetAutomatedEmailPreviewHandler, GetAutomatedEmailPrivacyFooterHandler, GetAutomatedEmailPublicPreviewHandler, GetAutomatedEmailSalesAnalyticsHandler, GetAutomatedEmailSalesLocationsAnalyticsHandler, GetAutomatedEmailSalesProductsAnalyticsHandler, GetAutomatedEmailSalesShopsAnalyticsHandler, GetAutomatedEmailsHandler, GetAutomatedEmailsRequest, GetAutomatedEmailStatusHandler, GetAutomatedEmailStatusRequest, GetAutomatedEmailStripoAuthTokenHandler, SendAutomatedTestEmailHandler, UpdateAutomatedEmailContentHandler, UpdateAutomatedEmailFontsHandler, UpdateAutomatedEmailHandler, UpdateAutomatedEmailRequest, ) from email_campaigns.automated.types import SourceCampaignConnection from email_campaigns.emails.exceptions import ( AutosaveUnauthorizedError, EmailContentRequiredFieldsError, EmailNotFoundError, EmailPrivacyFooterMissingError, EmailPublicPreviewTokenExpiredError, EmailSenderEmailNotValidError, InvalidEmailDomainIdError, ) from email_campaigns.emails.handlers import ( AutosaveEmailRequest, AutosaveEmailV2Request, GenerateEmailPublicPreviewUrlRequest, GetEmailAnalyticsRequest, GetEmailLinkAnalyticsRequest, GetEmailPreviewRequest, GetEmailPrivacyFooterRequest, GetEmailPublicPreviewRequest, GetEmailSalesAnalyticsRequest, GetEmailSalesLocationsAnalyticsRequest, GetEmailSalesProductsAnalyticsRequest, GetEmailSalesShopsAnalyticsRequest, GetEmailStripoAuthTokenRequest, SendTestEmailRequest, UpdateEmailContentRequest, UpdateEmailFontsRequest, ) from email_campaigns.emails.handlers.get_email import GetEmailRequest from email_campaigns.emails.templates import ( EMAIL_NOT_FOUND_HTML, EMAIL_PREVIEW_EXPIRED_HTML, ) from email_campaigns.emails.types import UrlEncodedJson from email_campaigns.fonts.exceptions import InvalidGoogleFontError from email_campaigns.rosters.exceptions import InvalidCustomListIdError basic_auth = HTTPBasic() router = APIRouter(tags=["Automated"]) @router.post( "/automated/autosave", operation_id="autosaveAutomatedEmailCampaign", response_class=Response, tags=["internal"], status_code=status.HTTP_200_OK, responses=response_errors( EmailNotFoundError, AutosaveUnauthorizedError, ), ) async def autosave_automated_email_campaign( credentials: Annotated[HTTPBasicCredentials, Depends(basic_auth)], data: schemas.AutosaveEmailInput, handler: Provide[AutosaveAutomatedEmailHandler], plugin_data: Annotated[ UrlEncodedJson | None, Header(alias="plugin-ui-data") ] = None, ) -> None: # Parse userId from plugin-ui-data header user_id = plugin_data.get("userId") if plugin_data else None handler.handle( AutosaveEmailRequest( username=credentials.username, password=credentials.password, key=data.key, email_id=data.email_id, html=data.html, css=data.css, user_id=user_id, ) ) @router.post( "/automated/autosave/v2", operation_id="autosaveAutomatedEmailCampaignV2", response_class=Response, tags=["internal"], status_code=status.HTTP_200_OK, responses=response_errors( EmailNotFoundError, AutosaveUnauthorizedError, ), ) async def autosave_automated_email_campaign_v2( credentials: Annotated[HTTPBasicCredentials, Depends(basic_auth)], data: schemas.AutosaveEmailV2Input, handler: Provide[AutosaveAutomatedEmailV2Handler], ) -> None: handler.handle( AutosaveEmailV2Request( email_id=data.email_id, user_id=data.user_id, username=credentials.username, password=credentials.password, ) ) @router.post( "/automated/list", operation_id="getAutomatedEmails", response_model=schemas.LimitOffsetPage[schemas.AutomatedEmail], response_model_by_alias=True, ) def get_automated_emails( identity_id: Annotated[str, Depends(auth.get_identity_id)], data: Annotated[schemas.GetAutomatedEmailsInput, Body(...)], handler: Provide[GetAutomatedEmailsHandler], ) -> Any: order_by = data.order_by or GetAutomatedEmailsRequest.DEFAULT_ORDER_BY limit = data.limit or GetAutomatedEmailsRequest.DEFAULT_LIMIT offset = data.offset or GetAutomatedEmailsRequest.DEFAULT_OFFSET response = handler.handle( GetAutomatedEmailsRequest( identity_id=identity_id, search=data.search, vendor_id=data.vendor_id, subaccount_id=data.subaccount_id, fandata_list_ids=data.fandata_list_ids, is_automated=data.is_automated, order_by=order_by, limit=limit, offset=offset, ) ) return schemas.LimitOffsetPage( total=response.total, items=response.items, limit=limit, offset=offset, ) @router.post( "/automated", operation_id="createAutomatedEmail", response_model=schemas.AutomatedEmail, response_model_by_alias=True, status_code=status.HTTP_201_CREATED, ) def create_automated_email( identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.CreateAutomatedEmailInput, handler: Provide[CreateAutomatedEmailHandler], ) -> Any: return handler.handle( CreateAutomatedEmailRequest( identity_id=identity_id, vendor_id=data.vendor_id, subaccount_id=data.subaccount_id, name=data.name, type=data.type, fandata_list_id=data.fandata_list_id, html_content=data.html_content, css_content=data.css_content, ) ) @router.get( "/automated/{emailId}", operation_id="getAutomatedEmail", response_model=schemas.AutomatedEmail, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_automated_email( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetAutomatedEmailHandler], ) -> Any: return handler.handle( GetEmailRequest( identity_id=identity_id, email_id=email_id, ) ) @router.get( "/automated/{emailId}/status", operation_id="getAutomatedEmailStatus", response_model=schemas.AutomatedEmailStatus, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_automated_email_status( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetAutomatedEmailStatusHandler], ) -> Any: return handler.handle( GetAutomatedEmailStatusRequest( identity_id=identity_id, email_id=email_id, ) ) @router.put( "/automated/{emailId}", operation_id="updateAutomatedEmail", response_model=schemas.AutomatedEmail, response_model_by_alias=True, responses=response_errors( AutomatedEmailNotFoundError, AutomatedEmailUniqueNameError, AutomatedEmailSenderEmailNotValidError, InvalidGlobalParticipantIdError, InvalidEmailDomainIdError, InvalidCustomListIdError, ), ) def update_automated_email( identity_id: Annotated[str, Depends(auth.get_identity_id)], email_id: Annotated[str, Path(alias="emailId")], data: schemas.UpdateAutomatedEmailInput, handler: Provide[UpdateAutomatedEmailHandler], ) -> Any: return handler.handle( UpdateAutomatedEmailRequest( identity_id=identity_id, email_id=email_id, name=data.name, vendor_id=data.vendor_id, subaccount_id=data.subaccount_id, fandata_list_id=data.fandata_list_id, type=data.type, sender_name=data.sender_name, email_domain_id=data.email_domain_id, email_username=data.email_username, subject=data.subject, preview_text=data.preview_text, ) ) @router.delete( "/automated/{emailId}", operation_id="deleteAutomatedEmail", response_class=Response, status_code=status.HTTP_204_NO_CONTENT, responses=response_errors( AutomatedEmailNotFoundError, AutomatedEmailHasTriggerError, ), ) def delete_automated_email( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[DeleteAutomatedEmailHandler], ) -> None: handler.handle( DeleteAutomatedEmailRequest( identity_id=identity_id, email_id=email_id, ) ) @router.post( "/automated/{emailId}/duplicate", operation_id="duplicateAutomatedEmail", response_model=schemas.AutomatedEmail, response_model_by_alias=True, status_code=status.HTTP_201_CREATED, responses=response_errors( AutomatedEmailNotFoundError, AutomatedEmailUniqueNameError, ), ) def duplicate_automated_email( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.DuplicateAutomatedEmailInput, handler: Provide[DuplicateAutomatedEmailHandler], ) -> Any: return handler.handle( DuplicateAutomatedEmailRequest( identity_id=identity_id, email_id=email_id, name=data.name, ) ) @router.get( "/automated/{emailId}/preview", operation_id="getAutomatedEmailPreview", response_model=schemas.EmailPreview, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_automated_email_preview( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetAutomatedEmailPreviewHandler], apply_substitutions: Annotated[bool, Query(alias="applySubstitutions")] = False, with_unapplied_changes: Annotated[ bool, Query(alias="withUnappliedChanges") ] = False, ) -> Any: return handler.handle( GetEmailPreviewRequest( identity_id=identity_id, email_id=email_id, apply_substitutions=apply_substitutions, with_unapplied_changes=with_unapplied_changes, ) ) @router.patch( "/automated/{emailId}/content", operation_id="updateAutomatedEmailContent", response_model=schemas.EmailPreview, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def update_automated_email_content( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.UpdateEmailContentInput, handler: Provide[UpdateAutomatedEmailContentHandler], ) -> Any: return handler.handle( UpdateEmailContentRequest( identity_id=identity_id, email_id=email_id, html=data.html, css=data.css, ) ) @router.patch( "/automated/{emailId}/fonts", operation_id="updateAutomatedEmailFonts", response_model=schemas.CustomFontsResponse, response_model_by_alias=True, responses=response_errors(EmailNotFoundError, InvalidGoogleFontError), ) def update_automated_email_fonts( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.UpdateEmailFontsInput, handler: Provide[UpdateAutomatedEmailFontsHandler], ) -> Any: return schemas.CustomFontsResponse( items=handler.handle( UpdateEmailFontsRequest( identity_id=identity_id, email_id=email_id, font=data.font, ) ) ) @router.get( "/automated/{emailId}/privacy-footer", operation_id="getAutomatedEmailPrivacyFooter", response_model=schemas.EmailPrivacyFooter, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_automated_email_privacy_footer( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetAutomatedEmailPrivacyFooterHandler], ) -> Any: privacy_footer = handler.handle( GetEmailPrivacyFooterRequest( identity_id=identity_id, email_id=email_id, ) ) return {"content": privacy_footer} @router.get( "/automated/{emailId}/analytics", operation_id="getAutomatedEmailAnalytics", response_model=schemas.GetEmailAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_automated_email_analytics( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetAutomatedEmailAnalyticsHandler], ) -> Any: return handler.handle( GetEmailAnalyticsRequest( identity_id=identity_id, email_id=email_id, ) ) @router.get( "/automated/{emailId}/analytics/links", operation_id="getAutomatedEmailLinkAnalytics", response_model=schemas.GetEmailLinkAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_automated_email_link_analytics( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetAutomatedEmailLinkAnalyticsHandler], ) -> Any: return handler.handle( GetEmailLinkAnalyticsRequest( identity_id=identity_id, email_id=email_id, ) ) @router.get( "/automated/{emailId}/analytics/sales", operation_id="getAutomatedEmailSalesAnalytics", response_model=schemas.GetEmailSalesAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_automated_email_sales_analytics( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetAutomatedEmailSalesAnalyticsHandler], ) -> Any: return handler.handle( GetEmailSalesAnalyticsRequest( identity_id=identity_id, email_id=email_id, ) ) @router.get( "/automated/{emailId}/analytics/sales/products", operation_id="getAutomatedEmailSalesProductsAnalytics", response_model=schemas.GetEmailSalesProductsAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_automated_email_sales_products_analytics( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetAutomatedEmailSalesProductsAnalyticsHandler], ) -> Any: return handler.handle( GetEmailSalesProductsAnalyticsRequest( identity_id=identity_id, email_id=email_id, ) ) @router.get( "/automated/{emailId}/analytics/sales/locations", operation_id="getAutomatedEmailSalesLocationsAnalytics", response_model=schemas.GetEmailSalesLocationsAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_automated_email_sales_locations_analytics( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetAutomatedEmailSalesLocationsAnalyticsHandler], ) -> Any: return handler.handle( GetEmailSalesLocationsAnalyticsRequest( identity_id=identity_id, email_id=email_id, ) ) @router.get( "/automated/{emailId}/analytics/sales/shops", operation_id="getAutomatedEmailSalesShopsAnalytics", response_model=schemas.GetEmailSalesShopsAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_automated_email_sales_shops_analytics( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetAutomatedEmailSalesShopsAnalyticsHandler], ) -> Any: return handler.handle( GetEmailSalesShopsAnalyticsRequest( identity_id=identity_id, email_id=email_id, ) ) @router.post( "/automated/{emailId}/send-test-email", operation_id="sendAutomatedTestEmail", response_class=Response, status_code=status.HTTP_204_NO_CONTENT, responses=response_errors( EmailNotFoundError, EmailContentRequiredFieldsError, EmailSenderEmailNotValidError, EmailPrivacyFooterMissingError, ), ) def send_automated_test_email( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.SendAutomatedTestEmailInput, handler: Provide[SendAutomatedTestEmailHandler], ) -> None: handler.handle( SendTestEmailRequest( identity_id=identity_id, email_id=email_id, emails=data.emails, content=data.content, with_unapplied_changes=data.with_unapplied_changes, ) ) @router.get( "/automated/{emailId}/stripo/token", operation_id="getAutomatedEmailStripoAuthToken", response_model=schemas.GetEmailStripoAuthTokenResponse, response_model_by_alias=True, ) def get_automated_email_stripo_auth_token( identity_id: Annotated[str, Depends(auth.get_identity_id)], email_id: Annotated[str, Path(alias="emailId")], handler: Provide[GetAutomatedEmailStripoAuthTokenHandler], ) -> Any: return handler.handle( GetEmailStripoAuthTokenRequest(identity_id=identity_id, email_id=email_id) ) @router.post( "/automated/{emailId}/public-preview-url", operation_id="generateAutomatedEmailPublicPreviewUrl", response_model=schemas.GenerateEmailPublicPreviewUrlResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def generate_automated_email_public_preview_url( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], input_data: schemas.GenerateEmailPublicPreviewUrlInput, handler: Provide[GenerateAutomatedEmailPublicPreviewUrlHandler], ) -> Any: response = handler.handle( GenerateEmailPublicPreviewUrlRequest( identity_id=identity_id, email_id=email_id, domain=input_data.domain, ) ) return {"url": response.url} @router.put( "/automated/{emailId}/source-campaigns/attach", operation_id="attachSourceCampaigns", response_model=schemas.AutomatedEmail, responses=response_errors( EmailNotFoundError, EmailContentRequiredFieldsError, EmailPrivacyFooterMissingError, EmailSenderEmailNotValidError, InvalidSourceCampaignIdError, SourceCampaignConnectionsEmptyError, SourceCampaignSubscriptionOnlyRequiredError, SourceCampaignVendorMismatchError, ), ) def attach_source_campaigns( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.AttachSourceCampaignsInput, handler: Provide[AttachSourceCampaignsHandler], ) -> Any: return handler.handle( AttachSourceCampaignsRequest( identity_id=identity_id, email_id=email_id, source_campaign_connections=[ SourceCampaignConnection( source_campaign_id=c.source_campaign_id, subscription_only=c.subscription_only, ) for c in data.source_campaign_connections ], ) ) @router.post( "/automated/{emailId}/apply-changes", operation_id="applyAutomatedEmailChanges", response_model=schemas.AutomatedEmailStatus, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def apply_automated_email_changes( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[ApplyAutomatedEmailChangesHandler], ) -> Any: return handler.handle( ApplyAutomatedEmailChangesRequest( identity_id=identity_id, email_id=email_id, ) ) @router.post( "/automated/{emailId}/source-campaigns/detach", operation_id="detachSourceCampaigns", response_model=schemas.AutomatedEmail, responses=response_errors( EmailNotFoundError, InvalidSourceCampaignIdError, SourceCampaignNotAttachedError, ), ) def detach_source_campaigns( email_id: Annotated[str, Path(alias="emailId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.DetachSourceCampaignsInput, handler: Provide[DetachSourceCampaignsHandler], ) -> Any: return handler.handle( DetachSourceCampaignsRequest( identity_id=identity_id, email_id=email_id, source_campaign_ids=data.source_campaign_ids, ) ) @router.get( "/automated-emails-public-preview/{emailId}", operation_id="getAutomatedEmailPublicPreview", response_class=HTMLResponse, ) def get_automated_email_public_preview( email_id: Annotated[str, Path(alias="emailId")], handler: Provide[GetAutomatedEmailPublicPreviewHandler], pwd: str | None = None, ) -> HTMLResponse: try: preview = handler.handle( GetEmailPublicPreviewRequest(email_id=email_id, token=pwd) ) return HTMLResponse(content=preview.html, status_code=200) except EmailPublicPreviewTokenExpiredError: return HTMLResponse(content=EMAIL_PREVIEW_EXPIRED_HTML, status_code=404) except EmailNotFoundError: return HTMLResponse(content=EMAIL_NOT_FOUND_HTML, status_code=404)