zkxw/backend/app/crud/inbound_sample_special.py

"""入库样本（采集协议 / 科研捐赠）· 特殊标记维护。"""
from __future__ import annotations

from typing import Any, Dict, Optional

from sqlalchemy.orm import Session

from app.models.models import BizProtocolSample
from app.models.research_donation import BizResearchDonationSample
from app.utils.sample_special import (
    apply_inbound_sample_special_mark_fields,
    sample_special_to_dict,
    validate_sample_special_raw,
)


def update_inbound_sample_special_mark(
    db: Session,
    *,
    tenant_no: str,
    reception_source: str,
    protocol_sample_id: str,
    is_special: bool,
    special_remark: Optional[str],
    user_id: str,
) -> Dict[str, Any]:
    sample_id = (protocol_sample_id or "").strip()
    if not sample_id:
        raise ValueError("缺少样本 ID")

    source = (reception_source or "cell_storage").strip()
    raw = {"is_special": is_special, "special_remark": special_remark or ""}
    validate_sample_special_raw(raw)

    if source == "research_donation":
        sample = (
            db.query(BizResearchDonationSample)
            .filter(
                BizResearchDonationSample.id == sample_id,
                BizResearchDonationSample.tenant_no == tenant_no,
                BizResearchDonationSample.del_flag == 0,
            )
            .first()
        )
        if not sample:
            raise ValueError("捐赠样本不存在")
    else:
        sample = (
            db.query(BizProtocolSample)
            .filter(
                BizProtocolSample.id == sample_id,
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
            )
            .first()
        )
        if not sample:
            raise ValueError("协议样本不存在")

    apply_inbound_sample_special_mark_fields(sample, raw)
    sample.update_by = user_id
    db.flush()
    return sample_special_to_dict(sample)