import datetime import decimal from dataclasses import dataclass from typing import Annotated, Literal, Required, Self, TypedDict from fansifter_common.utils.msgsegment import SegmentedMessage from pydantic import AwareDatetime, BaseModel, Field, Json from ows_text_campaigns.campaigns.enums import ( CampaignStatus, MessageChannel, ShorteningMethod, ) UrlPathLike = Annotated[str, Field(pattern="^[0-9a-zA-Z-]+$", min_length=5)] class ShortenedUrl(BaseModel): id: str url: str method: ShorteningMethod domain: str path: UrlPathLike @property def preview_url(self) -> str: if self.is_standard: return f"https://{self.domain}/{self.path}" return f"https://{self.domain}/{self.path}-abc123" @property def is_standard(self) -> bool: return self.method == ShorteningMethod.STANDARD @property def is_personalized(self) -> bool: return self.method == ShorteningMethod.PERSONALIZED CampaignCIFields: tuple[str, ...] = ("name",) CampaignOrderBy = Literal[ "name.asc", "name.desc", "sendAt.asc", "sendAt.desc", "status.asc", "status.desc", "createdAt.asc", "createdAt.desc", "updatedAt.asc", "updatedAt.desc", ] @dataclass(kw_only=True) class CampaignCriteria: search: str | None vendor_ids: set[int] | None subaccount_ids: set[int] | None global_participant_ids: set[str] | None statuses: list[CampaignStatus] | None @dataclass(kw_only=True) class CampaignAnalyticsCriteria: campaign_id: str class ShortenedUrlPersonalizedAttributes(TypedDict): fan_id: str phone_number: str channel: MessageChannel class RenderMessageAttributes(TypedDict, total=False): channel: Required[MessageChannel] is_test: bool @dataclass(kw_only=True) class MessageInfo: encoding: str number_of_characters: Annotated[int, Field(alias="numberOfCharacters")] number_of_unicode_scalars: Annotated[int, Field(alias="numberOfUnicodeScalars")] segments_count: Annotated[int, Field(alias="segmentsCount")] @classmethod def from_message(cls, message: str) -> Self: segmented_message = SegmentedMessage(message) return cls( encoding=segmented_message.encoding, number_of_characters=segmented_message.number_of_characters, number_of_unicode_scalars=segmented_message.number_of_unicode_scalars, segments_count=segmented_message.segments_count, ) @dataclass(kw_only=True) class ChannelMessageInfo: channel: MessageChannel info: MessageInfo @dataclass(kw_only=True) class MessagePreview: message: str info: MessageInfo @dataclass(kw_only=True) class FanRenderedMessage: fan_id: str channel: MessageChannel recipient: str sender: str message: str media_url: str | list[str] | None @dataclass(kw_only=True) class SendTimezone: country_code: str state_provinces: Json[list[str]] | list[str] | None = None iana_tz_id: str utc_offset_std: decimal.Decimal quiet_hours_min: str quiet_hours_max: str next_safe_time: AwareDatetime | None is_in_safe_hours: bool send_at_local: datetime.datetime is_adjusted: bool = False @property def adjustment_key(self) -> str: return f"{self.country_code}:{self.iana_tz_id}" @dataclass(kw_only=True) class CampaignSendCountry: country_code: str timezones: list[SendTimezone] class TimezoneAdjustment(BaseModel): country_code: str = Field(alias="countryCode") iana_tz_id: str = Field(alias="ianaTzId") @property def key(self) -> str: return f"{self.country_code}:{self.iana_tz_id}"