sqxp/backend/app/crud/user.py

"""用户CRUD"""
from typing import Optional, List, Tuple
from sqlalchemy.orm import Session
from sqlalchemy import and_

from app.models.models import SysUser, SysUserHis
from app.crud.base import CRUDBase
from app.core.security import get_password_hash
from app.core.utils import generate_random_password

# 运营端「重置密码」专用明文：仅服务端使用，禁止由 HTTP 请求传入
DEFAULT_ADMIN_RESET_PLAIN_PASSWORD = "111111"


class UserCRUD(CRUDBase[SysUser]):
    """用户CRUD操作"""

    def get_by_username(self, db: Session, username: str) -> Optional[SysUser]:
        """根据用户名获取用户"""
        return db.query(SysUser).filter(
            and_(SysUser.username == username, SysUser.is_del == 0)
        ).first()

    def get_by_phone(self, db: Session, phone: str) -> Optional[SysUser]:
        """根据手机号获取用户"""
        return db.query(SysUser).filter(
            and_(SysUser.phone == phone, SysUser.is_del == 0)
        ).first()

    def get_by_wx_mini_openid(self, db: Session, openid: str) -> Optional[SysUser]:
        """根据微信小程序 openid 获取用户"""
        if not openid:
            return None
        return db.query(SysUser).filter(
            and_(SysUser.wx_mini_openid == openid, SysUser.is_del == 0)
        ).first()

    def get_multi_by_role(self, db: Session, role: str, skip: int = 0, limit: int = 100) -> List[SysUser]:
        """根据角色获取用户列表"""
        return db.query(SysUser).filter(
            and_(SysUser.role == role, SysUser.is_del == 0)
        ).offset(skip).limit(limit).all()

    def get_multi_by_region(self, db: Session, region_id: str, skip: int = 0, limit: int = 100) -> List[SysUser]:
        """根据区域获取用户列表"""
        return db.query(SysUser).filter(
            and_(SysUser.region_id == region_id, SysUser.is_del == 0)
        ).offset(skip).limit(limit).all()

    def get_multi_with_filters(
        self,
        db: Session,
        role: str = None,
        region_id: str = None,
        keyword: str = None,
        is_active: bool = None,
        skip: int = 0,
        limit: int = 10
    ) -> tuple[List[SysUser], int]:
        """多条件查询用户"""
        query = db.query(SysUser).filter(SysUser.is_del == 0)
        
        if role:
            query = query.filter(SysUser.role == role)
        if region_id:
            query = query.filter(SysUser.region_id == region_id)
        if is_active is not None:
            query = query.filter(SysUser.is_active == is_active)
        if keyword:
            query = query.filter(
                (SysUser.username.like(f"%{keyword}%")) |
                (SysUser.real_name.like(f"%{keyword}%")) |
                (SysUser.phone.like(f"%{keyword}%"))
            )
        
        total = query.count()
        users = query.order_by(SysUser.create_at.desc()).offset(skip).limit(limit).all()
        return users, total

    def create_with_password(self, db: Session, obj_in: dict, creator: dict = None) -> Tuple[SysUser, Optional[str]]:
        """创建用户。

        若请求体包含非空 password（运营端表单），则使用该密码写入 password_hash；
        否则生成随机密码，并在返回值中给出明文 initialPassword（仅此种情况返回，供管理员告知用户）。
        """
        obj_in = dict(obj_in)
        raw_pwd = obj_in.pop("password", None)
        plain: Optional[str] = None
        if isinstance(raw_pwd, str):
            plain = raw_pwd.strip()
        if plain:
            if len(plain) < 6:
                raise ValueError("密码至少6位")
            obj_in["password_hash"] = get_password_hash(plain)
            initial_reveal: Optional[str] = None
        else:
            initial_reveal = generate_random_password()
            obj_in["password_hash"] = get_password_hash(initial_reveal)
        obj_in["is_active"] = obj_in.get("is_active", True)
        user = self.create(db, obj_in, creator)
        return user, initial_reveal

    def reset_password_to_admin_default(self, db: Session, id: str) -> bool:
        """将密码重置为服务端固定默认（DEFAULT_ADMIN_RESET_PLAIN_PASSWORD），不由请求传入。"""
        user = self.get(db, id)
        if not user:
            return False
        plain = DEFAULT_ADMIN_RESET_PLAIN_PASSWORD
        if len(plain) < 6:
            raise ValueError("默认重置密码配置无效：至少6位")
        user.password_hash = get_password_hash(plain)
        db.commit()
        return True

    def update_with_history(self, db: Session, id: str, obj_in: dict, updater: dict = None) -> Optional[SysUser]:
        """更新用户并记录历史"""
        original = self.get(db, id)
        if not original:
            return None

        his_data = {
            "id": original.id,
            "username": original.username,
            "password_hash": original.password_hash,
            "real_name": original.real_name,
            "phone": original.phone,
            "avatar_url": getattr(original, "avatar_url", None),
            "wx_mini_openid": getattr(original, "wx_mini_openid", None),
            "role": original.role,
            "region_id": original.region_id,
            "is_active": original.is_active,
            "create_at": original.create_at,
            "create_by": original.create_by,
            "create_name": original.create_name,
            "update_at": original.update_at,
            "update_by": original.update_by,
        }
        his = SysUserHis(**his_data)
        db.add(his)

        return self.update(db, id, obj_in, updater)

    def soft_delete_with_history(self, db: Session, id: str) -> bool:
        """删除用户并记录历史"""
        original = self.get(db, id)
        if not original:
            return False

        his_data = {
            "id": original.id,
            "username": original.username,
            "password_hash": original.password_hash,
            "real_name": original.real_name,
            "phone": original.phone,
            "avatar_url": getattr(original, "avatar_url", None),
            "wx_mini_openid": getattr(original, "wx_mini_openid", None),
            "role": original.role,
            "region_id": original.region_id,
            "is_active": original.is_active,
            "create_at": original.create_at,
            "create_by": original.create_by,
            "create_name": original.create_name,
            "update_at": original.update_at,
            "update_by": original.update_by,
        }
        his = SysUserHis(**his_data)
        db.add(his)

        return self.delete(db, id)


user_crud = UserCRUD(SysUser)