shanxia2.0/backend/app/crud/menu.py

from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional

from sqlalchemy import and_
from sqlalchemy.orm import Session

from app.core.delete_guard import assert_row_deletable_by_creator
from app.core.utils import generate_uuid
from app.models.models import SysMenu


def _row_to_dict(m: SysMenu) -> Dict[str, Any]:
    return {
        "id": m.id,
        "menuName": m.menu_name,
        "parentId": m.parent_id,
        "menuType": m.menu_type or "M",
        "path": m.path or "",
        "component": m.component or "",
        "perms": m.perms or "",
        "isFrame": m.is_frame if m.is_frame is not None else 1,
        "visible": m.visible or "1",
        "status": m.status or "1",
        "tenantNo": m.tenant_no,
        "orderNum": m.order_num if m.order_num is not None else 0,
        "remark": m.remark or "",
        "createBy": m.create_by or "",
        "createTime": m.create_time.strftime("%Y-%m-%d %H:%M:%S") if m.create_time else "",
        "updateTime": m.update_time.strftime("%Y-%m-%d %H:%M:%S") if m.update_time else "",
    }


def _build_tree(items: List[Dict[str, Any]], parent_id: str = "0") -> List[Dict[str, Any]]:
    children = [item for item in items if item.get("parentId") == parent_id]
    for child in children:
        child["children"] = _build_tree(items, child["id"])
    children.sort(key=lambda x: (x.get("orderNum", 0) or 0))
    return children


def list_menus(
    db: Session,
    *,
    keyword: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> List[Dict[str, Any]]:
    q = db.query(SysMenu).filter(SysMenu.del_flag == 0)
    if tenant_no:
        q = q.filter(SysMenu.tenant_no == tenant_no)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(SysMenu.menu_name.like(like))
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, SysMenu, user_info, db, tenant_no)
    rows = q.order_by(SysMenu.create_time.desc()).all()
    return [_row_to_dict(x) for x in rows]


def list_menus_tree(
    db: Session,
    *,
    keyword: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> List[Dict[str, Any]]:
    kw = (keyword or "").strip()
    # 关键词检索时：仅返回命中节点会导致其父节点缺失，从而无法组装为以 parentId=0 开始的树。
    # 这里改为：命中节点 + 其所有祖先节点，再组树。
    if not kw:
        flat = list_menus(db, keyword=None, tenant_no=tenant_no, user_info=user_info)
        return _build_tree(flat, "0")

    all_items = list_menus(db, keyword=None, tenant_no=tenant_no, user_info=user_info)
    id_map: Dict[str, Dict[str, Any]] = {x.get("id"): x for x in all_items if x.get("id")}

    like_items = [
        x for x in all_items if kw and kw in (x.get("menuName") or "")
    ]

    picked: Dict[str, Dict[str, Any]] = {}

    def pick_with_ancestors(item_id: str) -> None:
        cur = id_map.get(item_id)
        while cur and cur.get("id") and cur.get("id") not in picked:
            picked[cur["id"]] = cur
            parent_id = (cur.get("parentId") or "0").strip()
            if not parent_id or parent_id == "0":
                break
            cur = id_map.get(parent_id)

    for it in like_items:
        if it.get("id"):
            pick_with_ancestors(it["id"])

    return _build_tree(list(picked.values()), "0")


def get_menu(db: Session, menu_id: str) -> Optional[SysMenu]:
    return (
        db.query(SysMenu)
        .filter(and_(SysMenu.id == menu_id, SysMenu.del_flag == 0))
        .first()
    )


def menu_name_exists(
    db: Session,
    name: str,
    tenant_no: str,
    exclude_id: Optional[str] = None,
) -> bool:
    q = db.query(SysMenu).filter(
        and_(SysMenu.menu_name == name, SysMenu.tenant_no == tenant_no, SysMenu.del_flag == 0)
    )
    if exclude_id:
        q = q.filter(SysMenu.id != exclude_id)
    return q.first() is not None


def create_menu(
    db: Session,
    *,
    menu_name: str,
    parent_id: str = "0",
    menu_type: str = "M",
    path: str = "",
    component: Optional[str] = None,
    perms: Optional[str] = None,
    is_frame: int = 1,
    visible: str = "1",
    status: str = "1",
    tenant_no: Optional[str] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> SysMenu:
    name = (menu_name or "").strip()
    if not name:
        raise ValueError("菜单名称不能为空")
    obj = SysMenu(
        id=generate_uuid(),
        menu_name=name,
        parent_id=(parent_id or "0").strip(),
        menu_type=menu_type or "M",
        path=(path or "").strip(),
        component=(component or "").strip() or None,
        perms=(perms or "").strip() or None,
        is_frame=is_frame if is_frame is not None else 1,
        visible=visible or "1",
        status=status or "1",
        tenant_no=tenant_no or "shanxia2.0",
        order_num=order_num if order_num is not None else 0,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id="",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def update_menu(
    db: Session,
    menu_id: str,
    *,
    menu_name: Optional[str] = None,
    parent_id: Optional[str] = None,
    menu_type: Optional[str] = None,
    path: Optional[str] = None,
    component: Optional[str] = None,
    perms: Optional[str] = None,
    is_frame: Optional[int] = None,
    visible: Optional[str] = None,
    status: Optional[str] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> SysMenu:
    obj = get_menu(db, menu_id)
    if not obj:
        raise ValueError("菜单不存在")
    if menu_name is not None:
        obj.menu_name = (menu_name or "").strip() or obj.menu_name
    if parent_id is not None:
        obj.parent_id = (parent_id or "0").strip()
    if menu_type is not None:
        obj.menu_type = menu_type
    if path is not None:
        obj.path = (path or "").strip()
    if component is not None:
        obj.component = (component or "").strip() or None
    if perms is not None:
        obj.perms = (perms or "").strip() or None
    if is_frame is not None:
        obj.is_frame = is_frame
    if visible is not None:
        obj.visible = visible
    if status is not None:
        obj.status = status
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = (remark or "").strip() or None
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def _cascade_soft_delete(db: Session, menu_id: str) -> None:
    children = (
        db.query(SysMenu)
        .filter(and_(SysMenu.parent_id == menu_id, SysMenu.del_flag == 0))
        .all()
    )
    for child in children:
        _cascade_soft_delete(db, child.id)
        child.del_flag = 1
        child.update_time = datetime.now()
        db.add(child)


def soft_delete_menu(db: Session, menu_id: str, *, user_info: dict = None) -> None:
    obj = get_menu(db, menu_id)
    if not obj:
        raise ValueError("菜单不存在")
    assert_row_deletable_by_creator(obj, user_info)
    _cascade_soft_delete(db, menu_id)
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()