zkxw/backend/app/api/mini_menu.py

"""小程序菜单：当前用户在「移动端菜单」树下有权限的入口。"""
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})