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

from __future__ import annotations

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

from sqlalchemy import and_, or_
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 SysDept


def _row_to_dict(d: SysDept) -> Dict[str, Any]:
    return {
        "id": d.id,
        "parentId": d.parent_id,
        "deptName": d.dept_name,
        "deptCategoryCode": d.dept_category_code or "",
        "status": d.status,
        "leader": d.leader or "",
        "leaderPhone": d.leader_phone or "",
        "tenantNo": d.tenant_no,
        "orderNum": d.order_num if d.order_num is not None else 0,
        "remark": d.remark or "",
        "createBy": d.create_by or "",
        "createTime": d.create_time.strftime("%Y-%m-%d %H:%M:%S") if d.create_time else "",
        "updateTime": d.update_time.strftime("%Y-%m-%d %H:%M:%S") if d.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 _collect_with_ancestors(
    items: List[Dict[str, Any]], seed_ids: set[str]
) -> List[Dict[str, Any]]:
    """保留匹配节点及其全部上级，供关键词检索后仍能正确组树。"""
    by_id = {item["id"]: item for item in items if item.get("id")}
    keep: set[str] = set()
    for sid in seed_ids:
        cur = (sid or "").strip()
        while cur and cur != "0":
            if cur not in by_id:
                break
            keep.add(cur)
            cur = (by_id[cur].get("parentId") or "0").strip()
    return [by_id[i] for i in keep if i in by_id]


def list_depts(
    db: Session,
    *,
    keyword: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> List[Dict[str, Any]]:
    q = db.query(SysDept).filter(SysDept.del_flag == 0)
    if tenant_no:
        q = q.filter(SysDept.tenant_no == tenant_no)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(SysDept.dept_name.like(like))
    # 部门为组织主数据（选人/选部门/数据权限配置均需完整树），不做 create_by 过滤
    _ = user_info
    rows = q.order_by(SysDept.create_time.desc()).all()
    return [_row_to_dict(x) for x in rows]


def list_depts_tree(
    db: Session,
    *,
    keyword: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> List[Dict[str, Any]]:
    kw = (keyword or "").strip()
    if not kw:
        flat = list_depts(db, keyword=None, tenant_no=tenant_no, user_info=user_info)
        return _build_tree(flat, "0")

    matched = list_depts(db, keyword=keyword, tenant_no=tenant_no, user_info=user_info)
    if not matched:
        return []

    all_flat = list_depts(db, keyword=None, tenant_no=tenant_no, user_info=user_info)
    seed_ids = {m["id"] for m in matched if m.get("id")}
    expanded = _collect_with_ancestors(all_flat, seed_ids)
    return _build_tree(expanded, "0")


def get_dept(db: Session, dept_id: str) -> Optional[SysDept]:
    return (
        db.query(SysDept)
        .filter(and_(SysDept.id == dept_id, SysDept.del_flag == 0))
        .first()
    )


def create_dept(
    db: Session,
    *,
    parent_id: str = "0",
    dept_name: str,
    dept_category_code: Optional[str] = None,
    status: Optional[str] = "1",
    leader: Optional[str] = None,
    leader_phone: Optional[str] = None,
    tenant_no: Optional[str] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> SysDept:
    name = (dept_name or "").strip()
    if not name:
        raise ValueError("组织名称不能为空")
    obj = SysDept(
        id=generate_uuid(),
        parent_id=(parent_id or "0").strip(),
        dept_name=name,
        dept_category_code=(dept_category_code or "").strip() or None,
        status=status or "1",
        leader=(leader or "").strip() or None,
        leader_phone=(leader_phone or "").strip() or None,
        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_dept(
    db: Session,
    dept_id: str,
    *,
    dept_name: Optional[str] = None,
    dept_category_code: Optional[str] = None,
    status: Optional[str] = None,
    leader: Optional[str] = None,
    leader_phone: Optional[str] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> SysDept:
    obj = get_dept(db, dept_id)
    if not obj:
        raise ValueError("组织不存在")
    if dept_name is not None:
        obj.dept_name = (dept_name or "").strip() or obj.dept_name
    if dept_category_code is not None:
        obj.dept_category_code = (dept_category_code or "").strip() or None
    if status is not None:
        obj.status = status
    if leader is not None:
        obj.leader = (leader or "").strip() or None
    if leader_phone is not None:
        obj.leader_phone = (leader_phone or "").strip() or None
    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, dept_id: str) -> None:
    children = (
        db.query(SysDept)
        .filter(and_(SysDept.parent_id == dept_id, SysDept.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_dept(db: Session, dept_id: str, *, user_info: dict = None) -> None:
    obj = get_dept(db, dept_id)
    if not obj:
        raise ValueError("组织不存在")
    assert_row_deletable_by_creator(obj, user_info)
    _cascade_soft_delete(db, dept_id)
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()