wuzilongandCursor
bde19c0644
执行中的运输计划不再随意改单；库存流水记到装卸产品明细；手工派车日历区分酸碱并记录数量。同时收紧合同附件与子表的差量保存，并清理已废弃的 bak 备份目录。 Co-authored-by: Cursor <cursoragent@cursor.com>
298 lines
10 KiB
Python
298 lines
10 KiB
Python
"""角色 CRUD（sys_role / sys_role_menu / sys_role_data_scope_detail）"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from sqlalchemy import and_, func, or_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.delete_guard import assert_row_deletable_by_creator
|
||
from app.core.utils import generate_uuid
|
||
from app.crud.relation_sync import sync_relations, assert_relation_targets
|
||
from app.models.models import SysRole, SysRoleMenu, SysRoleDataScopeDetail, SysMenu, SysDept, SysUser
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _row_to_dict(r: SysRole) -> Dict[str, Any]:
|
||
return {
|
||
"id": r.id,
|
||
"roleName": r.role_name,
|
||
"dataScope": r.data_scope or "1",
|
||
"status": r.status or "1",
|
||
"tenantNo": r.tenant_no or "",
|
||
"orderNum": r.order_num if r.order_num is not None else 0,
|
||
"remark": r.remark or "",
|
||
"createBy": r.create_by or "",
|
||
"createTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
|
||
"updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "",
|
||
}
|
||
|
||
|
||
def total_pages(total: int, page_size: int) -> int:
|
||
if total <= 0:
|
||
return 0
|
||
return int(math.ceil(total / float(page_size)))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# list / get
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def list_roles(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
keyword: Optional[str] = None,
|
||
role_name: Optional[str] = None,
|
||
role_name_in: Optional[str] = None,
|
||
status: Optional[str] = None,
|
||
status_in: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
ids: Optional[List[str]] = None,
|
||
max_page_size: Optional[int] = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
def _filtered():
|
||
q0 = db.query(SysRole).filter(SysRole.del_flag == 0)
|
||
if tenant_no:
|
||
q0 = q0.filter(SysRole.tenant_no == tenant_no)
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
q0 = q0.filter(SysRole.role_name.like(like))
|
||
rn = (role_name or "").strip()
|
||
if rn:
|
||
q0 = q0.filter(SysRole.role_name.like(f"%{rn}%"))
|
||
rns_in = [s.strip() for s in str(role_name_in or "").split(",") if s.strip()]
|
||
if rns_in:
|
||
q0 = q0.filter(SysRole.role_name.in_(rns_in))
|
||
statuses = [s.strip() for s in str(status_in or "").split(",") if s.strip()]
|
||
if not statuses and status:
|
||
statuses = [status]
|
||
if statuses:
|
||
q0 = q0.filter(SysRole.status.in_(statuses))
|
||
if ids:
|
||
q0 = q0.filter(SysRole.id.in_(ids))
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q0 = apply_data_scope(q0, SysRole, user_info, db, tenant_no)
|
||
return q0
|
||
|
||
total = _filtered().order_by(None).with_entities(func.count(SysRole.id)).scalar() or 0
|
||
page = max(1, page)
|
||
cap = int(max_page_size) if max_page_size else 500
|
||
page_size = min(max(1, page_size), cap)
|
||
rows = (
|
||
_filtered()
|
||
.order_by(SysRole.create_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
return [_row_to_dict(x) for x in rows], int(total)
|
||
|
||
|
||
def get_role(db: Session, role_id: str) -> Optional[SysRole]:
|
||
return (
|
||
db.query(SysRole)
|
||
.filter(and_(SysRole.id == role_id, SysRole.del_flag == 0))
|
||
.first()
|
||
)
|
||
|
||
|
||
def role_name_exists(db: Session, name: str, tenant_no: str, exclude_id: Optional[str] = None) -> bool:
|
||
q = db.query(SysRole).filter(
|
||
and_(SysRole.role_name == name.strip(), SysRole.tenant_no == tenant_no, SysRole.del_flag == 0)
|
||
)
|
||
if exclude_id:
|
||
q = q.filter(SysRole.id != exclude_id)
|
||
return q.first() is not None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# create / update / delete
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def create_role(
|
||
db: Session,
|
||
*,
|
||
role_name: str,
|
||
data_scope: str = "4",
|
||
status: str = "1",
|
||
tenant_no: str = "shanxia2.0",
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> SysRole:
|
||
rn = (role_name or "").strip()
|
||
if not rn:
|
||
raise ValueError("角色名称不能为空")
|
||
if role_name_exists(db, rn, tenant_no):
|
||
raise ValueError("角色名称已存在")
|
||
obj = SysRole(
|
||
id=generate_uuid(),
|
||
role_name=rn,
|
||
data_scope=data_scope,
|
||
status=status,
|
||
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)
|
||
return obj
|
||
|
||
|
||
def update_role(
|
||
db: Session,
|
||
role_id: str,
|
||
*,
|
||
role_name: Optional[str] = None,
|
||
data_scope: Optional[str] = None,
|
||
status: Optional[str] = None,
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> SysRole:
|
||
obj = get_role(db, role_id)
|
||
if not obj:
|
||
raise ValueError("角色不存在")
|
||
if role_name is not None:
|
||
rn = role_name.strip()
|
||
if not rn:
|
||
raise ValueError("角色名称不能为空")
|
||
if rn != obj.role_name and role_name_exists(db, rn, obj.tenant_no, exclude_id=role_id):
|
||
raise ValueError("角色名称已存在")
|
||
obj.role_name = rn
|
||
if data_scope is not None:
|
||
obj.data_scope = data_scope
|
||
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
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def soft_delete_role(db: Session, role_id: str, *, user_info: dict = None) -> None:
|
||
obj = get_role(db, role_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)
|
||
# cascade soft-delete role_menu and role_data_scope_detail
|
||
db.query(SysRoleMenu).filter(
|
||
and_(SysRoleMenu.role_id == role_id, SysRoleMenu.del_flag == 0)
|
||
).update({"del_flag": 1, "update_time": datetime.now()})
|
||
db.query(SysRoleDataScopeDetail).filter(
|
||
and_(SysRoleDataScopeDetail.role_id == role_id, SysRoleDataScopeDetail.del_flag == 0)
|
||
).update({"del_flag": 1, "update_time": datetime.now()})
|
||
db.commit()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# role-menu
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def get_role_menu_ids(db: Session, role_id: str) -> List[str]:
|
||
rows = (
|
||
db.query(SysRoleMenu.menu_id)
|
||
.filter(and_(SysRoleMenu.role_id == role_id, SysRoleMenu.del_flag == 0))
|
||
.all()
|
||
)
|
||
return [r.menu_id for r in rows]
|
||
|
||
|
||
def update_role_menus(
|
||
db: Session,
|
||
role_id: str,
|
||
menu_ids: List[str],
|
||
tenant_no: str,
|
||
user_id: Optional[str] = None,
|
||
) -> None:
|
||
obj = db.query(SysRole).filter(
|
||
SysRole.id == role_id, SysRole.tenant_no == tenant_no, SysRole.del_flag == 0,
|
||
).with_for_update().first()
|
||
if not obj:
|
||
raise ValueError("角色不存在或不属于当前租户")
|
||
assert_relation_targets(db, SysMenu, tenant_no, menu_ids, "菜单")
|
||
sync_relations(
|
||
db, SysRoleMenu, tenant_no, {"role_id": role_id}, ("menu_id",),
|
||
[(mid,) for mid in menu_ids or [] if mid], user_id, unique_history=True,
|
||
)
|
||
db.commit()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# role data-scope
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def get_role_data_scope(db: Session, role_id: str) -> List[Dict[str, str]]:
|
||
rows = (
|
||
db.query(SysRoleDataScopeDetail)
|
||
.filter(and_(SysRoleDataScopeDetail.role_id == role_id, SysRoleDataScopeDetail.del_flag == 0))
|
||
.all()
|
||
)
|
||
return [{"scopeType": r.scope_type, "scopeValue": r.scope_value} for r in rows]
|
||
|
||
|
||
def update_role_data_scope(
|
||
db: Session,
|
||
role_id: str,
|
||
data_scope: str,
|
||
scope_details: Optional[List[Dict[str, str]]] = None,
|
||
tenant_no: str = "shanxia2.0",
|
||
user_id: Optional[str] = None,
|
||
) -> None:
|
||
obj = db.query(SysRole).filter(
|
||
SysRole.id == role_id, SysRole.tenant_no == tenant_no, SysRole.del_flag == 0,
|
||
).with_for_update().first()
|
||
if not obj:
|
||
raise ValueError("角色不存在或不属于当前租户")
|
||
# 本接口是数据权限弹窗的完整保存。未传 scopeDetails 与 [] 都表示当前没有自定义明细。
|
||
desired = []
|
||
if data_scope == "5":
|
||
for detail in scope_details or []:
|
||
scope_type = detail.get("scopeType", "dept")
|
||
scope_value = str(detail.get("scopeValue") or "").strip()
|
||
if scope_type not in ("dept", "user") or not scope_value:
|
||
raise ValueError("自定义权限须填写有效的部门或用户")
|
||
desired.append((scope_type, scope_value))
|
||
for scope_type, model, label in (("dept", SysDept, "部门"), ("user", SysUser, "用户")):
|
||
assert_relation_targets(db, model, tenant_no,
|
||
[value for kind, value in desired if kind == scope_type], label)
|
||
sync_relations(
|
||
db, SysRoleDataScopeDetail, tenant_no, {"role_id": role_id},
|
||
("scope_type", "scope_value"), desired, user_id,
|
||
)
|
||
obj.data_scope = data_scope
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.commit()
|