zkxw/backend/app/crud/sample_storage_cert.py

"""样本储存证办理 CRUD。"""
from __future__ import annotations

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.models import BizSampleStorageCert
from app.services.sample_info_progress import apply_handle_payload, storage_cert_handled


def _fmt_dt(t: Optional[datetime]) -> str:
    return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""


def get_by_sample_code(
    db: Session, tenant_no: str, sample_code: str
) -> Optional[BizSampleStorageCert]:
    code = (sample_code or "").strip()
    if not code:
        return None
    return (
        db.query(BizSampleStorageCert)
        .filter(
            BizSampleStorageCert.tenant_no == tenant_no,
            BizSampleStorageCert.sample_code == code,
            BizSampleStorageCert.del_flag == 0,
        )
        .first()
    )


def get_map_by_sample_codes(
    db: Session, tenant_no: str, codes: List[str]
) -> Dict[str, BizSampleStorageCert]:
    cleaned = list({(c or "").strip() for c in codes if (c or "").strip()})
    if not cleaned:
        return {}
    rows = (
        db.query(BizSampleStorageCert)
        .filter(
            BizSampleStorageCert.tenant_no == tenant_no,
            BizSampleStorageCert.sample_code.in_(cleaned),
            BizSampleStorageCert.del_flag == 0,
        )
        .all()
    )
    return {row.sample_code: row for row in rows}


def upsert_handle(
    db: Session,
    *,
    tenant_no: str,
    sample_code: str,
    collection_protocol_id: Optional[str],
    user_id: Optional[str],
    handle_time: Optional[datetime],
    handler_user_id: Optional[str],
    cert_image_urls: str = "",
    remark: str = "",
    mail_cert_time: Optional[datetime] = None,
    update_mail_cert_time: bool = False,
) -> BizSampleStorageCert:
    code = (sample_code or "").strip()
    if not code:
        raise ValueError("sample_code is required")

    payload = apply_handle_payload(
        handle_time=handle_time,
        handler_user_id=handler_user_id,
        attachments=cert_image_urls or "",
        remark=remark or "",
    )
    now = datetime.now()
    row = get_by_sample_code(db, tenant_no, code)
    if row is None:
        row = BizSampleStorageCert(
            id=generate_uuid(),
            sample_code=code,
            collection_protocol_id=(collection_protocol_id or "").strip() or None,
            tenant_no=tenant_no,
            order_num=0,
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(row)
    else:
        if collection_protocol_id and not row.collection_protocol_id:
            row.collection_protocol_id = (collection_protocol_id or "").strip() or None
        row.update_by = user_id
        row.update_time = now

    row.handle_time = payload["handle_time"]
    row.handler_user_id = payload["handler_user_id"]
    row.make_status = payload["make_or_todo_status"]
    row.make_date = payload["legacy_date"] or None
    row.cert_image_urls = payload["attachments"] or None
    row.remark = payload["remark"] or None
    if update_mail_cert_time:
        row.mail_cert_time = mail_cert_time

    db.flush()
    return row


def row_to_dict(row: BizSampleStorageCert) -> Dict[str, Any]:
    return {
        "id": row.id,
        "sampleCode": row.sample_code or "",
        "collectionProtocolId": row.collection_protocol_id or "",
        "makeStatus": row.make_status or "0",
        "certImageUrls": row.cert_image_urls or "",
        "makeDate": (row.make_date or "").strip() if row.make_date else "",
        "handleTime": _fmt_dt(row.handle_time) if getattr(row, "handle_time", None) else "",
        "handlerUserId": getattr(row, "handler_user_id", None) or "",
        "mailCertTime": _fmt_dt(row.mail_cert_time) if getattr(row, "mail_cert_time", None) else "",
        "storageCertHandled": storage_cert_handled(row),
        "tenantNo": row.tenant_no,
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": _fmt_dt(row.create_time),
        "updateTime": _fmt_dt(row.update_time),
    }