129 lines
3.8 KiB
Python
129 lines
3.8 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.config import settings
|
|
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) -> SysMenu | None:
|
|
tenant = settings.DEFAULT_TENANT_NO
|
|
rows = (
|
|
db.query(SysMenu)
|
|
.filter(
|
|
SysMenu.del_flag == 0,
|
|
SysMenu.parent_id == "0",
|
|
SysMenu.tenant_no == tenant,
|
|
)
|
|
.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) -> List[SysMenu]:
|
|
tenant = settings.DEFAULT_TENANT_NO
|
|
all_menus = (
|
|
db.query(SysMenu)
|
|
.filter(
|
|
SysMenu.del_flag == 0,
|
|
SysMenu.tenant_no == tenant,
|
|
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) -> Set[str]:
|
|
role_ids = get_effective_role_ids(db, user_id)
|
|
if not role_ids:
|
|
return set()
|
|
tenant = settings.DEFAULT_TENANT_NO
|
|
rows = (
|
|
db.query(SysRoleMenu.menu_id)
|
|
.filter(
|
|
SysRoleMenu.del_flag == 0,
|
|
SysRoleMenu.tenant_no == tenant,
|
|
SysRoleMenu.role_id.in_(role_ids),
|
|
)
|
|
.all()
|
|
)
|
|
return {str(r[0]).strip() for r in rows if r and r[0]}
|
|
|
|
|
|
@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()
|
|
if not user_id:
|
|
return error_response(401, "未授权")
|
|
|
|
root = _find_mobile_root(db)
|
|
if not root:
|
|
return success_response({"list": []})
|
|
|
|
descendants = _collect_descendants(db, root.id)
|
|
web_role = resolve_web_role_for_user(db, user_id)
|
|
allowed_ids = None if web_role == "super_admin" else _user_menu_ids(db, user_id)
|
|
|
|
items = []
|
|
for m in descendants:
|
|
if (m.menu_type or "").upper() == "F":
|
|
continue
|
|
path = (m.path or "").strip()
|
|
if not path:
|
|
continue
|
|
if allowed_ids is not None and m.id not in allowed_ids:
|
|
continue
|
|
items.append(
|
|
{
|
|
"id": m.id,
|
|
"title": m.menu_name,
|
|
"path": path if path.startswith("/") else f"/{path}",
|
|
"icon": "edit",
|
|
"perms": (m.perms or "").strip() or None,
|
|
}
|
|
)
|
|
return success_response({"list": items})
|