zkxw/backend/scripts/migrate_protocol_payment_v1_to_v2.py

#!/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())