shanxia2.0/backend/app/crud/user.py

"""用户 CRUD（sys_user + sys_user_role + sys_user_post）"""
from __future__ import annotations

import math
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

import bcrypt
from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.delete_guard import assert_row_deletable_by_creator
from app.core.utils import generate_uuid
from app.crud.base import CRUDBase
from app.models.models import SysUser, SysUserRole, SysUserPost, SysDept


def _hash_password(password: str) -> str:
    return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')


class _UserCRUDCompat(CRUDBase[SysUser]):
    def get_by_user_no(self, db: Session, user_no: str, tenant_no: Optional[str] = None) -> Optional[SysUser]:
        tenant = tenant_no or settings.DEFAULT_TENANT_NO
        un = (user_no or "").strip()
        if not un:
            return None
        return (
            db.query(SysUser)
            .filter(and_(SysUser.user_no == un, SysUser.tenant_no == tenant, SysUser.del_flag == 0))
            .first()
        )

    def get_by_user_no_for_web_login(self, db: Session, user_no: str) -> Optional[SysUser]:
        """Web 登录：优先 DEFAULT_TENANT_NO；未命中时按 user_no 回退（兼容未执行租户迁移的库）。"""
        user = self.get_by_user_no(db, user_no, settings.DEFAULT_TENANT_NO)
        if user:
            return user
        un = (user_no or "").strip()
        if not un:
            return None
        return (
            db.query(SysUser)
            .filter(and_(SysUser.user_no == un, SysUser.del_flag == 0))
            .order_by(SysUser.create_time.desc())
            .first()
        )


user_crud = _UserCRUDCompat(SysUser)


def _row_to_dict(u: SysUser, dept_name: str = "") -> Dict[str, Any]:
    return {
        "id": u.id,
        "deptId": u.dept_id or "",
        "userNo": u.user_no,
        "nickName": u.nick_name,
        "email": u.email or "",
        "phoneNo": u.phone_no or "",
        "sex": u.sex or "0",
        "avatar": u.avatar or "",
        "status": u.status,
        "tenantNo": u.tenant_no,
        "orderNum": u.order_num if u.order_num is not None else 0,
        "remark": u.remark or "",
        "createBy": u.create_by or "",
        "createTime": u.create_time.strftime("%Y-%m-%d %H:%M:%S") if u.create_time else "",
        "updateTime": u.update_time.strftime("%Y-%m-%d %H:%M:%S") if u.update_time else "",
        "deptName": dept_name,
    }


def list_users(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    dept_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
    status: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = db.query(SysUser).filter(SysUser.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(SysUser.tenant_no == tenant_no)
        if dept_id:
            q0 = q0.filter(SysUser.dept_id == dept_id)
        if status:
            q0 = q0.filter(SysUser.status == status)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    SysUser.user_no.like(like),
                    SysUser.nick_name.like(like),
                    SysUser.phone_no.like(like),
                )
            )
        return q0

    total = _filtered().order_by(None).with_entities(func.count(SysUser.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(SysUser.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )

    # batch resolve dept names
    dept_ids = list({r.dept_id for r in rows if r.dept_id})
    dept_map: Dict[str, str] = {}
    if dept_ids:
        dept_rows = db.query(SysDept).filter(SysDept.id.in_(dept_ids)).all()
        dept_map = {d.id: d.dept_name for d in dept_rows}

    return [_row_to_dict(r, dept_name=dept_map.get(r.dept_id or "", "")) for r in rows], int(total)


def get_user(db: Session, user_id: str) -> Optional[SysUser]:
    return (
        db.query(SysUser)
        .filter(and_(SysUser.id == user_id, SysUser.del_flag == 0))
        .first()
    )


def list_post_ids_for_user(db: Session, user_id: str) -> List[str]:
    """用户当前有效职位 ID 列表。"""
    return [
        r.post_id
        for r in db.query(SysUserPost)
        .filter(and_(SysUserPost.user_id == user_id, SysUserPost.del_flag == 0))
        .all()
    ]


def set_user_post_ids_only(
    db: Session,
    user_id: str,
    tenant_no: str,
    post_ids: List[str],
    updater_id: Optional[str],
) -> None:
    """仅替换 sys_user_post，不影响 sys_user_role。"""
    now = datetime.now()
    uniq = [p for p in dict.fromkeys(post_ids or []) if p]
    db.query(SysUserPost).filter(
        and_(SysUserPost.user_id == user_id, SysUserPost.del_flag == 0)
    ).update({"del_flag": 1, "update_time": now}, synchronize_session=False)
    for pid in uniq:
        db.add(
            SysUserPost(
                id=generate_uuid(),
                user_id=user_id,
                post_id=pid,
                tenant_no=tenant_no,
                order_num=0,
                remark=None,
                create_by=updater_id,
                create_dept_id="",
                create_time=now,
                update_by=updater_id,
                update_time=now,
                del_flag=0,
            )
        )


def list_user_ids_bound_to_post(db: Session, post_id: str, tenant_no: str) -> List[str]:
    rows = (
        db.query(SysUserPost.user_id)
        .filter(
            and_(
                SysUserPost.post_id == post_id,
                SysUserPost.tenant_no == tenant_no,
                SysUserPost.del_flag == 0,
            )
        )
        .all()
    )
    return [r[0] for r in rows]


def sync_post_user_bindings(
    db: Session,
    post_id: str,
    tenant_no: str,
    user_ids: List[str],
    updater_id: Optional[str],
) -> None:
    """将本职位绑定的用户集合设为 user_ids（只增删本职位，保留用户其它职位与角色）。"""
    ordered = list(dict.fromkeys([uid for uid in (user_ids or []) if uid]))
    for uid in ordered:
        u = get_user(db, uid)
        if not u or u.tenant_no != tenant_no:
            raise ValueError("存在无效用户或用户不属于当前租户")

    current = set(list_user_ids_bound_to_post(db, post_id, tenant_no))
    target = set(ordered)
    affected = current | target
    for uid in affected:
        u = get_user(db, uid)
        if not u or u.tenant_no != tenant_no:
            continue
        existing = list_post_ids_for_user(db, uid)
        without = [p for p in existing if p != post_id]
        new_posts = without + [post_id] if uid in target else without
        set_user_post_ids_only(db, uid, tenant_no, new_posts, updater_id)
    db.commit()


def get_user_with_relations(db: Session, user_id: str) -> Optional[Dict[str, Any]]:
    u = get_user(db, user_id)
    if not u:
        return None

    dept_name = ""
    if u.dept_id:
        dept = db.query(SysDept).filter(SysDept.id == u.dept_id).first()
        if dept:
            dept_name = dept.dept_name

    d = _row_to_dict(u, dept_name=dept_name)

    role_ids = [
        r.role_id
        for r in db.query(SysUserRole)
        .filter(and_(SysUserRole.user_id == user_id, SysUserRole.del_flag == 0))
        .all()
    ]
    post_ids = list_post_ids_for_user(db, user_id)
    d["roleIds"] = role_ids
    d["postIds"] = post_ids
    return d


def user_no_exists(db: Session, user_no: str, tenant_no: str, exclude_id: Optional[str] = None) -> bool:
    q = db.query(SysUser).filter(
        and_(SysUser.user_no == user_no.strip(), SysUser.tenant_no == tenant_no, SysUser.del_flag == 0)
    )
    if exclude_id:
        q = q.filter(SysUser.id != exclude_id)
    return q.first() is not None


def _replace_associations(
    db: Session,
    user_id: str,
    tenant_no: str,
    creator_id: Optional[str],
    role_ids: Optional[List[str]] = None,
    post_ids: Optional[List[str]] = None,
) -> None:
    now = datetime.now()
    db.query(SysUserRole).filter(
        and_(SysUserRole.user_id == user_id, SysUserRole.del_flag == 0)
    ).update({"del_flag": 1, "update_time": now})
    db.query(SysUserPost).filter(
        and_(SysUserPost.user_id == user_id, SysUserPost.del_flag == 0)
    ).update({"del_flag": 1, "update_time": now})

    if role_ids:
        for rid in role_ids:
            db.add(SysUserRole(
                id=generate_uuid(),
                user_id=user_id,
                role_id=rid,
                tenant_no=tenant_no,
                order_num=0,
                remark=None,
                create_by=creator_id,
                create_dept_id="",
                create_time=now,
                update_by=creator_id,
                update_time=now,
                del_flag=0,
            ))

    if post_ids:
        for pid in post_ids:
            db.add(SysUserPost(
                id=generate_uuid(),
                user_id=user_id,
                post_id=pid,
                tenant_no=tenant_no,
                order_num=0,
                remark=None,
                create_by=creator_id,
                create_dept_id="",
                create_time=now,
                update_by=creator_id,
                update_time=now,
                del_flag=0,
            ))


def create_user(
    db: Session,
    *,
    user_no: str,
    nick_name: str,
    password: str,
    dept_id: Optional[str] = None,
    email: str = "",
    phone_no: str = "",
    sex: str = "0",
    avatar: Optional[str] = None,
    status: str = "1",
    tenant_no: str = "shanxia2.0",
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    role_ids: Optional[List[str]] = None,
    post_ids: Optional[List[str]] = None,
) -> SysUser:
    un = (user_no or "").strip()
    if not un:
        raise ValueError("用户账号不能为空")
    nn = (nick_name or "").strip()
    if not nn:
        raise ValueError("用户名不能为空")
    pw = (password or "").strip()
    if not pw:
        raise ValueError("密码不能为空")

    obj = SysUser(
        id=generate_uuid(),
        dept_id=dept_id or None,
        user_no=un,
        nick_name=nn,
        email=(email or "").strip(),
        phone_no=(phone_no or "").strip(),
        sex=sex or "0",
        avatar=avatar,
        password=_hash_password(pw),
        status=status or "1",
        tenant_no=tenant_no,
        order_num=order_num if order_num is not None else 0,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id="",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)

    _replace_associations(db, obj.id, tenant_no, user_id, role_ids, post_ids)
    db.commit()
    return obj


def update_user(
    db: Session,
    user_id: str,
    *,
    nick_name: Optional[str] = None,
    dept_id: Optional[str] = None,
    email: Optional[str] = None,
    phone_no: Optional[str] = None,
    sex: Optional[str] = None,
    avatar: Optional[str] = None,
    status: Optional[str] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id_updater: Optional[str] = None,
    role_ids: Optional[List[str]] = None,
    post_ids: Optional[List[str]] = None,
) -> SysUser:
    obj = get_user(db, user_id)
    if not obj:
        raise ValueError("用户不存在")

    if nick_name is not None:
        obj.nick_name = (nick_name or "").strip() or obj.nick_name
    if dept_id is not None:
        obj.dept_id = dept_id or None
    if email is not None:
        obj.email = (email or "").strip()
    if phone_no is not None:
        obj.phone_no = (phone_no or "").strip()
    if sex is not None:
        obj.sex = sex
    if avatar is not None:
        obj.avatar = avatar or None
    if status is not None:
        obj.status = status
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = (remark or "").strip() or None

    obj.update_by = user_id_updater
    obj.update_time = datetime.now()
    db.add(obj)

    if role_ids is not None or post_ids is not None:
        _replace_associations(
            db, user_id, obj.tenant_no, user_id_updater,
            role_ids=role_ids, post_ids=post_ids,
        )

    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_user(db: Session, user_id: str, *, user_info: dict = None) -> None:
    obj = get_user(db, user_id)
    if not obj:
        raise ValueError("用户不存在")
    assert_row_deletable_by_creator(obj, user_info)
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


def change_password(db: Session, user_id: str, old_password: str, new_password: str) -> None:
    from app.core.security import verify_password

    obj = get_user(db, user_id)
    if not obj:
        raise ValueError("用户不存在")
    pwd_hash = getattr(obj, "password", None) or ""
    if not verify_password(old_password, pwd_hash):
        raise ValueError("原密码不正确")
    new_pw = (new_password or "").strip()
    if len(new_pw) < 6:
        raise ValueError("新密码长度不能少于6位")
    if old_password == new_pw:
        raise ValueError("新密码不能与原密码相同")
    obj.password = _hash_password(new_pw)
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


def reset_password(db: Session, user_id: str, new_password: str) -> None:
    obj = get_user(db, user_id)
    if not obj:
        raise ValueError("用户不存在")
    obj.password = _hash_password(new_password)
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


def total_pages(total: int, page_size: int) -> int:
    if total <= 0:
        return 0
    return int(math.ceil(total / float(page_size)))