zkxw/backend/app/crud/protocol_donation_convert.py

"""细胞存储协议 ⇄ 科研捐赠 转移关联与编号/样本编码豁免校验"""
from __future__ import annotations

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

from sqlalchemy import and_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.models import BizCollectionProtocol, BizProtocolSample
from app.models.research_donation import (
    BizProtocolDonationConvert,
    BizResearchDonation,
    BizResearchDonationSample,
)


def _active_link_query(db: Session, tenant_no: str):
    return db.query(BizProtocolDonationConvert).filter(
        and_(
            BizProtocolDonationConvert.tenant_no == tenant_no,
            BizProtocolDonationConvert.del_flag == 0,
        )
    )


def get_convert_partner(
    db: Session,
    tenant_no: str,
    *,
    protocol_id: Optional[str] = None,
    donation_id: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
    """按采集协议或科研捐赠 ID 查询转移配对信息。"""
    if protocol_id:
        row = (
            _active_link_query(db, tenant_no)
            .filter(BizProtocolDonationConvert.protocol_id == protocol_id)
            .first()
        )
    elif donation_id:
        row = (
            _active_link_query(db, tenant_no)
            .filter(BizProtocolDonationConvert.research_donation_id == donation_id)
            .first()
        )
    else:
        return None
    if not row:
        return None
    return {
        "id": row.id,
        "convert_type": row.convert_type,
        "protocol_id": row.protocol_id,
        "research_donation_id": row.research_donation_id,
        "biz_no": row.biz_no or "",
    }


def has_active_convert_link(
    db: Session,
    tenant_no: str,
    *,
    protocol_id: Optional[str] = None,
    donation_id: Optional[str] = None,
) -> bool:
    return get_convert_partner(
        db, tenant_no, protocol_id=protocol_id, donation_id=donation_id
    ) is not None


def create_convert_link(
    db: Session,
    *,
    convert_type: str,
    protocol_id: str,
    research_donation_id: str,
    biz_no: str,
    tenant_no: str,
    user_id: Optional[str] = None,
    now: Optional[datetime] = None,
) -> BizProtocolDonationConvert:
    """转移事务内写入关联；同一协议或捐赠不得重复建立有效关联。"""
    now = now or datetime.now()
    pid = (protocol_id or "").strip()
    did = (research_donation_id or "").strip()
    if not pid or not did:
        raise ValueError("转移关联参数不完整")

    if (
        _active_link_query(db, tenant_no)
        .filter(BizProtocolDonationConvert.protocol_id == pid)
        .first()
    ):
        raise ValueError("该采集协议已存在转移关联")
    if (
        _active_link_query(db, tenant_no)
        .filter(BizProtocolDonationConvert.research_donation_id == did)
        .first()
    ):
        raise ValueError("该科研捐赠已存在转移关联")

    obj = BizProtocolDonationConvert(
        id=generate_uuid(),
        convert_type=(convert_type or "").strip(),
        protocol_id=pid,
        research_donation_id=did,
        biz_no=(biz_no or "").strip(),
        tenant_no=tenant_no,
        order_num=0,
        remark=None,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    db.flush()
    return obj


def get_partner_sample_codes(
    db: Session,
    tenant_no: str,
    *,
    protocol_id: Optional[str] = None,
    donation_id: Optional[str] = None,
) -> Set[str]:
    """返回配对单据上未删除样本的编码集合（用于样本唯一性豁免）。"""
    partner = get_convert_partner(
        db, tenant_no, protocol_id=protocol_id, donation_id=donation_id
    )
    if not partner:
        return set()
    codes: Set[str] = set()
    protocol_ids: List[str] = []
    donation_ids: List[str] = []
    if partner.get("protocol_id"):
        protocol_ids.append(partner["protocol_id"])
    if partner.get("research_donation_id"):
        donation_ids.append(partner["research_donation_id"])
        don = (
            db.query(BizResearchDonation)
            .filter(
                and_(
                    BizResearchDonation.id == partner["research_donation_id"],
                    BizResearchDonation.tenant_no == tenant_no,
                    BizResearchDonation.del_flag == 0,
                )
            )
            .first()
        )
        if don and (don.source_protocol_id or "").strip():
            spid = don.source_protocol_id.strip()
            if spid not in protocol_ids:
                protocol_ids.append(spid)
    for pid in protocol_ids:
        rows = (
            db.query(BizProtocolSample.sample_code)
            .filter(
                and_(
                    BizProtocolSample.collection_protocol_id == pid,
                    BizProtocolSample.tenant_no == tenant_no,
                    BizProtocolSample.del_flag == 0,
                )
            )
            .all()
        )
        codes.update(r[0] for r in rows if r[0])
    for did in donation_ids:
        rows = (
            db.query(BizResearchDonationSample.sample_code)
            .filter(
                and_(
                    BizResearchDonationSample.research_donation_id == did,
                    BizResearchDonationSample.tenant_no == tenant_no,
                    BizResearchDonationSample.del_flag == 0,
                )
            )
            .all()
        )
        codes.update(r[0] for r in rows if r[0])
    return codes


def assert_biz_no_available(
    db: Session,
    *,
    tenant_no: str,
    biz_no: str,
    exclude_protocol_id: Optional[str] = None,
    exclude_donation_id: Optional[str] = None,
    for_donation: bool = False,
) -> None:
    """
    校验业务编号在采集协议与科研捐赠间可用。
    若当前单存在转移配对，则豁免配对单据上的编号占用。
    """
    pn = (biz_no or "").strip()
    if not pn:
        raise ValueError("协议编号不能为空")

    partner = None
    exempt_protocol_id: Optional[str] = None
    exempt_donation_id: Optional[str] = None
    if exclude_protocol_id:
        partner = get_convert_partner(
            db, tenant_no, protocol_id=exclude_protocol_id
        )
        if partner:
            exempt_donation_id = partner["research_donation_id"]
    elif exclude_donation_id:
        partner = get_convert_partner(
            db, tenant_no, donation_id=exclude_donation_id
        )
        if partner:
            exempt_protocol_id = partner["protocol_id"]

    proto_rows = (
        db.query(BizCollectionProtocol)
        .filter(
            and_(
                BizCollectionProtocol.protocol_no == pn,
                BizCollectionProtocol.tenant_no == tenant_no,
                BizCollectionProtocol.del_flag == 0,
            )
        )
        .all()
    )
    for row in proto_rows:
        if exclude_protocol_id and row.id == exclude_protocol_id:
            continue
        if exempt_protocol_id and row.id == exempt_protocol_id:
            continue
        if for_donation:
            raise ValueError("捐赠协议号已被采集协议单使用")
        raise ValueError("协议编号已存在")

    donation_rows = (
        db.query(BizResearchDonation)
        .filter(
            and_(
                BizResearchDonation.donation_no == pn,
                BizResearchDonation.tenant_no == tenant_no,
                BizResearchDonation.del_flag == 0,
            )
        )
        .all()
    )
    for row in donation_rows:
        if exclude_donation_id and row.id == exclude_donation_id:
            continue
        if exempt_donation_id and row.id == exempt_donation_id:
            continue
        if for_donation:
            raise ValueError("捐赠协议号已存在")
        raise ValueError("协议编号已被捐赠协议单使用")


def upsert_convert_link_for_to_storage(
    db: Session,
    *,
    protocol_id: str,
    research_donation_id: str,
    biz_no: str,
    tenant_no: str,
    user_id: Optional[str] = None,
    now: Optional[datetime] = None,
) -> BizProtocolDonationConvert:
    """
    转细胞存储时写入/更新关联。
    若捐赠单已有关联（如先转科研），则更新 protocol_id 指向新协议单。
    """
    now = now or datetime.now()
    did = (research_donation_id or "").strip()
    pid = (protocol_id or "").strip()
    existing = (
        _active_link_query(db, tenant_no)
        .filter(BizProtocolDonationConvert.research_donation_id == did)
        .first()
    )
    if existing:
        if (
            _active_link_query(db, tenant_no)
            .filter(
                BizProtocolDonationConvert.protocol_id == pid,
                BizProtocolDonationConvert.id != existing.id,
            )
            .first()
        ):
            raise ValueError("该采集协议已存在转移关联")
        existing.protocol_id = pid
        existing.convert_type = "research_to_storage"
        existing.update_by = user_id
        existing.update_time = now
        db.add(existing)
        db.flush()
        return existing
    return create_convert_link(
        db,
        convert_type="research_to_storage",
        protocol_id=pid,
        research_donation_id=did,
        biz_no=biz_no,
        tenant_no=tenant_no,
        user_id=user_id,
        now=now,
    )