zkxw/backend/app/crud/qc_item_param.py

from __future__ import annotations

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

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.qc_param import BizQcItemParam


def _row_to_dict(r: BizQcItemParam) -> Dict[str, Any]:
    return {
        "id": r.id,
        "paramName": r.param_name,
        "paramCode": r.param_code,
        "paramUnit": r.param_unit or "",
        "paramType": r.param_type or "text",
        "options": r.options or "",
        "defaultValue": r.default_value or "",
        "requiredFlag": r.required_flag or "0",
        "orderNum": r.order_num if r.order_num is not None else 0,
    }


def _parse_options_list(options_raw: Any) -> List[str]:
    if not options_raw:
        return []
    if isinstance(options_raw, list):
        arr = options_raw
    else:
        try:
            arr = json.loads(str(options_raw))
        except json.JSONDecodeError:
            raise ValueError("单选选项格式无效")
    if not isinstance(arr, list):
        raise ValueError("单选选项格式无效")
    cleaned = [str(x).strip() for x in arr if str(x).strip()]
    return cleaned


def _validate_param(p: Dict[str, Any], index: int) -> None:
    label = (p.get("paramName") or "").strip() or f"第{index + 1}个参数"
    param_type = (p.get("paramType") or "text").strip()
    if param_type != "select":
        return
    opts = _parse_options_list(p.get("options"))
    if not opts:
        raise ValueError(f"「{label}」为单选类型，请至少配置一个选项")
    if len(opts) != len(set(opts)):
        raise ValueError(f"「{label}」的选项不能重复")


def list_params_by_item(
    db: Session, item_source: str, item_id: str, tenant_no: str, *, user_info=None
) -> List[Dict[str, Any]]:
    q = db.query(BizQcItemParam).filter(
        BizQcItemParam.item_source == item_source,
        BizQcItemParam.item_id == item_id,
        BizQcItemParam.tenant_no == tenant_no,
        BizQcItemParam.del_flag == 0,
    )
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizQcItemParam, user_info, db, tenant_no)
    rows = q.order_by(BizQcItemParam.order_num.asc()).all()
    return [_row_to_dict(r) for r in rows]


def save_params_for_item(
    db: Session,
    item_source: str,
    item_id: str,
    params: List[Dict[str, Any]],
    tenant_no: str,
    user_id: Optional[str] = None,
) -> List[Dict[str, Any]]:
    for idx, p in enumerate(params):
        _validate_param(p, idx)

    now = datetime.now()
    db.query(BizQcItemParam).filter(
        BizQcItemParam.item_source == item_source,
        BizQcItemParam.item_id == item_id,
        BizQcItemParam.tenant_no == tenant_no,
        BizQcItemParam.del_flag == 0,
    ).update(
        {"del_flag": 1, "update_by": user_id, "update_time": now},
        synchronize_session="fetch",
    )
    db.flush()

    for idx, p in enumerate(params):
        param_type = (p.get("paramType") or "text").strip()
        options_val = None
        default_val = None
        if param_type == "select":
            opts = _parse_options_list(p.get("options"))
            options_val = json.dumps(opts, ensure_ascii=False)
            default_val = opts[0] if opts else None

        obj = BizQcItemParam(
            id=generate_uuid(),
            item_source=item_source,
            item_id=item_id,
            param_name=p.get("paramName", ""),
            param_code=p.get("paramCode", ""),
            param_unit=p.get("paramUnit") or None,
            param_type=param_type,
            options=options_val,
            default_value=default_val,
            required_flag=p.get("requiredFlag", "0"),
            tenant_no=tenant_no,
            order_num=p.get("orderNum", idx),
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(obj)

    db.commit()
    return list_params_by_item(db, item_source, item_id, tenant_no)