""" 消息通知相关的Pydantic Schema """ from typing import Optional, List, Dict, Any from datetime import datetime from pydantic import BaseModel, Field from enum import Enum class NotificationTypeEnum(str, Enum): """通知类型枚举""" SYSTEM = "system" # 系统通知 APPROVAL = "approval" # 审批通知 MAINTENANCE = "maintenance" # 维修通知 ALLOCATION = "allocation" # 调拨通知 ASSET = "asset" # 资产通知 WARRANTY = "warranty" # 保修到期通知 REMINDER = "reminder" # 提醒通知 class PriorityEnum(str, Enum): """优先级枚举""" LOW = "low" NORMAL = "normal" HIGH = "high" URGENT = "urgent" class NotificationBase(BaseModel): """消息通知基础Schema""" recipient_id: int = Field(..., description="接收人ID") title: str = Field(..., min_length=1, max_length=200, description="通知标题") content: str = Field(..., min_length=1, description="通知内容") notification_type: NotificationTypeEnum = Field(..., description="通知类型") priority: PriorityEnum = Field(default=PriorityEnum.NORMAL, description="优先级") related_entity_type: Optional[str] = Field(None, max_length=50, description="关联实体类型") related_entity_id: Optional[int] = Field(None, description="关联实体ID") action_url: Optional[str] = Field(None, max_length=500, description="操作链接") extra_data: Optional[Dict[str, Any]] = Field(None, description="额外数据") send_email: bool = Field(default=False, description="是否发送邮件") send_sms: bool = Field(default=False, description="是否发送短信") expire_at: Optional[datetime] = Field(None, description="过期时间") class NotificationCreate(NotificationBase): """创建消息通知Schema""" pass class NotificationUpdate(BaseModel): """更新消息通知Schema""" is_read: Optional[bool] = Field(None, description="是否已读") class NotificationInDB(BaseModel): """数据库中的消息通知Schema""" id: int recipient_id: int recipient_name: str title: str content: str notification_type: str priority: str is_read: bool read_at: Optional[datetime] related_entity_type: Optional[str] related_entity_id: Optional[int] action_url: Optional[str] extra_data: Optional[Dict[str, Any]] sent_via_email: bool sent_via_sms: bool created_at: datetime expire_at: Optional[datetime] class Config: from_attributes = True class NotificationResponse(NotificationInDB): """消息通知响应Schema""" pass class NotificationQueryParams(BaseModel): """消息通知查询参数""" recipient_id: Optional[int] = Field(None, description="接收人ID") notification_type: Optional[NotificationTypeEnum] = Field(None, description="通知类型") priority: Optional[PriorityEnum] = Field(None, description="优先级") is_read: Optional[bool] = Field(None, description="是否已读") start_time: Optional[datetime] = Field(None, description="开始时间") end_time: Optional[datetime] = Field(None, description="结束时间") keyword: Optional[str] = Field(None, description="关键词") page: int = Field(default=1, ge=1, description="页码") page_size: int = Field(default=20, ge=1, le=100, description="每页数量") class NotificationBatchCreate(BaseModel): """批量创建通知Schema""" recipient_ids: List[int] = Field(..., min_items=1, description="接收人ID列表") title: str = Field(..., min_length=1, max_length=200, description="通知标题") content: str = Field(..., min_length=1, description="通知内容") notification_type: NotificationTypeEnum = Field(..., description="通知类型") priority: PriorityEnum = Field(default=PriorityEnum.NORMAL, description="优先级") action_url: Optional[str] = Field(None, max_length=500, description="操作链接") extra_data: Optional[Dict[str, Any]] = Field(None, description="额外数据") class NotificationBatchUpdate(BaseModel): """批量更新通知Schema""" notification_ids: List[int] = Field(..., min_items=1, description="通知ID列表") is_read: bool = Field(..., description="是否已读") class NotificationStatistics(BaseModel): """通知统计Schema""" total_count: int = Field(..., description="总通知数") unread_count: int = Field(..., description="未读数") read_count: int = Field(..., description="已读数") high_priority_count: int = Field(..., description="高优先级数") urgent_count: int = Field(..., description="紧急通知数") type_distribution: List[Dict[str, Any]] = Field(default_factory=list, description="类型分布") # ===== 通知模板Schema ===== class NotificationTemplateBase(BaseModel): """通知模板基础Schema""" template_code: str = Field(..., min_length=1, max_length=50, description="模板编码") template_name: str = Field(..., min_length=1, max_length=200, description="模板名称") notification_type: NotificationTypeEnum = Field(..., description="通知类型") title_template: str = Field(..., min_length=1, max_length=200, description="标题模板") content_template: str = Field(..., min_length=1, description="内容模板") variables: Optional[Dict[str, str]] = Field(None, description="变量说明") priority: PriorityEnum = Field(default=PriorityEnum.NORMAL, description="默认优先级") send_email: bool = Field(default=False, description="是否发送邮件") send_sms: bool = Field(default=False, description="是否发送短信") is_active: bool = Field(default=True, description="是否启用") description: Optional[str] = Field(None, description="模板描述") class NotificationTemplateCreate(NotificationTemplateBase): """创建通知模板Schema""" pass class NotificationTemplateUpdate(BaseModel): """更新通知模板Schema""" template_name: Optional[str] = Field(None, min_length=1, max_length=200) title_template: Optional[str] = Field(None, min_length=1, max_length=200) content_template: Optional[str] = Field(None, min_length=1) variables: Optional[Dict[str, str]] = None priority: Optional[PriorityEnum] = None send_email: Optional[bool] = None send_sms: Optional[bool] = None is_active: Optional[bool] = None description: Optional[str] = None class NotificationTemplateInDB(BaseModel): """数据库中的通知模板Schema""" id: int template_code: str template_name: str notification_type: str title_template: str content_template: str variables: Optional[Dict[str, str]] priority: str send_email: bool send_sms: bool is_active: bool description: Optional[str] created_at: datetime updated_at: datetime class Config: from_attributes = True class NotificationTemplateResponse(NotificationTemplateInDB): """通知模板响应Schema""" pass class NotificationSendFromTemplate(BaseModel): """从模板发送通知Schema""" template_code: str = Field(..., description="模板编码") recipient_ids: List[int] = Field(..., min_items=1, description="接收人ID列表") variables: Dict[str, Any] = Field(default_factory=dict, description="模板变量") related_entity_type: Optional[str] = Field(None, description="关联实体类型") related_entity_id: Optional[int] = Field(None, description="关联实体ID") action_url: Optional[str] = Field(None, description="操作链接")