"""按样本缴费方案 / 应缴批次 / 转账笔 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_, 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, ) 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)