shanxia2.0/backend/scripts/migrate_20260808_pl_plan_legacy_party_contracts.py

#!/usr/bin/env python3
"""将历史运输计划的单一合同/客户，按单据类型落到供方或需方。

规则（与产品确认一致）：
- 出库单 outbound / 直销单 direct / 其它或空：旧合同 + 客户 → 需方
- 入库单 inbound：旧合同 + 客户 → 供方（customer_* 清空，写入 supplier_*）

处理范围：
1) 仍有 pl_plan.contract_id，且尚无 pl_plan_contract 有效关联 → 写入分侧合同
2) 无合同但有客户：按单据类型把客户落到供方/需方（便于用户继续完善）

用法（在 backend 目录、已激活 venv）：
  ENV_FILE=.env.production python scripts/migrate_20260808_pl_plan_legacy_party_contracts.py --dry-run
  ENV_FILE=.env.production python scripts/migrate_20260808_pl_plan_legacy_party_contracts.py
"""
from __future__ import annotations

import argparse
import sys
from datetime import datetime
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from app.core.database import SessionLocal  # noqa: E402
from app.core.utils import generate_uuid  # noqa: E402
from app.models.pl_models import PlPlan, PlPlanContract  # noqa: E402


def _resolve_side(plan_type: str) -> str:
    return "supplier" if (plan_type or "").strip() == "inbound" else "demander"


def _has_text(value) -> bool:
    return bool(str(value or "").strip())


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--dry-run", action="store_true", help="只统计不落库")
    args = parser.parse_args()

    db = SessionLocal()
    now = datetime.now()
    linked = 0
    party_updated = 0
    skipped = 0
    try:
        plans = db.query(PlPlan).filter(PlPlan.del_flag == 0).all()
        for plan in plans:
            side = _resolve_side(plan.plan_type or "")
            has_contract = _has_text(plan.contract_id)
            has_customer = _has_text(plan.customer_id) or _has_text(plan.customer_name)
            has_supplier = _has_text(plan.supplier_id) or _has_text(plan.supplier_name)

            exists_link = (
                db.query(PlPlanContract.id)
                .filter(
                    PlPlanContract.plan_id == plan.id,
                    PlPlanContract.del_flag == 0,
                )
                .first()
            )

            changed = False

            # 1) 合同落到分侧关联表
            if has_contract and not exists_link:
                if not args.dry_run:
                    db.add(
                        PlPlanContract(
                            id=generate_uuid(),
                            plan_id=plan.id,
                            side=side,
                            contract_id=plan.contract_id,
                            contract_no=plan.contract_no,
                            tenant_no=plan.tenant_no,
                            order_num=1,
                            create_by=plan.update_by or plan.create_by,
                            create_dept_id=plan.create_dept_id or "",
                            create_time=now,
                            update_by=plan.update_by or plan.create_by,
                            update_time=now,
                            del_flag=0,
                        )
                    )
                linked += 1
                changed = True

            # 2) 客户落到供方/需方
            if side == "supplier":
                # 入库：客户 → 供方；清空需方客户
                # - 供方为空：整段迁移
                # - 供方与客户同名：清掉重复的需方客户
                # - 供方与客户不同：保留两边，留给人工确认（不覆盖供方）
                if has_customer and not has_supplier:
                    if not args.dry_run:
                        plan.supplier_id = plan.customer_id
                        plan.supplier_name = plan.customer_name
                        plan.customer_id = None
                        plan.customer_name = None
                        plan.destination_id = None
                        plan.destination_name = None
                        plan.update_time = now
                    party_updated += 1
                    changed = True
                elif (
                    has_customer
                    and has_supplier
                    and str(plan.customer_name or "").strip()
                    == str(plan.supplier_name or "").strip()
                ):
                    if not args.dry_run:
                        plan.customer_id = None
                        plan.customer_name = None
                        plan.destination_id = None
                        plan.destination_name = None
                        plan.update_time = now
                    party_updated += 1
                    changed = True
            else:
                # 出库/直销：客户保留在需方；补齐 destination
                if has_customer and not _has_text(plan.destination_name):
                    if not args.dry_run:
                        plan.destination_id = plan.customer_id
                        plan.destination_name = plan.customer_name
                        plan.update_time = now
                    party_updated += 1
                    changed = True

            if not changed:
                skipped += 1

        if args.dry_run:
            db.rollback()
            print(
                f"[dry-run] would link contracts={linked}, "
                f"update party fields={party_updated}, skipped={skipped}"
            )
        else:
            db.commit()
            print(
                f"ok linked contracts={linked}, "
                f"updated party fields={party_updated}, skipped={skipped}"
            )
        return 0
    except Exception as exc:
        db.rollback()
        print(f"ERR {type(exc).__name__}: {exc}")
        return 1
    finally:
        db.close()


if __name__ == "__main__":
    raise SystemExit(main())