shanxia2.0/backend/scripts/migrate_crm_quote.py

#!/usr/bin/env python3
"""将 sx 库 CRM 报价单相关表迁入 shanxia2.0。

顺序：crm_supplier_info（依赖）→ crm_quote → crm_quote_product → crm_quote_approval → crm_quote_share_log
用户外键经 user_mapping；附件/pdf OSS ID 或绝对 URL → 相对路径。

用法：
  cd backend
  venv/bin/python scripts/migrate_crm_quote.py --dry-run --env-file .env.production
  venv/bin/python scripts/migrate_crm_quote.py --env-file .env.production
"""
from __future__ import annotations

import argparse
import csv
import os
import sys
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import urlparse, urlunparse

from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine

BACKEND_ROOT = Path(__file__).resolve().parents[1]
MIGRATION_DIR = BACKEND_ROOT / "scripts" / "migration"
if str(BACKEND_ROOT) not in sys.path:
    sys.path.insert(0, str(BACKEND_ROOT))
if str(MIGRATION_DIR) not in sys.path:
    sys.path.insert(0, str(MIGRATION_DIR))

from minio_url import extract_minio_object_key, join_relative_urls, resolve_oss_id_csv  # noqa: E402

DEFAULT_TENANT_NO = "shanxia2.0"
CRM_ADMIN_SOURCE_USER_ID = 1

MIGRATION_STEPS = (
    ("crm_supplier_info", "supplier_info"),
    ("crm_quote", "quote"),
    ("crm_quote_product", "quote_product"),
    ("crm_quote_approval", "quote_approval"),
    ("crm_quote_share_log", "quote_share_log"),
)

TABLES_WITHOUT_DEL_FLAG = frozenset({"crm_quote_share_log"})


@dataclass
class StepStats:
    source_count: int = 0
    inserted: int = 0
    updated: int = 0
    skipped: int = 0


@dataclass
class MigrationStats:
    steps: Dict[str, StepStats] = field(default_factory=dict)
    warnings: List[str] = field(default_factory=list)

    def step(self, name: str) -> StepStats:
        if name not in self.steps:
            self.steps[name] = StepStats()
        return self.steps[name]


def legacy_id_to_str(legacy_id: Any) -> Optional[str]:
    if legacy_id is None or str(legacy_id).strip() == "":
        return None
    try:
        return str(int(legacy_id))
    except (TypeError, ValueError):
        return str(legacy_id).strip()


def legacy_parent_id(raw: Any) -> str:
    if raw is None:
        return "0"
    try:
        val = int(raw)
        return "0" if val == 0 else str(val)
    except (TypeError, ValueError):
        s = str(raw).strip()
        return s if s else "0"


def derive_source_url(target_url: str, source_db: str = "sx") -> str:
    return urlunparse(urlparse(target_url)._replace(path=f"/{source_db}"))


def load_settings_env(env_file: Optional[str]) -> None:
    if env_file:
        os.environ["ENV_FILE"] = env_file
    from dotenv import load_dotenv

    candidates: List[Path] = []
    if env_file:
        p = Path(env_file)
        candidates.append(p if p.is_absolute() else BACKEND_ROOT / p)
    candidates.extend([BACKEND_ROOT / ".env.production", BACKEND_ROOT / ".env"])
    for path in candidates:
        if path.is_file():
            load_dotenv(path, override=True)
            break


def load_runtime_settings():
    from app.core.config import Settings

    return Settings()


def load_user_mapping(path: Path) -> Tuple[Dict[int, str], Optional[str]]:
    crm_map: Dict[int, str] = {}
    admin_id: Optional[str] = None
    with path.open(encoding="utf-8-sig", newline="") as f:
        for row in csv.DictReader(f):
            new_id = (row.get("new_user_id") or "").strip()
            user_no = (row.get("user_no") or "").strip()
            crm_raw = (row.get("crm_user_id") or "").strip()
            if user_no == "admin" and new_id:
                admin_id = new_id
            if crm_raw and new_id:
                try:
                    crm_map[int(crm_raw)] = new_id
                except ValueError:
                    pass
    return crm_map, admin_id


def fetch_portal_admin_id(target_engine: Engine) -> Optional[str]:
    with target_engine.connect() as conn:
        row = conn.execute(
            text("SELECT id FROM sys_user WHERE user_no='admin' AND del_flag=0 LIMIT 1")
        ).first()
        return str(row[0]) if row and row[0] else None


def resolve_portal_user_id(
    source_user_id: Any,
    crm_user_map: Dict[int, str],
    admin_id: Optional[str],
    stats: MigrationStats,
    label: str,
    *,
    required: bool = False,
) -> Optional[str]:
    if source_user_id is None or str(source_user_id).strip() == "":
        return admin_id if required else None
    try:
        uid = int(source_user_id)
    except (TypeError, ValueError):
        stats.warnings.append(f"{label}: 非法用户 id {source_user_id!r}")
        return admin_id if required else None
    if uid == CRM_ADMIN_SOURCE_USER_ID:
        return admin_id
    mapped = crm_user_map.get(uid)
    if not mapped:
        stats.warnings.append(f"{label}: crm_user_id={uid} 无映射")
        return admin_id if required else None
    return mapped


def load_oss_id_to_url(target_engine: Engine, tenant_no: str) -> Dict[str, str]:
    with target_engine.connect() as conn:
        rows = conn.execute(
            text("SELECT id, url FROM sys_oss WHERE tenant_no=:t AND del_flag=0"),
            {"t": tenant_no},
        ).mappings().all()
    return {str(r["id"]): (r["url"] or "").strip() for r in rows if r["id"]}


def normalize_attachment_csv(raw: Any, oss_map: Dict[str, str]) -> Optional[str]:
    if raw is None or str(raw).strip() == "":
        return None
    oss_resolved = resolve_oss_id_csv(raw, oss_map)
    if oss_resolved:
        return oss_resolved or None
    parts: List[str] = []
    for token in str(raw).split(","):
        token = token.strip()
        if not token:
            continue
        if token.isdigit() and token in oss_map:
            parts.append(oss_map[token])
        else:
            key = extract_minio_object_key(token)
            if key:
                parts.append(key)
    joined = join_relative_urls(parts)
    return joined or None


def normalize_pdf_url(raw: Any, oss_map: Dict[str, str]) -> Optional[str]:
    if raw is None or str(raw).strip() == "":
        return None
    value = str(raw).strip()
    if value.isdigit() and value in oss_map:
        return oss_map[value]
    key = extract_minio_object_key(value)
    return key or None


def _dec(value: Any, default: Optional[Decimal] = None) -> Optional[Decimal]:
    if value is None or value == "":
        return default
    return Decimal(str(value))


def _str_or_none(value: Any, max_len: int = 0) -> Optional[str]:
    if value is None:
        return None
    s = str(value).strip()
    if not s:
        return None
    return s[:max_len] if max_len else s


def fetch_source_rows(engine: Engine, table: str, include_deleted: bool) -> List[dict]:
    if table in TABLES_WITHOUT_DEL_FLAG:
        where = ""
    else:
        where = "" if include_deleted else "WHERE del_flag = 0"
    sql = f"SELECT * FROM `{table}` {where} ORDER BY id ASC"
    with engine.connect() as conn:
        return [dict(r._mapping) for r in conn.execute(text(sql))]


def target_row_exists(conn, table: str, row_id: str) -> bool:
    return int(conn.execute(text(f"SELECT COUNT(*) FROM `{table}` WHERE id=:id"), {"id": row_id}).scalar_one()) > 0


def transform_supplier_info(
    row: dict,
    tenant_no: str,
    crm_user_map: Dict[int, str],
    admin_id: Optional[str],
    stats: MigrationStats,
) -> dict:
    rid = legacy_id_to_str(row["id"])
    label = f"supplier_info#{rid}"

    def map_user(field_val: Any) -> Optional[str]:
        return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label)

    return {
        "id": rid,
        "supplier_name": (row.get("supplier_name") or "")[:200],
        "short_name": _str_or_none(row.get("short_name"), 100),
        "supplier_code": _str_or_none(row.get("supplier_code"), 50),
        "supplier_address": _str_or_none(row.get("supplier_address"), 500),
        "legal_person": _str_or_none(row.get("legal_person"), 100),
        "agent": _str_or_none(row.get("agent"), 100),
        "phone_fax": _str_or_none(row.get("phone_fax"), 100),
        "bank_name": _str_or_none(row.get("bank_name"), 200),
        "bank_account": _str_or_none(row.get("bank_account"), 100),
        "status": (row.get("status") or "0")[:1],
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": _str_or_none(row.get("remark"), 500),
        "create_by": map_user(row.get("create_by")),
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": map_user(row.get("update_by")),
        "update_time": row.get("update_time") or row.get("create_time") or datetime.now(),
        "del_flag": int(row.get("del_flag") or 0),
    }


def transform_quote(
    row: dict,
    tenant_no: str,
    crm_user_map: Dict[int, str],
    admin_id: Optional[str],
    oss_map: Dict[str, str],
    stats: MigrationStats,
) -> dict:
    rid = legacy_id_to_str(row["id"])
    label = f"quote#{rid}"

    def map_user(field_val: Any, *, required: bool = False) -> Optional[str]:
        return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label, required=required)

    owner_id = map_user(row.get("owner_user_id"), required=True)

    return {
        "id": rid,
        "quote_no": (row.get("quote_no") or "")[:50],
        "quote_title": (row.get("quote_title") or "")[:200],
        "quote_type": (row.get("quote_type") or "sales")[:20],
        "customer_id": legacy_id_to_str(row.get("customer_id")) or "",
        "customer_name": _str_or_none(row.get("customer_name"), 200),
        "opportunity_id": legacy_id_to_str(row.get("opportunity_id")),
        "contact_id": legacy_id_to_str(row.get("contact_id")),
        "contact_name": _str_or_none(row.get("contact_name"), 50),
        "contact_mobile": _str_or_none(row.get("contact_mobile"), 20),
        "contact_email": _str_or_none(row.get("contact_email"), 100),
        "version": (row.get("version") or "V1.0")[:20],
        "parent_id": legacy_parent_id(row.get("parent_id")),
        "is_latest": int(row.get("is_latest") if row.get("is_latest") is not None else 1),
        "total_amount": _dec(row.get("total_amount"), Decimal("0.00")),
        "discount_rate": _dec(row.get("discount_rate"), Decimal("0.00")),
        "discount_amount": _dec(row.get("discount_amount"), Decimal("0.00")),
        "final_amount": _dec(row.get("final_amount"), Decimal("0.00")),
        "validity_days": int(row.get("validity_days") or 30),
        "quote_date": row.get("quote_date"),
        "expire_date": row.get("expire_date"),
        "status": (row.get("status") or "draft")[:20],
        "reject_reason": _str_or_none(row.get("reject_reason"), 500),
        "pdf_url": normalize_pdf_url(row.get("pdf_url"), oss_map),
        "pdf_generated": int(row.get("pdf_generated") or 0),
        "share_link": _str_or_none(row.get("share_link"), 500),
        "share_code": _str_or_none(row.get("share_code"), 50),
        "share_password": _str_or_none(row.get("share_password"), 100),
        "share_expire_time": row.get("share_expire_time"),
        "view_count": int(row.get("view_count") or 0),
        "last_view_time": row.get("last_view_time"),
        "owner_user_id": owner_id or "",
        "owner_user_name": _str_or_none(row.get("owner_user_name"), 50),
        "supplier_id": legacy_id_to_str(row.get("supplier_id")),
        "supplier_name": _str_or_none(row.get("supplier_name"), 200),
        "contract_id": legacy_id_to_str(row.get("contract_id")),
        "payment_terms": _str_or_none(row.get("payment_terms"), 500),
        "delivery_terms": _str_or_none(row.get("delivery_terms"), 500),
        "freight": _dec(row.get("freight"), Decimal("0.00")),
        "attachments": normalize_attachment_csv(row.get("attachments"), oss_map),
        "template_content": row.get("template_content"),
        "version_no": int(row.get("version_no") or 0),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": _str_or_none(row.get("remark"), 500),
        "create_by": map_user(row.get("create_by")),
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": map_user(row.get("update_by")),
        "update_time": row.get("update_time") or row.get("create_time") or datetime.now(),
        "del_flag": int(row.get("del_flag") or 0),
    }


def transform_quote_product(
    row: dict,
    tenant_no: str,
    crm_user_map: Dict[int, str],
    admin_id: Optional[str],
    stats: MigrationStats,
) -> dict:
    rid = legacy_id_to_str(row["id"])
    label = f"quote_product#{rid}"

    def map_user(field_val: Any) -> Optional[str]:
        return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label)

    return {
        "id": rid,
        "quote_id": legacy_id_to_str(row.get("quote_id")) or "",
        "product_id": legacy_id_to_str(row.get("product_id")) or "",
        "product_name": (row.get("product_name") or "")[:200],
        "product_no": _str_or_none(row.get("product_no"), 50),
        "category_id": legacy_id_to_str(row.get("category_id")),
        "category_name": _str_or_none(row.get("category_name"), 100),
        "model": _str_or_none(row.get("model"), 100),
        "unit": (row.get("unit") or "件")[:20],
        "quantity": _dec(row.get("quantity"), Decimal("0")),
        "price": _dec(row.get("price"), Decimal("0.00")),
        "discount": _dec(row.get("discount"), Decimal("0.00")),
        "subtotal": _dec(row.get("subtotal"), Decimal("0.00")),
        "tax_amount": _dec(row.get("tax_amount"), Decimal("0.00")),
        "tax_rate": _dec(row.get("tax_rate"), Decimal("13.00")),
        "package_spec": _str_or_none(row.get("package_spec"), 200),
        "delivery_method": _str_or_none(row.get("delivery_method"), 200),
        "description": _str_or_none(row.get("description"), 1000),
        "sort": int(row.get("sort") or 0),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": _str_or_none(row.get("remark"), 500),
        "create_by": map_user(row.get("create_by")),
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": map_user(row.get("update_by")),
        "update_time": row.get("update_time") or row.get("create_time") or datetime.now(),
        "del_flag": int(row.get("del_flag") or 0),
    }


def transform_quote_approval(
    row: dict,
    tenant_no: str,
    crm_user_map: Dict[int, str],
    admin_id: Optional[str],
    stats: MigrationStats,
) -> dict:
    rid = legacy_id_to_str(row["id"])
    label = f"quote_approval#{rid}"

    def map_user(field_val: Any) -> Optional[str]:
        return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label)

    return {
        "id": rid,
        "quote_id": legacy_id_to_str(row.get("quote_id")) or "",
        "quote_no": _str_or_none(row.get("quote_no"), 50),
        "step": int(row.get("step") or 1),
        "approver_id": map_user(row.get("approver_id")) or "",
        "approver_name": _str_or_none(row.get("approver_name"), 50),
        "approval_result": (row.get("approval_result") or "pending")[:20],
        "approval_opinion": row.get("approval_opinion"),
        "approval_time": row.get("approval_time"),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": _str_or_none(row.get("remark"), 500),
        "create_by": map_user(row.get("create_by")),
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": map_user(row.get("update_by")),
        "update_time": row.get("update_time") or row.get("create_time") or datetime.now(),
        "del_flag": int(row.get("del_flag") or 0),
    }


def transform_quote_share_log(
    row: dict,
    tenant_no: str,
    stats: MigrationStats,
) -> dict:
    rid = legacy_id_to_str(row["id"])
    return {
        "id": rid,
        "quote_id": legacy_id_to_str(row.get("quote_id")) or "",
        "share_code": (row.get("share_code") or "")[:50],
        "visit_time": row.get("visit_time") or row.get("create_time") or datetime.now(),
        "ip_address": _str_or_none(row.get("ip_address"), 50),
        "location": _str_or_none(row.get("location"), 200),
        "device_type": _str_or_none(row.get("device_type"), 50),
        "browser": _str_or_none(row.get("browser"), 100),
        "stay_duration": int(row.get("stay_duration") or 0),
        "page_views": int(row.get("page_views") or 1),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": None,
        "create_by": None,
        "create_dept_id": "",
        "create_time": row.get("create_time") or row.get("visit_time") or datetime.now(),
        "update_by": None,
        "update_time": row.get("create_time") or row.get("visit_time") or datetime.now(),
        "del_flag": 0,
    }


INSERT_SUPPLIER_SQL = text(
    """
    INSERT INTO crm_supplier_info (
        id, supplier_name, short_name, supplier_code, supplier_address, legal_person, agent,
        phone_fax, bank_name, bank_account, status, tenant_no, order_num, remark,
        create_by, create_dept_id, create_time, update_by, update_time, del_flag
    ) VALUES (
        :id, :supplier_name, :short_name, :supplier_code, :supplier_address, :legal_person, :agent,
        :phone_fax, :bank_name, :bank_account, :status, :tenant_no, :order_num, :remark,
        :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
    )
    """
)

INSERT_QUOTE_SQL = text(
    """
    INSERT INTO crm_quote (
        id, quote_no, quote_title, quote_type, customer_id, customer_name, opportunity_id,
        contact_id, contact_name, contact_mobile, contact_email, version, parent_id, is_latest,
        total_amount, discount_rate, discount_amount, final_amount, validity_days, quote_date,
        expire_date, status, reject_reason, pdf_url, pdf_generated, share_link, share_code,
        share_password, share_expire_time, view_count, last_view_time, owner_user_id, owner_user_name,
        supplier_id, supplier_name, contract_id, payment_terms, delivery_terms, freight,
        attachments, template_content, version_no, tenant_no, order_num, remark,
        create_by, create_dept_id, create_time, update_by, update_time, del_flag
    ) VALUES (
        :id, :quote_no, :quote_title, :quote_type, :customer_id, :customer_name, :opportunity_id,
        :contact_id, :contact_name, :contact_mobile, :contact_email, :version, :parent_id, :is_latest,
        :total_amount, :discount_rate, :discount_amount, :final_amount, :validity_days, :quote_date,
        :expire_date, :status, :reject_reason, :pdf_url, :pdf_generated, :share_link, :share_code,
        :share_password, :share_expire_time, :view_count, :last_view_time, :owner_user_id, :owner_user_name,
        :supplier_id, :supplier_name, :contract_id, :payment_terms, :delivery_terms, :freight,
        :attachments, :template_content, :version_no, :tenant_no, :order_num, :remark,
        :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
    )
    """
)

INSERT_QUOTE_PRODUCT_SQL = text(
    """
    INSERT INTO crm_quote_product (
        id, quote_id, product_id, product_name, product_no, category_id, category_name,
        model, unit, quantity, price, discount, subtotal, tax_amount, tax_rate,
        package_spec, delivery_method, description, sort,
        tenant_no, order_num, remark, create_by, create_dept_id, create_time, update_by, update_time, del_flag
    ) VALUES (
        :id, :quote_id, :product_id, :product_name, :product_no, :category_id, :category_name,
        :model, :unit, :quantity, :price, :discount, :subtotal, :tax_amount, :tax_rate,
        :package_spec, :delivery_method, :description, :sort,
        :tenant_no, :order_num, :remark, :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
    )
    """
)

INSERT_QUOTE_APPROVAL_SQL = text(
    """
    INSERT INTO crm_quote_approval (
        id, quote_id, quote_no, step, approver_id, approver_name, approval_result,
        approval_opinion, approval_time, tenant_no, order_num, remark,
        create_by, create_dept_id, create_time, update_by, update_time, del_flag
    ) VALUES (
        :id, :quote_id, :quote_no, :step, :approver_id, :approver_name, :approval_result,
        :approval_opinion, :approval_time, :tenant_no, :order_num, :remark,
        :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
    )
    """
)

INSERT_QUOTE_SHARE_LOG_SQL = text(
    """
    INSERT INTO crm_quote_share_log (
        id, quote_id, share_code, visit_time, ip_address, location, device_type, browser,
        stay_duration, page_views, tenant_no, order_num, remark,
        create_by, create_dept_id, create_time, update_by, update_time, del_flag
    ) VALUES (
        :id, :quote_id, :share_code, :visit_time, :ip_address, :location, :device_type, :browser,
        :stay_duration, :page_views, :tenant_no, :order_num, :remark,
        :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
    )
    """
)

STEP_INSERT_SQL = {
    "crm_supplier_info": INSERT_SUPPLIER_SQL,
    "crm_quote": INSERT_QUOTE_SQL,
    "crm_quote_product": INSERT_QUOTE_PRODUCT_SQL,
    "crm_quote_approval": INSERT_QUOTE_APPROVAL_SQL,
    "crm_quote_share_log": INSERT_QUOTE_SHARE_LOG_SQL,
}


def write_mapping_csv(source_rows: List[dict], path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    migrated_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with path.open("w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=["source_id", "new_id", "migrated_at"])
        w.writeheader()
        for row in source_rows:
            sid = legacy_id_to_str(row["id"])
            w.writerow({"source_id": sid, "new_id": sid, "migrated_at": migrated_at})


def migrate_table(
    source_engine: Engine,
    target_engine: Engine,
    table: str,
    *,
    tenant_no: str,
    crm_user_map: Dict[int, str],
    admin_id: Optional[str],
    oss_map: Dict[str, str],
    stats: MigrationStats,
    dry_run: bool,
    update_existing: bool,
    include_deleted: bool,
) -> List[dict]:
    step = stats.step(table)
    source_rows = fetch_source_rows(source_engine, table, include_deleted)
    step.source_count = len(source_rows)

    transformers = {
        "crm_supplier_info": lambda r: transform_supplier_info(r, tenant_no, crm_user_map, admin_id, stats),
        "crm_quote": lambda r: transform_quote(r, tenant_no, crm_user_map, admin_id, oss_map, stats),
        "crm_quote_product": lambda r: transform_quote_product(r, tenant_no, crm_user_map, admin_id, stats),
        "crm_quote_approval": lambda r: transform_quote_approval(r, tenant_no, crm_user_map, admin_id, stats),
        "crm_quote_share_log": lambda r: transform_quote_share_log(r, tenant_no, stats),
    }
    payloads = [transformers[table](r) for r in source_rows]

    if dry_run:
        step.inserted = len(payloads)
        return source_rows

    insert_sql = STEP_INSERT_SQL[table]
    with target_engine.begin() as conn:
        for payload in payloads:
            rid = payload["id"]
            if target_row_exists(conn, table, rid):
                if update_existing:
                    step.skipped += 1
                else:
                    step.skipped += 1
            else:
                conn.execute(insert_sql, payload)
                step.inserted += 1
    return source_rows


def print_report(stats: MigrationStats, dry_run: bool) -> None:
    mode = "DRY-RUN" if dry_run else "COMMITTED"
    print(f"\n=== CRM 报价单迁移 [{mode}] ===")
    for table, _ in MIGRATION_STEPS:
        s = stats.step(table)
        print(f"{table}: 源={s.source_count} 插入={s.inserted} 跳过={s.skipped}")
    if stats.warnings:
        print(f"警告 ({len(stats.warnings)}):")
        for w in stats.warnings[:20]:
            print(f"  - {w}")
        if len(stats.warnings) > 20:
            print(f"  ... 另有 {len(stats.warnings) - 20} 条")


def run_migration(
    source_engine: Engine,
    target_engine: Engine,
    *,
    tenant_no: str,
    user_mapping_path: Path,
    mapping_dir: Path,
    dry_run: bool,
    update_existing: bool,
    include_deleted: bool,
    only_tables: Optional[List[str]],
    skip_supplier: bool,
) -> MigrationStats:
    stats = MigrationStats()
    crm_user_map, admin_id = load_user_mapping(user_mapping_path)
    if not admin_id:
        admin_id = fetch_portal_admin_id(target_engine)
    oss_map = load_oss_id_to_url(target_engine, tenant_no)
    print(f"sys_oss 映射条数: {len(oss_map)}")

    selected = [t for t, _ in MIGRATION_STEPS if not only_tables or t in only_tables]
    if skip_supplier and "crm_supplier_info" in selected:
        selected = [t for t in selected if t != "crm_supplier_info"]

    for table in selected:
        print(f"\n>> 迁移 {table} ...")
        source_rows = migrate_table(
            source_engine,
            target_engine,
            table,
            tenant_no=tenant_no,
            crm_user_map=crm_user_map,
            admin_id=admin_id,
            oss_map=oss_map,
            stats=stats,
            dry_run=dry_run,
            update_existing=update_existing,
            include_deleted=include_deleted,
        )
        if table == "crm_quote" and not dry_run:
            write_mapping_csv(source_rows, mapping_dir / "crm_quote_id_mapping.csv")

    print_report(stats, dry_run)
    return stats


def main() -> int:
    parser = argparse.ArgumentParser(description="Migrate CRM quote tables from sx to portal")
    parser.add_argument("--source-url", default="")
    parser.add_argument("--target-url", default="")
    parser.add_argument("--tenant-no", default=DEFAULT_TENANT_NO)
    parser.add_argument(
        "--user-mapping",
        default=str(BACKEND_ROOT.parent / "docs/data-migration/mappings/user_mapping_result.csv"),
    )
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--update", action="store_true")
    parser.add_argument("--include-deleted", action="store_true")
    parser.add_argument("--skip-supplier", action="store_true", help="跳过 crm_supplier_info 前置迁移")
    parser.add_argument("--tables", default="", help="逗号分隔，默认全部表")
    parser.add_argument("--env-file", default=os.environ.get("ENV_FILE", ""))
    args = parser.parse_args()

    load_settings_env(args.env_file or None)
    settings = load_runtime_settings()
    target_url = args.target_url or settings.DATABASE_URL
    source_url = args.source_url or os.environ.get("SX_SOURCE_DATABASE_URL") or derive_source_url(target_url)
    only_tables = [t.strip() for t in args.tables.split(",") if t.strip()] or None

    print(f"源库: {urlparse(source_url).path}")
    print(f"目标库: {urlparse(target_url).path}")

    run_migration(
        create_engine(source_url, pool_pre_ping=True),
        create_engine(target_url, pool_pre_ping=True),
        tenant_no=args.tenant_no,
        user_mapping_path=Path(args.user_mapping),
        mapping_dir=BACKEND_ROOT.parent / "docs/data-migration/mappings",
        dry_run=args.dry_run,
        update_existing=args.update,
        include_deleted=args.include_deleted,
        only_tables=only_tables,
        skip_supplier=args.skip_supplier,
    )
    return 0


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