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 (
    BizCellSampleType,
    BizCollectionProtocol,
    BizPresignTransfer,
    BizProtocolCollectionInfo,
    BizProtocolPartyCustomer,
    BizProtocolPaymentSchedule,
    BizProtocolPaymentTransfer,
    BizProtocolSample,
    BizProtocolSamplePayment,
    SysUser,
)
from app.services.payment_chase import (
    chase_period_money,
    is_chase_candidate,
    parse_due_date_ymd,
)
from app.services.protocol_transfer_status import (
    can_confirm_receipt,
    can_edit_transfer,
    can_submit_receipt,
    can_withdraw_receipt,
)
from app.services.sample_payment_fgd_convert import plan_fq_to_fgd_merge
from app.services.sample_payment_period_summary import summarize_period
from app.services.sample_payment_schedule_validate import schedule_dues_match_contract
from app.services.sample_payment_progress import (
    compute_sample_payment_progress,
    is_schedule_effectively_cleared,
)


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


def _build_periods_array(
    schedules: List[BizProtocolPaymentSchedule],
    transfers: List[BizProtocolPaymentTransfer],
) -> List[Optional[Dict[str, Any]]]:
    """固定长度 6；无对应 schedule_no 的槽位为 None。"""
    by_sch: Dict[str, List[BizProtocolPaymentTransfer]] = {}
    for t in transfers:
        by_sch.setdefault(t.schedule_id or "", []).append(t)
    by_no: Dict[int, BizProtocolPaymentSchedule] = {}
    for s in schedules:
        if (s.schedule_status or "") == "void":
            continue
        no = int(s.schedule_no or 0)
        if 1 <= no <= 6:
            by_no[no] = s
    periods: List[Optional[Dict[str, Any]]] = []
    for no in range(1, 7):
        s = by_no.get(no)
        if not s:
            periods.append(None)
            continue
        trs = by_sch.get(s.id, [])
        summary = summarize_period(
            due_amount=s.due_amount,
            transfers=[
                {
                    "transferAmount": t.transfer_amount,
                    "transferType": t.transfer_type or "",
                    "transferTime": t.transfer_time.strftime("%Y-%m-%d %H:%M:%S")
                    if t.transfer_time
                    else "",
                    "payStatus": t.pay_status or "",
                }
                for t in trs
            ],
        )
        periods.append(
            {
                "scheduleId": s.id,
                "scheduleNo": no,
                "dueAmount": float(s.due_amount) if s.due_amount is not None else None,
                "dueDate": _fmt_ymd8(s.due_date),
                "displayPaidAmount": float(summary["displayPaidAmount"]),
                "arrearsAmount": float(summary["arrearsAmount"]),
                "payDate": summary.get("payDate") or "",
                "payTypeSummary": summary.get("payTypeSummary") or "",
                "financeAudit": summary.get("financeAudit") or "dfj",
            }
        )
    return periods


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 _lock_sample_payment_graph(
    db: Session,
    *,
    sample_payment_id: str,
    tenant_no: str,
) -> Tuple[
    BizProtocolSamplePayment,
    List[BizProtocolPaymentSchedule],
    List[BizProtocolPaymentTransfer],
]:
    """按方案→批次→转账顺序加行锁，供普通方案保存做最终一致性校验。"""
    sp = (
        db.query(BizProtocolSamplePayment)
        .filter(
            BizProtocolSamplePayment.id == sample_payment_id,
            BizProtocolSamplePayment.tenant_no == tenant_no,
            BizProtocolSamplePayment.del_flag == 0,
        )
        .populate_existing()
        .with_for_update()
        .first()
    )
    if not sp:
        raise ValueError("样本缴费方案不存在")
    schedules = (
        db.query(BizProtocolPaymentSchedule)
        .filter(
            BizProtocolPaymentSchedule.sample_payment_id == sample_payment_id,
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
        )
        .order_by(BizProtocolPaymentSchedule.schedule_no.asc())
        .populate_existing()
        .with_for_update()
        .all()
    )
    schedule_ids = [row.id for row in schedules]
    transfers: List[BizProtocolPaymentTransfer] = []
    if schedule_ids:
        transfers = (
            db.query(BizProtocolPaymentTransfer)
            .filter(
                BizProtocolPaymentTransfer.schedule_id.in_(schedule_ids),
                BizProtocolPaymentTransfer.tenant_no == tenant_no,
                BizProtocolPaymentTransfer.del_flag == 0,
            )
            .order_by(BizProtocolPaymentTransfer.id.asc())
            .populate_existing()
            .with_for_update()
            .all()
        )
    return sp, schedules, transfers


def _locked_schedule_ids(
    schedules: List[BizProtocolPaymentSchedule],
    transfers: List[BizProtocolPaymentTransfer],
) -> set:
    locked = {
        row.id
        for row in schedules
        if (row.schedule_status or "").strip() != "open"
    }
    locked.update(
        row.schedule_id
        for row in transfers
        if (row.pay_status or "").strip() != "dfj"
    )
    return locked


def _assert_locked_schedule_structure(
    schedules: List[BizProtocolPaymentSchedule],
    transfers: List[BizProtocolPaymentTransfer],
    payload: List[Dict[str, Any]],
) -> None:
    locked_ids = _locked_schedule_ids(schedules, transfers)
    if not locked_ids:
        return
    current_nos = {int(row.schedule_no or 0) for row in schedules}
    submitted_no_list = [
        _schedule_no_of(raw, idx + 1)
        for idx, raw in enumerate(payload or [])
        if isinstance(raw, dict)
    ]
    if len(submitted_no_list) != len(set(submitted_no_list)):
        raise ValueError("应缴批次期号重复")
    submitted_nos = set(submitted_no_list)
    locked_nos = {
        int(row.schedule_no or 0) for row in schedules if row.id in locked_ids
    }
    if submitted_nos != current_nos or not locked_nos.issubset(submitted_nos):
        raise ValueError("存在锁定批次，不能省略、删除、重编号或重建")
    by_id = {row.id: int(row.schedule_no or 0) for row in schedules}
    for idx, raw in enumerate(payload or []):
        if not isinstance(raw, dict):
            continue
        schedule_id = str(raw.get("id") or raw.get("scheduleId") or "").strip()
        if schedule_id in locked_ids and _schedule_no_of(raw, idx + 1) != by_id[schedule_id]:
            raise ValueError("存在锁定批次，不能省略、删除、重编号或重建")


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 _get_transfer_with_schedule_lock(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_info: Optional[dict] = None,
) -> BizProtocolPaymentTransfer:
    """Authorize the complete parent chain, then lock schedule before transfer."""
    parent = (
        db.query(
            BizProtocolPaymentTransfer.schedule_id,
            BizProtocolPaymentTransfer.sample_payment_id,
            BizProtocolPaymentTransfer.collection_protocol_id,
        )
        .filter(
            BizProtocolPaymentTransfer.id == transfer_id,
            BizProtocolPaymentTransfer.tenant_no == tenant_no,
            BizProtocolPaymentTransfer.del_flag == 0,
        )
        .first()
    )
    if not parent:
        if user_info is not None:
            raise PermissionError("无权操作该协议")
        raise ValueError("转账笔不存在")
    schedule_id = parent.schedule_id
    schedule_query = (
        db.query(BizProtocolPaymentSchedule)
        .join(
            BizProtocolSamplePayment,
            and_(
                BizProtocolSamplePayment.id == parent.sample_payment_id,
                BizProtocolSamplePayment.id
                == BizProtocolPaymentSchedule.sample_payment_id,
                BizProtocolSamplePayment.collection_protocol_id
                == parent.collection_protocol_id,
                BizProtocolSamplePayment.tenant_no == tenant_no,
                BizProtocolSamplePayment.del_flag == 0,
            ),
        )
        .join(
            BizCollectionProtocol,
            and_(
                BizCollectionProtocol.id == parent.collection_protocol_id,
                BizCollectionProtocol.id
                == BizProtocolPaymentSchedule.collection_protocol_id,
                BizCollectionProtocol.tenant_no == tenant_no,
                BizCollectionProtocol.del_flag == 0,
            ),
        )
        .filter(
            BizProtocolPaymentSchedule.id == schedule_id,
            BizProtocolPaymentSchedule.sample_payment_id
            == parent.sample_payment_id,
            BizProtocolPaymentSchedule.collection_protocol_id
            == parent.collection_protocol_id,
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
        )
    )
    if user_info is not None:
        from app.core.data_scope import apply_data_scope

        schedule_query = apply_data_scope(
            schedule_query,
            BizCollectionProtocol,
            user_info,
            db,
            tenant_no,
        )
    schedule = schedule_query.populate_existing().with_for_update().first()
    if not schedule:
        if user_info is not None:
            raise PermissionError("无权操作该协议")
        raise ValueError("应缴批次不存在")
    transfer = (
        db.query(BizProtocolPaymentTransfer)
        .filter(
            BizProtocolPaymentTransfer.id == transfer_id,
            BizProtocolPaymentTransfer.schedule_id == schedule_id,
            BizProtocolPaymentTransfer.sample_payment_id
            == parent.sample_payment_id,
            BizProtocolPaymentTransfer.collection_protocol_id
            == parent.collection_protocol_id,
            BizProtocolPaymentTransfer.tenant_no == tenant_no,
            BizProtocolPaymentTransfer.del_flag == 0,
        )
        .populate_existing()
        .with_for_update()
        .first()
    )
    if not transfer:
        raise ValueError("转账笔不存在")
    return transfer


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_structure_editable(
    db: Session, *, sample_payment_id: str, tenant_no: str
) -> None:
    """已废弃：付款确认中不再阻断协议保存/改方案。保留空实现以兼容旧调用。"""
    return


# 兼容旧调用名
assert_sample_payment_editable = assert_sample_payment_structure_editable


def find_active_sample_payment(
    db: Session,
    *,
    tenant_no: str,
    sample_payment_id: Optional[str] = None,
    collection_protocol_id: Optional[str] = None,
    sample_code: Optional[str] = None,
) -> Optional[BizProtocolSamplePayment]:
    if sample_payment_id:
        query = db.query(BizProtocolSamplePayment).filter(
            BizProtocolSamplePayment.id == sample_payment_id,
            BizProtocolSamplePayment.tenant_no == tenant_no,
            BizProtocolSamplePayment.del_flag == 0,
        )
        if collection_protocol_id:
            query = query.filter(
                BizProtocolSamplePayment.collection_protocol_id
                == collection_protocol_id
            )
        if sample_code:
            query = query.filter(
                BizProtocolSamplePayment.sample_code == sample_code.strip()
            )
        return query.first()
    code = (sample_code or "").strip()
    pid = (collection_protocol_id or "").strip()
    if not code or not pid:
        return None
    return (
        db.query(BizProtocolSamplePayment)
        .filter(
            BizProtocolSamplePayment.collection_protocol_id == pid,
            BizProtocolSamplePayment.tenant_no == tenant_no,
            BizProtocolSamplePayment.sample_code == code,
            BizProtocolSamplePayment.del_flag == 0,
        )
        .first()
    )


def sample_payment_has_yjf(
    db: Session, *, sample_payment_id: str, tenant_no: str
) -> bool:
    """样本缴费方案下是否存在已财务确认（yjf）转账。"""
    q = (
        db.query(BizProtocolPaymentTransfer.id)
        .filter(
            BizProtocolPaymentTransfer.sample_payment_id == sample_payment_id,
            BizProtocolPaymentTransfer.tenant_no == tenant_no,
            BizProtocolPaymentTransfer.del_flag == 0,
            BizProtocolPaymentTransfer.pay_status == "yjf",
        )
        .limit(1)
        .first()
    )
    return q is not None


def apply_fq_to_fgd_merge(
    db: Session,
    *,
    sample_payment_id: str,
    tenant_no: str,
    user_id: Optional[str],
    target_pay_mode: str = "fgd",
) -> Dict[str, Any]:
    """将分期方案并期为趸交/分改趸：更新目标期应缴、软删后续期及其 dfj 转账。"""
    sp = _get_sample_payment_or_raise(
        db, sample_payment_id=sample_payment_id, tenant_no=tenant_no
    )
    schedules = (
        db.query(BizProtocolPaymentSchedule)
        .filter(
            BizProtocolPaymentSchedule.sample_payment_id == sample_payment_id,
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
        )
        .order_by(BizProtocolPaymentSchedule.schedule_no.asc())
        .all()
    )
    sch_ids = [s.id for s in schedules]
    transfers = []
    if sch_ids:
        transfers = (
            db.query(BizProtocolPaymentTransfer)
            .filter(
                BizProtocolPaymentTransfer.schedule_id.in_(sch_ids),
                BizProtocolPaymentTransfer.tenant_no == tenant_no,
                BizProtocolPaymentTransfer.del_flag == 0,
            )
            .all()
        )
    sch_dicts = [
        {
            "id": s.id,
            "scheduleNo": s.schedule_no,
            "dueAmount": s.due_amount,
            "scheduleStatus": s.schedule_status or "open",
        }
        for s in schedules
    ]
    tr_dicts = [
        {
            "scheduleId": t.schedule_id,
            "payStatus": t.pay_status or "",
            "transferAmount": t.transfer_amount,
        }
        for t in transfers
    ]
    plan = plan_fq_to_fgd_merge(schedules=sch_dicts, transfers=tr_dicts)
    now = _now()
    target = next((s for s in schedules if s.id == plan["targetScheduleId"]), None)
    if target is None:
        raise ValueError("并期目标期次不存在")
    target.due_amount = plan["newDueAmount"]
    target.update_by = user_id
    target.update_time = now
    db.add(target)

    soft_ids = set(plan["softDeleteScheduleIds"] or [])
    for s in schedules:
        if s.id not in soft_ids:
            continue
        s.del_flag = 1
        s.update_by = user_id
        s.update_time = now
        db.add(s)
    if soft_ids:
        for t in transfers:
            if t.schedule_id in soft_ids and (t.pay_status or "") == "dfj":
                t.del_flag = 1
                t.update_by = user_id
                t.update_time = now
                db.add(t)

    mode = (target_pay_mode or "fgd").strip()
    if mode not in ("dj", "fgd"):
        mode = "fgd"
    sp.pay_mode = mode
    sp.installment_count = plan["newInstallmentCount"]
    sp.update_by = user_id
    sp.update_time = now
    db.add(sp)
    db.flush()
    return plan


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", "fgd"):
        raise ValueError("缴费模式无效")

    obj: Optional[BizProtocolSamplePayment] = None
    if sample_payment_id:
        obj = find_active_sample_payment(
            db,
            sample_payment_id=sample_payment_id,
            collection_protocol_id=collection_protocol_id,
            sample_code=code,
            tenant_no=tenant_no,
        )
        if obj is None:
            raise ValueError("样本缴费方案不存在或不属于当前协议样本")
    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:
        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 _schedule_no_of(raw: Dict[str, Any], fallback: int) -> int:
    v = raw.get("schedule_no")
    if v is None:
        v = raw.get("scheduleNo")
    if v is None:
        return fallback
    try:
        return int(v)
    except (TypeError, ValueError):
        return fallback


def _apply_transfer_editable_fields(
    t: BizProtocolPaymentTransfer, raw: Dict[str, Any]
) -> None:
    """更新转账可编辑字段；不改 pay_status。"""
    if "transfer_type" in raw or "transferType" in raw:
        v = raw.get("transfer_type", raw.get("transferType"))
        t.transfer_type = (str(v) if v is not None else "").strip() or None
    if "transfer_amount" in raw or "transferAmount" in raw:
        amt = _dec(raw.get("transfer_amount", raw.get("transferAmount")))
        if amt is not None and amt < 0:
            raise ValueError("转账金额无效")
        t.transfer_amount = amt
    if "transfer_time" in raw or "transferTime" in raw:
        v = raw.get("transfer_time", raw.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 raw or "transferAttachments" in raw:
        v = raw.get("transfer_attachments", raw.get("transferAttachments"))
        t.transfer_attachments = (str(v) if v is not None else "").strip() or None
    if "remark" in raw:
        t.remark = (raw.get("remark") or "").strip() or None


def _sync_transfers_for_schedule(
    db: Session,
    *,
    sch: BizProtocolPaymentSchedule,
    tenant_no: str,
    user_id: Optional[str],
    transfers: List[Dict[str, Any]],
) -> None:
    now = _now()
    existing = (
        db.query(BizProtocolPaymentTransfer)
        .filter(
            BizProtocolPaymentTransfer.schedule_id == sch.id,
            BizProtocolPaymentTransfer.tenant_no == tenant_no,
            BizProtocolPaymentTransfer.del_flag == 0,
        )
        .order_by(BizProtocolPaymentTransfer.id.asc())
        .populate_existing()
        .with_for_update()
        .all()
    )
    by_id = {t.id: t for t in existing}
    keep_ids: set = set()
    for raw in transfers or []:
        if not isinstance(raw, dict):
            continue
        tid = (raw.get("id") or "").strip()
        if tid and tid in by_id:
            t = by_id[tid]
            keep_ids.add(tid)
            if (t.pay_status or "").strip() != "dfj":
                # 非草稿转账由专用状态流转接口维护，普通方案保存整行只读。
                continue
            _apply_transfer_editable_fields(t, raw)
            t.update_by = user_id
            t.update_time = now
            db.add(t)
            continue
        if tid:
            raise ValueError("转账笔不存在或不属于当前批次")
        # 新增转账笔（始终以待缴费入库；提交财务走独立接口）
        add_transfer(
            db,
            schedule_id=sch.id,
            tenant_no=tenant_no,
            user_id=user_id,
            transfer_amount=raw.get("transfer_amount", raw.get("transferAmount")),
            transfer_time=raw.get("transfer_time", raw.get("transferTime")),
            transfer_type=raw.get("transfer_type", raw.get("transferType")),
            transfer_attachments=raw.get(
                "transfer_attachments", raw.get("transferAttachments")
            ),
            remark=raw.get("remark"),
            commit=False,
        )
    for tid, t in by_id.items():
        if tid in keep_ids:
            continue
        if (t.pay_status or "").strip() != "dfj":
            continue
        t.del_flag = 1
        t.update_by = user_id
        t.update_time = now
        db.add(t)


def replace_schedules(
    db: Session,
    *,
    sample_payment_id: str,
    tenant_no: str,
    user_id: Optional[str],
    schedules: List[Dict[str, Any]],
) -> List[BizProtocolPaymentSchedule]:
    """仅在全部批次为 open、全部转账为 dfj 时重建批次。"""
    sp, old_schedules, transfers = _lock_sample_payment_graph(
        db,
        sample_payment_id=sample_payment_id,
        tenant_no=tenant_no,
    )
    if _locked_schedule_ids(old_schedules, transfers):
        raise ValueError("存在锁定批次，不能省略、删除、重编号或重建")
    now = _now()
    if old_schedules:
        for t in transfers:
            t.del_flag = 1
            t.update_by = user_id
            t.update_time = now
            db.add(t)
        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_no = _schedule_no_of(raw, idx + 1)
        obj = BizProtocolPaymentSchedule(
            id=generate_uuid(),
            sample_payment_id=sp.id,
            collection_protocol_id=sp.collection_protocol_id,
            sample_code=sp.sample_code,
            schedule_no=schedule_no,
            due_amount=_dec(raw.get("due_amount") if "due_amount" in raw else raw.get("dueAmount")),
            due_date=due_date,
            schedule_status="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 sync_schedules_and_transfers(
    db: Session,
    *,
    sample_payment_id: str,
    tenant_no: str,
    user_id: Optional[str],
    schedules: List[Dict[str, Any]],
) -> List[BizProtocolPaymentSchedule]:
    """
    同步应缴批次与转账笔：
    - 有非 open 批次或非 dfj 转账：结构固定，状态与非草稿转账只读保留
    - 全部为 open + dfj：允许重建批次，旧草稿转账随旧批次软删
    """
    _, old_schedules, locked_transfers = _lock_sample_payment_graph(
        db,
        sample_payment_id=sample_payment_id,
        tenant_no=tenant_no,
    )
    now = _now()
    by_no = {int(s.schedule_no or 0): s for s in old_schedules}
    payload = list(schedules or [])
    _assert_locked_schedule_structure(old_schedules, locked_transfers, payload)
    new_nos: List[int] = []
    for idx, raw in enumerate(payload):
        if not isinstance(raw, dict):
            continue
        new_nos.append(_schedule_no_of(raw, idx + 1))
    old_nos = set(by_no.keys())
    new_nos_set = set(new_nos)
    structure_changed = old_nos != new_nos_set

    if structure_changed:
        created = replace_schedules(
            db,
            sample_payment_id=sample_payment_id,
            tenant_no=tenant_no,
            user_id=user_id,
            schedules=payload,
        )
        for i, sch_raw in enumerate(payload):
            if i >= len(created) or not isinstance(sch_raw, dict):
                continue
            _sync_transfers_for_schedule(
                db,
                sch=created[i],
                tenant_no=tenant_no,
                user_id=user_id,
                transfers=sch_raw.get("transfers") or [],
            )
        return created

    result: List[BizProtocolPaymentSchedule] = []
    for idx, raw in enumerate(payload):
        if not isinstance(raw, dict):
            continue
        sno = _schedule_no_of(raw, idx + 1)
        sch = by_no.get(sno)
        if not sch:
            raise ValueError("应缴批次数据不一致，请刷新后重试")
        due_date = (raw.get("due_date") or raw.get("dueDate") or "").strip() or None
        sch.due_amount = _dec(
            raw.get("due_amount") if "due_amount" in raw else raw.get("dueAmount")
        )
        sch.due_date = due_date
        sch.order_num = idx
        if "remark" in raw:
            sch.remark = (raw.get("remark") or "").strip() or None
        sch.update_by = user_id
        sch.update_time = now
        db.add(sch)
        _sync_transfers_for_schedule(
            db,
            sch=sch,
            tenant_no=tenant_no,
            user_id=user_id,
            transfers=raw.get("transfers") or [],
        )
        result.append(sch)
    db.flush()
    return result


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,
    create_dept_id: str = "",
    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_dept_id or "").strip(),
        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],
    locked_transfer: Optional[BizProtocolPaymentTransfer] = None,
) -> BizProtocolPaymentTransfer:
    t = locked_transfer or _get_transfer_or_raise(
        db, transfer_id=transfer_id, tenant_no=tenant_no
    )
    if t.id != transfer_id or t.tenant_no != tenant_no or t.del_flag != 0:
        raise ValueError("转账笔不存在")
    if not can_edit_transfer(t.pay_status):
        raise ValueError("财务已确认收款的转账笔不可编辑")
    _apply_transfer_editable_fields(t, fields)
    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,
    locked_transfer: Optional[BizProtocolPaymentTransfer] = None,
) -> BizProtocolPaymentTransfer:
    t = locked_transfer or _get_transfer_or_raise(
        db, transfer_id=transfer_id, tenant_no=tenant_no
    )
    if t.id != transfer_id or t.tenant_no != tenant_no or t.del_flag != 0:
        raise ValueError("转账笔不存在")
    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,
    locked_transfer: Optional[BizProtocolPaymentTransfer] = None,
) -> BizProtocolPaymentTransfer:
    t = locked_transfer or _get_transfer_with_schedule_lock(
        db, transfer_id=transfer_id, tenant_no=tenant_no
    )
    if t.id != transfer_id or t.tenant_no != tenant_no or t.del_flag != 0:
        raise ValueError("转账笔不存在")
    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)
    # 批次已在 _get_transfer_with_schedule_lock 中加锁，此处不再二次 for_update
    schedule = (
        db.query(BizProtocolPaymentSchedule)
        .filter(
            BizProtocolPaymentSchedule.id == t.schedule_id,
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
        )
        .first()
    )
    _sync_schedule_cleared_after_confirm(
        db,
        schedule=schedule,
        tenant_no=tenant_no,
        user_id=user_id,
    )
    db.commit()
    db.refresh(t)
    return t


def _sync_schedule_cleared_after_confirm(
    db: Session,
    *,
    schedule: Optional[BizProtocolPaymentSchedule],
    tenant_no: str,
    user_id: Optional[str],
) -> None:
    """财务确认后：实缴覆盖应缴则回写 schedule_status=cleared（调用方须已持有批次锁）。"""
    if not schedule:
        return
    status = (schedule.schedule_status or "").strip()
    if status in ("cleared", "void"):
        return
    confirmed = (
        db.query(func.coalesce(func.sum(BizProtocolPaymentTransfer.transfer_amount), 0))
        .filter(
            BizProtocolPaymentTransfer.schedule_id == schedule.id,
            BizProtocolPaymentTransfer.tenant_no == tenant_no,
            BizProtocolPaymentTransfer.del_flag == 0,
            BizProtocolPaymentTransfer.pay_status == "yjf",
        )
        .scalar()
        or Decimal("0")
    )
    due = Decimal(str(schedule.due_amount or 0))
    if Decimal(str(confirmed)) < due:
        return
    schedule.schedule_status = "cleared"
    schedule.update_by = user_id
    schedule.update_time = _now()
    db.add(schedule)


def reject_receipt(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
    locked_transfer: Optional[BizProtocolPaymentTransfer] = None,
) -> BizProtocolPaymentTransfer:
    t = locked_transfer or _get_transfer_with_schedule_lock(
        db, transfer_id=transfer_id, tenant_no=tenant_no
    )
    if t.id != transfer_id or t.tenant_no != tenant_no or t.del_flag != 0:
        raise ValueError("转账笔不存在")
    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 withdraw_receipt(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
    locked_transfer: Optional[BizProtocolPaymentTransfer] = None,
) -> BizProtocolPaymentTransfer:
    """业务撤回：jfqrz → dfj，保留 finance_remark / finance_confirm_attachments。"""
    t = locked_transfer or _get_transfer_with_schedule_lock(
        db, transfer_id=transfer_id, tenant_no=tenant_no
    )
    if t.id != transfer_id or t.tenant_no != tenant_no or t.del_flag != 0:
        raise ValueError("转账笔不存在")
    if not can_withdraw_receipt(t.pay_status):
        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 soft_delete_payment_transfer(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
) -> None:
    t = _get_transfer_or_raise(db, transfer_id=transfer_id, tenant_no=tenant_no)
    if t.pay_status != "dfj":
        raise ValueError("仅已付款状态转账笔可删除")
    t.del_flag = 1
    t.update_by = user_id
    t.update_time = _now()
    db.add(t)
    db.commit()


def _money_value(value: Any, as_string: bool = False) -> Any:
    if value is None:
        return None
    amount = Decimal(str(value)).quantize(Decimal('0.01'))
    return format(amount, 'f') if as_string else float(amount)


def _transfer_to_dict(
    db: Session,
    t: BizProtocolPaymentTransfer,
    tenant_no: str,
    money_as_string: bool = False,
) -> 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": _money_value(t.transfer_amount, money_as_string),
        "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,
    money_as_string: bool = False,
) -> Dict[str, Any]:
    # 展示口径：已缴仅计 yjf（与 summarize_period / chase_period_money 一致）
    paid, arrears = chase_period_money(
        due_amount=s.due_amount,
        transfers=[
            {
                "transferAmount": t.transfer_amount,
                "transferType": t.transfer_type or "",
                "transferTime": t.transfer_time.strftime("%Y-%m-%d %H:%M:%S")
                if t.transfer_time
                else "",
                "payStatus": t.pay_status or "",
            }
            for t in transfers
        ],
    )
    schedule_status = s.schedule_status or ""
    if schedule_status != "void" and is_schedule_effectively_cleared(
        {"scheduleStatus": schedule_status, "dueAmount": s.due_amount},
        confirmed_paid=Decimal(str(paid or 0)),
    ):
        schedule_status = "cleared"
    return {
        "id": s.id,
        "samplePaymentId": s.sample_payment_id,
        "collectionProtocolId": s.collection_protocol_id,
        "sampleCode": s.sample_code or "",
        "scheduleNo": s.schedule_no,
        "dueAmount": _money_value(s.due_amount, money_as_string),
        "dueDate": s.due_date or "",
        "scheduleStatus": schedule_status,
        "paidAmount": _money_value(paid, money_as_string),
        "arrearsAmount": _money_value(arrears, money_as_string),
        "remark": s.remark or "",
        "transfers": [
            _transfer_to_dict(db, t, tenant_no, money_as_string) 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,
    money_as_string: bool = False,
) -> 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": _money_value(p.contract_amount, money_as_string),
        "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, money_as_string
            )
            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)):
        def normalize_decimal(value: Any) -> Any:
            if isinstance(value, Decimal):
                return format(value.quantize(Decimal("0.01")), "f")
            if isinstance(value, list):
                return [normalize_decimal(item) for item in value]
            if isinstance(value, dict):
                return {
                    key: normalize_decimal(item)
                    for key, item in value.items()
                }
            return value

        return json.dumps(normalize_decimal(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:
    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 in ("dfj", "jfqrz"):
                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:
    """按样本保存缴费方案；锁定批次/转账所在方案禁止结构性变更。"""
    existing = (
        db.query(BizProtocolSamplePayment)
        .filter(
            BizProtocolSamplePayment.collection_protocol_id == collection_protocol_id,
            BizProtocolSamplePayment.tenant_no == tenant_no,
            BizProtocolSamplePayment.del_flag == 0,
        )
        .order_by(BizProtocolSamplePayment.id.asc())
        .populate_existing()
        .with_for_update()
        .all()
    )
    payload_rows = [raw for raw in (sample_payments or []) if isinstance(raw, dict)]
    payload_by_id = {
        str(raw.get("id") or raw.get("samplePaymentId") or "").strip(): raw
        for raw in payload_rows
        if str(raw.get("id") or raw.get("samplePaymentId") or "").strip()
    }
    payload_by_code = {
        str(raw.get("sample_code") or raw.get("sampleCode") or "").strip(): raw
        for raw in payload_rows
        if str(raw.get("sample_code") or raw.get("sampleCode") or "").strip()
    }
    for current in existing:
        _, schedules, transfers = _lock_sample_payment_graph(
            db,
            sample_payment_id=current.id,
            tenant_no=tenant_no,
        )
        if not _locked_schedule_ids(schedules, transfers):
            continue
        raw = payload_by_id.get(current.id) or payload_by_code.get(current.sample_code)
        if raw is None:
            raise ValueError("存在锁定缴费方案，不能省略或删除")
        submitted_mode = str(raw.get("pay_mode") or raw.get("payMode") or "").strip()
        if submitted_mode and submitted_mode != (current.pay_mode or "").strip():
            raise ValueError("存在锁定缴费方案，不能变更缴费模式或重建")
        _assert_locked_schedule_structure(
            schedules,
            transfers,
            list(raw.get("schedules") or []),
        )

    keep_ids: set = set()
    for raw in payload_rows:
        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")
        prev = find_active_sample_payment(
            db,
            tenant_no=tenant_no,
            sample_payment_id=sp_id,
            collection_protocol_id=collection_protocol_id,
            sample_code=sample_code,
        )
        prev_mode = (prev.pay_mode or "").strip() if prev else ""
        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)
        if (
            prev_mode == "fq"
            and pay_mode in ("fgd", "dj")
            and sample_payment_has_yjf(
                db, sample_payment_id=sp.id, tenant_no=tenant_no
            )
        ):
            apply_fq_to_fgd_merge(
                db,
                sample_payment_id=sp.id,
                tenant_no=tenant_no,
                user_id=user_id,
                target_pay_mode=pay_mode,
            )
            continue
        schedules_payload = raw.get("schedules") or []
        require_fields = pay_mode in ("fq", "fgd", "djfq")
        ok, mismatch_msg = schedule_dues_match_contract(
            contract_amount=raw.get("contract_amount", raw.get("contractAmount")),
            schedules=schedules_payload,
            sample_code=sample_code,
            require_schedule_fields=require_fields,
        )
        if not ok:
            raise ValueError(mismatch_msg or "各期应缴合计与合同金额不一致")
        sync_schedules_and_transfers(
            db,
            sample_payment_id=sp.id,
            tenant_no=tenant_no,
            user_id=user_id,
            schedules=schedules_payload,
        )
    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 list_deposit_transfers_for_protocol(
    db: Session,
    *,
    source_presign_order_id: Optional[str],
    tenant_no: str,
    money_as_string: bool = False,
) -> List[Dict[str, Any]]:
    """正式协议意向金列表：始终读预签订单 biz_presign_transfer，保证财务审批两边一致。"""
    pid = (source_presign_order_id or "").strip()
    if not pid:
        return []
    from app.crud.presign_order import list_transfers

    result = list_transfers(db, presign_order_id=pid, tenant_no=tenant_no)
    if money_as_string and result:
        rows = (
            db.query(BizPresignTransfer)
            .filter(
                BizPresignTransfer.id.in_([item.get('id') for item in result]),
                BizPresignTransfer.tenant_no == tenant_no,
                BizPresignTransfer.del_flag == 0,
            )
            .all()
        )
        amount_by_id = {str(row.id): _money_value(row.transfer_amount, True) for row in rows}
        for item in result:
            item['transferAmount'] = amount_by_id.get(str(item.get('id') or ''))
    return result


def get_presign_deposit_hint(
    db: Session,
    *,
    source_presign_order_id: Optional[str],
    tenant_no: str,
    money_as_string: bool = False,
) -> 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": _money_value(total, money_as_string),
        "confirmedTransferCount": len(rows),
    }


def get_sample_payments_bundle(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    money_as_string: bool = False,
) -> 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,
            money_as_string,
        )
        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 = [
        {
            "id": s.id,
            "scheduleStatus": s.schedule_status or "",
            "dueAmount": s.due_amount,
        }
        for s in schedules
    ]
    tr_dicts = [
        {
            "scheduleId": t.schedule_id,
            "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],
    *,
    schedules: Optional[List[BizProtocolPaymentSchedule]] = None,
    transfers: Optional[List[BizProtocolPaymentTransfer]] = None,
    sample_meta: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    sch_list = schedules or []
    tr_list = transfers or []
    meta = sample_meta or {}
    active_dues = [
        s.due_amount
        for s in sch_list
        if (s.schedule_status or "") != "void" and s.due_amount is not None
    ]
    receivable = float(sum(active_dues, Decimal("0"))) if active_dues else 0.0
    discount = meta.get("discountAmount")
    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,
        "signDate": _fmt_ymd8(getattr(proto, "register_date", None) or ""),
        "storageYears": meta.get("storageYears"),
        "sampleTypeName": meta.get("sampleTypeName") or "",
        "discountAmount": discount,
        "receivableAmount": receivable,
        "remark": sp.remark or "",
        "periods": _build_periods_array(sch_list, tr_list),
    }
    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
    )
    sample_codes = list({(sp.sample_code or "").strip() for sp, _, _ in candidates if sp.sample_code})
    protocol_ids = list({sp.collection_protocol_id for sp, _, _ in candidates})
    sample_meta_by_key: Dict[str, Dict[str, Any]] = {}
    if sample_codes and protocol_ids:
        s_rows = (
            db.query(BizProtocolSample)
            .filter(
                BizProtocolSample.collection_protocol_id.in_(protocol_ids),
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.sample_code.in_(sample_codes),
            )
            .all()
        )
        type_ids = list({s.sample_type_id for s in s_rows if s.sample_type_id})
        type_name_map: Dict[str, str] = {}
        if type_ids:
            for t in (
                db.query(BizCellSampleType)
                .filter(BizCellSampleType.id.in_(type_ids), BizCellSampleType.del_flag == 0)
                .all()
            ):
                type_name_map[t.id] = t.type_name or ""
        for s in s_rows:
            key = f"{s.collection_protocol_id}|{(s.sample_code or '').strip()}"
            orig = float(s.original_price) if s.original_price is not None else None
            price = float(s.price) if s.price is not None else None
            discount = None
            if orig is not None and price is not None and orig > price:
                discount = round(orig - price, 2)
            sample_meta_by_key[key] = {
                "storageYears": s.storage_years,
                "sampleTypeName": type_name_map.get(s.sample_type_id or "", ""),
                "discountAmount": discount,
            }
    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)
        meta_key = f"{sp.collection_protocol_id}|{(sp.sample_code or '').strip()}"
        rows.append(
            _build_list_summary_row(
                sp,
                proto,
                party,
                progress,
                schedules=schedules,
                transfers=transfers,
                sample_meta=sample_meta_by_key.get(meta_key),
            )
        )
    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


def list_payment_chase_page(
    db: Session,
    *,
    tenant_no: str,
    keyword: Optional[str] = None,
    chase_type: Optional[str] = None,
    pay_mode: Optional[str] = None,
    sales_user_id: Optional[str] = None,
    page: int = 1,
    page_size: int = 10,
    user_info=None,
    today: Optional[Any] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    """催缴列表：一行 = 一个欠缴>0 且（逾期或未来30天内）的应缴期。"""
    from datetime import date as date_cls

    if not tenant_no:
        return [], 0
    today_d = today if isinstance(today, date_cls) else date_cls.today()

    q = (
        db.query(
            BizProtocolPaymentSchedule,
            BizProtocolSamplePayment,
            BizCollectionProtocol,
            BizProtocolPartyCustomer,
            BizProtocolCollectionInfo,
        )
        .join(
            BizProtocolSamplePayment,
            and_(
                BizProtocolSamplePayment.id
                == BizProtocolPaymentSchedule.sample_payment_id,
                BizProtocolSamplePayment.del_flag == 0,
            ),
        )
        .join(
            BizCollectionProtocol,
            and_(
                BizCollectionProtocol.id
                == BizProtocolPaymentSchedule.collection_protocol_id,
                BizCollectionProtocol.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolPartyCustomer,
            and_(
                BizProtocolPartyCustomer.collection_protocol_id
                == BizProtocolPaymentSchedule.collection_protocol_id,
                BizProtocolPartyCustomer.tenant_no == tenant_no,
                BizProtocolPartyCustomer.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolCollectionInfo,
            and_(
                BizProtocolCollectionInfo.collection_protocol_id
                == BizProtocolPaymentSchedule.collection_protocol_id,
                BizProtocolCollectionInfo.tenant_no == tenant_no,
                BizProtocolCollectionInfo.del_flag == 0,
            ),
        )
        .filter(
            BizProtocolPaymentSchedule.tenant_no == tenant_no,
            BizProtocolPaymentSchedule.del_flag == 0,
            BizProtocolPaymentSchedule.schedule_status != "void",
        )
    )
    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)
    sid = (sales_user_id or "").strip()
    if sid:
        q = q.filter(BizProtocolCollectionInfo.sales_user_id == sid)
    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.all()
    if not candidates:
        return [], 0

    sch_ids = [sch.id for sch, _, _, _, _ in candidates]
    transfers = (
        db.query(BizProtocolPaymentTransfer)
        .filter(
            BizProtocolPaymentTransfer.schedule_id.in_(sch_ids),
            BizProtocolPaymentTransfer.tenant_no == tenant_no,
            BizProtocolPaymentTransfer.del_flag == 0,
        )
        .all()
    )
    by_sch: Dict[str, List[BizProtocolPaymentTransfer]] = {}
    for t in transfers:
        by_sch.setdefault(t.schedule_id, []).append(t)

    sales_ids = list(
        {
            (ci.sales_user_id or "").strip()
            for _, _, _, _, ci in candidates
            if ci and ci.sales_user_id
        }
    )
    sales_name_map: Dict[str, str] = {}
    if sales_ids:
        for u in db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(sales_ids)).all():
            sales_name_map[u.id] = u.nick_name or ""

    ct = (chase_type or "").strip()
    rows: List[Dict[str, Any]] = []
    for sch, sp, proto, party, ci in candidates:
        due_d = parse_due_date_ymd(sch.due_date)
        if due_d is None:
            continue
        tr_maps = [
            {
                "transferAmount": t.transfer_amount,
                "transferType": t.transfer_type or "",
                "transferTime": t.transfer_time.strftime("%Y-%m-%d %H:%M:%S")
                if t.transfer_time
                else "",
                "payStatus": t.pay_status or "",
            }
            for t in by_sch.get(sch.id, [])
        ]
        paid, arr = chase_period_money(
            due_amount=sch.due_amount, transfers=tr_maps
        )
        if not is_chase_candidate(due_date=due_d, today=today_d, arrears=arr):
            continue
        overdue = due_d < today_d
        if ct == "overdue" and not overdue:
            continue
        if ct == "upcoming" and overdue:
            continue
        days_overdue = (today_d - due_d).days if overdue else 0
        days_until = (due_d - today_d).days if not overdue else 0
        rows.append(
            {
                "scheduleId": sch.id,
                "samplePaymentId": sp.id,
                "protocolId": proto.id if proto else sch.collection_protocol_id,
                "protocolNo": (proto.protocol_no 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 sch.sample_code or "",
                "salesUserId": (ci.sales_user_id if ci else "") or "",
                "salesUserName": sales_name_map.get(
                    (ci.sales_user_id if ci else "") or "", ""
                ),
                "payMode": sp.pay_mode or "",
                "scheduleNo": sch.schedule_no,
                "dueDate": _fmt_ymd8(sch.due_date),
                "dueAmount": float(sch.due_amount) if sch.due_amount is not None else None,
                "paidAmount": float(paid),
                "arrearsAmount": float(arr),
                "chaseStatus": "overdue" if overdue else "upcoming",
                "daysOverdue": days_overdue,
                "daysUntilDue": days_until,
                "protocolStatus": (proto.protocol_status if proto else "") or "",
                "voidType": (proto.void_type if proto else "") or "",
            }
        )

    rows.sort(
        key=lambda r: (
            0 if r["chaseStatus"] == "overdue" else 1,
            -int(r.get("daysOverdue") or 0),
            str(r.get("dueDate") or ""),
            str(r.get("protocolNo") or ""),
        )
    )
    total = len(rows)
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    start = (page - 1) * page_size
    return rows[start : start + page_size], total