zkxw/backend/app/crud/research_donation_sample_collection.py

"""科研捐赠样本采集信息列表 CRUD（样本信息管理 · 样本采集信息）"""
from __future__ import annotations

import math
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import and_, or_
from sqlalchemy.orm import Session

from app.core.utils import format_date
from app.utils.sample_special import sample_special_to_dict
from app.models.models import (
    BizCellSampleType,
    BizCollectionHospital,
    BizCollectionRegion,
)
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationCollection,
    BizResearchDonationParty,
    BizResearchDonationSample,
)


def total_pages(total: int, page_size: int) -> int:
    if total <= 0:
        return 0
    return int(math.ceil(total / float(page_size)))


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


def _sample_status_label(status: str) -> str:
    return {
        "ycj": "已采集",
        "jcz": "检测中",
        "yrk": "已入库",
        "yzf": "已作废",
    }.get(status, "未采集" if not status else status)


def _row_to_dict(
    *,
    donation_sample_id: str,
    research_donation_id: str,
    donation_no: str,
    donation_status: str,
    expected_delivery_date: str,
    mother_name: str,
    newborn_name: str,
    sample_code: str,
    sample_type_name: str,
    sample_status: str,
    sample_status_label: str,
    region_name: str,
    hospital_name: str,
    medical_staff: str,
    sample_collect_time: str,
    maternal_blood_collect_time: str,
) -> Dict[str, Any]:
    return {
        "donationSampleId": donation_sample_id,
        "researchDonationId": research_donation_id,
        "donationNo": donation_no,
        "donationStatus": donation_status,
        "expectedDeliveryDate": expected_delivery_date,
        "motherName": mother_name,
        "newbornName": newborn_name,
        "sampleCode": sample_code,
        "sampleTypeName": sample_type_name,
        "sampleStatus": sample_status,
        "sampleStatusLabel": sample_status_label,
        "regionName": region_name,
        "hospitalName": hospital_name,
        "medicalStaff": medical_staff,
        "sampleCollectTime": sample_collect_time,
        "maternalBloodCollectTime": maternal_blood_collect_time,
    }


def _base_query(db: Session, tenant_no: str):
    return (
        db.query(
            BizResearchDonationSample,
            BizResearchDonation,
            BizResearchDonationParty,
            BizResearchDonationCollection,
        )
        .join(
            BizResearchDonation,
            and_(
                BizResearchDonation.id == BizResearchDonationSample.research_donation_id,
                BizResearchDonation.del_flag == 0,
            ),
        )
        .outerjoin(
            BizResearchDonationParty,
            and_(
                BizResearchDonationParty.research_donation_id == BizResearchDonation.id,
                BizResearchDonationParty.del_flag == 0,
            ),
        )
        .outerjoin(
            BizResearchDonationCollection,
            and_(
                BizResearchDonationCollection.research_donation_id == BizResearchDonation.id,
                BizResearchDonationCollection.del_flag == 0,
            ),
        )
        .filter(
            BizResearchDonationSample.tenant_no == tenant_no,
            BizResearchDonationSample.del_flag == 0,
        )
    )


def _apply_sample_status_filter(q, sample_status: Optional[str]):
    st = (sample_status or "").strip()
    if not st or st == "all":
        return q
    if st == "uncollected":
        return q.filter(
            or_(
                BizResearchDonationSample.sample_status.is_(None),
                BizResearchDonationSample.sample_status == "",
            )
        )
    if st == "collected":
        return q.filter(BizResearchDonationSample.sample_status.in_(["ycj", "jcz", "yrk"]))
    if st == "voided":
        return q.filter(BizResearchDonationSample.sample_status == "yzf")
    return q.filter(BizResearchDonationSample.sample_status == st)


def list_donation_sample_collections(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    tenant_no: str,
    keyword: Optional[str] = None,
    sample_status: Optional[str] = None,
    collection_flag: Optional[str] = None,
    reception_flag: Optional[str] = None,
    qc_flag: Optional[str] = None,
    storage_flag: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0

    from app.crud.protocol_sample_collection import (
        _apply_qc_flag_filter,
        _apply_reception_flag_filter,
        _apply_storage_flag_filter,
    )

    q = _base_query(db, tenant_no)
    q = _apply_sample_status_filter(q, collection_flag or sample_status)
    q = _apply_reception_flag_filter(
        q,
        reception_flag,
        sample_code_col=BizResearchDonationSample.sample_code,
        tenant_no=tenant_no,
        source="research_donation",
    )
    q = _apply_qc_flag_filter(
        q,
        qc_flag,
        sample_code_col=BizResearchDonationSample.sample_code,
        tenant_no=tenant_no,
        source="research_donation",
    )
    q = _apply_storage_flag_filter(
        q,
        storage_flag,
        sample_code_col=BizResearchDonationSample.sample_code,
        tenant_no=tenant_no,
        source="research_donation",
    )

    if keyword and keyword.strip():
        kw = f"%{keyword.strip()}%"
        kw_conds = [
            BizResearchDonationParty.mother_name.like(kw),
            BizResearchDonationSample.sample_code.like(kw),
        ]
        from app.crud import sample_reception as reception_svc

        lab_codes = reception_svc.sample_codes_matching_lab_storage_keyword(
            db, tenant_no, keyword, reception_source="research_donation"
        )
        if lab_codes:
            kw_conds.append(BizResearchDonationSample.sample_code.in_(lab_codes))
        q = q.filter(or_(*kw_conds))

    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizResearchDonation, user_info, db, tenant_no)

    total = q.count()
    page = max(1, page)
    page_size = min(max(1, page_size), 100)

    rows = (
        q.order_by(
            BizResearchDonationParty.expected_delivery_date.is_(None),
            BizResearchDonationParty.expected_delivery_date.desc(),
            BizResearchDonation.create_time.desc(),
            BizResearchDonationSample.id.desc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )

    sample_type_ids = list({
        ds.sample_type_id for ds, _, _, _ in rows if ds and ds.sample_type_id
    })
    type_map: Dict[str, str] = {}
    if sample_type_ids:
        for t in (
            db.query(BizCellSampleType)
            .filter(BizCellSampleType.id.in_(sample_type_ids), BizCellSampleType.del_flag == 0)
            .all()
        ):
            type_map[t.id] = t.type_name or ""

    region_ids = list({
        (ci.region_id or "").strip()
        for _, _, _, ci in rows
        if ci and (ci.region_id or "").strip()
    })
    hospital_ids = list({
        (ci.hospital_id or "").strip()
        for _, _, _, ci in rows
        if ci and (ci.hospital_id or "").strip()
    })
    region_name_map: Dict[str, str] = {}
    hospital_name_map: Dict[str, str] = {}
    if region_ids:
        for r in (
            db.query(BizCollectionRegion.id, BizCollectionRegion.region_name)
            .filter(BizCollectionRegion.id.in_(region_ids), BizCollectionRegion.del_flag == 0)
            .all()
        ):
            region_name_map[r.id] = r.region_name or ""
    if hospital_ids:
        for h in (
            db.query(BizCollectionHospital.id, BizCollectionHospital.hospital_name)
            .filter(BizCollectionHospital.id.in_(hospital_ids), BizCollectionHospital.del_flag == 0)
            .all()
        ):
            hospital_name_map[h.id] = h.hospital_name or ""

    items: List[Dict[str, Any]] = []
    for dsample, donation, party, ci in rows:
        status = (dsample.sample_status or "").strip() if dsample else ""
        region_id = (ci.region_id or "").strip() if ci else ""
        hospital_id = (ci.hospital_id or "").strip() if ci else ""
        item = _row_to_dict(
                donation_sample_id=dsample.id,
                research_donation_id=donation.id if donation else "",
                donation_no=(donation.donation_no or "") if donation else "",
                donation_status=(donation.donation_status or "") if donation else "",
                expected_delivery_date=format_date(party.expected_delivery_date) if party else "",
                mother_name=(party.mother_name or "") if party else "",
                newborn_name=(party.newborn_name or "") if party else "",
                sample_code=dsample.sample_code or "",
                sample_type_name=type_map.get(dsample.sample_type_id or "", ""),
                sample_status=status,
                sample_status_label=_sample_status_label(status),
                region_name=region_name_map.get(region_id, ""),
                hospital_name=hospital_name_map.get(hospital_id, ""),
                medical_staff=(ci.medical_staff or "") if ci else "",
                sample_collect_time=_fmt_dt(ci.sample_collect_time) if ci else "",
                maternal_blood_collect_time=_fmt_dt(ci.maternal_blood_collect_time) if ci else "",
            )
        item.update(sample_special_to_dict(dsample))
        items.append(item)
    if items and tenant_no:
        from app.crud import sample_reception as reception_svc

        codes = [i["sampleCode"] for i in items if i.get("sampleCode")]
        code_map = reception_svc.get_lab_storage_code_map_by_sample_codes(
            db, tenant_no, codes, reception_source="research_donation"
        )
        for item in items:
            sc = (item.get("sampleCode") or "").strip()
            item["labStorageCode"] = code_map.get(sc, "") if sc else ""
        from app.services.sample_info_list_enrich import enrich_sample_info_items

        enrich_sample_info_items(
            db,
            tenant_no,
            items,
            reception_source="research_donation",
            include_handle_flags=False,
        )
    return items, total