zkxw/backend/app/services/lab_sample_processing.py

"""实验室样本处理数据 — 聚合市场/生产/质检已发生业务数据与部门字段回写。

Excel 仅作为字段目录示例；列表以协议样本为主表，关联样本接收与生产接收，
补充表只承载源系统没有的列，且不得覆盖源表字段。
"""
from __future__ import annotations

import json
import uuid
from datetime import datetime
from decimal import Decimal
from typing import Any, Dict, List, Optional, Tuple

from types import SimpleNamespace

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

from app.models.models import (
    BizCellSampleType,
    BizCollectionHospital,
    BizProtocolCollectionInfo,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizSampleProductionReception,
    BizSampleReception,
)
from app.services.lab_sample_processing_fields import (
    DATASET_HSC,
    DATASET_MSC,
    DATASET_UBMC,
    DEPT_EDIT_PERMS,
    PERM_LIST,
    PERM_MANAGER,
    SOURCE_EXTRA,
    SOURCE_PRODUCTION,
    SOURCE_PROTOCOL,
    SOURCE_RECEPTION,
    editable_depts_for_user,
    field_map,
    fields_for,
    resolve_dataset_by_type_name,
)


def _new_id() -> str:
    return uuid.uuid4().hex.upper()


def _fmt_dt(v: Any) -> Optional[str]:
    if v is None:
        return None
    if isinstance(v, datetime):
        return v.strftime("%Y-%m-%d %H:%M:%S")
    return str(v)


def _fmt_num(v: Any) -> Any:
    if v is None:
        return None
    if isinstance(v, Decimal):
        return float(v)
    return v


def _parse_extra(raw: Optional[str]) -> Dict[str, Any]:
    if not raw:
        return {}
    try:
        data = json.loads(raw)
    except Exception:
        return {}
    return data if isinstance(data, dict) else {}


def user_acl(db: Session, user: dict) -> Dict[str, Any]:
    from app.core.menu_guard import user_has_menu_perm

    is_manager = user_has_menu_perm(db, user, PERM_MANAGER)
    dept_flags = {
        dept: user_has_menu_perm(db, user, code) for dept, code in DEPT_EDIT_PERMS.items()
    }
    editable = editable_depts_for_user(is_manager=is_manager, dept_flags=dept_flags)
    return {
        "canList": user_has_menu_perm(db, user, PERM_LIST) or is_manager or any(dept_flags.values()),
        "isManager": is_manager,
        "editableDepts": sorted(editable),
        "deptFlags": dept_flags,
    }


def _type_name_filters(dataset: str):
    if dataset == DATASET_HSC:
        return "%造血%"
    if dataset == DATASET_MSC:
        return "%间充质%"
    if dataset == DATASET_UBMC:
        return "%单个核%"
    return None


def _code_prefix_filters(dataset: str) -> List[str]:
    if dataset == DATASET_HSC:
        return ["B%"]
    if dataset == DATASET_MSC:
        return ["BC%", "UC%"]
    if dataset == DATASET_UBMC:
        return ["UBMC%"]
    return []


def _match_keyword(row: Dict[str, Any], kw: str) -> bool:
    if not kw:
        return True
    text = f"{row.get('sampleCode') or ''} {row.get('motherName') or ''}"
    return kw.lower() in text.lower()


def _match_date_range(row: Dict[str, Any], date_from: Optional[str], date_to: Optional[str]) -> bool:
    if not date_from and not date_to:
        return True
    for key in ("receptionTime", "collectTime", "cryoTime", "storageDateP2"):
        val = row.get(key)
        if not val:
            continue
        day = str(val)[:10].replace(".", "-").replace("/", "-")
        # normalize compact dates like 2025-9-8
        parts = day.split("-")
        if len(parts) == 3:
            try:
                day = f"{int(parts[0]):04d}-{int(parts[1]):02d}-{int(parts[2]):02d}"
            except Exception:
                pass
        if date_from and day < date_from:
            continue
        if date_to and day > date_to:
            continue
        return True
    return False


def list_rows(
    db: Session,
    *,
    tenant_no: str,
    dataset: str,
    keyword: Optional[str] = None,
    date_from: Optional[str] = None,
    date_to: Optional[str] = None,
    page: int = 1,
    page_size: int = 20,
) -> Tuple[List[Dict[str, Any]], int]:
    from app.models.models import BizLabSampleProcessingExtra

    type_like = _type_name_filters(dataset)
    code_likes = _code_prefix_filters(dataset)

    q = (
        db.query(
            BizProtocolSample,
            BizCellSampleType,
            BizProtocolPartyCustomer,
            BizProtocolCollectionInfo,
            BizCollectionHospital,
        )
        .outerjoin(
            BizCellSampleType,
            (BizCellSampleType.id == BizProtocolSample.sample_type_id)
            & (BizCellSampleType.del_flag == 0),
        )
        .outerjoin(
            BizProtocolPartyCustomer,
            (BizProtocolPartyCustomer.collection_protocol_id == BizProtocolSample.collection_protocol_id)
            & (BizProtocolPartyCustomer.tenant_no == tenant_no)
            & (BizProtocolPartyCustomer.del_flag == 0),
        )
        .outerjoin(
            BizProtocolCollectionInfo,
            (BizProtocolCollectionInfo.collection_protocol_id == BizProtocolSample.collection_protocol_id)
            & (BizProtocolCollectionInfo.tenant_no == tenant_no)
            & (BizProtocolCollectionInfo.del_flag == 0),
        )
        .outerjoin(
            BizCollectionHospital,
            (BizCollectionHospital.id == BizProtocolCollectionInfo.hospital_id)
            & (BizCollectionHospital.del_flag == 0),
        )
        .filter(
            BizProtocolSample.tenant_no == tenant_no,
            BizProtocolSample.del_flag == 0,
            BizProtocolSample.sample_code.isnot(None),
            BizProtocolSample.sample_code != "",
        )
    )

    type_or_code = []
    if type_like:
        type_or_code.append(BizCellSampleType.type_name.like(type_like))
    for pref in code_likes:
        type_or_code.append(BizProtocolSample.sample_code.like(pref))
    if type_or_code:
        q = q.filter(or_(*type_or_code))

    rows = q.order_by(BizProtocolSample.create_time.desc()).all()
    filtered = []
    for sample, stype, party, coll, hospital in rows:
        type_name = (stype.type_name if stype else "") or ""
        code = sample.sample_code or ""
        if resolve_dataset_by_type_name(type_name, code) != dataset:
            continue
        filtered.append((sample, stype, party, coll, hospital))

    extras = {
        e.sample_code: e
        for e in db.query(BizLabSampleProcessingExtra)
        .filter(
            BizLabSampleProcessingExtra.tenant_no == tenant_no,
            BizLabSampleProcessingExtra.del_flag == 0,
            BizLabSampleProcessingExtra.dataset == dataset,
        )
        .all()
    }

    # 仅聚合真实协议样本；补充表只挂到已有样本上，不再单独展示 Excel 演示行
    codes = [s.sample_code for s, *_ in filtered if s.sample_code]
    sample_ids = [s.id for s, *_ in filtered if s.id]
    receptions_by_sample_id: Dict[str, BizSampleReception] = {}
    receptions_by_code: Dict[str, BizSampleReception] = {}
    productions: Dict[str, BizSampleProductionReception] = {}
    if sample_ids or codes:
        recv_filters = []
        if sample_ids:
            recv_filters.append(BizSampleReception.protocol_sample_id.in_(sample_ids))
        if codes:
            recv_filters.append(BizSampleReception.sample_code.in_(codes))
        for r in (
            db.query(BizSampleReception)
            .filter(
                BizSampleReception.tenant_no == tenant_no,
                BizSampleReception.del_flag == 0,
                or_(*recv_filters) if recv_filters else False,
            )
            .order_by(BizSampleReception.reception_time.desc())
            .all()
        ):
            if r.protocol_sample_id:
                receptions_by_sample_id.setdefault(r.protocol_sample_id, r)
            if r.sample_code:
                receptions_by_code.setdefault(r.sample_code, r)
        if receptions_by_sample_id or receptions_by_code:
            rids = list(
                {
                    *(r.id for r in receptions_by_sample_id.values()),
                    *(r.id for r in receptions_by_code.values()),
                }
            )
            for p in (
                db.query(BizSampleProductionReception)
                .filter(
                    BizSampleProductionReception.tenant_no == tenant_no,
                    BizSampleProductionReception.del_flag == 0,
                    BizSampleProductionReception.reception_id.in_(rids),
                )
                .order_by(BizSampleProductionReception.process_time.desc())
                .all()
            ):
                productions.setdefault(p.reception_id, p)

    built_map: Dict[str, Dict[str, Any]] = {}
    for sample, stype, party, coll, hospital in filtered:
        code = sample.sample_code
        reception = receptions_by_sample_id.get(sample.id) or receptions_by_code.get(code)
        production = productions.get(reception.id) if reception else None
        extra = _parse_extra(extras[code].payload_json if code in extras else None)
        built_map[code] = _build_row(
            dataset=dataset,
            sample=sample,
            type_name=(stype.type_name if stype else None),
            party=party,
            coll=coll,
            hospital=hospital,
            reception=reception,
            production=production,
            extra=extra,
        )

    # 单个核：业务已发生数据主要在科研捐献接收（lab_storage_code=UBMC*），无细胞储存协议样本时仍需展示
    if dataset == DATASET_UBMC:
        ubmc_recs = (
            db.query(BizSampleReception)
            .filter(
                BizSampleReception.tenant_no == tenant_no,
                BizSampleReception.del_flag == 0,
                BizSampleReception.lab_storage_code.like("UBMC%"),
            )
            .order_by(BizSampleReception.reception_time.desc())
            .all()
        )
        mother_by_donation: Dict[str, str] = {}
        donation_ids = sorted(
            {r.research_donation_id for r in ubmc_recs if r.research_donation_id}
        )
        if donation_ids:
            rows = db.execute(
                text(
                    "SELECT research_donation_id, mother_name FROM biz_research_donation_party "
                    "WHERE del_flag=0 AND tenant_no=:tenant_no AND research_donation_id IN ("
                    + ",".join(f":d{i}" for i in range(len(donation_ids)))
                    + ")"
                ),
                {
                    "tenant_no": tenant_no,
                    **{f"d{i}": did for i, did in enumerate(donation_ids)},
                },
            ).fetchall()
            for did, mother in rows:
                if did and mother and did not in mother_by_donation:
                    mother_by_donation[str(did)] = mother

        missing_rids = [
            r.id
            for r in ubmc_recs
            if r.id and r.id not in productions
        ]
        if missing_rids:
            for p in (
                db.query(BizSampleProductionReception)
                .filter(
                    BizSampleProductionReception.tenant_no == tenant_no,
                    BizSampleProductionReception.del_flag == 0,
                    BizSampleProductionReception.reception_id.in_(missing_rids),
                )
                .order_by(BizSampleProductionReception.process_time.desc())
                .all()
            ):
                productions.setdefault(p.reception_id, p)

        for reception in ubmc_recs:
            code = (reception.lab_storage_code or reception.sample_code or "").strip()
            if not code or code in built_map:
                continue
            production = productions.get(reception.id)
            extra = _parse_extra(extras[code].payload_json if code in extras else None)
            mother = mother_by_donation.get(str(reception.research_donation_id or ""))
            fake_sample = SimpleNamespace(
                id=reception.donation_sample_id or reception.id,
                sample_code=code,
                collection_protocol_id=None,
            )
            fake_party = SimpleNamespace(mother_name=mother) if mother else None
            built_map[code] = _build_row(
                dataset=dataset,
                sample=fake_sample,
                type_name="脐血单个核细胞",
                party=fake_party,
                coll=None,
                hospital=None,
                reception=reception,
                production=production,
                extra=extra,
            )

    kw = (keyword or "").strip()
    built = []
    for row in built_map.values():
        if not _match_keyword(row, kw):
            continue
        if not _match_date_range(row, date_from, date_to):
            continue
        built.append(row)

    # 稳定排序：优先按样本编码
    built.sort(key=lambda r: str(r.get("sampleCode") or ""))

    total = len(built)
    page = max(1, int(page or 1))
    page_size = max(1, min(5000, int(page_size or 20)))
    start = (page - 1) * page_size
    return built[start : start + page_size], total


def _build_row(
    *,
    dataset: str,
    sample: Optional[BizProtocolSample],
    type_name: Optional[str],
    party: Optional[BizProtocolPartyCustomer],
    coll: Optional[BizProtocolCollectionInfo],
    hospital: Optional[BizCollectionHospital],
    reception: Optional[BizSampleReception],
    production: Optional[BizSampleProductionReception],
    extra: Dict[str, Any],
) -> Dict[str, Any]:
    protocol_vals = {
        "sample_code": sample.sample_code if sample else extra.get("sampleCode"),
        "mother_name": party.mother_name if party else None,
        "sample_collect_time": coll.sample_collect_time if coll else None,
        "hospital_name": hospital.hospital_name if hospital else None,
    }
    reception_vals = {}
    if reception:
        reception_vals = {
            "reception_time": reception.reception_time,
            "receive_temperature": reception.receive_temperature,
            "handler_name": reception.handler_name,
            "remark": reception.remark,
            "lab_storage_code": reception.lab_storage_code,
        }
    production_vals = {}
    if production:
        production_vals = {
            "original_blood_weight_g": production.original_blood_weight_g,
            "hes_volume_ml": production.hes_volume_ml,
            "blood_type": production.blood_type,
            "infectious_four_test_result": production.infectious_four_test_result,
            "enriched_blood_weight_g": production.enriched_blood_weight_g,
            "cryopreservation_volume_ml": production.cryopreservation_volume_ml,
            "handler_name": production.handler_name,
            "remark": production.remark,
            "original_blood_volume_ml": production.original_blood_volume_ml,
            "nucleated_cell_count_1e8": production.nucleated_cell_count_1e8,
            "inoculation_amount": production.inoculation_amount,
            "cryopreservation_time": production.cryopreservation_time,
            "storage_total_tubes": production.storage_total_tubes,
            "available_cell_count_1e7": production.available_cell_count_1e7,
        }

    sample_code = (
        (sample.sample_code if sample else None)
        or extra.get("sampleCode")
        or ""
    )
    data: Dict[str, Any] = {
        "id": (sample.id if sample else f"extra:{dataset}:{sample_code}"),
        "dataset": dataset,
        "sampleCode": sample_code,
        "sampleTypeName": type_name,
        "protocolSampleId": sample.id if sample else None,
        "collectionProtocolId": sample.collection_protocol_id if sample else None,
        "receptionId": reception.id if reception else None,
        "productionReceptionId": production.id if production else None,
        "fromExtraOnly": sample is None,
    }
    for f in fields_for(dataset):
        val = None
        if f.source == SOURCE_PROTOCOL:
            val = protocol_vals.get(f.attr)
        elif f.source == SOURCE_RECEPTION:
            val = reception_vals.get(f.attr)
        elif f.source == SOURCE_PRODUCTION:
            val = production_vals.get(f.attr)
        elif f.source == SOURCE_EXTRA:
            # 仅源系统没有的列走补充表；业务源表字段不被补充表覆盖
            val = extra.get(f.key)

        if isinstance(val, datetime):
            val = _fmt_dt(val)
        elif isinstance(val, Decimal):
            val = _fmt_num(val)
        data[f.key] = val
    if not data.get("sampleCode"):
        data["sampleCode"] = sample_code
    return data


def upsert_extra_payload(
    db: Session,
    *,
    tenant_no: str,
    dataset: str,
    sample_code: str,
    payload: Dict[str, Any],
    operator_id: str = "import",
    replace: bool = False,
) -> None:
    from app.models.models import BizLabSampleProcessingExtra

    code = (sample_code or "").strip()
    if not code:
        raise ValueError("样本编码不能为空")
    now = datetime.now()
    row = (
        db.query(BizLabSampleProcessingExtra)
        .filter(
            BizLabSampleProcessingExtra.tenant_no == tenant_no,
            BizLabSampleProcessingExtra.dataset == dataset,
            BizLabSampleProcessingExtra.sample_code == code,
            BizLabSampleProcessingExtra.del_flag == 0,
        )
        .first()
    )
    current = {} if replace else _parse_extra(row.payload_json if row else None)
    for k, v in (payload or {}).items():
        if v in (None, ""):
            continue
        current[k] = v
    current["sampleCode"] = code
    if row is None:
        db.add(
            BizLabSampleProcessingExtra(
                id=_new_id(),
                dataset=dataset,
                sample_code=code,
                payload_json=json.dumps(current, ensure_ascii=False),
                tenant_no=tenant_no,
                create_by=operator_id,
                create_dept_id="",
                create_time=now,
                update_by=operator_id,
                update_time=now,
                del_flag=0,
            )
        )
    else:
        row.payload_json = json.dumps(current, ensure_ascii=False)
        row.update_by = operator_id
        row.update_time = now


def update_row(
    db: Session,
    *,
    tenant_no: str,
    dataset: str,
    sample_code: str,
    payload: Dict[str, Any],
    editable_depts: set,
    operator_id: str,
) -> Dict[str, Any]:
    code = (sample_code or "").strip()
    if not code:
        raise ValueError("样本编码不能为空")
    fmap = field_map(dataset)
    if not fmap:
        raise ValueError("未知数据集")

    denied = []
    accepted: Dict[str, Any] = {}
    for key, value in (payload or {}).items():
        f = fmap.get(key)
        if not f:
            continue
        if f.dept not in editable_depts:
            denied.append(key)
            continue
        accepted[key] = value
    if denied:
        raise PermissionError(f"无权编辑字段: {', '.join(denied)}")
    if not accepted:
        raise ValueError("没有可保存的字段")

    sample = (
        db.query(BizProtocolSample)
        .filter(
            BizProtocolSample.tenant_no == tenant_no,
            BizProtocolSample.sample_code == code,
            BizProtocolSample.del_flag == 0,
        )
        .first()
    )
    reception = (
        db.query(BizSampleReception)
        .filter(
            BizSampleReception.tenant_no == tenant_no,
            BizSampleReception.sample_code == code,
            BizSampleReception.del_flag == 0,
        )
        .order_by(BizSampleReception.reception_time.desc())
        .first()
    )
    production = None
    if reception:
        production = (
            db.query(BizSampleProductionReception)
            .filter(
                BizSampleProductionReception.tenant_no == tenant_no,
                BizSampleProductionReception.reception_id == reception.id,
                BizSampleProductionReception.del_flag == 0,
            )
            .order_by(BizSampleProductionReception.process_time.desc())
            .first()
        )

    party = None
    coll = None
    if sample:
        party = (
            db.query(BizProtocolPartyCustomer)
            .filter(
                BizProtocolPartyCustomer.collection_protocol_id == sample.collection_protocol_id,
                BizProtocolPartyCustomer.tenant_no == tenant_no,
                BizProtocolPartyCustomer.del_flag == 0,
            )
            .first()
        )
        coll = (
            db.query(BizProtocolCollectionInfo)
            .filter(
                BizProtocolCollectionInfo.collection_protocol_id == sample.collection_protocol_id,
                BizProtocolCollectionInfo.tenant_no == tenant_no,
                BizProtocolCollectionInfo.del_flag == 0,
            )
            .first()
        )

    now = datetime.now()
    extra_payload: Dict[str, Any] = {}

    def _set_dt(obj, attr, value):
        if value in (None, ""):
            setattr(obj, attr, None)
            return
        if isinstance(value, datetime):
            setattr(obj, attr, value)
            return
        text = str(value).strip().replace("/", "-").replace(".", "-")
        for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
            try:
                setattr(obj, attr, datetime.strptime(text, fmt))
                return
            except ValueError:
                continue
        setattr(obj, attr, value)

    for key, value in accepted.items():
        f = fmap[key]
        wrote = False
        if f.source == SOURCE_PROTOCOL and sample is not None:
            if f.attr == "mother_name" and party is not None:
                party.mother_name = None if value in (None, "") else str(value)
                party.update_by = operator_id
                party.update_time = now
                wrote = True
            elif f.attr == "sample_collect_time" and coll is not None:
                _set_dt(coll, "sample_collect_time", value)
                coll.update_by = operator_id
                coll.update_time = now
                wrote = True
        elif f.source == SOURCE_RECEPTION and reception is not None:
            if f.attr == "reception_time":
                _set_dt(reception, f.attr, value)
            elif f.attr in {"receive_temperature", "handler_name", "remark", "lab_storage_code"}:
                setattr(reception, f.attr, None if value in (None, "") else value)
            reception.update_by = operator_id
            reception.update_time = now
            wrote = True
        elif f.source == SOURCE_PRODUCTION and production is not None:
            numeric_attrs = {
                "original_blood_weight_g",
                "hes_volume_ml",
                "enriched_blood_weight_g",
                "cryopreservation_volume_ml",
                "original_blood_volume_ml",
                "nucleated_cell_count_1e8",
                "available_cell_count_1e7",
            }
            if f.attr == "cryopreservation_time":
                _set_dt(production, f.attr, value)
            elif f.attr == "storage_total_tubes":
                production.storage_total_tubes = None if value in (None, "") else int(value)
            elif f.attr in numeric_attrs:
                production_val = None if value in (None, "") else Decimal(str(value))
                setattr(production, f.attr, production_val)
            elif f.attr in {
                "blood_type",
                "infectious_four_test_result",
                "handler_name",
                "remark",
                "inoculation_amount",
            }:
                setattr(production, f.attr, None if value in (None, "") else value)
            production.update_by = operator_id
            production.update_time = now
            wrote = True

        # 补充表始终保留一份，保证导入台账与无权源表场景可编辑
        extra_payload[key] = value
        _ = wrote

    upsert_extra_payload(
        db,
        tenant_no=tenant_no,
        dataset=dataset,
        sample_code=code,
        payload=extra_payload,
        operator_id=operator_id,
        replace=False,
    )
    db.commit()

    rows, _ = list_rows(db, tenant_no=tenant_no, dataset=dataset, keyword=code, page=1, page_size=20)
    for row in rows:
        if row.get("sampleCode") == code:
            return row
    raise LookupError("保存成功但读取失败")


def export_rows(
    db: Session,
    *,
    tenant_no: str,
    dataset: str,
    keyword: Optional[str] = None,
    date_from: Optional[str] = None,
    date_to: Optional[str] = None,
) -> Tuple[List[str], List[List[Any]]]:
    rows, _ = list_rows(
        db,
        tenant_no=tenant_no,
        dataset=dataset,
        keyword=keyword,
        date_from=date_from,
        date_to=date_to,
        page=1,
        page_size=5000,
    )
    headers = [f.label for f in fields_for(dataset)]
    keys = [f.key for f in fields_for(dataset)]
    body = [[row.get(k) for k in keys] for row in rows]
    return headers, body