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

"""样本接收检查项结果 CRUD"""
from __future__ import annotations

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

from sqlalchemy import and_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.crud import sample_reception_check_item as check_item_svc
from app.models.models import (
    BizSampleReception,
    BizSampleReceptionCheckItem,
    BizSampleReceptionCheckResult,
)

_LEGACY_ITEMS = (
    ("packaging", "包装", "check_packaging"),
    ("sealing", "封口", "check_sealing"),
    ("no_damage", "无破损", "check_no_damage"),
    ("no_leakage", "无漏液", "check_no_leakage"),
    ("color", "颜色", "check_color"),
    ("tying_rope", "系绳", "check_tying_rope"),
    ("no_stratification", "无分层", "check_no_stratification"),
)


def _result_to_dict(r: BizSampleReceptionCheckResult) -> Dict[str, Any]:
    return {
        "id": r.id,
        "checkItemId": r.check_item_id,
        "itemCode": r.item_code,
        "itemName": r.item_name,
        "checked": bool(r.checked),
        "orderNum": r.order_num if r.order_num is not None else 0,
    }


def list_results_by_reception_ids(
    db: Session, reception_ids: List[str], tenant_no: str
) -> Dict[str, List[Dict[str, Any]]]:
    if not reception_ids:
        return {}
    rows = (
        db.query(BizSampleReceptionCheckResult)
        .filter(
            BizSampleReceptionCheckResult.reception_id.in_(reception_ids),
            BizSampleReceptionCheckResult.tenant_no == tenant_no,
            BizSampleReceptionCheckResult.del_flag == 0,
        )
        .order_by(
            BizSampleReceptionCheckResult.order_num.asc(),
            BizSampleReceptionCheckResult.create_time.asc(),
        )
        .all()
    )
    out: Dict[str, List[Dict[str, Any]]] = {}
    for r in rows:
        out.setdefault(r.reception_id, []).append(_result_to_dict(r))
    return out


def _legacy_results_from_reception(rec: BizSampleReception) -> List[Dict[str, Any]]:
    items: List[Dict[str, Any]] = []
    for order, (code, name, attr) in enumerate(_LEGACY_ITEMS, start=1):
        items.append(
            {
                "id": "",
                "checkItemId": "",
                "itemCode": code,
                "itemName": name,
                "checked": bool(getattr(rec, attr, 0)),
                "orderNum": order,
            }
        )
    return items


def get_check_results_for_reception(
    db: Session, rec: BizSampleReception, tenant_no: str
) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizSampleReceptionCheckResult)
        .filter(
            BizSampleReceptionCheckResult.reception_id == rec.id,
            BizSampleReceptionCheckResult.tenant_no == tenant_no,
            BizSampleReceptionCheckResult.del_flag == 0,
        )
        .order_by(
            BizSampleReceptionCheckResult.order_num.asc(),
            BizSampleReceptionCheckResult.create_time.asc(),
        )
        .all()
    )
    if rows:
        return [_result_to_dict(r) for r in rows]
    return _legacy_results_from_reception(rec)


def save_reception_check_results(
    db: Session,
    *,
    reception_id: str,
    tenant_no: str,
    user_id: Optional[str],
    create_dept_id: str,
    checks: List[Dict[str, Any]],
) -> None:
    """按当前租户启用检查项写入结果；checks: [{checkItemId, checked}]"""
    active_items = check_item_svc.list_active_check_items(db, tenant_no)
    if not active_items:
        raise ValueError("未配置启用的检查项，请先在「接收检查项配置」中维护")
    submitted = {
        str(c.get("checkItemId") or c.get("check_item_id") or ""): bool(c.get("checked"))
        for c in (checks or [])
    }
    active_ids = {str(x["id"]) for x in active_items}
    unknown = [k for k in submitted if k and k not in active_ids]
    if unknown:
        raise ValueError("存在无效的检查项")
    now = datetime.now()
    for item in active_items:
        item_id = str(item["id"])
        checked = submitted.get(item_id, False)
        row = BizSampleReceptionCheckResult(
            id=generate_uuid(),
            reception_id=reception_id,
            check_item_id=item_id,
            item_code=item["itemCode"],
            item_name=item["itemName"],
            checked=1 if checked else 0,
            tenant_no=tenant_no,
            order_num=item.get("orderNum") or 0,
            create_by=user_id,
            create_dept_id=create_dept_id or "",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(row)


def sync_legacy_columns_from_results(
    db: Session, rec: BizSampleReception, results: List[BizSampleReceptionCheckResult]
) -> None:
    """兼容旧列：按 item_code 回写 biz_sample_reception.check_*"""
    code_map = {r.item_code: bool(r.checked) for r in results}
    rec.check_packaging = 1 if code_map.get("packaging") else 0
    rec.check_sealing = 1 if code_map.get("sealing") else 0
    rec.check_no_damage = 1 if code_map.get("no_damage") else 0
    rec.check_no_leakage = 1 if code_map.get("no_leakage") else 0
    rec.check_color = 1 if code_map.get("color") else 0
    rec.check_tying_rope = 1 if code_map.get("tying_rope") else 0
    rec.check_no_stratification = 1 if code_map.get("no_stratification") else 0