zkxw/backend/app/crud/user.py

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

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

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

from app.core.config import settings
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()
        )


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 "",
        "supervisorUserId": getattr(u, "supervisor_user_id", None) or "",
        "userNo": u.user_no,
        "nickName": u.nick_name,
        "email": u.email or "",
        "phoneNo": u.phone_no or "",
        "hireDate": u.hire_date.strftime("%Y-%m-%d") if getattr(u, "hire_date", None) else "",
        "probationMonths": (
            u.probation_months if getattr(u, "probation_months", None) is not None else None
        ),
        "lastWorkDay": u.last_work_day.strftime("%Y-%m-%d")
        if getattr(u, "last_work_day", None)
        else "",
        "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 "",
        "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)
        .join(
            SysUser,
            and_(
                SysUser.id == SysUserPost.user_id,
                SysUser.del_flag == 0,
            ),
        )
        .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 list_user_ids_bound_to_posts(
    db: Session, post_ids: List[str], tenant_no: str
) -> List[str]:
    """多职位展开启用用户 ID（仅启用职位；用户启用过滤留给调用方或本函数后处理）。

    返回保序去重的 user_id 列表。仅包含 status=1 的职位下的关联。
    """
    from app.models.models import SysPost

    ordered = list(dict.fromkeys([pid for pid in (post_ids or []) if pid]))
    if not ordered:
        return []
    rows = (
        db.query(SysUserPost.user_id)
        .join(
            SysPost,
            and_(
                SysPost.id == SysUserPost.post_id,
                SysPost.del_flag == 0,
                SysPost.status == "1",
                SysPost.tenant_no == tenant_no,
            ),
        )
        .filter(
            and_(
                SysUserPost.post_id.in_(ordered),
                SysUserPost.tenant_no == tenant_no,
                SysUserPost.del_flag == 0,
            )
        )
        .all()
    )
    out: List[str] = []
    seen = set()
    for r in rows:
        uid = r[0]
        if uid and uid not in seen:
            seen.add(uid)
            out.append(uid)
    return out


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]]:
    from app.services.web_auth_roles import (
        get_effective_role_ids,
        get_post_inherited_role_ids,
        get_role_names_by_ids,
    )

    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)
    effective_role_ids = get_effective_role_ids(db, user_id)
    post_inherited_role_ids = get_post_inherited_role_ids(db, user_id)
    d["roleIds"] = role_ids
    d["postIds"] = post_ids
    d["postInheritedRoleIds"] = post_inherited_role_ids
    d["effectiveRoleIds"] = effective_role_ids
    d["effectiveRoleNames"] = get_role_names_by_ids(db, effective_role_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:
    """全量替换关联。仅当对应参数不为 None 时才替换该侧（避免只改职位时清空直接角色）。"""
    now = datetime.now()
    if role_ids is not None:
        db.query(SysUserRole).filter(
            and_(SysUserRole.user_id == user_id, SysUserRole.del_flag == 0)
        ).update({"del_flag": 1, "update_time": now})
        for rid in role_ids:
            if not rid:
                continue
            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 is not None:
        db.query(SysUserPost).filter(
            and_(SysUserPost.user_id == user_id, SysUserPost.del_flag == 0)
        ).update({"del_flag": 1, "update_time": now})
        for pid in post_ids:
            if not pid:
                continue
            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 _parse_hire_date(value: Optional[Union[str, date]]) -> Optional[date]:
    if value is None or value == "":
        return None
    if isinstance(value, date) and not isinstance(value, datetime):
        return value
    s = str(value).strip()
    if not s:
        return None
    return datetime.strptime(s[:10], "%Y-%m-%d").date()


def create_user(
    db: Session,
    *,
    user_no: str,
    nick_name: str,
    password: str,
    dept_id: Optional[str] = None,
    email: str = "",
    phone_no: str = "",
    hire_date: Optional[Union[str, date]] = None,
    probation_months: Optional[int] = None,
    last_work_day: Optional[Union[str, date]] = None,
    sex: str = "0",
    avatar: Optional[str] = None,
    status: str = "1",
    tenant_no: str = "zkxw",
    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,
    supervisor_user_id: Optional[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,
        supervisor_user_id=(supervisor_user_id or "").strip() or None,
        user_no=un,
        nick_name=nn,
        email=(email or "").strip(),
        phone_no=(phone_no or "").strip(),
        hire_date=_parse_hire_date(hire_date),
        probation_months=probation_months,
        last_work_day=_parse_hire_date(last_work_day),
        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,
    hire_date: Any = ...,
    probation_months: Any = ...,
    last_work_day: Any = ...,
    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,
    supervisor_user_id: Optional[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 supervisor_user_id is not None:
        obj.supervisor_user_id = (supervisor_user_id or "").strip() 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 hire_date is not ...:
        obj.hire_date = _parse_hire_date(hire_date)
    if probation_months is not ...:
        if probation_months is None or probation_months == "":
            obj.probation_months = None
        else:
            obj.probation_months = int(probation_months)
    if last_work_day is not ...:
        obj.last_work_day = _parse_hire_date(last_work_day)
    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) -> None:
    obj = get_user(db, user_id)
    if not obj:
        raise ValueError("用户不存在")
    obj.del_flag = 1
    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 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位")
    obj.password = _hash_password(new_pw)
    obj.update_by = user_id
    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)))