Files
zkxw/backend/scripts/migrate_protocol_payment_v1_to_v2.py
T
2026-10-15 23:21:07 +08:00

663 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""采集协议单缴费模型 v1 → v2 迁移(生产专用,可重复执行)。
规则(与 docs/superpowers/specs/2026-10-15-protocol-save-auto-submit-and-v1-to-v2-migration-design.md 一致):
- 单样本 + 无退费:1:1 迁到 sample_payment / schedule / transfer,并置 payment_model_version=2
- 多样本 + 无退费:默认只进人工清单;加 --multi-average 时按样本数平均分合同/应缴/转账
- 有退费痕迹 / 无有效样本 / 无样本编码 / 无 v1 缴费主表:只进人工清单
- 不物理删除 v1 payment / payment_detail
用法(在 backend 目录、已配置 .env):
python -m scripts.migrate_protocol_payment_v1_to_v2 --list-only
python -m scripts.migrate_protocol_payment_v1_to_v2 --dry-run
python -m scripts.migrate_protocol_payment_v1_to_v2
python -m scripts.migrate_protocol_payment_v1_to_v2 --multi-average
python -m scripts.migrate_protocol_payment_v1_to_v2 --multi-average --include-open-payments
python -m scripts.migrate_protocol_payment_v1_to_v2 --protocol-id <ID>
"""
from __future__ import annotations
import argparse
import csv
import os
import sys
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal, ROUND_DOWN, ROUND_HALF_UP
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.core.database import SessionLocal
from app.core.utils import generate_uuid
from app.models.models import (
BizCollectionProtocol,
BizProtocolPayment,
BizProtocolPaymentDetail,
BizProtocolPaymentSchedule,
BizProtocolPaymentTransfer,
BizProtocolSample,
BizProtocolSamplePayment,
)
REFUND_STATUSES = frozenset({"tfqrz", "ytf"})
OPEN_STATUSES = frozenset({"dfj", "jfqrz"})
VOID_SAMPLE_STATUS = "yzf"
CENTS = Decimal("0.01")
def _dec(value: Any) -> Optional[Decimal]:
if value is None or value == "":
return None
try:
return Decimal(str(value))
except Exception:
return None
def _money(value: Any) -> Optional[Decimal]:
d = _dec(value)
if d is None:
return None
return d.quantize(CENTS, rounding=ROUND_HALF_UP)
def split_amount_across(total: Optional[Decimal], n: int) -> List[Optional[Decimal]]:
"""金额按分均分,余数落在最后一项;total 为空则返回全 None。"""
if n <= 0:
raise ValueError("n must be positive")
if total is None:
return [None] * n
cents = int((total.quantize(CENTS, rounding=ROUND_HALF_UP) * 100).to_integral_value(rounding=ROUND_DOWN))
base, rem = divmod(cents, n)
out: List[Optional[Decimal]] = []
for i in range(n):
c = base + (rem if i == n - 1 else 0)
out.append((Decimal(c) / Decimal(100)).quantize(CENTS))
return out
def detail_has_refund(d: BizProtocolPaymentDetail) -> bool:
status = (d.pay_status or "").strip()
if status in REFUND_STATUSES:
return True
amt = _dec(d.refund_amount)
if amt is not None and amt > 0:
return True
for attr in (
"refund_attachments",
"refund_remark",
"refund_finance_confirm_attachments",
"refund_finance_remark",
):
if str(getattr(d, attr, None) or "").strip():
return True
return False
def _is_active_sample(s: BizProtocolSample) -> bool:
if int(s.del_flag or 0) != 0:
return False
status = (s.sample_status or "").strip()
if status == VOID_SAMPLE_STATUS:
return False
return True
def _schedule_no_of(d: BizProtocolPaymentDetail, *, pay_mode: str) -> int:
if (pay_mode or "").strip() != "fq":
return 1
no = int(d.installment_no or 0)
return no if no > 0 else 1
def _append_remark(existing: Optional[str], extra: str) -> str:
base = (existing or "").strip()
if not base:
return extra[:512]
joined = f"{base} | {extra}"
return joined[:512]
@dataclass
class ProtocolCandidate:
protocol: BizCollectionProtocol
samples: List[BizProtocolSample]
payment: Optional[BizProtocolPayment]
details: List[BizProtocolPaymentDetail]
skip_reason: Optional[str] = None
has_open_payment: bool = False
@property
def sample_count(self) -> int:
return len(self.samples)
@property
def protocol_id(self) -> str:
return str(self.protocol.id)
@dataclass
class MigrateStats:
scanned: int = 0
migrated: int = 0
skipped_already_v2: int = 0
listed: List[Dict[str, Any]] = field(default_factory=list)
errors: List[str] = field(default_factory=list)
def load_v1_protocols(
db,
*,
protocol_id: Optional[str] = None,
) -> List[BizCollectionProtocol]:
q = db.query(BizCollectionProtocol).filter(
BizCollectionProtocol.del_flag == 0,
BizCollectionProtocol.payment_model_version == 1,
)
if protocol_id:
q = q.filter(BizCollectionProtocol.id == protocol_id)
return q.order_by(BizCollectionProtocol.create_time.asc()).all()
def has_v2_sample_payment(db, protocol_id: str) -> bool:
row = (
db.query(BizProtocolSamplePayment.id)
.filter(
BizProtocolSamplePayment.collection_protocol_id == protocol_id,
BizProtocolSamplePayment.del_flag == 0,
)
.first()
)
return row is not None
def build_candidate(db, protocol: BizCollectionProtocol) -> ProtocolCandidate:
samples = (
db.query(BizProtocolSample)
.filter(
BizProtocolSample.collection_protocol_id == protocol.id,
BizProtocolSample.del_flag == 0,
)
.order_by(BizProtocolSample.order_num.asc(), BizProtocolSample.create_time.asc())
.all()
)
active = [s for s in samples if _is_active_sample(s)]
payment = (
db.query(BizProtocolPayment)
.filter(
BizProtocolPayment.collection_protocol_id == protocol.id,
BizProtocolPayment.del_flag == 0,
)
.order_by(BizProtocolPayment.create_time.asc())
.first()
)
details: List[BizProtocolPaymentDetail] = []
if payment:
details = (
db.query(BizProtocolPaymentDetail)
.filter(
BizProtocolPaymentDetail.payment_id == payment.id,
BizProtocolPaymentDetail.del_flag == 0,
)
.order_by(
BizProtocolPaymentDetail.installment_no.asc(),
BizProtocolPaymentDetail.create_time.asc(),
)
.all()
)
cand = ProtocolCandidate(
protocol=protocol,
samples=active,
payment=payment,
details=details,
has_open_payment=any((d.pay_status or "").strip() in OPEN_STATUSES for d in details),
)
if has_v2_sample_payment(db, protocol.id):
cand.skip_reason = "already_has_v2"
return cand
if not payment:
cand.skip_reason = "no_v1_payment"
return cand
if not active:
cand.skip_reason = "no_sample"
return cand
if any(not str(s.sample_code or "").strip() for s in active):
cand.skip_reason = "no_sample_code"
return cand
if any(detail_has_refund(d) for d in details):
cand.skip_reason = "has_refund"
return cand
if len(active) != 1:
cand.skip_reason = "multi_sample"
return cand
return cand
def _list_row(cand: ProtocolCandidate) -> Dict[str, Any]:
p = cand.protocol
pay = cand.payment
return {
"protocol_id": p.id,
"protocol_no": p.protocol_no,
"tenant_no": p.tenant_no,
"sample_count": cand.sample_count,
"sample_codes": ",".join(str(s.sample_code or "") for s in cand.samples),
"contract_amount": str(pay.contract_amount) if pay and pay.contract_amount is not None else "",
"detail_count": len(cand.details),
"has_open_payment": "1" if cand.has_open_payment else "0",
"skip_reason": cand.skip_reason or "",
}
def _new_common(
*,
tenant_no: str,
user_id: Optional[str],
create_dept_id: str,
create_by: Optional[str],
create_time: Optional[datetime],
order_num: Optional[int],
remark: Optional[str],
) -> Dict[str, Any]:
now = datetime.now()
return {
"tenant_no": tenant_no,
"order_num": order_num,
"remark": remark,
"create_by": create_by or user_id,
"create_dept_id": (create_dept_id or "").strip(),
"create_time": create_time or now,
"update_by": user_id,
"update_time": now,
"del_flag": 0,
}
def _group_details(
details: Sequence[BizProtocolPaymentDetail], *, pay_mode: str
) -> Dict[int, List[BizProtocolPaymentDetail]]:
groups: Dict[int, List[BizProtocolPaymentDetail]] = defaultdict(list)
for d in details:
groups[_schedule_no_of(d, pay_mode=pay_mode)].append(d)
return dict(sorted(groups.items(), key=lambda x: x[0]))
def _due_from_group(
rows: Sequence[BizProtocolPaymentDetail],
*,
payment: BizProtocolPayment,
schedule_no: int,
) -> Tuple[Optional[Decimal], Optional[str]]:
due_amount = None
due_date = None
for d in rows:
if due_amount is None and d.due_amount is not None:
due_amount = _money(d.due_amount)
if not due_date and str(d.installment_due_date or "").strip():
due_date = str(d.installment_due_date).strip()
if due_amount is None and schedule_no == 1 and (payment.pay_mode or "").strip() != "fq":
due_amount = _money(payment.contract_amount)
if not due_date and (payment.pay_mode or "").strip() != "fq":
due_date = (payment.lump_sum_deadline or "").strip() or None
return due_amount, due_date
def migrate_single_sample(
db,
cand: ProtocolCandidate,
*,
user_id: Optional[str],
dry_run: bool,
) -> None:
assert cand.payment is not None
assert len(cand.samples) == 1
sample = cand.samples[0]
payment = cand.payment
protocol = cand.protocol
pay_mode = (payment.pay_mode or "dj").strip() or "dj"
common = _new_common(
tenant_no=protocol.tenant_no,
user_id=user_id,
create_dept_id=payment.create_dept_id or protocol.create_dept_id or "",
create_by=payment.create_by,
create_time=payment.create_time,
order_num=payment.order_num,
remark=payment.remark,
)
sp_id = generate_uuid()
sp = BizProtocolSamplePayment(
id=sp_id,
collection_protocol_id=protocol.id,
protocol_sample_id=sample.id,
sample_code=str(sample.sample_code).strip(),
pay_mode=pay_mode,
contract_amount=payment.contract_amount,
installment_count=payment.installment_count,
installment_interval_months=payment.installment_interval_months,
installment_start_date=payment.installment_start_date,
installment_advance_days=payment.installment_advance_days,
installment_ratios=payment.installment_ratios,
lump_sum_deadline=payment.lump_sum_deadline,
**common,
)
if not dry_run:
db.add(sp)
groups = _group_details(cand.details, pay_mode=pay_mode)
if not groups:
# 无明细:仍建一期空 schedule,保证方案可编辑
groups = {1: []}
for schedule_no, rows in groups.items():
due_amount, due_date = _due_from_group(rows, payment=payment, schedule_no=schedule_no)
sch_id = generate_uuid()
sch = BizProtocolPaymentSchedule(
id=sch_id,
sample_payment_id=sp_id,
collection_protocol_id=protocol.id,
sample_code=str(sample.sample_code).strip(),
schedule_no=schedule_no,
due_amount=due_amount,
due_date=due_date,
schedule_status="open",
**_new_common(
tenant_no=protocol.tenant_no,
user_id=user_id,
create_dept_id=payment.create_dept_id or protocol.create_dept_id or "",
create_by=payment.create_by,
create_time=payment.create_time,
order_num=schedule_no,
remark=None,
),
)
if not dry_run:
db.add(sch)
for d in rows:
tr = BizProtocolPaymentTransfer(
id=generate_uuid(),
schedule_id=sch_id,
sample_payment_id=sp_id,
collection_protocol_id=protocol.id,
sample_code=str(sample.sample_code).strip(),
transfer_amount=d.transfer_amount,
transfer_time=d.transfer_time,
transfer_type=d.transfer_type,
transfer_attachments=d.transfer_attachments,
pay_status=(d.pay_status or "dfj").strip() or "dfj",
finance_confirm_attachments=d.finance_confirm_attachments,
finance_remark=d.finance_remark,
**_new_common(
tenant_no=protocol.tenant_no,
user_id=user_id,
create_dept_id=d.create_dept_id or payment.create_dept_id or "",
create_by=d.create_by,
create_time=d.create_time,
order_num=d.order_num,
remark=_append_remark(d.remark, f"migrated_from_v1_detail:{d.id}"),
),
)
if not dry_run:
db.add(tr)
if not dry_run:
protocol.payment_model_version = 2
protocol.update_by = user_id
protocol.update_time = datetime.now()
db.add(protocol)
def migrate_multi_average(
db,
cand: ProtocolCandidate,
*,
user_id: Optional[str],
dry_run: bool,
) -> None:
"""多样本:合同/应缴/转账金额均分;状态与凭证复制;备注标注来源明细。"""
assert cand.payment is not None
assert len(cand.samples) >= 2
payment = cand.payment
protocol = cand.protocol
samples = cand.samples
n = len(samples)
pay_mode = (payment.pay_mode or "dj").strip() or "dj"
contract_parts = split_amount_across(_money(payment.contract_amount), n)
groups = _group_details(cand.details, pay_mode=pay_mode)
if not groups:
groups = {1: []}
# 每期应缴:取组代表金额后均分
due_parts_by_no: Dict[int, List[Optional[Decimal]]] = {}
due_date_by_no: Dict[int, Optional[str]] = {}
for schedule_no, rows in groups.items():
due_amount, due_date = _due_from_group(rows, payment=payment, schedule_no=schedule_no)
due_parts_by_no[schedule_no] = split_amount_across(due_amount, n)
due_date_by_no[schedule_no] = due_date
for idx, sample in enumerate(samples):
code = str(sample.sample_code).strip()
sp_id = generate_uuid()
sp = BizProtocolSamplePayment(
id=sp_id,
collection_protocol_id=protocol.id,
protocol_sample_id=sample.id,
sample_code=code,
pay_mode=pay_mode,
contract_amount=contract_parts[idx],
installment_count=payment.installment_count,
installment_interval_months=payment.installment_interval_months,
installment_start_date=payment.installment_start_date,
installment_advance_days=payment.installment_advance_days,
installment_ratios=payment.installment_ratios,
lump_sum_deadline=payment.lump_sum_deadline,
**_new_common(
tenant_no=protocol.tenant_no,
user_id=user_id,
create_dept_id=payment.create_dept_id or protocol.create_dept_id or "",
create_by=payment.create_by,
create_time=payment.create_time,
order_num=idx + 1,
remark=_append_remark(payment.remark, f"v1_multi_average:{n}"),
),
)
if not dry_run:
db.add(sp)
for schedule_no, rows in groups.items():
sch_id = generate_uuid()
sch = BizProtocolPaymentSchedule(
id=sch_id,
sample_payment_id=sp_id,
collection_protocol_id=protocol.id,
sample_code=code,
schedule_no=schedule_no,
due_amount=due_parts_by_no[schedule_no][idx],
due_date=due_date_by_no[schedule_no],
schedule_status="open",
**_new_common(
tenant_no=protocol.tenant_no,
user_id=user_id,
create_dept_id=payment.create_dept_id or protocol.create_dept_id or "",
create_by=payment.create_by,
create_time=payment.create_time,
order_num=schedule_no,
remark=None,
),
)
if not dry_run:
db.add(sch)
for d in rows:
amt_parts = split_amount_across(_money(d.transfer_amount), n)
tr = BizProtocolPaymentTransfer(
id=generate_uuid(),
schedule_id=sch_id,
sample_payment_id=sp_id,
collection_protocol_id=protocol.id,
sample_code=code,
transfer_amount=amt_parts[idx],
transfer_time=d.transfer_time,
transfer_type=d.transfer_type,
transfer_attachments=d.transfer_attachments,
pay_status=(d.pay_status or "dfj").strip() or "dfj",
finance_confirm_attachments=d.finance_confirm_attachments,
finance_remark=d.finance_remark,
**_new_common(
tenant_no=protocol.tenant_no,
user_id=user_id,
create_dept_id=d.create_dept_id or payment.create_dept_id or "",
create_by=d.create_by,
create_time=d.create_time,
order_num=d.order_num,
remark=_append_remark(
d.remark, f"拆自原明细:{d.id} ({idx + 1}/{n})"
),
),
)
if not dry_run:
db.add(tr)
if not dry_run:
protocol.payment_model_version = 2
protocol.update_by = user_id
protocol.update_time = datetime.now()
db.add(protocol)
def run_migration(
db,
*,
dry_run: bool,
list_only: bool,
multi_average: bool,
include_open_payments: bool,
protocol_id: Optional[str],
user_id: Optional[str],
list_csv_path: Optional[str],
) -> MigrateStats:
stats = MigrateStats()
protocols = load_v1_protocols(db, protocol_id=protocol_id)
stats.scanned = len(protocols)
for protocol in protocols:
if has_v2_sample_payment(db, protocol.id):
stats.skipped_already_v2 += 1
continue
cand = build_candidate(db, protocol)
# 可自动迁:单样本无 skip;或多平均且 skip 仅为 multi_sample
can_single = cand.skip_reason is None
can_multi = (
multi_average
and cand.skip_reason == "multi_sample"
and cand.payment is not None
and cand.sample_count >= 2
and all(str(s.sample_code or "").strip() for s in cand.samples)
)
if can_multi and cand.has_open_payment and not include_open_payments:
cand.skip_reason = "multi_sample_open_payment"
can_multi = False
if list_only or (not can_single and not can_multi):
if cand.skip_reason and cand.skip_reason != "already_has_v2":
stats.listed.append(_list_row(cand))
continue
try:
if can_single:
migrate_single_sample(db, cand, user_id=user_id, dry_run=dry_run)
else:
# 临时清掉 skip,便于日志
cand.skip_reason = None
migrate_multi_average(db, cand, user_id=user_id, dry_run=dry_run)
if not dry_run:
db.commit()
else:
db.rollback()
stats.migrated += 1
print(
f"[OK] {'DRY ' if dry_run else ''}migrated "
f"{protocol.protocol_no} ({protocol.id}) samples={cand.sample_count}"
)
except Exception as exc: # noqa: BLE001
db.rollback()
msg = f"[ERR] {protocol.protocol_no} ({protocol.id}): {exc}"
stats.errors.append(msg)
print(msg)
if stats.listed:
print(f"\n=== 人工清单 {len(stats.listed)} 条 ===")
writer = csv.DictWriter(
sys.stdout,
fieldnames=list(stats.listed[0].keys()),
extrasaction="ignore",
)
writer.writeheader()
writer.writerows(stats.listed)
if list_csv_path:
with open(list_csv_path, "w", encoding="utf-8", newline="") as f:
w = csv.DictWriter(f, fieldnames=list(stats.listed[0].keys()))
w.writeheader()
w.writerows(stats.listed)
print(f"\n清单已写入: {list_csv_path}")
return stats
def main(argv: Optional[Sequence[str]] = None) -> int:
parser = argparse.ArgumentParser(description="Migrate protocol payment v1 → v2")
parser.add_argument("--dry-run", action="store_true", help="只演练,不写库")
parser.add_argument("--list-only", action="store_true", help="只输出人工清单,不迁移")
parser.add_argument(
"--multi-average",
action="store_true",
help="多样本无退费时按平均分自动迁移(默认:多样本仅进清单)",
)
parser.add_argument(
"--include-open-payments",
action="store_true",
help="与 --multi-average 联用:允许 dfj/jfqrz 明细一并均分(有财务重复确认风险)",
)
parser.add_argument("--protocol-id", default=None, help="只处理指定协议 ID")
parser.add_argument("--user-id", default="migrate_v1_v2", help="写入 update_by")
parser.add_argument(
"--list-csv",
default=None,
help="人工清单另存 CSV 路径",
)
args = parser.parse_args(argv)
db = SessionLocal()
try:
stats = run_migration(
db,
dry_run=bool(args.dry_run),
list_only=bool(args.list_only),
multi_average=bool(args.multi_average),
include_open_payments=bool(args.include_open_payments),
protocol_id=(args.protocol_id or "").strip() or None,
user_id=(args.user_id or "").strip() or None,
list_csv_path=(args.list_csv or "").strip() or None,
)
finally:
db.close()
print(
"\n=== 汇总 ===\n"
f"scanned_v1={stats.scanned}\n"
f"migrated={stats.migrated}\n"
f"skipped_already_v2_data={stats.skipped_already_v2}\n"
f"manual_list={len(stats.listed)}\n"
f"errors={len(stats.errors)}"
)
return 1 if stats.errors else 0
if __name__ == "__main__":
raise SystemExit(main())