from __future__ import annotations from datetime import datetime from typing import Any, ClassVar, Literal from pydantic import BaseModel class Campaign(BaseModel): """Active campaign record — from local file (dev) or Snowflake dim_campaign (prod).""" campaign_key: str campaign_nm: str = "" campaign_id: str | None = None start_date: str | None = None end_date: str | None = None status: str = "active" artist_key: str = "" artist_name: str | None = None tiktok_handle: str | None = None is_client: bool = False tags: list[str] = [] track_names: list[str] = [] questions: list[str] = [] sample_size: int = 0 class SentimentRequest(BaseModel): """Slim API input for simple discovery — one artist, one prompt, N tracks.""" prompt: str artist_name: str tiktok_handle: str | None = None tracks: list[str] sample_size: int = 20 @property def artist_key(self) -> str: import re return re.sub(r"[^a-z0-9]+", "_", self.artist_name.lower()).strip("_") def track_to_tag(self, track: str) -> str: import re return re.sub(r"[^a-z0-9]", "", track.lower()) def to_campaign(self) -> Campaign: return Campaign( campaign_key=self.artist_key, campaign_nm=self.artist_name, campaign_id=self.artist_key, artist_key=self.artist_key, artist_name=self.artist_name, tiktok_handle=self.tiktok_handle, tags=[self.track_to_tag(t) for t in self.tracks], track_names=self.tracks, questions=[self.prompt], sample_size=self.sample_size, status="active", ) class CampaignPost(BaseModel): """One post in campaign config — tracked for monitoring.""" post_id: str url: str = "" sound_id: str | None = None campaign_config_key: str = "" is_active: bool = True class VideoRef(BaseModel): video_id: str video_url: str author: str class VideoData(BaseModel): video_id: str | None = None url: str author: str | None = None created_at: datetime | None = None views: int | None = None likes: int | None = None comments: int | None = None shares: int | None = None favorites: int | None = None caption: str | None = None hashtags: list[str] = [] sound_url: str | None = None sound_id: str | None = None sound_title: str | None = None class PostMetrics(BaseModel): """One time-series point per post per run → fact_campaign_snapshot.""" campaign_config_key: str post_id: str sound_id: str | None = None run_id: str observed_at: str views: int | None = None likes: int | None = None comments: int | None = None shares: int | None = None favorites: int | None = None delta_views: int = 0 delta_likes: int = 0 delta_comments: int = 0 delta_shares: int = 0 delta_favorites: int = 0 _METRICS: ClassVar[tuple[str, ...]] = ( "views", "likes", "comments", "shares", "favorites", ) @classmethod def from_observation( cls, *, campaign_config_key: str, post_id: str, run_id: str, observed_at: str, prev: dict[str, Any] | None = None, **fields: Any, ) -> PostMetrics: """Build a metrics record computing deltas against the previous snapshot (0 when first).""" prev = prev or {} deltas = {} for metric in cls._METRICS: curr_value, prev_value = fields.get(metric), prev.get(metric) deltas[f"delta_{metric}"] = ( curr_value - prev_value if curr_value is not None and prev_value is not None else 0 ) return cls.model_validate( { "campaign_config_key": campaign_config_key, "post_id": post_id, "run_id": run_id, "observed_at": observed_at, **fields, **deltas, } ) class RunRecord(BaseModel): """dim_run — one record per discovery or monitoring run.""" run_id: str run_type: Literal["discovery", "watch", "artist_snapshot"] campaign_key: str | None = None started_at: str finished_at: str | None = None posts_ingested: int | None = None is_sample: bool = False sample_size_target: int | None = None class SoundRecord(BaseModel): """dim_sound — upsert payload, written alongside campaign post.""" sound_id: str title: str | None = None artist_key: str | None = None isrc: str | None = None url: str | None = None first_seen_at: str class CampaignConfigRecord(BaseModel): """fact_campaign_config — one row per tracked post.""" campaign_key: str artist_key: str sound_key: str | None = None post_id: str content_tier: int hashtags: str | None = None start_date: str | None = None end_date: str | None = None is_active: bool = True class ArtistSnapshotRecord(BaseModel): """fact_artist_snapshot — TikTok profile stats at a point in time.""" artist_key: str run_id: str observed_at: str tiktok_followers: int | None = None total_creates: int | None = None total_views: int | None = None total_likes: int | None = None total_comments: int | None = None total_shares: int | None = None engagement_rate: float | None = None class CampaignDiscoveryRunReasoning(BaseModel): """Agent reasoning batch — persisted alongside metrics for a discovery run.""" campaign_config_key: str run_id: str count: int reasoning: str | None = None items: list[dict[str, Any]] class TrendSignal(BaseModel): signal: str evidence: str supporting_video_ids: list[str] class CampaignSentiment(BaseModel): """fact_campaign_sentiment — AI-generated sentiment analysis for a campaign run.""" campaign_config_key: str run_id: str observed_at: str sentiment: Literal["positive", "neutral", "negative"] confidence: float summary: str reasoning: str how_sound_is_used: str confidence_rationale: str detected_topics: list[str] = [] primary_hashtags: list[str] = [] trend_signals: list[TrendSignal] = [] virality_factors: list[str] = [] video_ids: list[str] = []