opencloud-backend/app/schemas.py

import uuid
from datetime import datetime
from typing import Generic, Literal, TypeVar

from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator, model_validator


Role = Literal["user", "admin"]
CloudStatus = Literal["pending", "approved", "rejected"]
Rarity = Literal["common", "uncommon", "rare"]


class MessageOut(BaseModel):
    message: str


class RegisterIn(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8, max_length=128)
    username: str = Field(min_length=2, max_length=32)

    @field_validator("username")
    @classmethod
    def strip_username(cls, value: str) -> str:
        value = value.strip()
        if len(value) < 2:
            raise ValueError("昵称至少需要 2 个字符")
        return value


class LoginIn(BaseModel):
    email: EmailStr
    password: str = Field(min_length=1, max_length=128)


class TokenIn(BaseModel):
    token: str = Field(min_length=20, max_length=512)


class ForgotPasswordIn(BaseModel):
    email: EmailStr


class ResetPasswordIn(TokenIn):
    password: str = Field(min_length=8, max_length=128)


class ChangePasswordIn(BaseModel):
    password: str = Field(min_length=8, max_length=128)


class ProfileOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: uuid.UUID
    username: str
    avatar_url: str | None = None
    role: Role
    is_disabled: bool
    created_at: datetime


class PublicProfileOut(BaseModel):
    id: uuid.UUID
    username: str
    avatar_url: str | None = None
    created_at: datetime


class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: uuid.UUID
    email: EmailStr
    email_verified: bool
    created_at: datetime


class AuthOut(BaseModel):
    access_token: str
    token_type: Literal["bearer"] = "bearer"
    expires_in: int
    user: UserOut
    profile: ProfileOut


class MeOut(BaseModel):
    user: UserOut
    profile: ProfileOut


class ProfileUpdateIn(BaseModel):
    username: str = Field(min_length=2, max_length=32)

    @field_validator("username")
    @classmethod
    def strip_username(cls, value: str) -> str:
        return value.strip()


class CloudTypeSummary(BaseModel):
    id: int
    name: str
    name_en: str
    rarity: Rarity


class CloudTypeOut(CloudTypeSummary):
    genus: str
    description: str | None
    icon_url: str | None
    created_at: datetime


class CloudOwnerOut(BaseModel):
    id: uuid.UUID
    username: str


class CloudOut(BaseModel):
    id: uuid.UUID
    user_id: uuid.UUID
    cloud_type_id: int | None
    custom_cloud_type: str | None
    image_url: str
    thumbnail_url: str
    latitude: float | None
    longitude: float | None
    location_name: str | None
    description: str | None
    captured_at: datetime | None
    status: CloudStatus
    is_hidden: bool
    created_at: datetime
    updated_at: datetime
    cloud_type: CloudTypeSummary | None
    owner: CloudOwnerOut | None
    cloud_type_name: str
    cloud_type_rarity: Rarity
    username: str


class CloudUpdateIn(BaseModel):
    cloud_type_id: int | None = None
    custom_cloud_type: str | None = Field(default=None, max_length=80)
    latitude: float | None = Field(default=None, ge=-90, le=90)
    longitude: float | None = Field(default=None, ge=-180, le=180)
    location_name: str | None = Field(default=None, max_length=120)
    description: str | None = Field(default=None, max_length=2000)
    captured_at: datetime | None = None
    is_hidden: bool | None = None

    @model_validator(mode="after")
    def validate_coordinates(self) -> "CloudUpdateIn":
        provided = self.model_fields_set
        if ("latitude" in provided) != ("longitude" in provided):
            raise ValueError("经纬度必须同时提供")
        if "latitude" in provided and ((self.latitude is None) != (self.longitude is None)):
            raise ValueError("经纬度必须同时为空或同时有值")
        return self


class BadgeOut(BaseModel):
    cloud_type_id: int
    cloud_name: str
    cloud_name_en: str
    rarity: Rarity
    unlocked_at: datetime


class CloudCreateOut(BaseModel):
    cloud: CloudOut
    unlocked_badge: BadgeOut | None


T = TypeVar("T")


class PageOut(BaseModel, Generic[T]):
    items: list[T]
    page: int
    page_size: int
    total: int
    total_pages: int


class BatchIdsIn(BaseModel):
    ids: list[uuid.UUID] = Field(min_length=1, max_length=100)

    @field_validator("ids")
    @classmethod
    def unique_ids(cls, value: list[uuid.UUID]) -> list[uuid.UUID]:
        return list(dict.fromkeys(value))


class DeleteResultOut(BaseModel):
    deleted: int


class CollectionCloudOut(BaseModel):
    id: uuid.UUID
    image_url: str
    thumbnail_url: str
    captured_at: datetime | None
    created_at: datetime
    location_name: str | None


class CollectionOut(BaseModel):
    id: uuid.UUID
    user_id: uuid.UUID
    cloud_type_id: int
    first_cloud_id: uuid.UUID | None
    unlocked_at: datetime
    cloud_type: CloudTypeSummary
    first_cloud: CollectionCloudOut | None


class AdminStatsOut(BaseModel):
    users: int
    images: int
    today_uploads: int
    pending: int
    approved: int
    rejected: int
    hidden: int


class AdminUserOut(BaseModel):
    user: UserOut
    profile: ProfileOut


class AdminUserUpdateIn(BaseModel):
    role: Role | None = None
    is_disabled: bool | None = None

    @model_validator(mode="after")
    def require_change(self) -> "AdminUserUpdateIn":
        if not self.model_fields_set:
            raise ValueError("至少需要提供一个修改字段")
        return self


class AdminCloudStatusIn(BatchIdsIn):
    status: CloudStatus


class AdminCloudVisibilityIn(BatchIdsIn):
    is_hidden: bool


class BatchUpdateOut(BaseModel):
    updated: int


class ProfileStatsOut(BaseModel):
    user_id: uuid.UUID
    username: str
    email: str
    created_at: datetime
    cloud_count: int
    public_cloud_count: int
    collection_count: int