shanxia2.0/backend/app/services/bx/expense_item.py

from __future__ import annotations

from typing import List, Optional

from sqlalchemy.orm import Session

from app.core.bx_auth import BxUserContext
from app.core.tenant_context import get_tenant_no
from app.models.bx_models import BxExpenseItem
from app.services.bx.base import BxCrudService


class ExpenseItemService:
    @staticmethod
    def _tenant_no(db: Session, user: BxUserContext) -> str:
        return get_tenant_no(db, user.to_tenant_user_dict())

    @staticmethod
    def get_by_id(
        db: Session,
        item_id: str,
        tenant_no: str,
        *,
        user: Optional[BxUserContext] = None,
        skip_scope: bool = True,
    ) -> Optional[BxExpenseItem]:
        # 费用项目为主数据：默认跳过 create_by 数据范围
        return BxCrudService.get_by_id(
            BxExpenseItem, db, item_id, tenant_no, user=user, skip_scope=skip_scope
        )

    @staticmethod
    def create(
        db: Session,
        obj_in: dict,
        *,
        user: BxUserContext,
    ) -> BxExpenseItem:
        tenant_no = ExpenseItemService._tenant_no(db, user)
        data = dict(obj_in)
        parent_id = data.get("parent_id") or "0"
        data["parent_id"] = parent_id
        if parent_id and parent_id != "0":
            parent = ExpenseItemService.get_by_id(db, parent_id, tenant_no, user=user)
            if not parent:
                raise ValueError("父级费用项目不存在")
            data.setdefault("level", (parent.level or 1) + 1)
        else:
            data["parent_id"] = "0"
            data.setdefault("level", 1)
        return BxCrudService.create(
            BxExpenseItem,
            db,
            data,
            tenant_no=tenant_no,
            user_id=user.id,
            dept_id=user.dept_id,
        )

    @staticmethod
    def update(
        db: Session,
        item_id: str,
        obj_in: dict,
        *,
        user: BxUserContext,
    ) -> Optional[BxExpenseItem]:
        tenant_no = ExpenseItemService._tenant_no(db, user)
        data = dict(obj_in)
        if "parent_id" in data and data["parent_id"] is not None:
            parent_id = data["parent_id"] or "0"
            data["parent_id"] = parent_id
            if parent_id != "0":
                if parent_id == item_id:
                    raise ValueError("不能将项目设置为自己的父级")
                parent = ExpenseItemService.get_by_id(db, parent_id, tenant_no, user=user)
                if not parent:
                    raise ValueError("父级费用项目不存在")
        return BxCrudService.update(
            BxExpenseItem,
            db,
            item_id,
            tenant_no,
            data,
            user_id=user.id,
            user=user,
            skip_scope=True,
        )

    @staticmethod
    def soft_delete(db: Session, item_id: str, *, user: BxUserContext) -> bool:
        tenant_no = ExpenseItemService._tenant_no(db, user)
        return BxCrudService.soft_delete(
            BxExpenseItem,
            db,
            item_id,
            tenant_no,
            user_id=user.id,
            user=user,
            skip_scope=True,
        )

    @staticmethod
    def get_item_tree(db: Session, tenant_no: str, *, user: BxUserContext) -> List[BxExpenseItem]:
        # 费用项目是租户级主数据，供报销/备用金选择；不做 create_by 数据范围过滤，
        # 否则「仅本人」角色看不到管理员维护的费用项目。
        _ = user
        query = db.query(BxExpenseItem).filter(
            BxExpenseItem.del_flag == 0, BxExpenseItem.tenant_no == tenant_no
        )
        items = query.order_by(
            BxExpenseItem.sort_order.asc(), BxExpenseItem.create_time.desc()
        ).all()
        return ExpenseItemService._build_tree(items)

    @staticmethod
    def _build_tree(items: List[BxExpenseItem]) -> List[BxExpenseItem]:
        item_map = {item.id: item for item in items}
        roots: List[BxExpenseItem] = []
        for item in items:
            parent_id = item.parent_id or "0"
            if not parent_id or parent_id == "0":
                roots.append(item)
            else:
                parent = item_map.get(parent_id)
                if parent:
                    if not hasattr(parent, "children") or parent.children is None:
                        parent.children = []
                    parent.children.append(item)
        for item in items:
            if hasattr(item, "children") and item.children:
                item.child_count = len(item.children)
            else:
                item.child_count = 0
        return roots