zkxw/backend/app/crud/qc_task_result.py

from __future__ import annotations

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

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.crud import qc_item_param as param_svc
from app.crud.qc_task import get_task
from app.models.models import BizQcWorkOrderLabItem, BizQcWorkOrderThirdItem
from app.models.qc_param import BizQcTaskResult


def _row_to_dict(r: BizQcTaskResult) -> Dict[str, Any]:
    return {
        "id": r.id,
        "paramId": r.param_id or "",
        "paramName": r.param_name,
        "paramCode": r.param_code or "",
        "paramUnit": r.param_unit or "",
        "paramType": r.param_type or "text",
        "paramValue": r.param_value or "",
    }


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


def _resolve_task_item_ref(db: Session, task) -> Tuple[Optional[str], Optional[str]]:
    """返回 (item_source, item_id) 供 biz_qc_item_param 查询。"""
    if task.task_type == "sysz":
        line = (
            db.query(BizQcWorkOrderLabItem)
            .filter(
                BizQcWorkOrderLabItem.id == task.source_line_id,
                BizQcWorkOrderLabItem.del_flag == 0,
            )
            .first()
        )
        if line:
            return "lab", line.lab_test_item_id
    elif task.task_type == "dsfjc":
        line = (
            db.query(BizQcWorkOrderThirdItem)
            .filter(
                BizQcWorkOrderThirdItem.id == task.source_line_id,
                BizQcWorkOrderThirdItem.del_flag == 0,
            )
            .first()
        )
        if line:
            return "third_party", line.third_party_item_id
    return None, None


def _merge_saved_into_template(
    templates: List[Dict[str, Any]], saved: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    by_id = {s["paramId"]: s for s in saved if s.get("paramId")}
    by_code = {s["paramCode"]: s for s in saved if s.get("paramCode")}
    out: List[Dict[str, Any]] = []
    for p in templates:
        pid = p.get("id") or ""
        code = p.get("paramCode") or ""
        hit = by_id.get(pid) or (by_code.get(code) if code else None)
        saved_value = (hit.get("paramValue") if hit else "") or ""
        default_value = (p.get("defaultValue") or "").strip()
        param_type = p.get("paramType", "text")
        if not saved_value and param_type == "select" and default_value:
            saved_value = default_value
        out.append(
            {
                "paramId": pid,
                "paramName": p.get("paramName", ""),
                "paramCode": code,
                "paramUnit": p.get("paramUnit", ""),
                "paramType": param_type,
                "options": p.get("options", ""),
                "defaultValue": default_value,
                "paramValue": saved_value,
                "requiredFlag": p.get("requiredFlag", "0"),
            }
        )
    return out


def list_saved_results_by_task_ids(
    db: Session, task_ids: List[str], tenant_no: str
) -> Dict[str, List[Dict[str, Any]]]:
    """按任务 ID 批量返回已保存检测参数（只读展示用，不合并模板空项）。"""
    ids = list({(x or "").strip() for x in task_ids if (x or "").strip()})
    if not ids or not tenant_no:
        return {}
    rows = (
        db.query(BizQcTaskResult)
        .filter(
            BizQcTaskResult.task_id.in_(ids),
            BizQcTaskResult.tenant_no == tenant_no,
            BizQcTaskResult.del_flag == 0,
        )
        .order_by(BizQcTaskResult.create_time.asc())
        .all()
    )
    grouped: Dict[str, List[Dict[str, Any]]] = {i: [] for i in ids}
    for r in rows:
        tid = (r.task_id or "").strip()
        if tid in grouped:
            grouped[tid].append(_row_to_dict(r))
    return grouped


def list_results_by_task(
    db: Session, task_id: str, tenant_no: str, *, user_info=None
) -> List[Dict[str, Any]]:
    """返回任务结果填写表单：项目参数模板 + 已保存值（无模板时仅返回已保存记录）。"""
    saved = _list_saved_results_by_task(db, task_id, tenant_no, user_info=user_info)
    task = get_task(db, task_id, tenant_no=tenant_no)
    if not task:
        return saved
    item_source, item_id = _resolve_task_item_ref(db, task)
    if not item_source or not item_id:
        return saved
    templates = param_svc.list_params_by_item(db, item_source, item_id, tenant_no)
    if not templates:
        return saved
    return _merge_saved_into_template(templates, saved)


def save_results_for_task(
    db: Session,
    task_id: str,
    results: List[Dict[str, Any]],
    tenant_no: str,
    user_id: Optional[str] = None,
    *,
    commit: bool = True,
) -> List[Dict[str, Any]]:
    task = get_task(db, task_id, tenant_no=tenant_no)
    if not task:
        raise ValueError("任务不存在")
    now = datetime.now()
    db.query(BizQcTaskResult).filter(
        BizQcTaskResult.task_id == task_id,
        BizQcTaskResult.tenant_no == tenant_no,
        BizQcTaskResult.del_flag == 0,
    ).update(
        {"del_flag": 1, "update_by": user_id, "update_time": now},
        synchronize_session="fetch",
    )
    db.flush()

    for res in results:
        obj = BizQcTaskResult(
            id=generate_uuid(),
            task_id=task_id,
            param_id=res.get("paramId") or None,
            param_name=res.get("paramName", ""),
            param_code=res.get("paramCode") or None,
            param_unit=res.get("paramUnit") or None,
            param_type=res.get("paramType", "text"),
            param_value=res.get("paramValue") or None,
            tenant_no=tenant_no,
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(obj)

    if commit:
        db.commit()
    else:
        db.flush()
    return list_results_by_task(db, task_id, tenant_no)