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

"""字典数据 CRUD（sys_dict_data）"""
from __future__ import annotations

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

from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.models import SysDictData


def _row_to_dict(d: SysDictData) -> Dict[str, Any]:
    return {
        "id": d.id,
        "dictLabel": d.dict_label,
        "dictValue": d.dict_value,
        "dictType": d.dict_type,
        "tenantNo": d.tenant_no,
        "orderNum": d.order_num if d.order_num is not None else 0,
        "remark": d.remark 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 list_dict_data(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    dict_type_code: Optional[str] = None,
    keyword: Optional[str] = None,
    tenant_no: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = db.query(SysDictData).filter(SysDictData.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(SysDictData.tenant_no == tenant_no)
        if dict_type_code:
            q0 = q0.filter(SysDictData.dict_type == dict_type_code)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(SysDictData.dict_label.like(like))
        return q0

    total = _filtered().order_by(None).with_entities(func.count(SysDictData.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(SysDictData.order_num.asc(), SysDictData.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return [_row_to_dict(x) for x in rows], int(total)


def get_dict_data(db: Session, data_id: str) -> Optional[SysDictData]:
    return (
        db.query(SysDictData)
        .filter(and_(SysDictData.id == data_id, SysDictData.del_flag == 0))
        .first()
    )


def create_dict_data(
    db: Session,
    *,
    dict_label: str,
    dict_value: str,
    dict_type: str,
    tenant_no: str,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> SysDictData:
    obj = SysDictData(
        id=generate_uuid(),
        dict_label=(dict_label or "").strip(),
        dict_value=(dict_value or "").strip(),
        dict_type=(dict_type or "").strip(),
        tenant_no=tenant_no,
        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_dict_data(
    db: Session,
    data_id: str,
    *,
    dict_label: Optional[str] = None,
    dict_value: Optional[str] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> SysDictData:
    obj = get_dict_data(db, data_id)
    if not obj:
        raise ValueError("字典数据不存在")
    if dict_label is not None:
        obj.dict_label = (dict_label or "").strip()
    if dict_value is not None:
        obj.dict_value = (dict_value or "").strip()
    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 soft_delete_dict_data(db: Session, data_id: str) -> None:
    obj = get_dict_data(db, data_id)
    if not obj:
        raise ValueError("字典数据不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


def list_dict_data_options(
    db: Session, *, dict_type_code: str, tenant_no: Optional[str] = None
) -> List[Dict[str, str]]:
    q = db.query(SysDictData).filter(
        and_(SysDictData.dict_type == dict_type_code, SysDictData.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(SysDictData.tenant_no == tenant_no)
    rows = q.order_by(SysDictData.order_num.asc(), SysDictData.create_time.desc()).all()
    return [{"label": r.dict_label, "value": r.dict_value} for r in rows]


def total_pages(total: int, page_size: int) -> int:
    if total <= 0:
        return 0
    return int(math.ceil(total / float(page_size)))