"""小程序菜单:当前用户在「移动端菜单」树下有权限的入口。""" 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]} def _mobile_menu_icon(path: str) -> str: """按路由匹配移动端宫格图标(与 mobile-h5 van-icon 字形名对齐)。""" p = (path or "").lower() 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 "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", ) ): return {"iconBg": "transparent", "iconFg": ""} if "form-sink" in p: return {"iconBg": "#eef2f1", "iconFg": "#3d524a"} return {"iconBg": "#eef2f1", "iconFg": "#3d524a"} @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 colors = _mobile_menu_icon_colors(path) items.append( { "id": m.id, "title": m.menu_name, "path": path if path.startswith("/") else f"/{path}", "icon": _mobile_menu_icon(path), "iconBg": colors["iconBg"], "iconFg": colors["iconFg"], "perms": (m.perms or "").strip() or None, } ) return success_response({"list": items})