zkxw/backend/app/services/qc_work_order_fee_summary.py

"""质检工单 · 用户费用摘要（采集协议 / 基础实验）"""
from __future__ import annotations

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

from sqlalchemy.orm import Session

from app.models.clinical_payment import BizPilotPayment
from app.models.models import (
    BizPilotInfusionSchedule,
    BizProtocolPayment,
    BizProtocolPaymentDetail,
    BizProtocolPaymentSchedule,
    BizProtocolPaymentTransfer,
    BizProtocolSamplePayment,
)

_PAY_MODE_LABELS = {"dj": "趸交", "fq": "分期", "djfq": "趸交+分期"}
_PAY_STATUS_LABELS = {
    "dfj": "已付款",
    "jfqrz": "付款确认中",
    "yjf": "付款完成",
    "tfqrz": "退费确认中",
    "ytf": "已退费",
}

# 出库工单：输注计划缴费（基础实验 / 临床应用共用 biz_pilot_payment）
_OUTBOUND_FEE_BIZ_TYPES = frozenset({"basic_experiment", "clinical_application"})


def _is_outbound_fee_biz_type(biz_type: Optional[str]) -> bool:
    return (biz_type or "").strip() in _OUTBOUND_FEE_BIZ_TYPES


def _dec(val: Any) -> Optional[float]:
    if val is None:
        return None
    try:
        return float(val)
    except (TypeError, ValueError):
        return None


def _received_amount_from_details(details: List[BizProtocolPaymentDetail]) -> float:
    total = Decimal("0")
    for d in details:
        st = (d.pay_status or "").strip()
        if st in ("yjf", "jfqrz"):
            total += d.transfer_amount or Decimal("0")
    return float(total)


def _protocol_status_hint(pay_mode: str, statuses: List[str]) -> str:
    counts: Dict[str, int] = {}
    for st in statuses:
        if st:
            counts[st] = counts.get(st, 0) + 1
    paid = counts.get("yjf", 0)
    confirm = counts.get("jfqrz", 0)
    pending = counts.get("dfj", 0)
    if pay_mode in ("fq", "djfq"):
        parts: List[str] = ["分期"]
        if paid:
            parts.append(f"{paid} 期已缴")
        if confirm:
            parts.append(f"{confirm} 期待确认")
        if pending:
            parts.append(f"{pending} 期待缴费")
        return " · ".join(parts) if len(parts) > 1 else parts[0]
    if confirm:
        return f"{confirm} 期待确认" if confirm > 1 else "收款确认中"
    if pending:
        return "待缴费"
    return ""


def _protocol_overall_status(
    statuses: List[str], contract: Optional[float], pending: Optional[float]
) -> Dict[str, str]:
    if "tfqrz" in statuses:
        return {"overallPayStatus": "tfqrz", "overallPayStatusLabel": "退费确认中", "overallPayStatusTagType": "warning"}
    if statuses and all(s == "ytf" for s in statuses):
        return {"overallPayStatus": "ytf", "overallPayStatusLabel": "已退费", "overallPayStatusTagType": "info"}
    if (
        contract is not None
        and pending is not None
        and pending <= 0
        and "dfj" not in statuses
        and "jfqrz" not in statuses
    ):
        return {"overallPayStatus": "cleared", "overallPayStatusLabel": "已缴清", "overallPayStatusTagType": "success"}
    if "jfqrz" in statuses:
        return {"overallPayStatus": "jfqrz", "overallPayStatusLabel": "收款确认中", "overallPayStatusTagType": "warning"}
    if "dfj" in statuses:
        return {"overallPayStatus": "dfj", "overallPayStatusLabel": "待缴费", "overallPayStatusTagType": "info"}
    return {"overallPayStatus": "", "overallPayStatusLabel": "—", "overallPayStatusTagType": "info"}


def _fmt_ymd(val: Any) -> str:
    s = (str(val or "")).strip()
    if len(s) == 8 and s.isdigit():
        return f"{s[:4]}-{s[4:6]}-{s[6:8]}"
    return s


def _fmt_dt(v):
    if v is None:
        return ""
    if isinstance(v, datetime):
        return v.strftime("%Y-%m-%d %H:%M:%S")
    s = str(v).strip()
    return s


def _payment_detail_row(d: BizProtocolPaymentDetail, pay_mode: str, idx: int) -> Dict[str, Any]:
    st = (d.pay_status or "").strip()
    inst_no = d.installment_no
    if pay_mode in ("fq", "djfq") and inst_no is not None:
        label = f"第{inst_no}期"
    else:
        label = f"缴费{idx + 1}"
    return {
        "installmentNo": inst_no,
        "installmentLabel": label,
        "installmentDueDate": _fmt_ymd(d.installment_due_date),
        "dueAmount": _dec(d.due_amount),
        "transferAmount": _dec(d.transfer_amount),
        "payStatus": st,
        "payStatusLabel": _PAY_STATUS_LABELS.get(st, st or "—"),
        "transferTime": _fmt_dt(getattr(d, "transfer_time", None)),
    }


def _sort_payment_details(details: List[BizProtocolPaymentDetail]) -> List[BizProtocolPaymentDetail]:
    return sorted(
        details,
        key=lambda d: (
            d.installment_no if d.installment_no is not None else 10_000,
            d.order_num if d.order_num is not None else 0,
            d.create_time or datetime.min,
        ),
    )


def build_protocol_payment_summary(
    pay: Optional[BizProtocolPayment],
    details: List[BizProtocolPaymentDetail],
) -> Optional[Dict[str, Any]]:
    if not pay:
        return None
    pay_mode = (pay.pay_mode or "dj").strip() or "dj"
    contract = _dec(pay.contract_amount)
    sorted_details = _sort_payment_details(details)
    detail_rows = [_payment_detail_row(d, pay_mode, i) for i, d in enumerate(sorted_details)]
    received = _received_amount_from_details(sorted_details)
    pending = max(contract - received, 0) if contract is not None else None
    statuses = [(d.pay_status or "").strip() for d in sorted_details]
    overall = _protocol_overall_status(statuses, contract, pending)
    hint = _protocol_status_hint(pay_mode, statuses) or overall["overallPayStatusLabel"]
    return {
        "payMode": pay_mode,
        "payModeLabel": _PAY_MODE_LABELS.get(pay_mode, pay_mode),
        "contractAmount": contract,
        "receivedAmount": received,
        "pendingAmount": pending,
        "statusHint": hint,
        "installmentCount": len(detail_rows) if pay_mode in ("fq", "djfq") else None,
        "paymentDetails": detail_rows,
        **overall,
    }


def pick_earliest_transfer_time_by_protocol(
    rows: Sequence[Tuple[Any, Any]],
) -> Dict[str, datetime]:
    """从 (protocol_id, transfer_time) 行中按协议取最早 transfer_time。"""
    earliest: Dict[str, datetime] = {}
    for pid, t in rows:
        if not pid or t is None:
            continue
        prev = earliest.get(pid)
        if prev is None or t < prev:
            earliest[pid] = t
    return earliest


def _first_period_first_transfer_times(
    db: Session,
    protocol_ids: List[str],
    tenant_no: str,
) -> Dict[str, str]:
    """协议 → 第 1 期（schedule_no=1）多笔转账中 transfer_time 最早的一笔。"""
    ids = [x.strip() for x in protocol_ids if (x or "").strip()]
    if not ids:
        return {}
    rows = (
        db.query(
            BizProtocolPaymentTransfer.collection_protocol_id,
            BizProtocolPaymentTransfer.transfer_time,
        )
        .join(
            BizProtocolPaymentSchedule,
            BizProtocolPaymentSchedule.id == BizProtocolPaymentTransfer.schedule_id,
        )
        .filter(
            BizProtocolPaymentTransfer.collection_protocol_id.in_(ids),
            BizProtocolPaymentTransfer.tenant_no == tenant_no,
            BizProtocolPaymentTransfer.del_flag == 0,
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
            BizProtocolPaymentSchedule.schedule_no == 1,
            BizProtocolPaymentTransfer.transfer_time.isnot(None),
        )
        .all()
    )
    earliest = pick_earliest_transfer_time_by_protocol(rows)
    return {pid: _fmt_dt(t) for pid, t in earliest.items() if _fmt_dt(t)}


def build_protocol_payment_summaries(
    db: Session,
    protocol_ids: List[str],
    tenant_no: str,
    *,
    include_sample_payment_modes: bool = False,
) -> Dict[str, Dict[str, Any]]:
    ids = [x.strip() for x in protocol_ids if (x or "").strip()]
    if not ids:
        return {}
    payments = (
        db.query(BizProtocolPayment)
        .filter(
            BizProtocolPayment.collection_protocol_id.in_(ids),
            BizProtocolPayment.tenant_no == tenant_no,
            BizProtocolPayment.del_flag == 0,
        )
        .all()
    )
    pay_by_proto = {p.collection_protocol_id: p for p in payments}
    payment_ids = [p.id for p in payments]
    details_by_payment: Dict[str, List[BizProtocolPaymentDetail]] = {}
    if payment_ids:
        for d in (
            db.query(BizProtocolPaymentDetail)
            .filter(
                BizProtocolPaymentDetail.payment_id.in_(payment_ids),
                BizProtocolPaymentDetail.tenant_no == tenant_no,
                BizProtocolPaymentDetail.del_flag == 0,
            )
            .all()
        ):
            details_by_payment.setdefault(d.payment_id, []).append(d)
    sample_modes: Dict[str, set] = {}
    if include_sample_payment_modes:
        for pid, mode in db.query(
            BizProtocolSamplePayment.collection_protocol_id,
            BizProtocolSamplePayment.pay_mode,
        ).filter(
            BizProtocolSamplePayment.collection_protocol_id.in_(ids),
            BizProtocolSamplePayment.tenant_no == tenant_no,
            BizProtocolSamplePayment.del_flag == 0,
        ).all():
            if mode:
                sample_modes.setdefault(pid, set()).add(mode)
    first_pay_map = _first_period_first_transfer_times(db, ids, tenant_no)
    out: Dict[str, Dict[str, Any]] = {}
    for pid in ids:
        pay = pay_by_proto.get(pid)
        summary = (
            build_protocol_payment_summary(pay, details_by_payment.get(pay.id, []))
            if pay
            else None
        )
        first_pay = first_pay_map.get(pid) or ""
        modes = sample_modes.get(pid, set())
        if not summary and not first_pay and not modes:
            continue
        if not summary:
            summary = {
                "payMode": "",
                "payModeLabel": "",
                "contractAmount": None,
                "receivedAmount": None,
                "pendingAmount": None,
                "statusHint": "",
                "installmentCount": None,
                "paymentDetails": [],
                "overallPayStatus": "",
                "overallPayStatusLabel": "—",
                "overallPayStatusTagType": "info",
            }
        if modes:
            # 协议列表按有效样本方案展示；未开启的质检费用摘要保持原语义。
            ordered_modes = sorted(modes, key=lambda mode: (
                {"dj": 0, "fq": 1, "djfq": 2, "fgd": 3}.get(mode, 99), mode
            ))
            labels = {**_PAY_MODE_LABELS, "fgd": "分改趸"}
            summary["payMode"] = "djfq" if modes == {"dj", "fq"} else "+".join(ordered_modes)
            summary["payModeLabel"] = "+".join(labels.get(mode, mode) for mode in ordered_modes)
        if first_pay:
            summary["firstPayTransferTime"] = first_pay
        out[pid] = summary
    return out


def build_basic_experiment_fee_summary(
    schedule: Optional[BizPilotInfusionSchedule],
    payment: Optional[BizPilotPayment],
) -> Optional[Dict[str, Any]]:
    sale = _dec(schedule.amount) if schedule else None
    if not schedule and not payment:
        return None
    pay_status = (payment.pay_status or "").strip() if payment else ""
    transfer = None
    if payment and pay_status in ("yjf", "jfqrz"):
        transfer = _dec(payment.transfer_amount)
    return {
        "saleAmount": sale,
        "paymentAmount": _dec(payment.amount) if payment else None,
        "transferAmount": transfer,
        "payStatus": pay_status or None,
        "payStatusLabel": _PAY_STATUS_LABELS.get(pay_status, "—") if pay_status else None,
    }


def build_basic_experiment_fee_summaries(
    db: Session,
    schedule_ids: List[str],
    tenant_no: str,
) -> Dict[str, Dict[str, Any]]:
    ids = [x.strip() for x in schedule_ids if (x or "").strip()]
    if not ids:
        return {}
    schedules = {
        s.id: s
        for s in db.query(BizPilotInfusionSchedule)
        .filter(
            BizPilotInfusionSchedule.id.in_(ids),
            BizPilotInfusionSchedule.tenant_no == tenant_no,
            BizPilotInfusionSchedule.del_flag == 0,
        )
        .all()
    }
    payments = {
        p.infusion_schedule_id: p
        for p in db.query(BizPilotPayment)
        .filter(
            BizPilotPayment.infusion_schedule_id.in_(ids),
            BizPilotPayment.tenant_no == tenant_no,
            BizPilotPayment.del_flag == 0,
        )
        .all()
    }
    out: Dict[str, Dict[str, Any]] = {}
    for sid in ids:
        summary = build_basic_experiment_fee_summary(schedules.get(sid), payments.get(sid))
        if summary:
            out[sid] = summary
    return out


def _is_inbound_cell_storage_item(item: Dict[str, Any]) -> bool:
    wo_type = (item.get("workOrderType") or "inbound").strip() or "inbound"
    if wo_type in ("outbound", "temporary"):
        return False
    return (item.get("receptionSource") or "cell_storage").strip() in ("", "cell_storage")


def _load_reception_map(db: Session, reception_ids: List[str]) -> Dict[str, Any]:
    from app.models.models import BizSampleReception

    ids = [x for x in reception_ids if x]
    if not ids:
        return {}
    return {
        r.id: r
        for r in db.query(BizSampleReception)
        .filter(BizSampleReception.id.in_(ids), BizSampleReception.del_flag == 0)
        .all()
    }


def _load_protocol_id_by_no(db: Session, protocol_nos: List[str], tenant_no: str) -> Dict[str, str]:
    from app.models.models import BizCollectionProtocol

    nos = [x for x in protocol_nos if x]
    if not nos:
        return {}
    rows = (
        db.query(BizCollectionProtocol.protocol_no, BizCollectionProtocol.id)
        .filter(
            BizCollectionProtocol.protocol_no.in_(nos),
            BizCollectionProtocol.tenant_no == tenant_no,
            BizCollectionProtocol.del_flag == 0,
        )
        .all()
    )
    return {(r.protocol_no or "").strip(): r.id for r in rows if (r.protocol_no or "").strip()}


def _effective_collection_protocol_id(
    item: Dict[str, Any],
    rec: Optional[Any],
    proto_no_map: Dict[str, str],
) -> str:
    pid = (item.get("collectionProtocolId") or "").strip()
    if pid:
        return pid
    if rec is not None:
        pid = (getattr(rec, "collection_protocol_id", None) or "").strip()
        if pid:
            return pid
    pno = (item.get("protocolNo") or "").strip()
    return proto_no_map.get(pno, "")


def enrich_work_order_fee_summaries(
    db: Session,
    items: List[Dict[str, Any]],
    tenant_no: str,
) -> List[Dict[str, Any]]:
    if not items or not tenant_no:
        return items

    inbound_items = [x for x in items if _is_inbound_cell_storage_item(x)]
    rec_ids = [(x.get("receptionId") or "").strip() for x in inbound_items if (x.get("receptionId") or "").strip()]
    rec_map = _load_reception_map(db, rec_ids)
    proto_nos = list(
        dict.fromkeys(
            (x.get("protocolNo") or "").strip()
            for x in inbound_items
            if (x.get("protocolNo") or "").strip()
        )
    )
    proto_no_map = _load_protocol_id_by_no(db, proto_nos, tenant_no)

    protocol_ids: List[str] = []
    for item in inbound_items:
        rec = rec_map.get((item.get("receptionId") or "").strip())
        pid = _effective_collection_protocol_id(item, rec, proto_no_map)
        if pid:
            protocol_ids.append(pid)

    schedule_ids = list(
        dict.fromkeys(
            (x.get("infusionScheduleId") or "").strip()
            for x in items
            if (x.get("workOrderType") or "") == "outbound"
            and _is_outbound_fee_biz_type(x.get("bizType"))
            and (x.get("infusionScheduleId") or "").strip()
        )
    )

    payment_map = build_protocol_payment_summaries(db, list(dict.fromkeys(protocol_ids)), tenant_no)
    fee_map = build_basic_experiment_fee_summaries(db, schedule_ids, tenant_no)

    for item in items:
        wo_type = item.get("workOrderType") or "inbound"
        if wo_type == "outbound":
            item["paymentSummary"] = None
            if _is_outbound_fee_biz_type(item.get("bizType")):
                sid = (item.get("infusionScheduleId") or "").strip()
                item["feeSummary"] = fee_map.get(sid) if sid else None
            else:
                item["feeSummary"] = None
        elif wo_type == "temporary":
            item["paymentSummary"] = None
            item["feeSummary"] = None
        else:
            item["feeSummary"] = None
            if _is_inbound_cell_storage_item(item):
                rec = rec_map.get((item.get("receptionId") or "").strip())
                pid = _effective_collection_protocol_id(item, rec, proto_no_map)
                item["paymentSummary"] = payment_map.get(pid) if pid else None
            else:
                item["paymentSummary"] = None
    return items