zkxw/backend/app/services/sample_info_quick_register.py

"""样本信息列表快捷登记：去重与入库字段规范化。"""
from __future__ import annotations

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

from app.services.sample_info_progress import apply_handle_payload


def unique_nonempty_ids(ids: Iterable[Optional[str]]) -> List[str]:
    seen = set()
    out: List[str] = []
    for raw in ids:
        s = (raw or "").strip()
        if not s or s in seen:
            continue
        seen.add(s)
        out.append(s)
    return out


def unique_sample_items(
    items: Optional[Iterable[Dict[str, str]]],
) -> List[tuple]:
    """按 sampleCode 去重，保留首次出现的 (sampleCode, protocolId)。"""
    seen = set()
    out: List[tuple] = []
    for it in items or []:
        code = (it.get("sampleCode") or "").strip()
        pid = (it.get("protocolId") or "").strip()
        if not code or code in seen:
            continue
        seen.add(code)
        out.append((code, pid))
    return out


def apply_quick_storage_payload(
    *,
    handle_time: Optional[datetime],
    handler_user_id: Optional[str],
    attachments: str = "",
    remark: str = "",
) -> Dict[str, Any]:
    """快捷登记入库必须填写办理时间；不走生产/质检双门禁。"""
    if handle_time is None:
        raise ValueError("请填写办理时间")
    payload = apply_handle_payload(
        handle_time=handle_time,
        handler_user_id=handler_user_id,
        attachments=attachments,
        remark=remark,
    )
    return {
        "storage_status": "yrk",
        "storage_time": payload["handle_time"],
        "storage_by": payload["handler_user_id"],
        "storage_in_date": payload["legacy_date"],
        "attachments": payload["attachments"],
        "remark": payload["remark"],
    }