115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
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) -> Optional[BxExpenseItem]:
|
|
return BxCrudService.get_by_id(BxExpenseItem, db, item_id, tenant_no)
|
|
|
|
@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)
|
|
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)
|
|
if not parent:
|
|
raise ValueError("父级费用项目不存在")
|
|
return BxCrudService.update(
|
|
BxExpenseItem,
|
|
db,
|
|
item_id,
|
|
tenant_no,
|
|
data,
|
|
user_id=user.id,
|
|
)
|
|
|
|
@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
|
|
)
|
|
|
|
@staticmethod
|
|
def get_item_tree(db: Session, tenant_no: str) -> List[BxExpenseItem]:
|
|
items = (
|
|
db.query(BxExpenseItem)
|
|
.filter(BxExpenseItem.del_flag == 0, BxExpenseItem.tenant_no == tenant_no)
|
|
.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
|