from sqlalchemy import Column, String, DateTime, Integer, Boolean, Enum, ForeignKey, DECIMAL from sqlalchemy.sql import func from sqlalchemy.dialects.mysql import JSON from pydantic import BaseModel, Field from .database import Base, get_db from typing import Optional, List, Union from datetime import datetime import enum import random import string class UserRole(str, enum.Enum): USER = "user" DELIVERYMAN = "deliveryman" ADMIN = "admin" class Gender(str, enum.Enum): MALE = "MALE" FEMALE = "FEMALE" UNKNOWN = "UNKNOWN" # 数据库模型 class UserDB(Base): __tablename__ = "users" userid = Column(Integer, primary_key=True,autoincrement=True, index=True) username = Column(String(50)) phone = Column(String(11), unique=True, index=True) user_code = Column(String(6), unique=True, nullable=False) referral_code = Column(String(6), ForeignKey("users.user_code"), nullable=True) password = Column(String(128), nullable=True) # 加密后的密码 avatar = Column(String(200), nullable=True) # 头像URL地址 gender = Column(Enum(Gender), nullable=False, default=Gender.UNKNOWN) points = Column(DECIMAL(10,1), nullable=False, default=0.0) roles = Column(JSON, default=lambda: [UserRole.USER]) # 存储角色列表 create_time = Column(DateTime(timezone=True), server_default=func.now()) update_time = Column(DateTime(timezone=True), onupdate=func.now()) # Pydantic 模型 class UserLogin(BaseModel): phone: str = Field(..., pattern="^1[3-9]\d{9}$") verify_code: str = Field(..., min_length=6, max_length=6) referral_code: Optional[str] = Field(None, min_length=6, max_length=6) class UserInfo(BaseModel): userid: int username: str phone: str user_code: str referral_code: Optional[str] = None avatar: Optional[str] = None gender: Gender = Gender.UNKNOWN points: int = 0 roles: List[UserRole] create_time: datetime class Config: from_attributes = True class PhoneLoginRequest(BaseModel): phone: str = Field(..., pattern="^1[3-9]\d{9}$") referral_code: Optional[str] = Field(None, min_length=6, max_length=6) class ResetPasswordRequest(BaseModel): user_id: int new_password: str = Field(..., min_length=6, max_length=20) class VerifyCodeRequest(BaseModel): phone: str = Field(..., pattern="^1[3-9]\d{9}$") class UserUpdate(BaseModel): username: Optional[str] = Field(None, min_length=2, max_length=50) avatar: Optional[str] = Field(None, max_length=200) gender: Optional[Gender] = None class Config: extra = "forbid" # 禁止额外字段 class UserPasswordLogin(BaseModel): phone: str = Field(..., pattern="^1[3-9]\d{9}$") password: str = Field(..., min_length=6, max_length=20) def generate_user_code(db=None) -> str: """生成6位大写字母+数字的用户编码""" chars = string.ascii_uppercase + string.digits while True: code = ''.join(random.choices(chars, k=6)) # 检查是否已存在 if db: exists = db.query(UserDB).filter(UserDB.user_code == code).first() else: db = next(get_db()) exists = db.query(UserDB).filter(UserDB.user_code == code).first() if not exists: return code class ReferralUserInfo(BaseModel): username: str phone: str # 会在API中处理脱敏 create_time: datetime class Config: from_attributes = True