from __future__ import annotations import datetime import sqlalchemy as sa from fansifter_common.adapters.db.types import ChoiceType from fansifter_common.auth.account import Account from fansifter_common.utils.uuid import uuid_string from sqlalchemy.orm import Mapped, mapped_column, relationship from ows_text_campaigns.adapters.db import Model, PgModel from ows_text_campaigns.audiences.enums import AudienceTarget, FanDataListType class Audience(PgModel, kw_only=True): __tablename__ = "audience" id: Mapped[str] = mapped_column(primary_key=True, default_factory=uuid_string) name: Mapped[str] vendor_id: Mapped[int] subaccount_id: Mapped[int] target: Mapped[AudienceTarget] = mapped_column(ChoiceType(AudienceTarget)) fan_count: Mapped[int] = mapped_column(default=0) archived_at: Mapped[datetime.datetime | None] = mapped_column(default=None) # relations fandata_lists: Mapped[list[AudienceFanDataList]] = relationship( default_factory=list ) @property def is_archived(self) -> bool: return self.archived_at is not None @property def account(self) -> Account: return Account(vendor_id=self.vendor_id, subaccount_id=self.subaccount_id) @property def is_text(self) -> bool: return self.target == AudienceTarget.TEXT @property def global_participant_ids(self) -> list[str]: return [ fandata_item.list_id for fandata_item in self.fandata_lists if fandata_item.type == FanDataListType.ARTIST ] @property def global_participant_id(self) -> str | None: try: return self.global_participant_ids[0] except IndexError: return None class AudienceFanDataList(PgModel, kw_only=True): __tablename__ = "audience_fandata_list" __table_args__ = (sa.UniqueConstraint("audience_id", "list_id", "type"),) id: Mapped[str] = mapped_column(primary_key=True, default_factory=uuid_string) audience_id: Mapped[str] = mapped_column(sa.ForeignKey("audience.id"), init=False) list_id: Mapped[str] type: Mapped[FanDataListType] = mapped_column(ChoiceType(FanDataListType)) class AudienceTextFan(Model, kw_only=True): __tablename__ = "audience_text_fan" audience_id: Mapped[str] = mapped_column(primary_key=True) fan_id: Mapped[str] = mapped_column(primary_key=True) fan_phone_number: Mapped[str] fan_country: Mapped[str] fan_state: Mapped[str | None] = mapped_column(default=None) fan_area_code: Mapped[str | None] = mapped_column(default=None) channel: Mapped[str]