from __future__ import annotations import datetime from dataclasses import field from typing import Any, ClassVar import sqlalchemy as sa from fansifter_common.adapters.db.types import ChoiceType, PydanticType from fansifter_common.auth.account import Account from fansifter_common.core.enums import Brand from fansifter_common.email import EmailType from fansifter_common.utils import timezone from fansifter_common.utils.uuid import uuid_string from sqlalchemy.orm import Mapped, declared_attr, mapped_column, relationship from sqlalchemy_utils.types import ScalarListType from email_campaigns.adapters.db import Model from email_campaigns.api.schemas import CustomFont from email_campaigns.emails.types import EmailAnalytics, EmailPreview from email_campaigns.rosters.types import FanDataListId class EmailDomain(Model, kw_only=True): __tablename__ = "email_domain" id: Mapped[str] = mapped_column(primary_key=True, default_factory=uuid_string) domain: Mapped[str] brand: Mapped[Brand] = mapped_column(ChoiceType(Brand)) vendor_id: Mapped[int | None] = mapped_column(default=None) subaccount_id: Mapped[int | None] = mapped_column(default=None) @property def account(self) -> Account | None: if self.vendor_id is not None and self.subaccount_id is not None: return Account(vendor_id=self.vendor_id, subaccount_id=self.subaccount_id) return None class BaseEmail(Model, kw_only=True, repr=False): """Abstract base class for all email types (campaigns, automated emails, etc.).""" __abstract__ = True email_type: ClassVar[EmailType] id: Mapped[str] name: Mapped[str] vendor_id: Mapped[int] subaccount_id: Mapped[int] global_participant_id: Mapped[str | None] = mapped_column(default=None) custom_list_id: Mapped[str | None] = mapped_column(default=None) sender_name: Mapped[str | None] = mapped_column(default=None) email_username: Mapped[str | None] = mapped_column(default=None) email_domain_id: Mapped[str | None] = mapped_column( sa.ForeignKey(EmailDomain.id), default=None, init=False ) subject: Mapped[str | None] = mapped_column(default=None) html_content: Mapped[str | None] = mapped_column(default=None) css_content: Mapped[str | None] = mapped_column(default=None) custom_fonts: Mapped[list[CustomFont] | None] = mapped_column( PydanticType(list[CustomFont] | None), default=None ) compressed_content: Mapped[str | None] = mapped_column(default=None) has_unapplied_changes: Mapped[bool] = mapped_column(default=False) preview_text: Mapped[str | None] = mapped_column(default=None) created_at: Mapped[datetime.datetime] = mapped_column(default_factory=timezone.now) created_by: Mapped[str] updated_at: Mapped[datetime.datetime] = mapped_column( default_factory=timezone.now, onupdate=timezone.now ) updated_by: Mapped[str] deleted_at: Mapped[datetime.datetime | None] = mapped_column(default=None) deleted_by: Mapped[str | None] = mapped_column(default=None) analytics: EmailAnalytics | None = field(default=None, init=False, compare=False) @declared_attr @classmethod def email_domain(cls) -> Mapped[EmailDomain | None]: """Email domain relationship inherited by subclasses.""" return relationship(EmailDomain, lazy="joined", default=None, repr=False) @property def account(self) -> Account: return Account(vendor_id=self.vendor_id, subaccount_id=self.subaccount_id) @account.setter def account(self, account: Account) -> None: self.vendor_id = account.vendor_id self.subaccount_id = account.subaccount_id @property def preview(self) -> EmailPreview: return EmailPreview( id=self.id, name=self.name, subject=self.subject, preview_text=self.preview_text, sender_name=self.sender_name, html=self.html_content, css=self.css_content, custom_fonts=self.custom_fonts, ) @property def sender_email(self) -> str: if self.email_domain and self.email_username: return f"{self.email_username}@{self.email_domain.domain}" return "" @property def fandata_list_id(self) -> FanDataListId | None: if self.global_participant_id is not None: return FanDataListId.from_artist_id(self.global_participant_id) elif self.custom_list_id is not None: return FanDataListId.from_custom_list_id(self.custom_list_id) return None @fandata_list_id.setter def fandata_list_id(self, fandata_list_id: FanDataListId) -> None: if fandata_list_id.is_artist: self.global_participant_id = fandata_list_id.id self.custom_list_id = None elif fandata_list_id.is_custom_list: self.custom_list_id = fandata_list_id.id self.global_participant_id = None @property def filled_data(self) -> dict[str, Any]: return { name: value for name in self.__table__.columns.keys() if (value := getattr(self, name)) is not None } class EmailTestSend(Model, kw_only=True): __tablename__ = "email_test_send" id: Mapped[str] = mapped_column(primary_key=True, default_factory=uuid_string) email_id: Mapped[str | None] = mapped_column(default=None) email_type: Mapped[EmailType | None] = mapped_column( ChoiceType(EmailType), default=None ) recipients: Mapped[list[str]] = mapped_column( ScalarListType(), default_factory=list ) created_at: Mapped[datetime.datetime] = mapped_column(default_factory=timezone.now) created_by: Mapped[str] # Analytics class EmailAnalyticsV3Dbt(Model, kw_only=True): __tablename__ = "v_email_analytics_v3_dbt" email_id: Mapped[str] = mapped_column(primary_key=True) email_type: Mapped[EmailType] = mapped_column(primary_key=True) last_event_at: Mapped[datetime.datetime] = mapped_column() sends: Mapped[int] drops: Mapped[int] delivered: Mapped[int] opens: Mapped[int] clicks: Mapped[int] unique_opens: Mapped[int] unique_clicks: Mapped[int] unique_bounces: Mapped[int] unique_spam_complaints: Mapped[int] unsubscribes: Mapped[int] open_rate: Mapped[float] delivered_rate: Mapped[float] click_through_rate: Mapped[float] click_to_open_rate: Mapped[float] bounce_rate: Mapped[float] spam_complaints_rate: Mapped[float] unsubscribe_rate: Mapped[float] unique_opens_dropoff: Mapped[float] unique_clicks_dropoff: Mapped[float] class EmailAnalyticsBenchmarkByAccountArtistV2Dbt(Model, kw_only=True): __tablename__ = "email_analytics_benchmark_by_account_artist_v2_dbt" global_participant_id: Mapped[str | None] custom_list_id: Mapped[str | None] vendor_id: Mapped[int] = mapped_column(primary_key=True) subaccount_id: Mapped[int] = mapped_column(primary_key=True) email_type: Mapped[EmailType] = mapped_column(primary_key=True) avg_open_rate: Mapped[float] avg_click_through_rate: Mapped[float] avg_bounce_rate: Mapped[float] avg_delivered_rate: Mapped[float] avg_unique_clicks: Mapped[int] avg_click_to_open_rate: Mapped[float] avg_spam_complaints_rate: Mapped[float] avg_unsubscribe_rate: Mapped[float] class EmailAnalyticsBenchmarkByLinktV3Dbt(Model, kw_only=True): __tablename__ = "v_email_analytics_by_link_v3_dbt" email_id: Mapped[str] = mapped_column(primary_key=True) email_type: Mapped[EmailType] = mapped_column(primary_key=True) url: Mapped[str] = mapped_column(primary_key=True) total_clicks: Mapped[int] unique_clicks: Mapped[int] percentage_of_total_clicks: Mapped[float] class ShopifyEmailOrdersDbt(Model, kw_only=True): __tablename__ = "shopify_email_orders_dbt" email_id: Mapped[str] = mapped_column(primary_key=True) shop_id: Mapped[str] = mapped_column(primary_key=True) currency: Mapped[str] = mapped_column(primary_key=True) country_code: Mapped[str | None] = mapped_column(primary_key=True, default=None) shop_name: Mapped[str] total_orders: Mapped[int] items_sold: Mapped[int] total_sales_usd: Mapped[float] total_discounts_usd: Mapped[float] total_taxes_usd: Mapped[float] total_delivery_usd: Mapped[float] total_net_sales_usd: Mapped[float] total_refunds_usd: Mapped[float] = mapped_column(default=0.0) first_time_buyer_rate: Mapped[float] = mapped_column(default=0.0) class ShopifyEmailProductsDbt(Model, kw_only=True): __tablename__ = "shopify_email_products_dbt" email_id: Mapped[str] = mapped_column(primary_key=True) product_title: Mapped[str | None] = mapped_column(primary_key=True, default=None) product_type: Mapped[str | None] = mapped_column(primary_key=True, default=None) currency: Mapped[str] = mapped_column(primary_key=True) total_quantity_sold: Mapped[int] total_net_sales_usd: Mapped[float] class ShopifyProductTypesDbt(Model, kw_only=True): __tablename__ = "shopify_product_types_dbt" product_type: Mapped[str] = mapped_column(primary_key=True) chart_eligible: Mapped[bool]