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.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})