""" 兑换码相关 Schema 定义兑换码数据的验证和序列化规则。 """ from datetime import datetime from typing import Annotated from pydantic import Field, field_validator from app.schemas.base import BaseSchema, PaginatedResponse from app.schemas.balance import UNITS_PER_DISPLAY, display_to_units, format_display from app.models.redeem_code import RedeemCodeStatus # ============================================================ # 用户兑换 Schema # ============================================================ class RedeemRequest(BaseSchema): """兑换请求""" code: Annotated[ str, Field( min_length=1, max_length=32, description="兑换码", examples=["ABCD-EFGH-JKLM-NPQR"], ), ] @field_validator("code") @classmethod def normalize_code(cls, v: str) -> str: """标准化兑换码格式""" # 移除空格,转大写 return v.strip().upper().replace(" ", "") class RedeemResponse(BaseSchema): """兑换响应""" success: bool = True message: str = "兑换成功" face_value: str = Field(description="兑换金额") balance_before: str = Field(description="兑换前余额") balance_after: str = Field(description="兑换后余额") # ============================================================ # 兑换码信息 Schema # ============================================================ class RedeemCodeResponse(BaseSchema): """兑换码信息响应""" id: str code: str face_value_units: int = Field(description="面值(单位额度)") status: RedeemCodeStatus max_uses: int used_count: int expires_at: datetime | None used_at: datetime | None created_at: datetime @property def face_value(self) -> str: """显示面值(2 位小数)""" return format_display(self.face_value_units) @property def is_valid(self) -> bool: """是否有效""" if self.status != RedeemCodeStatus.ACTIVE: return False if self.used_count >= self.max_uses: return False if self.expires_at and self.expires_at < datetime.now(self.expires_at.tzinfo): return False return True class RedeemCodeDetailResponse(RedeemCodeResponse): """兑换码详细信息响应(管理员用)""" batch_id: str | None batch_name: str | None = None remark: str | None created_by: str | None used_by: str | None class RedeemCodeListResponse(PaginatedResponse[RedeemCodeDetailResponse]): """兑换码列表响应""" pass # ============================================================ # 批次 Schema # ============================================================ class BatchCreateRequest(BaseSchema): """创建兑换码批次请求""" name: Annotated[ str, Field( min_length=1, max_length=128, description="批次名称", examples=["2024年1月活动"], ), ] description: Annotated[ str | None, Field( default=None, max_length=500, description="批次描述", ), ] face_value: Annotated[ float, Field( gt=0, description="面值(显示金额)", examples=[10.00, 50.00], ), ] count: Annotated[ int, Field( gt=0, le=10000, description="生成数量", examples=[100], ), ] max_uses: Annotated[ int, Field( default=1, gt=0, le=100, description="每个兑换码最大使用次数", ), ] expires_at: Annotated[ datetime | None, Field( default=None, description="过期时间", ), ] @field_validator("face_value") @classmethod def validate_face_value(cls, v: float) -> float: """验证面值精度""" if round(v, 3) != v: raise ValueError("面值精度不能超过 3 位小数") return v @property def face_value_units(self) -> int: """转换为单位额度""" return display_to_units(self.face_value) class BatchResponse(BaseSchema): """批次信息响应""" id: str name: str description: str | None face_value_units: int total_count: int used_count: int created_by: str | None created_at: datetime @property def face_value(self) -> str: """显示面值(2 位小数)""" return format_display(self.face_value_units) @property def unused_count(self) -> int: """未使用数量""" return self.total_count - self.used_count class BatchDetailResponse(BatchResponse): """批次详细信息响应""" codes: list[RedeemCodeResponse] = [] class BatchListResponse(PaginatedResponse[BatchResponse]): """批次列表响应""" pass # ============================================================ # 导入导出 Schema # ============================================================ class ImportCodeRequest(BaseSchema): """导入兑换码请求""" code: Annotated[ str, Field( min_length=1, max_length=32, description="兑换码", ), ] face_value: Annotated[ float, Field( gt=0, description="面值", ), ] max_uses: Annotated[ int, Field( default=1, gt=0, ), ] expires_at: datetime | None = None remark: str | None = None @field_validator("code") @classmethod def normalize_code(cls, v: str) -> str: """标准化兑换码""" return v.strip().upper() @property def face_value_units(self) -> int: """转换为单位额度""" return display_to_units(self.face_value) class BulkImportRequest(BaseSchema): """批量导入兑换码请求""" codes: Annotated[ list[ImportCodeRequest], Field( min_length=1, max_length=1000, description="兑换码列表", ), ] batch_name: Annotated[ str | None, Field( default=None, max_length=128, description="批次名称(可选)", ), ] class BulkImportResponse(BaseSchema): """批量导入响应""" success_count: int failed_count: int failed_codes: list[str] = [] batch_id: str | None = None class ExportCodeItem(BaseSchema): """导出兑换码条目""" code: str face_value: str status: str max_uses: int used_count: int expires_at: str | None created_at: str used_at: str | None used_by: str | None class ExportResponse(BaseSchema): """导出响应""" total: int codes: list[ExportCodeItem] # ============================================================ # 使用日志 Schema # ============================================================ class UsageLogResponse(BaseSchema): """兑换码使用日志响应""" id: str redeem_code_id: str code_snapshot: str user_id: str username: str | None = None face_value: str ip_address: str | None created_at: datetime class UsageLogListResponse(PaginatedResponse[UsageLogResponse]): """使用日志列表响应""" pass # ============================================================ # 查询参数 Schema # ============================================================ class RedeemCodeQueryParams(BaseSchema): """兑换码查询参数""" status: RedeemCodeStatus | None = None batch_id: str | None = None code: str | None = None created_after: datetime | None = None created_before: datetime | None = None class UsageLogQueryParams(BaseSchema): """使用日志查询参数""" redeem_code_id: str | None = None user_id: str | None = None code: str | None = None created_after: datetime | None = None created_before: datetime | None = None