import base64 from datetime import UTC, datetime from enum import StrEnum from typing import Any from pydantic import BaseModel, Field, field_serializer class ConversationState(StrEnum): GREETING = "greeting" OPT_IN_ASKED = "opt_in_asked" OPTED_IN = "opted_in" OPTED_OUT = "opted_out" class TextMessageAttachment(BaseModel): type: str content: bytes filename: str | None = None mimetype: str | None = None @field_serializer("content", mode="plain") def encode(self, content: bytes) -> str: return base64.b64encode(content).decode("ascii") class SenderType(StrEnum): artist = "artist" fan = "fan" twilio = "twilio" fansifter = "fansifter" class TextMessage(BaseModel): sender: SenderType text: str = Field(..., min_length=1) # ISO 8601 timestamp for precise client formatting ts: str = Field( default_factory=lambda: datetime.now(UTC) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z") ) attachments: list[TextMessageAttachment] = Field(default_factory=list) class MessageType(StrEnum): init = "init" chat = "chat" reset = "reset" settings = "settings" settings_ack = "settings_ack" log_event = "log_event" error = "error" ping = "ping" pong = "pong" class Message(BaseModel): type: MessageType payload: dict[str, Any] class UserAppSettings(BaseModel): delay_min: float = 3 delay_max: float = 7 artist_name: str = "Mozart" show_events: bool = True class UserTwilioSettings(BaseModel): opt_out_keywords: list[str] = ["stop"] opt_out_message: str = "" opt_in_keywords: list[str] = ["start"] opt_in_message: str = "" help_keywords: list[str] = ["help"] help_message: str = "" class UserFansifterSettings(BaseModel): welcome: str = "" opted_in: str = "" info: str = "" class UserSettings(BaseModel): app_settings: UserAppSettings = Field(default_factory=UserAppSettings) twilio_settings: UserTwilioSettings = Field(default_factory=UserTwilioSettings) fansifter_settings: UserFansifterSettings = Field(default_factory=UserFansifterSettings) class User(BaseModel): model_config = {"arbitrary_types_allowed": True} id: str messages: list[TextMessage] = [] settings: UserSettings = Field(default_factory=UserSettings) new_user: bool = True connector: Any @property def state(self) -> ConversationState: return self.connector.fansifter_connector.state