From 8827ec38048108d4aec167bd9b959f1bd2754566 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Sat, 3 Oct 2026 10:58:21 +0800 Subject: [PATCH] feat: add sample payment schedule and transfer status rules Co-authored-by: Cursor --- backend/app/crud/protocol_sample_payment.py | 701 ++++++++++++++++++ .../app/services/protocol_transfer_status.py | 25 + backend/tests/test_protocol_sample_payment.py | 44 ++ 3 files changed, 770 insertions(+) create mode 100644 backend/app/crud/protocol_sample_payment.py create mode 100644 backend/app/services/protocol_transfer_status.py create mode 100644 backend/tests/test_protocol_sample_payment.py diff --git a/backend/app/crud/protocol_sample_payment.py b/backend/app/crud/protocol_sample_payment.py new file mode 100644 index 00000000..81a8d61a --- /dev/null +++ b/backend/app/crud/protocol_sample_payment.py @@ -0,0 +1,701 @@ +"""按样本缴费方案 / 应缴批次 / 转账笔 CRUD（payment_model_version=2）""" + +from __future__ import annotations + +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, + 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, +) -> 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) + db.commit() + db.refresh(obj) + 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 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, +) -> Tuple[List[Dict[str, Any]], int]: + if not tenant_no: + return [], 0 + st = (pay_status or "").strip() + q = ( + db.query(BizProtocolPaymentTransfer, BizCollectionProtocol, BizProtocolPartyCustomer) + .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, + ), + ) + .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), + ) + ) + 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 in rows: + item = _transfer_to_dict(db, t, tenant_no) + item["protocolNo"] = (proto.protocol_no if proto else "") or "" + item["motherName"] = (party.mother_name if party else "") or "" + out.append(item) + return out, int(total) diff --git a/backend/app/services/protocol_transfer_status.py b/backend/app/services/protocol_transfer_status.py new file mode 100644 index 00000000..a54220e4 --- /dev/null +++ b/backend/app/services/protocol_transfer_status.py @@ -0,0 +1,25 @@ +"""转账笔缴费状态规则（正式样本缴费 / 预签约意向金共用）""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Iterable, Optional, Tuple, Union + + +def can_submit_receipt(pay_status: Optional[str]) -> bool: + return (pay_status or "").strip() == "dfj" + + +def can_confirm_receipt(pay_status: Optional[str]) -> bool: + return (pay_status or "").strip() == "jfqrz" + + +def schedule_paid_and_arrears( + *, + due: Union[Decimal, float, int, str, None], + confirmed_transfers: Iterable[Union[Decimal, float, int, str, None]], +) -> Tuple[Decimal, Decimal]: + """已缴 = Σ 已确认转账；欠缴 = 应缴 − 已缴（可为负表示超缴）。""" + due_amt = Decimal(str(due or 0)) + paid = sum((Decimal(str(x or 0)) for x in confirmed_transfers), Decimal("0")) + return paid, due_amt - paid diff --git a/backend/tests/test_protocol_sample_payment.py b/backend/tests/test_protocol_sample_payment.py new file mode 100644 index 00000000..9b9aa367 --- /dev/null +++ b/backend/tests/test_protocol_sample_payment.py @@ -0,0 +1,44 @@ +"""按样本缴费：转账状态机与批次已缴/欠缴计算""" + +from decimal import Decimal +import unittest + +from app.services.protocol_transfer_status import ( + can_confirm_receipt, + can_submit_receipt, + schedule_paid_and_arrears, +) + + +class TestTransferStatusRules(unittest.TestCase): + def test_submit_only_dfj(self): + self.assertTrue(can_submit_receipt("dfj")) + self.assertFalse(can_submit_receipt("jfqrz")) + self.assertFalse(can_submit_receipt("yjf")) + self.assertFalse(can_submit_receipt("")) + self.assertFalse(can_submit_receipt(None)) + + def test_confirm_only_jfqrz(self): + self.assertTrue(can_confirm_receipt("jfqrz")) + self.assertFalse(can_confirm_receipt("dfj")) + self.assertFalse(can_confirm_receipt("yjf")) + + def test_schedule_arrears(self): + paid, arrears = schedule_paid_and_arrears( + due=Decimal("1000"), + confirmed_transfers=[Decimal("300"), Decimal("200")], + ) + self.assertEqual(paid, Decimal("500")) + self.assertEqual(arrears, Decimal("500")) + + def test_schedule_arrears_overpay(self): + paid, arrears = schedule_paid_and_arrears( + due=Decimal("100"), + confirmed_transfers=[Decimal("60"), Decimal("50")], + ) + self.assertEqual(paid, Decimal("110")) + self.assertEqual(arrears, Decimal("-10")) + + +if __name__ == "__main__": + unittest.main()