PayrollManagementSystem/backend/app/services/authorization_management.py

"""管理员维护授权；审计、版本检查和权限写入在同一业务事务中。"""
from sqlalchemy import delete, select

from app.core.exceptions import BadRequestError, ConflictError, ForbiddenError, NotFoundError
from app.models import Permission, Role, RolePermission, User, UserRole
from app.repositories import user_repository
from app.services import audit_service
from app.services import authorization_service as auth
from app.services.audit_service import AuditRecord
from app.services.permission_catalog import CATALOG


def require_admin(session, actor):
    if actor is None or actor.status != 1 or "admin" not in auth.role_codes(session, actor.id, for_update=True):
        raise ForbiddenError("仅系统管理员可以配置权限")


def lock_policy(session, actor, expected=None):
    policy = auth.setting(session, lock=True)
    require_admin(session, actor)
    if expected is not None and policy.revision != expected:
        raise ConflictError()
    return policy


def bump(policy, actor):
    policy.revision += 1
    policy.updated_by = actor.id


def _user(session, user_id):
    user = session.get(User, user_id)
    if user is None:
        raise NotFoundError("用户不存在")
    return user


def _roles(session, codes, *, existing=()):
    if len(set(codes)) != len(codes):
        raise BadRequestError("角色不能重复")
    rows = list(session.scalars(select(Role).where(Role.role_code.in_(codes))))
    if len(rows) != len(codes) or any(r.status != 1 and r.role_code not in existing for r in rows):
        raise BadRequestError("角色不存在或已停用")
    return rows


def _overrides(payload):
    if payload.overrides:
        raise BadRequestError("已改为角色授权，请在角色管理中配置权限")
    return {}


def user_authorization(session, user_id):
    user = _user(session, user_id)
    result = auth.calculate(session, user)
    return {**result, "id": user.id, "username": user.username,
            "role_codes": [link.role.role_code for link in user.user_roles],
            "overrides": [], "revision": auth.setting(session).revision,
            "is_payroll_maker": auth.is_maker(session, user)}


def preview(session, user_id, payload):
    user = _user(session, user_id)
    _roles(session, payload.role_codes, existing=[link.role.role_code for link in user.user_roles])
    overrides = _overrides(payload)
    if "admin" in payload.role_codes and overrides:
        raise BadRequestError("管理员权限为固定基线，不能设置个人覆盖")
    proposed = auth.calculate(session, user, roles=payload.role_codes, overrides=overrides)
    before = set(auth.calculate(session, user)["permissions"])
    after = set(proposed["permissions"])
    return {**proposed, "added": sorted(after - before), "removed": sorted(before - after)}


def guard_user_change(session, user, actor, *, roles=None, status=None, deleting=False):
    """调用方已持全局授权锁；防止两个管理员并发把系统管理资格撤空。"""
    session.refresh(user, with_for_update=True)
    current = auth.role_codes(session, user.id, for_update=True)
    removes_admin = "admin" in current and (
        deleting or status == 0 or (roles is not None and "admin" not in roles)
    )
    if removes_admin:
        if actor.id == user.id:
            raise ForbiddenError("不能撤销自己的管理员资格或停用自己")
        active = list(session.scalars(select(User.id).join(UserRole).join(Role).where(
            User.status == 1, Role.role_code == "admin", Role.status == 1
        ).with_for_update()))
        if len(active) <= 1:
            raise ForbiddenError("不能移除最后一个启用的管理员")


def save_user(session, user_id, payload, actor, meta=None):
    audit_data = payload.model_dump(mode="json")
    with audit_service.audit(session, operator=actor, meta=meta, record=AuditRecord(
        "权限管理", "用户授权", f"修改用户 #{user_id} 的权限", audit_data
    )):
        policy = lock_policy(session, actor, payload.expected_revision)
        user = _user(session, user_id)
        before = user_authorization(session, user_id)
        changes = preview(session, user_id, payload)
        guard_user_change(session, user, actor, roles=payload.role_codes)
        roles = _roles(session, payload.role_codes, existing=[link.role.role_code for link in user.user_roles])
        user_repository.replace_roles(session, user, roles)
        bump(policy, actor)
        session.flush()
        result = user_authorization(session, user_id)
        result["changes"] = {"added": changes["added"], "removed": changes["removed"],
                             "before_roles": before["role_codes"], "after_roles": result["role_codes"]}
        audit_data["changes"] = result["changes"]
        session.commit()
        return result


def role_permissions(session, role_id):
    role = session.get(Role, role_id)
    if role is None:
        raise NotFoundError("角色不存在")
    codes = sorted(CATALOG) if role.role_code == "admin" else list(session.scalars(
        select(Permission.perm_code).join(RolePermission, RolePermission.permission_id == Permission.id)
        .where(RolePermission.role_id == role.id, Permission.perm_code.in_(CATALOG))
    ))
    return {"role_id": role.id, "role_code": role.role_code, "role_name": role.role_name,
            "permission_codes": codes, "revision": auth.setting(session).revision,
            "fixed": role.role_code == "admin"}


def validate_role_permissions(permission_codes):
    """新建与修改角色共用同一权限目录、管理员边界及查看依赖校验。"""
    codes = set(permission_codes)
    if codes - CATALOG.keys() or len(codes) != len(permission_codes):
        raise BadRequestError("权限未知或重复，请刷新目录")
    if any(CATALOG[c]["admin_only"] for c in codes):
        raise BadRequestError("管理员专用权限不能分配给普通角色")
    if any(CATALOG[c]["view_code"] not in codes for c in codes):
        raise BadRequestError("操作权限必须同时授予相应查看权限")
    return codes


def save_role(session, role_id, payload, actor, meta=None):
    audit_data = payload.model_dump(mode="json")
    with audit_service.audit(session, operator=actor, meta=meta, record=AuditRecord(
        "权限管理", "角色授权", f"修改角色 #{role_id} 的权限", audit_data
    )):
        policy = lock_policy(session, actor, payload.expected_revision)
        role = session.get(Role, role_id)
        if role is None:
            raise NotFoundError("角色不存在")
        if role.role_code == "admin":
            raise ForbiddenError("管理员角色权限固定，不能修改")
        codes = validate_role_permissions(payload.permission_codes)
        audit_data["before"] = role_permissions(session, role.id)["permission_codes"]
        audit_data["after"] = sorted(codes)
        session.execute(delete(RolePermission).where(RolePermission.role_id == role.id))
        ids = dict(session.execute(select(Permission.perm_code, Permission.id)).all())
        session.add_all([RolePermission(role_id=role.id, permission_id=ids[c]) for c in codes])
        bump(policy, actor)
        session.flush()
        result = role_permissions(session, role.id)
        session.commit()
        return result


def settings_out(session):
    policy = auth.setting(session)
    return {"mode": "roles", "revision": policy.revision, "maker_user_id": None,
            "maker_username": None,
            "maker_status": None, "updated_by": policy.updated_by,
            "updated_at": policy.update_time}


def save_settings(session, payload, actor, meta=None):
    raise BadRequestError("已取消单独的制作人绑定，请在角色管理配置权限，再在用户管理分配角色")