zkxw/backend/app/services/sample_info_progress.py

"""样本信息管理：进度展示码、证/保险已办理判定、办理字段同步。"""
from __future__ import annotations

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


def collection_progress(sample_status: str) -> Tuple[str, str]:
    st = (sample_status or "").strip()
    if not st:
        return "uncollected", "未采集"
    if st == "yzf":
        return "voided", "已作废"
    if st in ("ycj", "jcz", "yrk"):
        return "collected", "已采集"
    return "uncollected", "未采集"


def reception_progress(reception: Optional[Dict[str, Any]]) -> Tuple[str, str]:
    if not reception:
        return "none", "未接收"
    c = (reception.get("acceptConclusion") or "").strip()
    if c == "js":
        return "js", "已接收"
    if c == "bj":
        return "bj", "拒收"
    return "none", "未接收"


def qc_progress(qc: Optional[Dict[str, Any]]) -> Tuple[str, str]:
    if not qc:
        return "incomplete", "未完成"
    if (qc.get("qcWorkOrderStatus") or "").strip() == "yzf":
        return "voided", "已作废"
    conc = (qc.get("qcConclusion") or "").strip()
    if conc in ("tg", "btg"):
        return "complete", "已完成"
    return "incomplete", "未完成"


def storage_progress(review: Optional[Dict[str, Any]]) -> Tuple[str, str]:
    if not review:
        return "none", "未入库"
    if (review.get("storageStatus") or "").strip() == "yrk":
        return "yrk", "已入库"
    if (review.get("reviewConclusion") or "").strip() == "btg":
        return "rejected", "拒绝入库"
    return "none", "未入库"


def storage_cert_handled(row: Any) -> bool:
    if row is None:
        return False
    if getattr(row, "handle_time", None):
        return True
    if (getattr(row, "make_status", None) or "").strip() == "1":
        return True
    return bool((getattr(row, "make_date", None) or "").strip())


def insurance_handled(row: Any) -> bool:
    if row is None:
        return False
    if getattr(row, "handle_time", None):
        return True
    if (getattr(row, "todo_status", None) or "").strip() == "1":
        return True
    return bool((getattr(row, "handle_date", None) or "").strip())


def sync_handle_status_fields(handle_time: Optional[datetime]) -> Tuple[str, str]:
    if handle_time is None:
        return "0", ""
    return "1", handle_time.strftime("%Y%m%d")


def apply_handle_payload(
    *,
    handle_time: Optional[datetime],
    handler_user_id: Optional[str],
    attachments: str = "",
    remark: str = "",
) -> Dict[str, Any]:
    """规范化办理写入字段。有办理时间则办理人员必填。"""
    uid = (handler_user_id or "").strip() or None
    if handle_time is not None and not uid:
        raise ValueError("填写办理时间时必须指定办理人员")
    status, legacy_date = sync_handle_status_fields(handle_time)
    if handle_time is None:
        uid = None
    return {
        "handle_time": handle_time,
        "handler_user_id": uid,
        "attachments": (attachments or "").strip(),
        "remark": (remark or "").strip(),
        "make_or_todo_status": status,
        "legacy_date": legacy_date,
    }