79 lines
3.0 KiB
Python
79 lines
3.0 KiB
Python
|
"""不依赖外部组件的查询角色映射辅助函数。"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
|
||
|
|
_ROLE_MAP = {
|
||
|
|
"ADMIN": "admin",
|
||
|
|
"KNOWLEDGE_ADMIN": "knowledge_admin",
|
||
|
|
"KNOWLEDGE_OPERATOR": "knowledge_operator",
|
||
|
|
"投顾": "advisor",
|
||
|
|
"理财顾问": "advisor",
|
||
|
|
"风控专员": "risk",
|
||
|
|
"客户经理": "customer_manager",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def map_employee_role(employee_role: str | None) -> str | None:
|
||
|
|
if not employee_role:
|
||
|
|
return None
|
||
|
|
return _ROLE_MAP.get(employee_role, employee_role.lower())
|
||
|
|
|
||
|
|
|
||
|
|
def build_query_permission(
|
||
|
|
user,
|
||
|
|
role,
|
||
|
|
table_permissions,
|
||
|
|
column_permissions,
|
||
|
|
sensitive_fields,
|
||
|
|
) -> dict:
|
||
|
|
"""将用户、角色、表权限和字段权限合并成一次请求的权限快照。"""
|
||
|
|
permission = {
|
||
|
|
"can_query": False,
|
||
|
|
"role": map_employee_role(getattr(user, "employee_role", None)),
|
||
|
|
"tables": set(),
|
||
|
|
"columns": {},
|
||
|
|
"masks": {},
|
||
|
|
"row_scopes": {},
|
||
|
|
"max_rows": getattr(role, "max_rows", 0) if role else 0,
|
||
|
|
"daily_quota": getattr(role, "daily_quota", 0) if role else 0,
|
||
|
|
}
|
||
|
|
if getattr(user, "user_type", None) != "EMPLOYEE" or not role:
|
||
|
|
return permission
|
||
|
|
if not getattr(role, "can_query", False):
|
||
|
|
return permission
|
||
|
|
|
||
|
|
permission["can_query"] = True
|
||
|
|
for item in table_permissions or []:
|
||
|
|
if getattr(item, "permission", "") == "SELECT" and getattr(item, "table_name", None):
|
||
|
|
permission["tables"].add(item.table_name)
|
||
|
|
scope_type = getattr(item, "row_scope_type", "none")
|
||
|
|
scope_column = getattr(item, "row_scope_column", None)
|
||
|
|
if scope_type != "none" and scope_column:
|
||
|
|
permission["row_scopes"][item.table_name] = {
|
||
|
|
"type": scope_type,
|
||
|
|
"column": scope_column,
|
||
|
|
}
|
||
|
|
|
||
|
|
for item in column_permissions or []:
|
||
|
|
table_name = getattr(item, "table_name", None)
|
||
|
|
column_name = getattr(item, "column_name", None)
|
||
|
|
if table_name not in permission["tables"] or not column_name:
|
||
|
|
continue
|
||
|
|
if getattr(item, "access_mode", "allow") in {"allow", "mask"}:
|
||
|
|
permission["columns"].setdefault(table_name, set()).add(column_name)
|
||
|
|
if getattr(item, "access_mode", "allow") == "mask":
|
||
|
|
permission["masks"][(table_name, column_name)] = getattr(
|
||
|
|
item, "mask_type", None
|
||
|
|
) or "partial"
|
||
|
|
elif getattr(item, "access_mode", "allow") == "deny":
|
||
|
|
permission["columns"].setdefault(table_name, set()).discard(column_name)
|
||
|
|
|
||
|
|
for item in sensitive_fields or []:
|
||
|
|
table_name = getattr(item, "table_name", None)
|
||
|
|
column_name = getattr(item, "column_name", None)
|
||
|
|
if table_name in permission["tables"] and column_name in permission["columns"].get(table_name, set()):
|
||
|
|
permission["masks"][(table_name, column_name)] = getattr(
|
||
|
|
item, "mask_type", None
|
||
|
|
) or "partial"
|
||
|
|
return permission
|