zkxw/backend/app/crud/protocol_sample_payment.py

"""按样本缴费方案 / 应缴批次 / 转账笔 CRUD（payment_model_version=2）"""

from __future__ import annotations

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

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

from app.core.utils import generate_uuid, parse_datetime
from app.crud.collection_protocol import urls_to_attachment_list
from app.models.models import (
    BizCollectionProtocol,
    BizPresignTransfer,
    BizProtocolPartyCustomer,
    BizProtocolPaymentSchedule,
    BizProtocolPaymentTransfer,
    BizProtocolSamplePayment,
)
from app.services.protocol_transfer_status import (
    can_confirm_receipt,
    can_submit_receipt,
    schedule_paid_and_arrears,
)
from app.services.sample_payment_progress import compute_sample_payment_progress


def _dec(v: Any) -> Optional[Decimal]:
    if v is None or v == "":
        return None
    return Decimal(str(v))


def _now() -> datetime:
    return datetime.now()


def _get_sample_payment_or_raise(
    db: Session, *, sample_payment_id: str, tenant_no: str
) -> BizProtocolSamplePayment:
    obj = (
        db.query(BizProtocolSamplePayment)
        .filter(
            BizProtocolSamplePayment.id == sample_payment_id,
            BizProtocolSamplePayment.tenant_no == tenant_no,
            BizProtocolSamplePayment.del_flag == 0,
        )
        .first()
    )
    if not obj:
        raise ValueError("样本缴费方案不存在")
    return obj


def _get_schedule_or_raise(
    db: Session, *, schedule_id: str, tenant_no: str
) -> BizProtocolPaymentSchedule:
    obj = (
        db.query(BizProtocolPaymentSchedule)
        .filter(
            BizProtocolPaymentSchedule.id == schedule_id,
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
        )
        .first()
    )
    if not obj:
        raise ValueError("应缴批次不存在")
    return obj


def _get_transfer_or_raise(
    db: Session, *, transfer_id: str, tenant_no: str
) -> BizProtocolPaymentTransfer:
    obj = (
        db.query(BizProtocolPaymentTransfer)
        .filter(
            BizProtocolPaymentTransfer.id == transfer_id,
            BizProtocolPaymentTransfer.tenant_no == tenant_no,
            BizProtocolPaymentTransfer.del_flag == 0,
        )
        .first()
    )
    if not obj:
        raise ValueError("转账笔不存在")
    return obj


def has_blocking_transfer_for_sample_payment(
    db: Session, *, sample_payment_id: str, tenant_no: str
) -> bool:
    row = (
        db.query(BizProtocolPaymentTransfer.id)
        .filter(
            BizProtocolPaymentTransfer.sample_payment_id == sample_payment_id,
            BizProtocolPaymentTransfer.tenant_no == tenant_no,
            BizProtocolPaymentTransfer.del_flag == 0,
            BizProtocolPaymentTransfer.pay_status == "jfqrz",
        )
        .first()
    )
    return row is not None


def assert_sample_payment_editable(
    db: Session, *, sample_payment_id: str, tenant_no: str
) -> None:
    if has_blocking_transfer_for_sample_payment(
        db, sample_payment_id=sample_payment_id, tenant_no=tenant_no
    ):
        raise ValueError("存在缴费确认中的转账笔，无法修改缴费方案或批次")


def upsert_sample_payment(
    db: Session,
    *,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    sample_code: str,
    pay_mode: str,
    contract_amount: Any = None,
    protocol_sample_id: Optional[str] = None,
    installment_count: Optional[int] = None,
    installment_interval_months: Any = None,
    installment_start_date: Optional[str] = None,
    installment_advance_days: Optional[int] = None,
    installment_ratios: Optional[str] = None,
    lump_sum_deadline: Optional[str] = None,
    remark: Optional[str] = None,
    sample_payment_id: Optional[str] = None,
) -> BizProtocolSamplePayment:
    code = (sample_code or "").strip()
    if not code:
        raise ValueError("样本编码不能为空")
    mode = (pay_mode or "").strip()
    if mode not in ("dj", "fq", "djfq"):
        raise ValueError("缴费模式无效")

    obj: Optional[BizProtocolSamplePayment] = None
    if sample_payment_id:
        obj = _get_sample_payment_or_raise(
            db, sample_payment_id=sample_payment_id, tenant_no=tenant_no
        )
    else:
        obj = (
            db.query(BizProtocolSamplePayment)
            .filter(
                BizProtocolSamplePayment.collection_protocol_id == collection_protocol_id,
                BizProtocolSamplePayment.tenant_no == tenant_no,
                BizProtocolSamplePayment.sample_code == code,
                BizProtocolSamplePayment.del_flag == 0,
            )
            .first()
        )

    now = _now()
    if obj:
        assert_sample_payment_editable(db, sample_payment_id=obj.id, tenant_no=tenant_no)
        if (obj.pay_mode or "") != mode:
            assert_sample_payment_editable(db, sample_payment_id=obj.id, tenant_no=tenant_no)
        obj.protocol_sample_id = (protocol_sample_id or "").strip() or None
        obj.sample_code = code
        obj.pay_mode = mode
        obj.contract_amount = _dec(contract_amount)
        obj.installment_count = installment_count
        obj.installment_interval_months = _dec(installment_interval_months)
        obj.installment_start_date = (installment_start_date or "").strip() or None
        obj.installment_advance_days = installment_advance_days
        obj.installment_ratios = (installment_ratios or "").strip() or None
        obj.lump_sum_deadline = (lump_sum_deadline or "").strip() or None
        if remark is not None:
            obj.remark = (remark or "").strip() or None
        obj.update_by = user_id
        obj.update_time = now
        db.add(obj)
        db.flush()
        return obj

    obj = BizProtocolSamplePayment(
        id=generate_uuid(),
        collection_protocol_id=collection_protocol_id,
        protocol_sample_id=(protocol_sample_id or "").strip() or None,
        sample_code=code,
        pay_mode=mode,
        contract_amount=_dec(contract_amount),
        installment_count=installment_count,
        installment_interval_months=_dec(installment_interval_months),
        installment_start_date=(installment_start_date or "").strip() or None,
        installment_advance_days=installment_advance_days,
        installment_ratios=(installment_ratios or "").strip() or None,
        lump_sum_deadline=(lump_sum_deadline or "").strip() or None,
        tenant_no=tenant_no,
        order_num=0,
        remark=(remark or "").strip() or 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 replace_schedules(
    db: Session,
    *,
    sample_payment_id: str,
    tenant_no: str,
    user_id: Optional[str],
    schedules: List[Dict[str, Any]],
) -> List[BizProtocolPaymentSchedule]:
    """软删原批次（及其下 dfj 转账），重建批次。存在 jfqrz 时禁止。"""
    sp = _get_sample_payment_or_raise(
        db, sample_payment_id=sample_payment_id, tenant_no=tenant_no
    )
    assert_sample_payment_editable(
        db, sample_payment_id=sample_payment_id, tenant_no=tenant_no
    )
    now = _now()
    old_schedules = (
        db.query(BizProtocolPaymentSchedule)
        .filter(
            BizProtocolPaymentSchedule.sample_payment_id == sample_payment_id,
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
        )
        .all()
    )
    old_ids = [s.id for s in old_schedules]
    if old_ids:
        transfers = (
            db.query(BizProtocolPaymentTransfer)
            .filter(
                BizProtocolPaymentTransfer.schedule_id.in_(old_ids),
                BizProtocolPaymentTransfer.tenant_no == tenant_no,
                BizProtocolPaymentTransfer.del_flag == 0,
            )
            .all()
        )
        for t in transfers:
            if t.pay_status == "jfqrz":
                raise ValueError("存在缴费确认中的转账笔，无法重建批次")
            if t.pay_status == "dfj":
                t.del_flag = 1
                t.update_by = user_id
                t.update_time = now
                db.add(t)
            # yjf 转账保留：所属旧批次软删后仍可历史追溯；新批次不含这些笔
        for s in old_schedules:
            s.del_flag = 1
            s.update_by = user_id
            s.update_time = now
            db.add(s)

    created: List[BizProtocolPaymentSchedule] = []
    for idx, raw in enumerate(schedules or []):
        due_date = (raw.get("due_date") or raw.get("dueDate") or "").strip() or None
        schedule_status = (
            raw.get("schedule_status") or raw.get("scheduleStatus") or "open"
        )
        obj = BizProtocolPaymentSchedule(
            id=generate_uuid(),
            sample_payment_id=sp.id,
            collection_protocol_id=sp.collection_protocol_id,
            sample_code=sp.sample_code,
            schedule_no=raw.get("schedule_no")
            if raw.get("schedule_no") is not None
            else raw.get("scheduleNo"),
            due_amount=_dec(raw.get("due_amount") if "due_amount" in raw else raw.get("dueAmount")),
            due_date=due_date,
            schedule_status=(str(schedule_status).strip() or "open"),
            tenant_no=tenant_no,
            order_num=idx,
            remark=(raw.get("remark") or "").strip() or None,
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(obj)
        created.append(obj)
    db.flush()
    return created


def add_transfer(
    db: Session,
    *,
    schedule_id: str,
    tenant_no: str,
    user_id: Optional[str],
    transfer_amount: Any = None,
    transfer_time: Any = None,
    transfer_type: Optional[str] = None,
    transfer_attachments: Optional[str] = None,
    remark: Optional[str] = None,
    commit: bool = True,
) -> BizProtocolPaymentTransfer:
    sch = _get_schedule_or_raise(db, schedule_id=schedule_id, tenant_no=tenant_no)
    now = _now()
    tt = None
    if transfer_time not in (None, ""):
        tt = (
            transfer_time
            if isinstance(transfer_time, datetime)
            else parse_datetime(str(transfer_time))
        )
    obj = BizProtocolPaymentTransfer(
        id=generate_uuid(),
        schedule_id=sch.id,
        sample_payment_id=sch.sample_payment_id,
        collection_protocol_id=sch.collection_protocol_id,
        sample_code=sch.sample_code,
        transfer_amount=_dec(transfer_amount),
        transfer_time=tt,
        transfer_type=(transfer_type or "").strip() or None,
        transfer_attachments=(transfer_attachments or "").strip() or None,
        pay_status="dfj",
        tenant_no=tenant_no,
        order_num=0,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    if commit:
        db.commit()
        db.refresh(obj)
    else:
        db.flush()
    return obj


def update_transfer_draft(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
    fields: Dict[str, Any],
) -> BizProtocolPaymentTransfer:
    t = _get_transfer_or_raise(db, transfer_id=transfer_id, tenant_no=tenant_no)
    if t.pay_status != "dfj":
        raise ValueError("仅已缴费状态可编辑转账笔")
    if "transfer_type" in fields or "transferType" in fields:
        v = fields.get("transfer_type", fields.get("transferType"))
        t.transfer_type = (str(v) if v is not None else "").strip() or None
    if "transfer_amount" in fields or "transferAmount" in fields:
        v = fields.get("transfer_amount", fields.get("transferAmount"))
        amt = _dec(v)
        if amt is not None and amt < 0:
            raise ValueError("转账金额无效")
        t.transfer_amount = amt
    if "transfer_time" in fields or "transferTime" in fields:
        v = fields.get("transfer_time", fields.get("transferTime"))
        if v in (None, ""):
            t.transfer_time = None
        elif isinstance(v, datetime):
            t.transfer_time = v
        else:
            t.transfer_time = parse_datetime(str(v))
    if "transfer_attachments" in fields or "transferAttachments" in fields:
        v = fields.get("transfer_attachments", fields.get("transferAttachments"))
        t.transfer_attachments = (str(v) if v is not None else "").strip() or None
    if "remark" in fields:
        t.remark = (fields.get("remark") or "").strip() or None
    t.update_by = user_id
    t.update_time = _now()
    db.add(t)
    db.commit()
    db.refresh(t)
    return t


def _validate_transfer_submit_ready(t: BizProtocolPaymentTransfer) -> None:
    if not (t.transfer_type or "").strip():
        raise ValueError("请选择转账方式")
    if t.transfer_amount is None or Decimal(str(t.transfer_amount)) <= 0:
        raise ValueError("请填写转账金额")


def submit_receipt_finance(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
    draft_fields: Optional[Dict[str, Any]] = None,
) -> BizProtocolPaymentTransfer:
    t = _get_transfer_or_raise(db, transfer_id=transfer_id, tenant_no=tenant_no)
    if draft_fields:
        if t.pay_status != "dfj":
            raise ValueError("仅已缴费状态可编辑转账笔")
        # 就地改草稿字段（不单独 commit）
        if "transfer_type" in draft_fields or "transferType" in draft_fields:
            v = draft_fields.get("transfer_type", draft_fields.get("transferType"))
            t.transfer_type = (str(v) if v is not None else "").strip() or None
        if "transfer_amount" in draft_fields or "transferAmount" in draft_fields:
            t.transfer_amount = _dec(
                draft_fields.get("transfer_amount", draft_fields.get("transferAmount"))
            )
        if "transfer_time" in draft_fields or "transferTime" in draft_fields:
            v = draft_fields.get("transfer_time", draft_fields.get("transferTime"))
            if v in (None, ""):
                t.transfer_time = None
            elif isinstance(v, datetime):
                t.transfer_time = v
            else:
                t.transfer_time = parse_datetime(str(v))
        if "transfer_attachments" in draft_fields or "transferAttachments" in draft_fields:
            v = draft_fields.get(
                "transfer_attachments", draft_fields.get("transferAttachments")
            )
            t.transfer_attachments = (str(v) if v is not None else "").strip() or None
    if not can_submit_receipt(t.pay_status):
        raise ValueError("仅已缴费状态可提交财务确认")
    _validate_transfer_submit_ready(t)
    t.pay_status = "jfqrz"
    t.update_by = user_id
    t.update_time = _now()
    db.add(t)
    db.commit()
    db.refresh(t)
    return t


def confirm_receipt(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
    finance_confirm_attachments: Optional[str] = None,
    finance_remark: Optional[str] = None,
) -> BizProtocolPaymentTransfer:
    t = _get_transfer_or_raise(db, transfer_id=transfer_id, tenant_no=tenant_no)
    if t.pay_status == "yjf":
        raise ValueError("该转账笔已确认收款")
    if not can_confirm_receipt(t.pay_status):
        raise ValueError("当前状态不可确认收款")
    t.pay_status = "yjf"
    if finance_confirm_attachments is not None:
        t.finance_confirm_attachments = finance_confirm_attachments.strip() or None
    if finance_remark is not None:
        t.finance_remark = finance_remark.strip() or None
    t.update_by = user_id
    t.update_time = _now()
    db.add(t)
    db.commit()
    db.refresh(t)
    return t


def reject_receipt(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
) -> BizProtocolPaymentTransfer:
    t = _get_transfer_or_raise(db, transfer_id=transfer_id, tenant_no=tenant_no)
    if t.pay_status != "jfqrz":
        raise ValueError("当前状态不可退回")
    t.pay_status = "dfj"
    t.update_by = user_id
    t.update_time = _now()
    db.add(t)
    db.commit()
    db.refresh(t)
    return t


def _transfer_to_dict(
    db: Session, t: BizProtocolPaymentTransfer, tenant_no: str
) -> Dict[str, Any]:
    return {
        "id": t.id,
        "scheduleId": t.schedule_id,
        "samplePaymentId": t.sample_payment_id,
        "collectionProtocolId": t.collection_protocol_id,
        "sampleCode": t.sample_code or "",
        "transferAmount": float(t.transfer_amount) if t.transfer_amount is not None else None,
        "transferTime": t.transfer_time.strftime("%Y-%m-%d %H:%M:%S") if t.transfer_time else "",
        "transferType": t.transfer_type or "",
        "transferAttachments": t.transfer_attachments or "",
        "attachments": urls_to_attachment_list(t.transfer_attachments, db, tenant_no),
        "payStatus": t.pay_status or "",
        "financeConfirmAttachments": t.finance_confirm_attachments or "",
        "financeConfirmAttachmentFiles": urls_to_attachment_list(
            t.finance_confirm_attachments, db, tenant_no
        ),
        "financeRemark": t.finance_remark or "",
        "remark": t.remark or "",
        "source": "formal",
        "createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "",
        "updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "",
    }


def _schedule_to_dict(
    db: Session,
    s: BizProtocolPaymentSchedule,
    transfers: List[BizProtocolPaymentTransfer],
    tenant_no: str,
) -> Dict[str, Any]:
    confirmed = [
        t.transfer_amount
        for t in transfers
        if (t.pay_status or "") == "yjf"
    ]
    paid, arrears = schedule_paid_and_arrears(
        due=s.due_amount, confirmed_transfers=confirmed
    )
    return {
        "id": s.id,
        "samplePaymentId": s.sample_payment_id,
        "collectionProtocolId": s.collection_protocol_id,
        "sampleCode": s.sample_code or "",
        "scheduleNo": s.schedule_no,
        "dueAmount": float(s.due_amount) if s.due_amount is not None else None,
        "dueDate": s.due_date or "",
        "scheduleStatus": s.schedule_status or "",
        "paidAmount": float(paid),
        "arrearsAmount": float(arrears),
        "remark": s.remark or "",
        "transfers": [_transfer_to_dict(db, t, tenant_no) for t in transfers],
    }


def _sample_payment_to_dict(
    db: Session,
    p: BizProtocolSamplePayment,
    schedules: List[BizProtocolPaymentSchedule],
    transfers_by_schedule: Dict[str, List[BizProtocolPaymentTransfer]],
    tenant_no: str,
) -> Dict[str, Any]:
    return {
        "id": p.id,
        "collectionProtocolId": p.collection_protocol_id,
        "protocolSampleId": p.protocol_sample_id or "",
        "sampleCode": p.sample_code or "",
        "payMode": p.pay_mode or "",
        "contractAmount": float(p.contract_amount) if p.contract_amount is not None else None,
        "installmentCount": p.installment_count,
        "installmentIntervalMonths": float(p.installment_interval_months)
        if p.installment_interval_months is not None
        else None,
        "installmentStartDate": p.installment_start_date or "",
        "installmentAdvanceDays": p.installment_advance_days,
        "installmentRatios": p.installment_ratios or "",
        "lumpSumDeadline": p.lump_sum_deadline or "",
        "remark": p.remark or "",
        "schedules": [
            _schedule_to_dict(
                db, s, transfers_by_schedule.get(s.id, []), tenant_no
            )
            for s in schedules
        ],
    }


def _normalize_installment_ratios(v: Any) -> Optional[str]:
    if v is None or v == "":
        return None
    if isinstance(v, (list, dict)):
        return json.dumps(v, ensure_ascii=False)
    return str(v).strip() or None


def _soft_delete_sample_payment(
    db: Session,
    *,
    sp: BizProtocolSamplePayment,
    tenant_no: str,
    user_id: Optional[str],
) -> None:
    assert_sample_payment_editable(db, sample_payment_id=sp.id, tenant_no=tenant_no)
    now = _now()
    schedules = (
        db.query(BizProtocolPaymentSchedule)
        .filter(
            BizProtocolPaymentSchedule.sample_payment_id == sp.id,
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
        )
        .all()
    )
    sids = [s.id for s in schedules]
    if sids:
        transfers = (
            db.query(BizProtocolPaymentTransfer)
            .filter(
                BizProtocolPaymentTransfer.schedule_id.in_(sids),
                BizProtocolPaymentTransfer.tenant_no == tenant_no,
                BizProtocolPaymentTransfer.del_flag == 0,
            )
            .all()
        )
        for t in transfers:
            if t.pay_status == "jfqrz":
                raise ValueError("存在缴费确认中的转账笔，无法删除样本缴费方案")
            if t.pay_status == "dfj":
                t.del_flag = 1
                t.update_by = user_id
                t.update_time = now
                db.add(t)
    for s in schedules:
        s.del_flag = 1
        s.update_by = user_id
        s.update_time = now
        db.add(s)
    sp.del_flag = 1
    sp.update_by = user_id
    sp.update_time = now
    db.add(sp)


def save_sample_payments_for_protocol(
    db: Session,
    *,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    sample_payments: List[Dict[str, Any]],
) -> None:
    """按样本保存缴费方案：upsert 方案 → 重建批次 → 写入待缴费转账笔。"""
    existing = (
        db.query(BizProtocolSamplePayment)
        .filter(
            BizProtocolSamplePayment.collection_protocol_id == collection_protocol_id,
            BizProtocolSamplePayment.tenant_no == tenant_no,
            BizProtocolSamplePayment.del_flag == 0,
        )
        .all()
    )
    keep_ids: set = set()
    for raw in sample_payments or []:
        if not isinstance(raw, dict):
            continue
        sample_code = (raw.get("sample_code") or raw.get("sampleCode") or "").strip()
        if not sample_code:
            raise ValueError("样本缴费方案缺少样本编码")
        pay_mode = (raw.get("pay_mode") or raw.get("payMode") or "").strip()
        sp_id = (raw.get("id") or raw.get("samplePaymentId") or "").strip() or None
        ratios = raw.get("installment_ratios")
        if ratios is None:
            ratios = raw.get("installmentRatios")
        sp = upsert_sample_payment(
            db,
            collection_protocol_id=collection_protocol_id,
            tenant_no=tenant_no,
            user_id=user_id,
            sample_code=sample_code,
            pay_mode=pay_mode,
            contract_amount=raw.get("contract_amount", raw.get("contractAmount")),
            protocol_sample_id=raw.get("protocol_sample_id") or raw.get("protocolSampleId"),
            installment_count=raw.get("installment_count", raw.get("installmentCount")),
            installment_interval_months=raw.get(
                "installment_interval_months", raw.get("installmentIntervalMonths")
            ),
            installment_start_date=raw.get(
                "installment_start_date", raw.get("installmentStartDate")
            ),
            installment_advance_days=raw.get(
                "installment_advance_days", raw.get("installmentAdvanceDays")
            ),
            installment_ratios=_normalize_installment_ratios(ratios),
            lump_sum_deadline=raw.get("lump_sum_deadline", raw.get("lumpSumDeadline")),
            remark=raw.get("remark"),
            sample_payment_id=sp_id,
        )
        keep_ids.add(sp.id)
        schedules_payload = raw.get("schedules") or []
        created = replace_schedules(
            db,
            sample_payment_id=sp.id,
            tenant_no=tenant_no,
            user_id=user_id,
            schedules=schedules_payload,
        )
        for i, sch_raw in enumerate(schedules_payload):
            if i >= len(created) or not isinstance(sch_raw, dict):
                continue
            for t_raw in sch_raw.get("transfers") or []:
                if not isinstance(t_raw, dict):
                    continue
                status = (
                    t_raw.get("pay_status") or t_raw.get("payStatus") or "dfj"
                ).strip()
                # 仅重建待缴费草稿；jfqrz/yjf 由既有流程保留在旧批次或独立接口维护
                if status != "dfj":
                    continue
                add_transfer(
                    db,
                    schedule_id=created[i].id,
                    tenant_no=tenant_no,
                    user_id=user_id,
                    transfer_amount=t_raw.get(
                        "transfer_amount", t_raw.get("transferAmount")
                    ),
                    transfer_time=t_raw.get(
                        "transfer_time", t_raw.get("transferTime")
                    ),
                    transfer_type=t_raw.get(
                        "transfer_type", t_raw.get("transferType")
                    ),
                    transfer_attachments=t_raw.get(
                        "transfer_attachments", t_raw.get("transferAttachments")
                    ),
                    remark=t_raw.get("remark"),
                    commit=False,
                )
    for sp in existing:
        if sp.id not in keep_ids:
            _soft_delete_sample_payment(
                db, sp=sp, tenant_no=tenant_no, user_id=user_id
            )
    db.commit()


def get_presign_deposit_hint(
    db: Session, *, source_presign_order_id: Optional[str], tenant_no: str
) -> Optional[Dict[str, Any]]:
    pid = (source_presign_order_id or "").strip()
    if not pid:
        return None
    rows = (
        db.query(BizPresignTransfer)
        .filter(
            BizPresignTransfer.presign_order_id == pid,
            BizPresignTransfer.tenant_no == tenant_no,
            BizPresignTransfer.del_flag == 0,
            BizPresignTransfer.pay_status == "yjf",
        )
        .all()
    )
    total = Decimal("0")
    for r in rows:
        if r.transfer_amount is not None:
            total += Decimal(str(r.transfer_amount))
    return {
        "presignOrderId": pid,
        "confirmedDepositAmount": float(total),
        "confirmedTransferCount": len(rows),
    }


def get_sample_payments_bundle(
    db: Session, collection_protocol_id: str, tenant_no: str
) -> List[Dict[str, Any]]:
    payments = (
        db.query(BizProtocolSamplePayment)
        .filter(
            BizProtocolSamplePayment.collection_protocol_id == collection_protocol_id,
            BizProtocolSamplePayment.tenant_no == tenant_no,
            BizProtocolSamplePayment.del_flag == 0,
        )
        .order_by(
            BizProtocolSamplePayment.order_num.asc(),
            BizProtocolSamplePayment.create_time.asc(),
        )
        .all()
    )
    if not payments:
        return []
    pids = [p.id for p in payments]
    schedules = (
        db.query(BizProtocolPaymentSchedule)
        .filter(
            BizProtocolPaymentSchedule.sample_payment_id.in_(pids),
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
        )
        .order_by(
            BizProtocolPaymentSchedule.order_num.asc(),
            BizProtocolPaymentSchedule.schedule_no.asc(),
        )
        .all()
    )
    sids = [s.id for s in schedules]
    transfers: List[BizProtocolPaymentTransfer] = []
    if sids:
        transfers = (
            db.query(BizProtocolPaymentTransfer)
            .filter(
                BizProtocolPaymentTransfer.schedule_id.in_(sids),
                BizProtocolPaymentTransfer.tenant_no == tenant_no,
                BizProtocolPaymentTransfer.del_flag == 0,
            )
            .order_by(BizProtocolPaymentTransfer.create_time.asc())
            .all()
        )
    by_schedule: Dict[str, List[BizProtocolPaymentTransfer]] = {}
    for t in transfers:
        by_schedule.setdefault(t.schedule_id, []).append(t)
    schedules_by_payment: Dict[str, List[BizProtocolPaymentSchedule]] = {}
    for s in schedules:
        schedules_by_payment.setdefault(s.sample_payment_id, []).append(s)
    return [
        _sample_payment_to_dict(
            db,
            p,
            schedules_by_payment.get(p.id, []),
            by_schedule,
            tenant_no,
        )
        for p in payments
    ]


def list_transfers_for_finance(
    db: Session,
    *,
    pay_status: str,
    tenant_no: str,
    keyword: Optional[str] = None,
    page: int = 1,
    page_size: int = 10,
    user_info=None,
    due_date_start: Optional[str] = None,
    due_date_end: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0
    st = (pay_status or "").strip()
    q = (
        db.query(
            BizProtocolPaymentTransfer,
            BizCollectionProtocol,
            BizProtocolPartyCustomer,
            BizProtocolPaymentSchedule,
            BizProtocolSamplePayment,
        )
        .join(
            BizCollectionProtocol,
            and_(
                BizCollectionProtocol.id
                == BizProtocolPaymentTransfer.collection_protocol_id,
                BizCollectionProtocol.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolPartyCustomer,
            and_(
                BizProtocolPartyCustomer.collection_protocol_id
                == BizProtocolPaymentTransfer.collection_protocol_id,
                BizProtocolPartyCustomer.tenant_no == tenant_no,
                BizProtocolPartyCustomer.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolPaymentSchedule,
            and_(
                BizProtocolPaymentSchedule.id == BizProtocolPaymentTransfer.schedule_id,
                BizProtocolPaymentSchedule.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolSamplePayment,
            and_(
                BizProtocolSamplePayment.id
                == BizProtocolPaymentTransfer.sample_payment_id,
                BizProtocolSamplePayment.del_flag == 0,
            ),
        )
        .filter(
            BizProtocolPaymentTransfer.tenant_no == tenant_no,
            BizProtocolPaymentTransfer.del_flag == 0,
            BizProtocolPaymentTransfer.pay_status == st,
        )
    )
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(
            or_(
                BizCollectionProtocol.protocol_no.like(like),
                BizProtocolPaymentTransfer.sample_code.like(like),
                BizProtocolPartyCustomer.mother_name.like(like),
                BizProtocolPartyCustomer.mother_id_card.like(like),
                BizProtocolPartyCustomer.newborn_name.like(like),
                BizProtocolPartyCustomer.newborn_id_card.like(like),
            )
        )
    ds = (due_date_start or "").strip().replace("-", "")[:8]
    de = (due_date_end or "").strip().replace("-", "")[:8]
    if ds:
        q = q.filter(BizProtocolPaymentSchedule.due_date >= ds)
    if de:
        q = q.filter(BizProtocolPaymentSchedule.due_date <= de)
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(
            q, BizProtocolPaymentTransfer, user_info, db, tenant_no
        )
    total = q.order_by(None).with_entities(func.count(BizProtocolPaymentTransfer.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q.order_by(BizProtocolPaymentTransfer.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    out: List[Dict[str, Any]] = []
    for t, proto, party, sch, sp in rows:
        item = _transfer_to_dict(db, t, tenant_no)
        item["protocolNo"] = (proto.protocol_no if proto else "") or ""
        item["protocolStatus"] = (proto.protocol_status if proto else "") or ""
        item["voidType"] = (proto.void_type if proto else "") or ""
        item["motherName"] = (party.mother_name if party else "") or ""
        item["motherIdCard"] = (party.mother_id_card if party else "") or ""
        item["newbornName"] = (party.newborn_name if party else "") or ""
        item["newbornIdCard"] = (party.newborn_id_card if party else "") or ""
        item["installmentNo"] = sch.schedule_no if sch else None
        item["installmentDueDate"] = (sch.due_date if sch else "") or ""
        item["dueAmount"] = (
            float(sch.due_amount) if sch and sch.due_amount is not None else None
        )
        item["payMode"] = (sp.pay_mode if sp else "") or ""
        out.append(item)
    return out, int(total)


def _filter_and_page_progress(
    rows: List[Dict[str, Any]],
    progress_status: Optional[str],
    page: int,
    page_size: int,
) -> Tuple[List[Dict[str, Any]], int]:
    """先按 progressStatus 过滤，再分页切片（禁止先 LIMIT 再筛进度）。"""
    filtered = rows
    ps = (progress_status or "").strip()
    if ps:
        filtered = [r for r in rows if (r.get("progressStatus") or "") == ps]
    total = len(filtered)
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    start = (page - 1) * page_size
    return filtered[start : start + page_size], total


def _progress_to_json_fields(progress: Dict[str, Any]) -> Dict[str, Any]:
    paid = progress.get("paidAmount")
    due_remain = progress.get("dueRemainAmount")
    return {
        "progressStatus": progress.get("progressStatus") or "",
        "paidAmount": float(paid) if paid is not None else 0.0,
        "dueRemainAmount": float(due_remain) if due_remain is not None else 0.0,
        "clearedScheduleCount": progress.get("clearedScheduleCount") or 0,
        "activeScheduleCount": progress.get("activeScheduleCount") or 0,
        "canConfirm": bool(progress.get("canConfirm")),
    }


def _load_payment_children_batch(
    db: Session,
    sample_payment_ids: List[str],
    tenant_no: str,
) -> Tuple[
    Dict[str, List[BizProtocolPaymentSchedule]],
    Dict[str, List[BizProtocolPaymentTransfer]],
]:
    schedules_by_payment: Dict[str, List[BizProtocolPaymentSchedule]] = {}
    transfers_by_payment: Dict[str, List[BizProtocolPaymentTransfer]] = {}
    if not sample_payment_ids:
        return schedules_by_payment, transfers_by_payment
    schedules = (
        db.query(BizProtocolPaymentSchedule)
        .filter(
            BizProtocolPaymentSchedule.sample_payment_id.in_(sample_payment_ids),
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
        )
        .order_by(
            BizProtocolPaymentSchedule.order_num.asc(),
            BizProtocolPaymentSchedule.schedule_no.asc(),
        )
        .all()
    )
    sids = [s.id for s in schedules]
    transfers: List[BizProtocolPaymentTransfer] = []
    if sids:
        transfers = (
            db.query(BizProtocolPaymentTransfer)
            .filter(
                BizProtocolPaymentTransfer.schedule_id.in_(sids),
                BizProtocolPaymentTransfer.tenant_no == tenant_no,
                BizProtocolPaymentTransfer.del_flag == 0,
            )
            .order_by(BizProtocolPaymentTransfer.create_time.asc())
            .all()
        )
    for s in schedules:
        schedules_by_payment.setdefault(s.sample_payment_id, []).append(s)
    schedule_to_payment = {s.id: s.sample_payment_id for s in schedules}
    for t in transfers:
        pid = schedule_to_payment.get(t.schedule_id) or t.sample_payment_id
        if pid:
            transfers_by_payment.setdefault(pid, []).append(t)
    return schedules_by_payment, transfers_by_payment


def _compute_progress_for_payment(
    schedules: List[BizProtocolPaymentSchedule],
    transfers: List[BizProtocolPaymentTransfer],
) -> Dict[str, Any]:
    sch_dicts = [
        {
            "scheduleStatus": s.schedule_status or "",
            "dueAmount": s.due_amount,
        }
        for s in schedules
    ]
    tr_dicts = [
        {
            "payStatus": t.pay_status or "",
            "transferAmount": t.transfer_amount,
        }
        for t in transfers
    ]
    return compute_sample_payment_progress(schedules=sch_dicts, transfers=tr_dicts)


def _build_list_summary_row(
    sp: BizProtocolSamplePayment,
    proto: BizCollectionProtocol,
    party: Optional[BizProtocolPartyCustomer],
    progress: Dict[str, Any],
) -> Dict[str, Any]:
    row: Dict[str, Any] = {
        "id": sp.id,
        "collectionProtocolId": sp.collection_protocol_id,
        "protocolNo": (proto.protocol_no if proto else "") or "",
        "protocolStatus": (proto.protocol_status if proto else "") or "",
        "voidType": (proto.void_type if proto else "") or "",
        "motherName": (party.mother_name if party else "") or "",
        "motherPhone": (party.mother_phone if party else "") or "",
        "sampleCode": sp.sample_code or "",
        "payMode": sp.pay_mode or "",
        "contractAmount": float(sp.contract_amount)
        if sp.contract_amount is not None
        else None,
    }
    row.update(_progress_to_json_fields(progress))
    return row


def list_sample_payments_page(
    db: Session,
    *,
    tenant_no: str,
    keyword: Optional[str] = None,
    pay_mode: Optional[str] = None,
    progress_status: Optional[str] = None,
    due_date_start: Optional[str] = None,
    due_date_end: Optional[str] = None,
    page: int = 1,
    page_size: int = 10,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0
    q = (
        db.query(
            BizProtocolSamplePayment,
            BizCollectionProtocol,
            BizProtocolPartyCustomer,
        )
        .join(
            BizCollectionProtocol,
            and_(
                BizCollectionProtocol.id
                == BizProtocolSamplePayment.collection_protocol_id,
                BizCollectionProtocol.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolPartyCustomer,
            and_(
                BizProtocolPartyCustomer.collection_protocol_id
                == BizProtocolSamplePayment.collection_protocol_id,
                BizProtocolPartyCustomer.tenant_no == tenant_no,
                BizProtocolPartyCustomer.del_flag == 0,
            ),
        )
        .filter(
            BizProtocolSamplePayment.tenant_no == tenant_no,
            BizProtocolSamplePayment.del_flag == 0,
        )
    )
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(
            or_(
                BizCollectionProtocol.protocol_no.like(like),
                BizProtocolSamplePayment.sample_code.like(like),
                BizProtocolPartyCustomer.mother_name.like(like),
                BizProtocolPartyCustomer.mother_phone.like(like),
            )
        )
    mode = (pay_mode or "").strip()
    if mode:
        q = q.filter(BizProtocolSamplePayment.pay_mode == mode)
    ds = (due_date_start or "").strip().replace("-", "")[:8]
    de = (due_date_end or "").strip().replace("-", "")[:8]
    if ds or de:
        due_conditions = [
            BizProtocolPaymentSchedule.sample_payment_id
            == BizProtocolSamplePayment.id,
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
            BizProtocolPaymentSchedule.schedule_status != "void",
        ]
        if ds:
            due_conditions.append(BizProtocolPaymentSchedule.due_date >= ds)
        if de:
            due_conditions.append(BizProtocolPaymentSchedule.due_date <= de)
        q = q.filter(exists().where(and_(*due_conditions)))
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizCollectionProtocol, user_info, db, tenant_no)
    candidates = q.order_by(BizProtocolSamplePayment.create_time.desc()).all()
    if not candidates:
        return [], 0
    pids = [sp.id for sp, _, _ in candidates]
    schedules_by_payment, transfers_by_payment = _load_payment_children_batch(
        db, pids, tenant_no
    )
    rows: List[Dict[str, Any]] = []
    for sp, proto, party in candidates:
        schedules = schedules_by_payment.get(sp.id, [])
        transfers = transfers_by_payment.get(sp.id, [])
        progress = _compute_progress_for_payment(schedules, transfers)
        rows.append(_build_list_summary_row(sp, proto, party, progress))
    return _filter_and_page_progress(rows, progress_status, page, page_size)


def get_sample_payment_detail(
    db: Session,
    *,
    sample_payment_id: str,
    tenant_no: str,
    user_info=None,
) -> Optional[Dict[str, Any]]:
    q = (
        db.query(
            BizProtocolSamplePayment,
            BizCollectionProtocol,
            BizProtocolPartyCustomer,
        )
        .join(
            BizCollectionProtocol,
            and_(
                BizCollectionProtocol.id
                == BizProtocolSamplePayment.collection_protocol_id,
                BizCollectionProtocol.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolPartyCustomer,
            and_(
                BizProtocolPartyCustomer.collection_protocol_id
                == BizProtocolSamplePayment.collection_protocol_id,
                BizProtocolPartyCustomer.tenant_no == tenant_no,
                BizProtocolPartyCustomer.del_flag == 0,
            ),
        )
        .filter(
            BizProtocolSamplePayment.id == sample_payment_id,
            BizProtocolSamplePayment.tenant_no == tenant_no,
            BizProtocolSamplePayment.del_flag == 0,
        )
    )
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizCollectionProtocol, user_info, db, tenant_no)
    row = q.first()
    if not row:
        return None
    sp, proto, party = row
    schedules_by_payment, transfers_by_payment = _load_payment_children_batch(
        db, [sp.id], tenant_no
    )
    schedules = schedules_by_payment.get(sp.id, [])
    transfers = transfers_by_payment.get(sp.id, [])
    by_schedule: Dict[str, List[BizProtocolPaymentTransfer]] = {}
    for t in transfers:
        by_schedule.setdefault(t.schedule_id, []).append(t)
    detail = _sample_payment_to_dict(db, sp, schedules, by_schedule, tenant_no)
    detail["protocolNo"] = (proto.protocol_no if proto else "") or ""
    detail["protocolStatus"] = (proto.protocol_status if proto else "") or ""
    detail["voidType"] = (proto.void_type if proto else "") or ""
    detail["motherName"] = (party.mother_name if party else "") or ""
    detail["motherPhone"] = (party.mother_phone if party else "") or ""
    progress = _compute_progress_for_payment(schedules, transfers)
    detail.update(_progress_to_json_fields(progress))
    return detail