zkxw/backend/app/crud/qc_material_amount_auth.py

"""质检物料金额查看授权 CRUD（sys_user_role 绑定专用角色）。"""
from __future__ import annotations

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

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

from app.core.qc_material_amount_auth_constants import (
    QC_MATERIAL_AMOUNT_VIEW_ROLE_ID,
    QC_MATERIAL_AMOUNT_VIEW_ROLE_NAME,
    QC_MATERIAL_AMOUNT_AUTH_DEPT_NAME,
)
from app.core.data_scope import _get_child_dept_ids
from app.core.utils import generate_uuid
from app.models.models import SysDept, SysRole, SysUser, SysUserRole


def get_amount_view_role(db: Session, tenant_no: str) -> Optional[SysRole]:
    row = (
        db.query(SysRole)
        .filter(
            SysRole.id == QC_MATERIAL_AMOUNT_VIEW_ROLE_ID,
            SysRole.tenant_no == tenant_no,
            SysRole.del_flag == 0,
            SysRole.status == "1",
        )
        .first()
    )
    if row:
        return row
    return (
        db.query(SysRole)
        .filter(
            SysRole.role_name == QC_MATERIAL_AMOUNT_VIEW_ROLE_NAME,
            SysRole.tenant_no == tenant_no,
            SysRole.del_flag == 0,
            SysRole.status == "1",
        )
        .first()
    )


def _get_auth_dept_id(db: Session, tenant_no: str) -> Optional[str]:
    row = (
        db.query(SysDept.id)
        .filter(
            SysDept.dept_name == QC_MATERIAL_AMOUNT_AUTH_DEPT_NAME,
            SysDept.tenant_no == tenant_no,
            SysDept.del_flag == 0,
            SysDept.status == "1",
        )
        .first()
    )
    return str(row[0]).strip() if row and row[0] else None


def _assert_user_in_auth_dept(db: Session, user: SysUser, tenant_no: str) -> None:
    dept_id = _get_auth_dept_id(db, tenant_no)
    if not dept_id:
        raise ValueError(f"未找到部门「{QC_MATERIAL_AMOUNT_AUTH_DEPT_NAME}」")
    user_dept_id = (user.dept_id or "").strip()
    if not user_dept_id:
        raise ValueError(f"仅可授权「{QC_MATERIAL_AMOUNT_AUTH_DEPT_NAME}」下的用户")
    allowed = set(_get_child_dept_ids(db, dept_id, tenant_no))
    if user_dept_id not in allowed:
        raise ValueError(f"仅可授权「{QC_MATERIAL_AMOUNT_AUTH_DEPT_NAME}」下的用户")


def _user_has_role_binding(db: Session, user_id: str, role_id: str, tenant_no: str) -> bool:
    return (
        db.query(SysUserRole)
        .filter(
            SysUserRole.user_id == user_id,
            SysUserRole.role_id == role_id,
            SysUserRole.tenant_no == tenant_no,
            SysUserRole.del_flag == 0,
        )
        .first()
        is not None
    )


def _row_to_dict(user: SysUser, dept_name: str, grant_time: Optional[datetime]) -> Dict[str, Any]:
    return {
        "id": user.id,
        "userId": user.id,
        "userNo": user.user_no,
        "nickName": user.nick_name,
        "phoneNo": user.phone_no or "",
        "deptName": dept_name,
        "status": user.status,
        "grantTime": grant_time.strftime("%Y-%m-%d %H:%M:%S") if grant_time else "",
    }


def list_authorized_users(
    db: Session,
    *,
    tenant_no: str,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    role = get_amount_view_role(db, tenant_no)
    if not role:
        return [], 0

    q = (
        db.query(SysUser, SysUserRole.create_time)
        .join(
            SysUserRole,
            and_(
                SysUserRole.user_id == SysUser.id,
                SysUserRole.role_id == role.id,
                SysUserRole.tenant_no == tenant_no,
                SysUserRole.del_flag == 0,
            ),
        )
        .filter(SysUser.del_flag == 0, SysUser.tenant_no == tenant_no)
    )
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(
            or_(
                SysUser.user_no.like(like),
                SysUser.nick_name.like(like),
                SysUser.phone_no.like(like),
            )
        )
    total = q.count()
    rows = (
        q.order_by(SysUserRole.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    dept_ids = list({u.dept_id for u, _ in rows if u.dept_id})
    dept_map: Dict[str, str] = {}
    if dept_ids:
        for d in db.query(SysDept).filter(SysDept.id.in_(dept_ids), SysDept.del_flag == 0).all():
            dept_map[d.id] = d.dept_name or ""
    items = [
        _row_to_dict(u, dept_map.get(u.dept_id or "", ""), grant_time)
        for u, grant_time in rows
    ]
    return items, total


def grant_amount_view_role(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    operator_id: Optional[str],
) -> bool:
    """绑定专用角色；已绑定则幂等返回 False（表示未新增）。"""
    role = get_amount_view_role(db, tenant_no)
    if not role:
        raise ValueError("未找到「质检物料金额查看」角色，请先执行 migrate-qc-material-amount-auth.sql")

    user = (
        db.query(SysUser)
        .filter(SysUser.id == user_id, SysUser.tenant_no == tenant_no, SysUser.del_flag == 0)
        .first()
    )
    if not user:
        raise ValueError("用户不存在")
    if user.status != "1":
        raise ValueError("用户已停用，无法授权")

    _assert_user_in_auth_dept(db, user, tenant_no)

    if _user_has_role_binding(db, user_id, role.id, tenant_no):
        return False

    now = datetime.now()
    db.add(
        SysUserRole(
            id=generate_uuid(),
            user_id=user_id,
            role_id=role.id,
            tenant_no=tenant_no,
            order_num=0,
            remark=None,
            create_by=operator_id,
            create_dept_id="",
            create_time=now,
            update_by=operator_id,
            update_time=now,
            del_flag=0,
        )
    )
    db.commit()
    return True


def revoke_amount_view_role(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    operator_id: Optional[str],
) -> bool:
    """解除专用角色绑定；未绑定则返回 False。"""
    role = get_amount_view_role(db, tenant_no)
    if not role:
        raise ValueError("未找到「质检物料金额查看」角色")

    now = datetime.now()
    updated = (
        db.query(SysUserRole)
        .filter(
            SysUserRole.user_id == user_id,
            SysUserRole.role_id == role.id,
            SysUserRole.tenant_no == tenant_no,
            SysUserRole.del_flag == 0,
        )
        .update({"del_flag": 1, "update_by": operator_id, "update_time": now})
    )
    if updated:
        db.commit()
        return True
    return False