| """
|
| Event-Centric Pydantic Models for MongoDB
|
| Author: AI Generated
|
| Created: 2025-11-24
|
| Purpose: Define schemas for event-specific analysis results
|
| """
|
|
|
| from pydantic import BaseModel, Field
|
| from typing import List, Dict, Optional, Any
|
| from datetime import datetime
|
| from bson import ObjectId
|
|
|
|
|
| class PyObjectId(ObjectId):
|
| """Custom ObjectId type for Pydantic v2"""
|
|
|
| @classmethod
|
| def __get_pydantic_core_schema__(cls, source_type, handler):
|
| from pydantic_core import core_schema
|
|
|
| return core_schema.union_schema([
|
| core_schema.is_instance_schema(ObjectId),
|
| core_schema.chain_schema([
|
| core_schema.str_schema(),
|
| core_schema.no_info_plain_validator_function(cls.validate),
|
| ])
|
| ],
|
| serialization=core_schema.plain_serializer_function_ser_schema(
|
| lambda x: str(x)
|
| ))
|
|
|
| @classmethod
|
| def validate(cls, v):
|
| if isinstance(v, ObjectId):
|
| return v
|
| if isinstance(v, str):
|
| if not ObjectId.is_valid(v):
|
| raise ValueError(f"Invalid ObjectId: {v}")
|
| return ObjectId(v)
|
| raise ValueError(f"Expected ObjectId or string, got {type(v)}")
|
|
|
|
|
|
|
| class MarketingContent(BaseModel):
|
| """Marketing email content generated by AI"""
|
| email_subject: str
|
| email_body: str
|
| status: str = "Draft"
|
| generated_at: datetime = Field(default_factory=datetime.utcnow)
|
| approved_at: Optional[datetime] = None
|
| approved_by: Optional[str] = None
|
|
|
|
|
| class EventAudienceSegment(BaseModel):
|
| """
|
| Audience segment specific to an event.
|
| Stores clustering results and marketing content for Event Owner review.
|
| """
|
| id: Optional[PyObjectId] = Field(default=None, alias="_id")
|
| event_code: str = Field(..., description="Event identifier")
|
| segment_name: str = Field(..., description="Human-readable segment name in Vietnamese")
|
| segment_type: str = Field(..., description="Segment category (e.g., VIP, Potential, Dormant)")
|
| user_count: int = Field(..., description="Number of users in this segment")
|
| user_ids: List[PyObjectId] = Field(default_factory=list, description="List of user ObjectIds in this segment")
|
|
|
| criteria: Dict[str, Any] = Field(
|
| default_factory=dict,
|
| description="Average statistics for this segment (e.g., avg_spend, avg_tickets, avg_recency)"
|
| )
|
|
|
| marketing_content: Optional[MarketingContent] = Field(
|
| default=None,
|
| description="AI-generated marketing email (Draft, pending approval)"
|
| )
|
|
|
| created_at: datetime = Field(default_factory=datetime.utcnow)
|
| last_updated: datetime = Field(default_factory=datetime.utcnow)
|
|
|
| class Config:
|
| populate_by_name = True
|
| arbitrary_types_allowed = True
|
| json_encoders = {ObjectId: str}
|
|
|
|
|
| class AIInsights(BaseModel):
|
| """AI-generated insights from sentiment analysis"""
|
| summary: str = Field(..., description="Overall sentiment summary in Vietnamese")
|
| top_issues: List[str] = Field(default_factory=list, description="Top 5 recurring issues")
|
| improvement_suggestions: List[str] = Field(default_factory=list, description="Actionable suggestions")
|
| predicted_nps: Optional[float] = Field(None, description="Predicted Net Promoter Score (0-100)")
|
|
|
|
|
| class EventSentimentSummary(BaseModel):
|
| """
|
| Aggregated sentiment analysis summary for an event.
|
| Provides Event Owner with quick insights about attendee feedback.
|
| """
|
| id: Optional[PyObjectId] = Field(default=None, alias="_id")
|
| event_code: str = Field(..., description="Event identifier")
|
|
|
| total_comments: int = Field(default=0, description="Total number of comments analyzed")
|
|
|
| sentiment_distribution: Dict[str, int] = Field(
|
| default_factory=dict,
|
| description="Count of Positive, Negative, Neutral comments"
|
| )
|
|
|
| avg_confidence: float = Field(default=0.0, description="Average confidence score of sentiment predictions")
|
|
|
| top_keywords: List[str] = Field(
|
| default_factory=list,
|
| description="Most frequently mentioned keywords/phrases"
|
| )
|
|
|
| ai_insights: Optional[AIInsights] = Field(
|
| default=None,
|
| description="AI-generated insights and recommendations"
|
| )
|
|
|
| last_updated: datetime = Field(default_factory=datetime.utcnow)
|
|
|
| class Config:
|
| populate_by_name = True
|
| arbitrary_types_allowed = True
|
| json_encoders = {ObjectId: str}
|
|
|
