zkxw/backend/app/utils/sample_special.py

"""协议/捐赠样本 · 特殊标记字段工具。"""
from __future__ import annotations

from typing import Any, Dict, Optional


def _truthy_flag(value: Any) -> bool:
    if value is True or value == 1:
        return True
    if isinstance(value, str):
        return value.strip().lower() in ("1", "true", "yes")
    return False


def sample_special_to_dict(sample: Any) -> Dict[str, Any]:
    if sample is None:
        return {"isSpecial": False, "specialRemark": ""}
    return {
        "isSpecial": bool(getattr(sample, "is_special", 0)),
        "specialRemark": (getattr(sample, "special_remark", None) or "").strip(),
    }


def validate_sample_special_raw(raw: dict, *, index: Optional[int] = None) -> None:
    is_special = _truthy_flag(raw.get("is_special"))
    remark = (raw.get("special_remark") or "").strip()
    prefix = f"样本第{index + 1}行：" if index is not None else ""
    if is_special and not remark:
        raise ValueError(f"{prefix}特殊样本须填写特殊说明")
    if len(remark) > 256:
        raise ValueError(f"{prefix}特殊说明不能超过256个字符")


def apply_sample_special_fields(obj: Any, raw: dict) -> None:
    is_special = _truthy_flag(raw.get("is_special"))
    remark = (raw.get("special_remark") or "").strip()
    obj.is_special = 1 if is_special else 0
    obj.special_remark = remark if is_special and remark else None


def apply_inbound_sample_special_mark_fields(obj: Any, raw: dict) -> None:
    """质检等场景：取消勾选时保留 special_remark，仅清除 is_special 标记。"""
    is_special = _truthy_flag(raw.get("is_special"))
    remark = (raw.get("special_remark") or "").strip()
    obj.is_special = 1 if is_special else 0
    if is_special:
        obj.special_remark = remark if remark else None
    elif remark:
        obj.special_remark = remark


def apply_special_to_item_from_samples(
    item: Dict[str, Any],
    *,
    source: str,
    protocol_sample_id: str,
    donation_sample_id: str,
    protocol_samples: Dict[str, Any],
    donation_samples: Dict[str, Any],
) -> None:
    sample = None
    if source == "research_donation":
        sample = donation_samples.get(donation_sample_id or "")
    else:
        sample = protocol_samples.get(protocol_sample_id or "")
    item.update(sample_special_to_dict(sample))