File size: 4,666 Bytes
34b2632 ea06065 34b2632 ea06065 34b2632 ea06065 34b2632 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | """
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" # Draft, Approved, Sent
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}
|