feat: add sample payment schedule and transfer status rules
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed
1 parent
ca3d464761
commit
8827ec3804
3 files changed
+770
No files matched your search
@@ -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)
|
||||
@@ -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
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user