from typing import Annotated, Any from anydi import Provide from anydi.ext.fastapi import Inject from fansifter_common.api.schemas import response_errors from fansifter_common.artist.exceptions import InvalidGlobalParticipantIdError from fansifter_common.auth.exceptions import InvalidAccountError from fastapi import APIRouter, Body, Depends, Header, Path, Query, status from fastapi.security import HTTPBasic, HTTPBasicCredentials from pydantic import AwareDatetime from starlette.responses import HTMLResponse, Response from email_campaigns.api import auth, schemas from email_campaigns.audiences.exceptions import ( ArchivedAudienceError, EmptyAudienceError, InvalidAudienceIdError, NotSupportedAudienceTargetError, ) from email_campaigns.campaigns.exceptions import ( CampaignAlreadyQueuedError, EmailCampaignMustBeDraftError, EmailCampaignMustBeScheduledError, EmailCampaignNotFoundError, EmailCampaignRequiredFieldsError, EmailCampaignSendAtMustBeFutureError, EmailCampaignSenderEmailNotValidError, EmailCampaignUniqueNameError, MalformedEmailCampaignError, ) from email_campaigns.campaigns.handlers import ( AutosaveCampaignHandler, AutosaveCampaignV2Handler, CreateCampaignHandler, CreateCampaignRequest, DeleteCampaignHandler, DeleteCampaignRequest, DuplicateCampaignHandler, DuplicateCampaignRequest, EditScheduleCampaignHandler, EditScheduleCampaignRequest, EstimateSendTimeHandler, EstimateSendTimeRequest, GenerateCampaignPublicPreviewUrlHandler, GetCampaignAnalyticsHandler, GetCampaignHandler, GetCampaignInternalHandler, GetCampaignInternalRequest, GetCampaignLinkAnalyticsHandler, GetCampaignLiveStatusHandler, GetCampaignLiveStatusRequest, GetCampaignPreviewHandler, GetCampaignPrivacyFooterHandler, GetCampaignPublicPreviewHandler, GetCampaignRequest, GetCampaignSalesAnalyticsHandler, GetCampaignSalesLocationsAnalyticsHandler, GetCampaignSalesProductsAnalyticsHandler, GetCampaignSalesShopsAnalyticsHandler, GetCampaignsHandler, GetCampaignsRequest, GetCampaignStripoAuthTokenHandler, ScheduleCampaignHandler, ScheduleCampaignRequest, SearchCampaignsHandler, SearchCampaignsRequest, SendCampaignHandler, SendCampaignRequest, SendTestCampaignEmailHandler, UnscheduleCampaignHandler, UnscheduleCampaignRequest, UpdateCampaignContentHandler, UpdateCampaignFontsHandler, UpdateCampaignHandler, UpdateCampaignRequest, ) from email_campaigns.campaigns.handlers.bulk_estimate_send_time import ( BulkEstimateSendTimeHandler, BulkEstimateSendTimeRequest, ) from email_campaigns.campaigns.models import EmailCampaign from email_campaigns.emails.exceptions import ( AutosaveUnauthorizedError, EmailDomainMissingError, EmailNotFoundError, EmailPublicPreviewTokenExpiredError, 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.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 ( FanDataListIdIsRequiredError, InvalidCustomListIdError, ) router = APIRouter(tags=["Campaigns"]) basic_auth = HTTPBasic() @router.post( "/campaigns/list", operation_id="getCampaigns", response_model=schemas.LimitOffsetPage[schemas.EmailCampaign], response_model_by_alias=True, ) def get_campaigns( identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.GetEmailCampaignsInput, handler: Provide[GetCampaignsHandler], ) -> Any: order_by = data.order_by or GetCampaignsRequest.DEFAULT_ORDER_BY limit = data.limit or GetCampaignsRequest.DEFAULT_LIMIT offset = data.offset or GetCampaignsRequest.DEFAULT_OFFSET response = handler.handle( GetCampaignsRequest( identity_id=identity_id, search=data.search, vendor_id=data.vendor_id, subaccount_id=data.subaccount_id, fandata_list_ids=data.fandata_list_ids, statuses=data.statuses, audience_id=data.audience_id, order_by=order_by, limit=limit, offset=offset, ) ) return schemas.LimitOffsetPage( total=response.total, items=response.items, limit=limit, offset=offset, ) @router.post( "/campaigns/search", operation_id="searchCampaigns", response_model=list[schemas.CampaignPreview], response_model_by_alias=True, ) def search_campaigns( identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.SearchCampaignsInput, handler: Annotated[SearchCampaignsHandler, Inject()], ) -> Any: return handler.handle( SearchCampaignsRequest( identity_id=identity_id, campaign_ids=data.campaign_ids, fandata_list_ids=data.fandata_list_ids, vendor_id=data.vendor_id, subaccount_id=data.subaccount_id, search=data.search, search_context=data.search_context, statuses=data.statuses, audience_id=data.audience_id, order_by=data.order_by, limit=data.limit or SearchCampaignsRequest.DEFAULT_LIMIT, ) ) @router.post( "/campaigns", operation_id="createEmailCampaign", response_model=schemas.EmailCampaign, response_model_by_alias=True, status_code=status.HTTP_201_CREATED, responses=response_errors( InvalidGlobalParticipantIdError, InvalidAccountError, EmailCampaignUniqueNameError, InvalidCustomListIdError, FanDataListIdIsRequiredError, ), ) def create_email_campaign( identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.CreateEmailCampaignInput, handler: Provide[CreateCampaignHandler], ) -> EmailCampaign: return handler.handle( CreateCampaignRequest( identity_id=identity_id, name=data.name, vendor_id=data.vendor_id, subaccount_id=data.subaccount_id, fandata_list_id=data.fandata_list_id, html_content=data.html_content, css_content=data.css_content, ) ) @router.put( "/campaigns/{campaignId}", operation_id="updateEmailCampaign", response_model=schemas.EmailCampaign, response_model_by_alias=True, responses=response_errors( EmailCampaignNotFoundError, EmailCampaignMustBeDraftError, EmailCampaignUniqueNameError, InvalidGlobalParticipantIdError, InvalidAudienceIdError, MalformedEmailCampaignError, ArchivedAudienceError, NotSupportedAudienceTargetError, EmptyAudienceError, InvalidEmailDomainIdError, InvalidCustomListIdError, FanDataListIdIsRequiredError, ), ) def update_email_campaign( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.UpdateEmailCampaignInput, handler: Provide[UpdateCampaignHandler], ) -> Any: return handler.handle( UpdateCampaignRequest( identity_id=identity_id, campaign_id=campaign_id, name=data.name, vendor_id=data.vendor_id, subaccount_id=data.subaccount_id, fandata_list_id=data.fandata_list_id, 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, audience_id=data.audience_id, ) ) @router.delete( "/campaigns/{campaignId}", operation_id="deleteEmailCampaignContent", response_class=Response, status_code=status.HTTP_204_NO_CONTENT, responses=response_errors( EmailCampaignNotFoundError, EmailCampaignMustBeDraftError, ), ) def delete_email_campaign( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[DeleteCampaignHandler], ) -> None: handler.handle( DeleteCampaignRequest( identity_id=identity_id, campaign_id=campaign_id, ) ) @router.get( "/campaigns/{campaignId}", operation_id="getEmailCampaign", response_model=schemas.EmailCampaign, response_model_by_alias=True, responses=response_errors( EmailCampaignNotFoundError, ), ) def get_email_campaign( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetCampaignHandler], ) -> Any: return handler.handle( GetCampaignRequest( identity_id=identity_id, campaign_id=campaign_id, ) ) @router.get( "/campaigns/{campaignId}/preview", operation_id="getCampaignEmailPreview", response_model=schemas.EmailPreview, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_campaign_email_preview( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetCampaignPreviewHandler], apply_substitutions: Annotated[bool, Query(alias="applySubstitutions")] = False, ) -> Any: return handler.handle( GetEmailPreviewRequest( identity_id=identity_id, email_id=campaign_id, apply_substitutions=apply_substitutions, ) ) @router.patch( "/campaigns/{campaignId}/content", operation_id="updateEmailCampaignContent", response_model=schemas.EmailPreview, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, EmailCampaignMustBeDraftError, ), ) def update_email_campaign_content( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.UpdateEmailContentInput, handler: Provide[UpdateCampaignContentHandler], ) -> Any: return handler.handle( UpdateEmailContentRequest( identity_id=identity_id, email_id=campaign_id, html=data.html, css=data.css, ) ) @router.patch( "/campaigns/{campaignId}/fonts", operation_id="updateEmailCampaignFonts", response_model=list[schemas.CustomFont], response_model_by_alias=True, responses=response_errors(EmailNotFoundError, InvalidGoogleFontError), ) def update_email_campaign_fonts( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.UpdateEmailFontsInput, handler: Provide[UpdateCampaignFontsHandler], ) -> Any: return handler.handle( UpdateEmailFontsRequest( identity_id=identity_id, email_id=campaign_id, font=data.font, ) ) @router.post( "/campaigns/{campaignId}/send-test-email", operation_id="sendTestEmail", response_class=Response, status_code=status.HTTP_204_NO_CONTENT, responses=response_errors( EmailCampaignNotFoundError, EmailCampaignRequiredFieldsError, EmailCampaignSenderEmailNotValidError, ), ) def send_test_email_campaign( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.SendTestEmailCampaignInput, handler: Provide[SendTestCampaignEmailHandler], ) -> None: handler.handle( SendTestEmailRequest( identity_id=identity_id, email_id=campaign_id, emails=data.emails, content=data.content, ) ) @router.post( "/campaigns/{campaignId}/schedule", operation_id="scheduleEmailCampaign", response_class=Response, status_code=status.HTTP_204_NO_CONTENT, responses=response_errors( EmailCampaignNotFoundError, EmailCampaignMustBeDraftError, EmailCampaignSendAtMustBeFutureError, EmailCampaignRequiredFieldsError, InvalidAudienceIdError, ArchivedAudienceError, NotSupportedAudienceTargetError, EmptyAudienceError, ), ) def schedule_email_campaign( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.ScheduleEmailCampaignInput, handler: Provide[ScheduleCampaignHandler], ) -> None: handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign_id, send_at=data.send_at, timezone=data.timezone, ) ) @router.put( "/campaigns/{campaignId}/schedule", operation_id="editScheduleEmailCampaign", response_class=Response, status_code=status.HTTP_204_NO_CONTENT, responses=response_errors( EmailCampaignNotFoundError, EmailCampaignMustBeScheduledError, EmailCampaignSendAtMustBeFutureError, EmailCampaignRequiredFieldsError, InvalidAudienceIdError, ArchivedAudienceError, NotSupportedAudienceTargetError, EmptyAudienceError, EmailDomainMissingError, ), ) def edit_schedule_email_campaign( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.ScheduleEmailCampaignInput, handler: Provide[EditScheduleCampaignHandler], ) -> None: handler.handle( EditScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign_id, send_at=data.send_at, timezone=data.timezone, ) ) @router.post( "/campaigns/{campaignId}/unschedule", operation_id="unscheduleEmailCampaign", response_class=Response, status_code=status.HTTP_204_NO_CONTENT, responses=response_errors( EmailCampaignNotFoundError, EmailCampaignMustBeScheduledError, CampaignAlreadyQueuedError, ), ) def unschedule_email_campaign( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[UnscheduleCampaignHandler], ) -> None: handler.handle( UnscheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign_id, ) ) @router.post( "/campaigns/{campaignId}/send", operation_id="sendEmailCampaign", response_class=Response, status_code=status.HTTP_204_NO_CONTENT, responses=response_errors( EmailCampaignNotFoundError, EmailCampaignMustBeDraftError, EmailCampaignRequiredFieldsError, InvalidAudienceIdError, ArchivedAudienceError, NotSupportedAudienceTargetError, EmptyAudienceError, ), ) def send_email_campaign( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[SendCampaignHandler], ) -> None: handler.handle( SendCampaignRequest( identity_id=identity_id, campaign_id=campaign_id, ) ) @router.post( "/campaigns/{campaignId}/duplicate", operation_id="duplicateEmailCampaign", response_model=schemas.EmailCampaign, response_model_by_alias=True, status_code=status.HTTP_201_CREATED, responses=response_errors( EmailCampaignNotFoundError, EmailCampaignUniqueNameError, ), ) def duplicate_email_campaign( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], data: schemas.DuplicateEmailCampaignInput, handler: Provide[DuplicateCampaignHandler], ) -> Any: return handler.handle( DuplicateCampaignRequest( identity_id=identity_id, campaign_id=campaign_id, name=data.name, ) ) @router.get( "/campaigns/{campaignId}/status", operation_id="getCampaignLiveStatus", status_code=status.HTTP_200_OK, response_model=schemas.GetCampaignLiveStatusOutput, responses=response_errors(EmailCampaignNotFoundError), ) def get_campaign_live_status( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetCampaignLiveStatusHandler], ) -> Any: return handler.handle( GetCampaignLiveStatusRequest( identity_id=identity_id, campaign_id=campaign_id, ) ) @router.get( "/campaigns/{campaignId}/stripo/token", operation_id="getEmailCampaignStripoAuthToken", response_model=schemas.GetEmailStripoAuthTokenResponse, response_model_by_alias=True, ) def get_stripo_auth_token( identity_id: Annotated[str, Depends(auth.get_identity_id)], campaign_id: Annotated[str, Path(alias="campaignId")], handler: Provide[GetCampaignStripoAuthTokenHandler], ) -> Any: return handler.handle( GetEmailStripoAuthTokenRequest(identity_id=identity_id, email_id=campaign_id) ) @router.get( "/campaigns/{campaignId}/internal", operation_id="getEmailCampaignInternal", tags=["internal"], response_model=schemas.EmailCampaignInternal, response_model_by_alias=True, responses=response_errors( EmailCampaignNotFoundError, EmailCampaignRequiredFieldsError, ), ) def get_email_campaign_internal( campaign_id: Annotated[str, Path(alias="campaignId")], handler: Provide[GetCampaignInternalHandler], ) -> Any: return handler.handle( GetCampaignInternalRequest( campaign_id=campaign_id, ) ) @router.get( "/campaigns/{campaignId}/analytics", operation_id="getEmailCampaignAnalytics", response_model=schemas.GetEmailAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_email_campaign_analytics( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetCampaignAnalyticsHandler], ) -> Any: return handler.handle( GetEmailAnalyticsRequest( identity_id=identity_id, email_id=campaign_id, ) ) @router.get( "/campaigns/{campaignId}/analytics/links", operation_id="getEmailCampaignLinkAnalytics", response_model=schemas.GetEmailLinkAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_email_campaign_link_analytics( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetCampaignLinkAnalyticsHandler], ) -> Any: return handler.handle( GetEmailLinkAnalyticsRequest( identity_id=identity_id, email_id=campaign_id, ) ) @router.get( "/campaigns/{campaignId}/analytics/sales", operation_id="getEmailCampaignSalesAnalytics", response_model=schemas.GetEmailSalesAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_email_campaign_sales_analytics( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetCampaignSalesAnalyticsHandler], ) -> Any: return handler.handle( GetEmailSalesAnalyticsRequest( identity_id=identity_id, email_id=campaign_id, ) ) @router.get( "/campaigns/{campaignId}/analytics/sales/products", operation_id="getEmailCampaignSalesProductsAnalytics", response_model=schemas.GetEmailSalesProductsAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_email_campaign_sales_products_analytics( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetCampaignSalesProductsAnalyticsHandler], ) -> Any: return handler.handle( GetEmailSalesProductsAnalyticsRequest( identity_id=identity_id, email_id=campaign_id, ) ) @router.get( "/campaigns/{campaignId}/analytics/sales/locations", operation_id="getEmailCampaignSalesLocationsAnalytics", response_model=schemas.GetEmailSalesLocationsAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_email_campaign_sales_locations_analytics( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetCampaignSalesLocationsAnalyticsHandler], ) -> Any: return handler.handle( GetEmailSalesLocationsAnalyticsRequest( identity_id=identity_id, email_id=campaign_id, ) ) @router.get( "/campaigns/{campaignId}/analytics/sales/shops", operation_id="getEmailCampaignSalesShopsAnalytics", response_model=schemas.GetEmailSalesShopsAnalyticsResponse, response_model_by_alias=True, responses=response_errors( EmailNotFoundError, ), ) def get_email_campaign_sales_shops_analytics( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetCampaignSalesShopsAnalyticsHandler], ) -> Any: return handler.handle( GetEmailSalesShopsAnalyticsRequest( identity_id=identity_id, email_id=campaign_id, ) ) @router.post( "/campaigns/autosave", operation_id="autosaveEmailCampaign", response_class=Response, tags=["internal"], status_code=status.HTTP_200_OK, responses=response_errors( EmailCampaignNotFoundError, EmailCampaignMustBeDraftError, AutosaveUnauthorizedError, ), ) async def autosave_email_campaign( credentials: Annotated[HTTPBasicCredentials, Depends(basic_auth)], data: schemas.AutosaveEmailInput, handler: Provide[AutosaveCampaignHandler], 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( "/campaigns/autosave/v2", operation_id="autosaveEmailCampaignV2", response_class=Response, tags=["internal"], status_code=status.HTTP_200_OK, responses=response_errors( EmailCampaignNotFoundError, EmailCampaignMustBeDraftError, AutosaveUnauthorizedError, ), ) async def autosave_email_campaign_v2( credentials: Annotated[HTTPBasicCredentials, Depends(basic_auth)], data: schemas.AutosaveEmailV2Input, handler: Provide[AutosaveCampaignV2Handler], ) -> None: handler.handle( AutosaveEmailV2Request( email_id=data.email_id, user_id=data.user_id, username=credentials.username, password=credentials.password, ) ) @router.get( "/campaigns/{campaignId}/send-estimation", operation_id="estimateEmailCampaignSendTime", response_model=schemas.EstimateSendTimeResponse, response_model_by_alias=True, responses=response_errors( EmailCampaignNotFoundError, ), ) def estimate_email_campaign_send_time( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[EstimateSendTimeHandler], send_at: Annotated[AwareDatetime | None, Query(alias="sendAt")] = None, ) -> Any: response = handler.handle( EstimateSendTimeRequest( identity_id=identity_id, campaign_id=campaign_id, send_at=send_at, ) ) return response @router.post( "/campaigns/send-estimation", operation_id="bulkEstimateEmailCampaignSendTime", response_model=list[schemas.CampaignEstimate], response_model_by_alias=True, ) def bulk_estimate_email_campaign_send_time( identity_id: Annotated[str, Depends(auth.get_identity_id)], params: Annotated[schemas.BulkEstimateSendTimeRequest, Body()], handler: Provide[BulkEstimateSendTimeHandler], ) -> Any: return handler.handle( BulkEstimateSendTimeRequest( identity_id=identity_id, campaign_ids=params.campaign_ids, ) ) @router.get( "/campaigns/{campaignId}/privacy-footer", operation_id="getEmailCampaignPrivacyFooter", response_model=schemas.EmailPrivacyFooter, response_model_by_alias=True, responses=response_errors( EmailCampaignNotFoundError, ), ) def get_email_campaign_privacy_footer( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], handler: Provide[GetCampaignPrivacyFooterHandler], ) -> Any: privacy_footer = handler.handle( GetEmailPrivacyFooterRequest( identity_id=identity_id, email_id=campaign_id, ) ) return {"content": privacy_footer} @router.post( "/campaigns/{campaignId}/public-preview-url", operation_id="generateEmailCampaignPublicPreviewUrl", response_model=schemas.GenerateEmailPublicPreviewUrlResponse, response_model_by_alias=True, responses=response_errors( EmailCampaignNotFoundError, ), ) def generate_email_campaign_public_preview_url( campaign_id: Annotated[str, Path(alias="campaignId")], identity_id: Annotated[str, Depends(auth.get_identity_id)], input_data: schemas.GenerateEmailPublicPreviewUrlInput, handler: Annotated[GenerateCampaignPublicPreviewUrlHandler, Inject()], ) -> Any: response = handler.handle( GenerateEmailPublicPreviewUrlRequest( identity_id=identity_id, email_id=campaign_id, domain=input_data.domain, ) ) return {"url": response.url} @router.get( "/campaigns-public-preview/hello/", response_model=schemas.HelloResponse, ) def get_email_campaign_public_hello() -> Any: return {"status": "ok"} @router.get( "/campaigns-public-preview/{campaignId}", operation_id="getEmailCampaignPublicPreview", response_class=HTMLResponse, ) def get_email_campaign_public_preview( campaign_id: Annotated[str, Path(alias="campaignId")], handler: Annotated[GetCampaignPublicPreviewHandler, Inject()], pwd: str | None = None, ) -> HTMLResponse: try: preview = handler.handle( GetEmailPublicPreviewRequest( email_id=campaign_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)