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 from app.core.imageprocessor import process_image, ImageFormat from sqlalchemy.orm import relationship, backref class UserRole(str, enum.Enum): USER = "user" DELIVERYMAN = "deliveryman" MERCHANT = "merchant" ADMIN = "admin" PARTNER = "partner" 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) # 微信用户信息 openid = Column(String(64), unique=True, nullable=True) unionid = Column(String(64), unique=True, nullable=True) mp_openid = Column(String(64), unique=True, nullable=True) # 企业微信用户信息 wecom_userid = Column(String(64), unique=True, nullable=True) wecom_pending_id = Column(String(64), unique=True, nullable=True) nickname = 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(Integer, default=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()) community_id = Column(Integer, ForeignKey("communities.id"), nullable=True) # 归属小区 is_auth = Column(Boolean, nullable=False, default=False) is_delivering = Column(Boolean, nullable=False, default=False) # 是否在配送中 qr_code_url = Column(String(200), nullable=True) # 二维码URL地址 # 用户所属小区关系 community = relationship("CommunityDB", foreign_keys=[community_id], backref=backref("residents", lazy="dynamic")) @property def optimized_avatar(self): return process_image(self.avatar).thumbnail(width=800, height=800).format(ImageFormat.WEBP).build() # Pydantic 模型 class UserLogin(BaseModel): phone: str = Field(..., pattern="^1[3-9]\d{9}$") verify_code: str = Field(..., min_length=4, max_length=6) referral_code: Optional[str] = Field(None, min_length=6, max_length=6) class UserInfo(BaseModel): userid: int openid: Optional[str] = None unionid: Optional[str] = None mp_openid: Optional[str] = None wecom_userid: Optional[str] = None wecom_pending_id: Optional[str] = None nickname: str phone: str user_code: str referral_code: Optional[str] = None avatar: Optional[str] = None optimized_avatar: Optional[str] = None gender: Gender = Gender.UNKNOWN points: int = 0 roles: List[UserRole] create_time: datetime coupon_count: Optional[int] = 0 community_id: Optional[int] = None community_name: Optional[str] = None is_auth: bool = False is_delivering: bool = False qr_code_url: Optional[str] = None 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): nickname: 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}$", description="手机号") password: str = Field(..., min_length=6, max_length=20, description="密码") role: UserRole = Field(default=UserRole.ADMIN, description="角色") class ChangePasswordRequest(BaseModel): phone: str = Field(..., pattern="^1[3-9]\d{9}$") verify_code: str = Field(..., min_length=4, max_length=6) new_password: str = Field(..., min_length=6, max_length=20) class UserUpdateRoles(BaseModel): user_id: int roles: List[UserRole] class DeliverymanSetDelivering(BaseModel): is_delivering: bool 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): nickname: str phone: str # 会在API中处理脱敏 create_time: datetime class Config: from_attributes = True