217 lines
6.6 KiB
Python
217 lines
6.6 KiB
Python
"""小程序菜单：当前用户在「移动端菜单」树下有权限的入口。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Dict, List, Set
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.auth import get_current_user_by_token
|
|
from app.core.database import get_db
|
|
from app.core.response import error_response, success_response
|
|
from app.models.models import SysMenu, SysRoleMenu
|
|
from app.services.web_auth_roles import get_effective_role_ids, resolve_web_role_for_user
|
|
|
|
router = APIRouter(prefix="/mini", tags=["小程序菜单"])
|
|
|
|
|
|
async def _current_user(request: Request, db: Session) -> dict:
|
|
authorization = request.headers.get("Authorization", "")
|
|
if not authorization.startswith("Bearer "):
|
|
raise ValueError("未授权")
|
|
token = authorization.replace("Bearer ", "", 1).strip()
|
|
return await get_current_user_by_token(token=token, db=db)
|
|
|
|
|
|
def _find_mobile_root(db: Session, tenant_no: str) -> SysMenu | None:
|
|
rows = (
|
|
db.query(SysMenu)
|
|
.filter(
|
|
SysMenu.del_flag == 0,
|
|
SysMenu.parent_id == "0",
|
|
SysMenu.tenant_no == tenant_no,
|
|
)
|
|
.all()
|
|
)
|
|
for m in rows:
|
|
name = (m.menu_name or "").strip()
|
|
if "移动端" in name:
|
|
return m
|
|
return None
|
|
|
|
|
|
def _collect_descendants(
|
|
db: Session, root_id: str, tenant_no: str
|
|
) -> List[SysMenu]:
|
|
all_menus = (
|
|
db.query(SysMenu)
|
|
.filter(
|
|
SysMenu.del_flag == 0,
|
|
SysMenu.tenant_no == tenant_no,
|
|
SysMenu.status == "1",
|
|
SysMenu.visible == "1",
|
|
)
|
|
.order_by(SysMenu.order_num.asc(), SysMenu.create_time.asc())
|
|
.all()
|
|
)
|
|
by_parent: Dict[str, List[SysMenu]] = {}
|
|
for m in all_menus:
|
|
pid = (m.parent_id or "0").strip() or "0"
|
|
by_parent.setdefault(pid, []).append(m)
|
|
|
|
out: List[SysMenu] = []
|
|
|
|
def walk(pid: str) -> None:
|
|
for child in by_parent.get(pid, []):
|
|
out.append(child)
|
|
walk(child.id)
|
|
|
|
walk(root_id)
|
|
return out
|
|
|
|
|
|
def _user_menu_ids(db: Session, user_id: str, tenant_no: str) -> Set[str]:
|
|
role_ids = get_effective_role_ids(db, user_id, tenant_no=tenant_no)
|
|
if not role_ids:
|
|
return set()
|
|
rows = (
|
|
db.query(SysRoleMenu.menu_id)
|
|
.filter(
|
|
SysRoleMenu.del_flag == 0,
|
|
SysRoleMenu.tenant_no == tenant_no,
|
|
SysRoleMenu.role_id.in_(role_ids),
|
|
)
|
|
.all()
|
|
)
|
|
return {str(r[0]).strip() for r in rows if r and r[0]}
|
|
|
|
|
|
def _mobile_menu_icon(path: str) -> str:
|
|
"""按路由匹配移动端宫格图标（与 mobile-h5 van-icon 字形名对齐）。"""
|
|
p = (path or "").lower()
|
|
# 兼容有无前导斜杠、query 分隔写法
|
|
if "market/webview" in p and "entry=" in p:
|
|
# payment-chase 须先于 payment 匹配
|
|
if "payment-chase" in p:
|
|
return "market-payment-chase"
|
|
if "presign" in p:
|
|
return "market-presign"
|
|
if "protocol" in p:
|
|
return "market-protocol"
|
|
if "payment" in p:
|
|
return "market-payment"
|
|
if "freight" in p:
|
|
return "market-freight"
|
|
return ""
|
|
if "oa/start" in p:
|
|
return "approve"
|
|
if "oa/drafts" in p:
|
|
return "oa-drafts"
|
|
if "oa/pending" in p:
|
|
return "oa-pending"
|
|
if "oa/done" in p:
|
|
return "oa-done"
|
|
if "oa/mine" in p:
|
|
return "oa-mine"
|
|
if "oa/cc" in p:
|
|
return "oa-cc"
|
|
if "oa/mentions" in p:
|
|
return "oa-mentions"
|
|
if "oa/wechat-notify" in p:
|
|
return "oa-wechat-notify"
|
|
if "form-sink" in p:
|
|
return "orders-o"
|
|
return "apps-o"
|
|
|
|
|
|
def _mobile_menu_icon_colors(path: str) -> dict:
|
|
"""宫格图标底色/图形色（可按入口自定义，不强制品牌绿）。"""
|
|
p = (path or "").lower()
|
|
# OA / 市场菜单 PNG 自带颜色，宫格不套色底
|
|
if any(
|
|
x in p
|
|
for x in (
|
|
"oa/start",
|
|
"oa/drafts",
|
|
"oa/pending",
|
|
"oa/done",
|
|
"oa/mine",
|
|
"oa/cc",
|
|
"oa/mentions",
|
|
"oa/wechat-notify",
|
|
)
|
|
) or ("market/webview" in p and "entry=" in p):
|
|
return {"iconBg": "transparent", "iconFg": ""}
|
|
if "form-sink" in p:
|
|
return {"iconBg": "#eef2f1", "iconFg": "#3d524a"}
|
|
return {"iconBg": "#eef2f1", "iconFg": "#3d524a"}
|
|
|
|
|
|
def list_mobile_menu_items(
|
|
db: Session,
|
|
*,
|
|
user_id: str,
|
|
tenant_no: str,
|
|
is_super_admin: bool = False,
|
|
) -> List[dict]:
|
|
"""列出当前租户下已授权的移动端叶子入口。"""
|
|
tenant = (tenant_no or "").strip()
|
|
uid = (user_id or "").strip()
|
|
if not tenant or not uid:
|
|
return []
|
|
|
|
root = _find_mobile_root(db, tenant)
|
|
if not root:
|
|
return []
|
|
|
|
descendants = _collect_descendants(db, root.id, tenant)
|
|
by_id = {menu.id: menu for menu in descendants}
|
|
allowed_ids = None if is_super_admin else _user_menu_ids(db, uid, tenant)
|
|
items: List[dict] = []
|
|
for menu in descendants:
|
|
if (menu.menu_type or "").upper() == "F":
|
|
continue
|
|
path = (menu.path or "").strip()
|
|
if not path:
|
|
continue
|
|
if "form-sink" in path.lower() or "水槽" in str(menu.menu_name or ""):
|
|
continue
|
|
if allowed_ids is not None and menu.id not in allowed_ids:
|
|
continue
|
|
colors = _mobile_menu_icon_colors(path)
|
|
parent = by_id.get((menu.parent_id or "").strip())
|
|
item = {
|
|
"id": menu.id,
|
|
"title": menu.menu_name,
|
|
"path": path if path.startswith("/") else f"/{path}",
|
|
"icon": _mobile_menu_icon(path),
|
|
"iconBg": colors["iconBg"],
|
|
"iconFg": colors["iconFg"],
|
|
"perms": (menu.perms or "").strip() or None,
|
|
}
|
|
if parent is not None and parent.id != root.id:
|
|
item["groupTitle"] = parent.menu_name
|
|
items.append(item)
|
|
return items
|
|
|
|
|
|
@router.get("/menus")
|
|
async def list_my_mobile_menus(request: Request, db: Session = Depends(get_db)):
|
|
try:
|
|
user = await _current_user(request, db)
|
|
except Exception:
|
|
return error_response(401, "未授权")
|
|
|
|
user_id = str(user.get("user_id") or user.get("sub") or "").strip()
|
|
tenant_no = str(user.get("tenant_no") or "").strip()
|
|
if not user_id or not tenant_no:
|
|
return error_response(401, "未授权")
|
|
web_role = resolve_web_role_for_user(db, user_id, tenant_no=tenant_no)
|
|
items = list_mobile_menu_items(
|
|
db,
|
|
user_id=user_id,
|
|
tenant_no=tenant_no,
|
|
is_super_admin=web_role == "super_admin",
|
|
)
|
|
return success_response({"list": items})
|