from datetime import datetime from typing import Any, Literal from uuid import UUID from pydantic import BaseModel, EmailStr, Field, field_validator, model_validator class RegisterRequest(BaseModel): username: str = Field(min_length=3, max_length=32, pattern=r"^[A-Za-z0-9_\-]+$") email: EmailStr | None = None phone: str | None = Field(default=None, min_length=11, max_length=20) verification_channel: Literal["email", "phone"] captcha_id: str = Field(min_length=10, max_length=100) captcha_code: str = Field(min_length=4, max_length=8) password: str = Field(min_length=8, max_length=128) @field_validator("phone") @classmethod def normalize_phone(cls, value: str | None) -> str | None: if value is None: return None normalized = value.strip().replace(" ", "").replace("-", "") if normalized.startswith("+86"): normalized = normalized[3:] if not normalized.isdigit() or len(normalized) != 11 or not normalized.startswith("1"): raise ValueError("请输入有效的中国大陆手机号") return normalized @model_validator(mode="after") def validate_channel(self) -> "RegisterRequest": if not self.email and not self.phone: raise ValueError("邮箱或手机号至少填写一项") if self.verification_channel == "email" and not self.email: raise ValueError("邮箱验证需要填写邮箱") if self.verification_channel == "phone" and not self.phone: raise ValueError("短信验证需要填写手机号") return self class LoginRequest(BaseModel): identifier: str = Field(min_length=3, max_length=120) captcha_id: str = Field(min_length=10, max_length=100) captcha_code: str = Field(min_length=4, max_length=8) password: str = Field(min_length=1, max_length=128) class PasswordChangeRequest(BaseModel): current_password: str = Field(min_length=1, max_length=128) new_password: str = Field(min_length=8, max_length=128) class AdminUserCreateRequest(BaseModel): username: str = Field(min_length=3, max_length=32, pattern=r"^[A-Za-z0-9_\-]+$") email: EmailStr | None = None phone: str | None = Field(default=None, min_length=11, max_length=20) password: str = Field(min_length=8, max_length=128) email_verified: bool = False phone_verified: bool = False reason: str = Field(min_length=1, max_length=500) @field_validator("phone") @classmethod def normalize_phone(cls, value: str | None) -> str | None: if value is None: return None normalized = value.strip().replace(" ", "").replace("-", "") if normalized.startswith("+86"): normalized = normalized[3:] if not normalized.isdigit() or len(normalized) != 11 or not normalized.startswith("1"): raise ValueError("请输入有效的中国大陆手机号") return normalized @model_validator(mode="after") def validate_contact(self) -> "AdminUserCreateRequest": if not self.email and not self.phone: raise ValueError("邮箱或手机号至少填写一项") return self class AdminUserUpdateRequest(BaseModel): username: str = Field(min_length=3, max_length=32, pattern=r"^[A-Za-z0-9_\-]+$") email: EmailStr | None = None phone: str | None = Field(default=None, min_length=11, max_length=20) email_verified: bool = False phone_verified: bool = False reason: str = Field(min_length=1, max_length=500) @field_validator("phone") @classmethod def normalize_phone(cls, value: str | None) -> str | None: if value is None: return None normalized = value.strip().replace(" ", "").replace("-", "") if normalized.startswith("+86"): normalized = normalized[3:] if not normalized.isdigit() or len(normalized) != 11 or not normalized.startswith("1"): raise ValueError("请输入有效的中国大陆手机号") return normalized @model_validator(mode="after") def validate_contact(self) -> "AdminUserUpdateRequest": if not self.email and not self.phone: raise ValueError("邮箱或手机号至少填写一项") return self class UserProfileUpdateRequest(BaseModel): username: str = Field(min_length=3, max_length=32, pattern=r"^[A-Za-z0-9_\-]+$") email: EmailStr | None = None phone: str | None = Field(default=None, min_length=11, max_length=20) @field_validator("phone") @classmethod def normalize_phone(cls, value: str | None) -> str | None: if value is None: return None normalized = value.strip().replace(" ", "").replace("-", "") if normalized.startswith("+86"): normalized = normalized[3:] if not normalized.isdigit() or len(normalized) != 11 or not normalized.startswith("1"): raise ValueError("请输入有效的中国大陆手机号") return normalized @model_validator(mode="after") def validate_contact(self) -> "UserProfileUpdateRequest": if not self.email and not self.phone: raise ValueError("邮箱或手机号至少填写一项") return self class UserPublic(BaseModel): id: UUID username: str email: str | None phone: str | None role: Literal["user", "admin"] plan: Literal["free", "vip"] status: Literal["active", "disabled"] email_verified: bool phone_verified: bool created_at: datetime last_login_at: datetime | None class AuthResponse(BaseModel): user: UserPublic class RegisterResponse(AuthResponse): verification_required: bool challenge_id: UUID class MembershipRequest(BaseModel): plan: Literal["free", "vip"] starts_at: datetime | None = None expires_at: datetime | None = None reason: str = Field(min_length=1, max_length=500) class StatusRequest(BaseModel): status: Literal["active", "disabled"] reason: str = Field(min_length=1, max_length=500) class QuotaAdjustmentRequest(BaseModel): amount: int reason: str = Field(min_length=1, max_length=500) idempotency_key: str = Field(min_length=8, max_length=120) class MembershipRevokeRequest(BaseModel): reason: str = Field(min_length=1, max_length=500) class AdminTaskFilter(BaseModel): user_id: UUID | None = None status: Literal["queued", "running", "succeeded", "failed"] | None = None created_after: datetime | None = None created_before: datetime | None = None page: int = Field(default=1, ge=1) limit: int = Field(default=20, ge=1, le=100) class AdminUserFilter(BaseModel): email: str | None = None phone: str | None = None status: Literal["active", "disabled"] | None = None page: int = Field(default=1, ge=1) limit: int = Field(default=20, ge=1, le=100) class VerificationConfirmRequest(BaseModel): challenge_id: UUID code: str = Field(min_length=4, max_length=12) class VerificationSendRequest(BaseModel): channel: Literal["email", "phone"] purpose: Literal["registration", "contact_binding", "contact_change", "password_reset"] class VerificationResendRequest(BaseModel): challenge_id: UUID class TtsTaskRequest(BaseModel): text: str = Field(min_length=1, max_length=10000) voice_id: str = Field(min_length=1, max_length=120) parameters: dict[str, Any] = Field(default_factory=dict) class TtsVoicePublic(BaseModel): id: UUID provider_voice_id: str name: str language: str | None description: str | None supported_parameters: dict[str, Any] class TtsTaskPublic(BaseModel): id: UUID status: str text_length: int voice_id: str parameters: dict[str, Any] error_code: str | None = None audio_available: bool = False audio_expires_at: datetime | None = None created_at: datetime started_at: datetime | None = None finished_at: datetime | None = None def normalize_email(value: str) -> str: return value.strip().lower()