shanxia2.0/backend/scripts/migrate_bx_data.py

#!/usr/bin/env python3
"""将 sx 库报销（bx_*）数据迁入 shanxia2.0。

阶段：
  base      — 费用项目 + 工作流定义/节点（替换门户预置工作流）
  advance   — 备用金申请 + 审批
  reimburse — 报销单 + 明细 + 审批 + 冲抵
  attachment— 附件（依赖业务单 biz_id）
  all       — 按依赖顺序全部执行

用户外键：docs/data-migration/mappings/user_mapping_result.csv（bx_user_id）；
bx_user_id=1（原系统管理员）→ 门户 admin。

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

import argparse
import csv
import os
import re
import sys
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path
from typing import Any, Callable, 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]
if str(BACKEND_ROOT) not in sys.path:
    sys.path.insert(0, str(BACKEND_ROOT))

DEFAULT_TENANT_NO = "shanxia2.0"
BX_ADMIN_SOURCE_USER_ID = 1

PHASE_STEPS: Dict[str, Tuple[str, ...]] = {
    "base": ("bx_expense_item", "bx_workflow_def", "bx_workflow_node"),
    "advance": ("bx_advance_apply", "bx_advance_approval"),
    "reimburse": (
        "bx_reimburse_apply",
        "bx_reimburse_detail",
        "bx_reimburse_approval",
        "bx_reimburse_offset",
    ),
    "attachment": ("bx_attachment",),
}

ALL_STEPS = PHASE_STEPS["base"] + PHASE_STEPS["advance"] + PHASE_STEPS["reimburse"] + PHASE_STEPS["attachment"]

APPLY_CODE_RE = re.compile(r"^BX(\d{8})(\d{3})$")


@dataclass
class StepStats:
    source_count: int = 0
    inserted: 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 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_bx_user_mapping(path: Path) -> Tuple[Dict[int, str], Optional[str]]:
    bx_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()
            bx_raw = (row.get("bx_user_id") or "").strip()
            if user_no == "admin" and new_id:
                admin_id = new_id
            if bx_raw and new_id:
                try:
                    bx_map[int(bx_raw)] = new_id
                except ValueError:
                    pass
    return bx_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 map_bx_user_id(
    raw: Any,
    bx_user_map: Dict[int, str],
    admin_id: Optional[str],
    stats: MigrationStats,
    label: str,
    *,
    required: bool = False,
) -> Optional[str]:
    if raw is None or str(raw).strip() == "":
        return None
    text_val = str(raw).strip()
    if len(text_val) >= 32 and not text_val.isdigit():
        return text_val
    try:
        uid = int(text_val)
    except (TypeError, ValueError):
        stats.warnings.append(f"{label}: 非法用户引用 {raw!r}")
        return admin_id if required else None
    if uid == BX_ADMIN_SOURCE_USER_ID:
        return admin_id
    mapped = bx_user_map.get(uid)
    if not mapped:
        stats.warnings.append(f"{label}: bx_user_id={uid} 无映射")
        return admin_id if required else None
    return mapped


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 _dec(value: Any) -> Optional[Decimal]:
    if value is None or value == "":
        return None
    return Decimal(str(value))


def parse_reimburse_meta(apply_code: str) -> Tuple[date, int]:
    m = APPLY_CODE_RE.match(apply_code or "")
    if not m:
        raise ValueError(f"无法解析报销单号: {apply_code!r}")
    ymd, seq = m.group(1), int(m.group(2))
    d = date(int(ymd[0:4]), int(ymd[4:6]), int(ymd[6:8]))
    return d, seq


def fetch_source_rows(
    engine: Engine, table: str, *, include_deleted: bool, no_del_flag: bool = False
) -> List[dict]:
    if no_del_flag:
        where = ""
    elif include_deleted:
        where = ""
    else:
        where = "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 clear_portal_workflow(conn) -> None:
    conn.execute(text("DELETE FROM bx_workflow_node"))
    conn.execute(text("DELETE FROM bx_workflow_def"))


def transform_expense_item(
    row: dict, tenant_no: str, bx_user_map: Dict[int, str], admin_id: Optional[str], stats: MigrationStats
) -> dict:
    rid = legacy_id_to_str(row["id"])
    label = f"expense#{rid}"
    return {
        "id": rid,
        "parent_id": legacy_id_to_str(row.get("parent_id")) or "0",
        "item_name": (row.get("item_name") or "")[:128],
        "item_code": _str_or_none(row.get("item_code"), 64),
        "level": int(row.get("level") or 1),
        "sort_order": int(row.get("sort_order") or 0),
        "status": int(row.get("status") or 1),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": _str_or_none(row.get("remark"), 512),
        "create_by": map_bx_user_id(row.get("create_by"), bx_user_map, admin_id, stats, label),
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": map_bx_user_id(row.get("update_by"), bx_user_map, admin_id, stats, label),
        "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_workflow_def(row: dict, tenant_no: str, stats: MigrationStats) -> dict:
    rid = legacy_id_to_str(row["id"])
    return {
        "id": rid,
        "biz_type": (row.get("biz_type") or "")[:32],
        "wf_name": (row.get("wf_name") or "")[:128],
        "status": int(row.get("status") or 1),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": _str_or_none(row.get("remark"), 512),
        "create_by": None,
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": None,
        "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_workflow_node(
    row: dict, tenant_no: str, bx_user_map: Dict[int, str], admin_id: Optional[str], stats: MigrationStats
) -> dict:
    rid = legacy_id_to_str(row["id"])
    label = f"wf_node#{rid}"
    return {
        "id": rid,
        "wf_id": legacy_id_to_str(row.get("wf_id")),
        "node_name": (row.get("node_name") or "")[:128],
        "node_order": int(row.get("node_order") or 0),
        "approver_id": map_bx_user_id(
            row.get("approver_id"), bx_user_map, admin_id, stats, label, required=True
        ),
        "approver_name": (row.get("approver_name") or "")[:64],
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": None,
        "create_by": None,
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": None,
        "update_time": row.get("update_time") or row.get("create_time") or datetime.now(),
        "del_flag": 0,
    }


def transform_advance_apply(
    row: dict, tenant_no: str, bx_user_map: Dict[int, str], admin_id: Optional[str], stats: MigrationStats
) -> dict:
    rid = legacy_id_to_str(row["id"])
    label = f"advance#{rid}"
    expense_item_id = legacy_id_to_str(row.get("expense_item_id"))
    return {
        "id": rid,
        "apply_code": (row.get("apply_code") or "")[:32],
        "applicant_id": map_bx_user_id(
            row.get("applicant_id"), bx_user_map, admin_id, stats, label, required=True
        ),
        "applicant_name": (row.get("applicant_name") or "")[:64],
        "reason": (row.get("reason") or "")[:1024],
        "purpose": _str_or_none(row.get("purpose"), 1024),
        "expense_item_id": expense_item_id,
        "amount": _dec(row.get("amount")) or Decimal("0"),
        "status": (row.get("status") or "draft")[:32],
        "current_node": _str_or_none(row.get("current_node"), 128),
        "wf_id": legacy_id_to_str(row.get("wf_id")),
        "issue_status": _str_or_none(row.get("issue_status"), 32),
        "issue_time": row.get("issue_time"),
        "issue_user_id": map_bx_user_id(row.get("issue_user_id"), bx_user_map, admin_id, stats, label),
        "issue_user_name": _str_or_none(row.get("issue_user_name"), 64),
        "issue_remark": _str_or_none(row.get("issue_remark"), 512),
        "actual_amount": _dec(row.get("actual_amount")),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": _str_or_none(row.get("remark"), 512),
        "create_by": map_bx_user_id(row.get("create_by"), bx_user_map, admin_id, stats, label),
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": map_bx_user_id(row.get("update_by"), bx_user_map, admin_id, stats, label),
        "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_advance_approval(
    row: dict, tenant_no: str, bx_user_map: Dict[int, str], admin_id: Optional[str], stats: MigrationStats
) -> dict:
    rid = legacy_id_to_str(row["id"])
    label = f"advance_approval#{rid}"
    now = datetime.now()
    return {
        "id": rid,
        "apply_id": legacy_id_to_str(row.get("apply_id")),
        "node_order": int(row.get("node_order") or 0),
        "node_name": (row.get("node_name") or "")[:128],
        "approver_id": map_bx_user_id(
            row.get("approver_id"), bx_user_map, admin_id, stats, label, required=True
        ),
        "approver_name": (row.get("approver_name") or "")[:64],
        "approval_status": (row.get("approval_status") or "")[:32],
        "approval_opinion": _str_or_none(row.get("approval_opinion"), 1024),
        "approval_time": row.get("approval_time") or now,
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": None,
        "create_by": None,
        "create_dept_id": "",
        "create_time": row.get("approval_time") or now,
        "update_by": None,
        "update_time": row.get("approval_time") or now,
        "del_flag": 0,
    }


def transform_reimburse_apply(
    row: dict, tenant_no: str, bx_user_map: Dict[int, str], admin_id: Optional[str], stats: MigrationStats
) -> dict:
    rid = legacy_id_to_str(row["id"])
    label = f"reimburse#{rid}"
    apply_code = (row.get("apply_code") or "")[:32]
    try:
        reimburse_date, title_seq = parse_reimburse_meta(apply_code)
    except ValueError as exc:
        stats.warnings.append(f"{label}: {exc}")
        reimburse_date = (row.get("create_time") or datetime.now()).date()
        title_seq = int(rid or 0) % 1000 or 1
    return {
        "id": rid,
        "apply_code": apply_code,
        "title": (row.get("title") or "")[:256],
        "reimburse_date": reimburse_date,
        "title_seq": title_seq,
        "applicant_id": map_bx_user_id(
            row.get("applicant_id"), bx_user_map, admin_id, stats, label, required=True
        ),
        "applicant_name": (row.get("applicant_name") or "")[:64],
        "expense_id": legacy_id_to_str(row.get("expense_id")),
        "expense_name": (row.get("expense_name") or "")[:128],
        "amount": _dec(row.get("amount")) or Decimal("0"),
        "invoice_count": int(row.get("invoice_count") or 0),
        "attach_count": int(row.get("attach_count") or 0),
        "status": (row.get("status") or "draft")[:32],
        "current_node": _str_or_none(row.get("current_node"), 128),
        "wf_id": legacy_id_to_str(row.get("wf_id")),
        "pay_status": _str_or_none(row.get("pay_status"), 32),
        "pay_time": row.get("pay_time"),
        "pay_user_id": map_bx_user_id(row.get("pay_user_id"), bx_user_map, admin_id, stats, label),
        "pay_user_name": _str_or_none(row.get("pay_user_name"), 64),
        "actual_amount": _dec(row.get("actual_amount")),
        "offset_advance_id": legacy_id_to_str(row.get("offset_advance_id")),
        "offset_amount": _dec(row.get("offset_amount")),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": _str_or_none(row.get("remark"), 512),
        "create_by": map_bx_user_id(row.get("create_by"), bx_user_map, admin_id, stats, label),
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": map_bx_user_id(row.get("update_by"), bx_user_map, admin_id, stats, label),
        "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_reimburse_detail(row: dict, tenant_no: str, stats: MigrationStats) -> dict:
    rid = legacy_id_to_str(row["id"])
    now = row.get("create_time") or datetime.now()
    return {
        "id": rid,
        "reimburse_id": legacy_id_to_str(row.get("reimburse_id")),
        "expense_id": legacy_id_to_str(row.get("expense_id")),
        "expense_name": (row.get("expense_name") or "")[:128],
        "amount": _dec(row.get("amount")) or Decimal("0"),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": _str_or_none(row.get("remark"), 512),
        "create_by": None,
        "create_dept_id": "",
        "create_time": now,
        "update_by": None,
        "update_time": row.get("update_time") or now,
        "del_flag": int(row.get("del_flag") or 0),
    }


def transform_reimburse_approval(
    row: dict, tenant_no: str, bx_user_map: Dict[int, str], admin_id: Optional[str], stats: MigrationStats
) -> dict:
    rid = legacy_id_to_str(row["id"])
    label = f"reimburse_approval#{rid}"
    now = datetime.now()
    return {
        "id": rid,
        "apply_id": legacy_id_to_str(row.get("apply_id")),
        "node_order": int(row.get("node_order") or 0),
        "node_name": (row.get("node_name") or "")[:128],
        "approver_id": map_bx_user_id(
            row.get("approver_id"), bx_user_map, admin_id, stats, label, required=True
        ),
        "approver_name": (row.get("approver_name") or "")[:64],
        "approval_status": (row.get("approval_status") or "")[:32],
        "approval_opinion": _str_or_none(row.get("approval_opinion"), 1024),
        "approval_time": row.get("approval_time") or now,
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": None,
        "create_by": None,
        "create_dept_id": "",
        "create_time": row.get("approval_time") or now,
        "update_by": None,
        "update_time": row.get("approval_time") or now,
        "del_flag": 0,
    }


def transform_reimburse_offset(row: dict, tenant_no: str, stats: MigrationStats) -> dict:
    rid = legacy_id_to_str(row["id"])
    now = row.get("create_time") or datetime.now()
    return {
        "id": rid,
        "reimburse_id": legacy_id_to_str(row.get("reimburse_id")),
        "advance_id": legacy_id_to_str(row.get("advance_id")),
        "offset_amount": _dec(row.get("offset_amount")) or Decimal("0"),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": _str_or_none(row.get("remark"), 512),
        "create_by": None,
        "create_dept_id": "",
        "create_time": now,
        "update_by": None,
        "update_time": now,
        "del_flag": 0,
    }


def transform_attachment(
    row: dict, tenant_no: str, bx_user_map: Dict[int, str], admin_id: Optional[str], stats: MigrationStats
) -> dict:
    rid = legacy_id_to_str(row["id"])
    label = f"attachment#{rid}"
    biz_id = legacy_id_to_str(row.get("biz_id")) or "0"
    return {
        "id": rid,
        "biz_type": (row.get("biz_type") or "")[:32],
        "biz_id": biz_id,
        "file_name": (row.get("file_name") or "")[:256],
        "file_path": (row.get("file_path") or "")[:512],
        "file_type": _str_or_none(row.get("file_type"), 128),
        "file_size": row.get("file_size"),
        "attach_type": (row.get("attach_type") or "attachment")[:32],
        "sort_order": int(row.get("sort_order") or 0),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": None,
        "create_by": map_bx_user_id(row.get("create_by"), bx_user_map, admin_id, stats, label),
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": None,
        "update_time": row.get("create_time") or datetime.now(),
        "del_flag": int(row.get("del_flag") or 0),
    }


INSERT_SQL = {
    "bx_expense_item": text(
        """
        INSERT INTO bx_expense_item (
            id, parent_id, item_name, item_code, level, sort_order, status,
            tenant_no, order_num, remark, create_by, create_dept_id,
            create_time, update_by, update_time, del_flag
        ) VALUES (
            :id, :parent_id, :item_name, :item_code, :level, :sort_order, :status,
            :tenant_no, :order_num, :remark, :create_by, :create_dept_id,
            :create_time, :update_by, :update_time, :del_flag
        )
        """
    ),
    "bx_workflow_def": text(
        """
        INSERT INTO bx_workflow_def (
            id, biz_type, wf_name, status, tenant_no, order_num, remark,
            create_by, create_dept_id, create_time, update_by, update_time, del_flag
        ) VALUES (
            :id, :biz_type, :wf_name, :status, :tenant_no, :order_num, :remark,
            :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
        )
        """
    ),
    "bx_workflow_node": text(
        """
        INSERT INTO bx_workflow_node (
            id, wf_id, node_name, node_order, approver_id, approver_name,
            tenant_no, order_num, remark, create_by, create_dept_id,
            create_time, update_by, update_time, del_flag
        ) VALUES (
            :id, :wf_id, :node_name, :node_order, :approver_id, :approver_name,
            :tenant_no, :order_num, :remark, :create_by, :create_dept_id,
            :create_time, :update_by, :update_time, :del_flag
        )
        """
    ),
    "bx_advance_apply": text(
        """
        INSERT INTO bx_advance_apply (
            id, apply_code, applicant_id, applicant_name, reason, purpose, expense_item_id,
            amount, status, current_node, wf_id, issue_status, issue_time, issue_user_id,
            issue_user_name, issue_remark, actual_amount, tenant_no, order_num, remark,
            create_by, create_dept_id, create_time, update_by, update_time, del_flag
        ) VALUES (
            :id, :apply_code, :applicant_id, :applicant_name, :reason, :purpose, :expense_item_id,
            :amount, :status, :current_node, :wf_id, :issue_status, :issue_time, :issue_user_id,
            :issue_user_name, :issue_remark, :actual_amount, :tenant_no, :order_num, :remark,
            :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
        )
        """
    ),
    "bx_advance_approval": text(
        """
        INSERT INTO bx_advance_approval (
            id, apply_id, node_order, node_name, approver_id, approver_name,
            approval_status, approval_opinion, approval_time, tenant_no, order_num, remark,
            create_by, create_dept_id, create_time, update_by, update_time, del_flag
        ) VALUES (
            :id, :apply_id, :node_order, :node_name, :approver_id, :approver_name,
            :approval_status, :approval_opinion, :approval_time, :tenant_no, :order_num, :remark,
            :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
        )
        """
    ),
    "bx_reimburse_apply": text(
        """
        INSERT INTO bx_reimburse_apply (
            id, apply_code, title, reimburse_date, title_seq, applicant_id, applicant_name,
            expense_id, expense_name, amount, invoice_count, attach_count, status, current_node,
            wf_id, pay_status, pay_time, pay_user_id, pay_user_name, actual_amount,
            offset_advance_id, offset_amount, tenant_no, order_num, remark,
            create_by, create_dept_id, create_time, update_by, update_time, del_flag
        ) VALUES (
            :id, :apply_code, :title, :reimburse_date, :title_seq, :applicant_id, :applicant_name,
            :expense_id, :expense_name, :amount, :invoice_count, :attach_count, :status, :current_node,
            :wf_id, :pay_status, :pay_time, :pay_user_id, :pay_user_name, :actual_amount,
            :offset_advance_id, :offset_amount, :tenant_no, :order_num, :remark,
            :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
        )
        """
    ),
    "bx_reimburse_detail": text(
        """
        INSERT INTO bx_reimburse_detail (
            id, reimburse_id, expense_id, expense_name, amount, tenant_no, order_num, remark,
            create_by, create_dept_id, create_time, update_by, update_time, del_flag
        ) VALUES (
            :id, :reimburse_id, :expense_id, :expense_name, :amount, :tenant_no, :order_num, :remark,
            :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
        )
        """
    ),
    "bx_reimburse_approval": text(
        """
        INSERT INTO bx_reimburse_approval (
            id, apply_id, node_order, node_name, approver_id, approver_name,
            approval_status, approval_opinion, approval_time, tenant_no, order_num, remark,
            create_by, create_dept_id, create_time, update_by, update_time, del_flag
        ) VALUES (
            :id, :apply_id, :node_order, :node_name, :approver_id, :approver_name,
            :approval_status, :approval_opinion, :approval_time, :tenant_no, :order_num, :remark,
            :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
        )
        """
    ),
    "bx_reimburse_offset": text(
        """
        INSERT INTO bx_reimburse_offset (
            id, reimburse_id, advance_id, offset_amount, tenant_no, order_num, remark,
            create_by, create_dept_id, create_time, update_by, update_time, del_flag
        ) VALUES (
            :id, :reimburse_id, :advance_id, :offset_amount, :tenant_no, :order_num, :remark,
            :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
        )
        """
    ),
    "bx_attachment": text(
        """
        INSERT INTO bx_attachment (
            id, biz_type, biz_id, file_name, file_path, file_type, file_size, attach_type,
            sort_order, tenant_no, order_num, remark, create_by, create_dept_id,
            create_time, update_by, update_time, del_flag
        ) VALUES (
            :id, :biz_type, :biz_id, :file_name, :file_path, :file_type, :file_size, :attach_type,
            :sort_order, :tenant_no, :order_num, :remark, :create_by, :create_dept_id,
            :create_time, :update_by, :update_time, :del_flag
        )
        """
    ),
}


TABLE_OPTIONS: Dict[str, Dict[str, Any]] = {
    "bx_expense_item": {"include_deleted": True},
    "bx_workflow_def": {"include_deleted": False},
    "bx_workflow_node": {"include_deleted": False, "no_del_flag": True},
    "bx_advance_apply": {"include_deleted": False},
    "bx_advance_approval": {"include_deleted": False, "no_del_flag": True},
    "bx_reimburse_apply": {"include_deleted": False},
    "bx_reimburse_detail": {"include_deleted": False},
    "bx_reimburse_approval": {"include_deleted": False, "no_del_flag": True},
    "bx_reimburse_offset": {"include_deleted": False, "no_del_flag": True},
    "bx_attachment": {"include_deleted": False},
}


def write_mapping_csv(table: str, 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 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,
    bx_user_map: Dict[int, str],
    admin_id: Optional[str],
    stats: MigrationStats,
    dry_run: bool,
    replace_workflow: bool,
) -> List[dict]:
    step = stats.step(table)
    opts = TABLE_OPTIONS.get(table, {})
    source_rows = fetch_source_rows(
        source_engine,
        table,
        include_deleted=bool(opts.get("include_deleted")),
        no_del_flag=bool(opts.get("no_del_flag")),
    )
    step.source_count = len(source_rows)

    transformers: Dict[str, Callable[..., dict]] = {
        "bx_expense_item": lambda r: transform_expense_item(r, tenant_no, bx_user_map, admin_id, stats),
        "bx_workflow_def": lambda r: transform_workflow_def(r, tenant_no, stats),
        "bx_workflow_node": lambda r: transform_workflow_node(r, tenant_no, bx_user_map, admin_id, stats),
        "bx_advance_apply": lambda r: transform_advance_apply(r, tenant_no, bx_user_map, admin_id, stats),
        "bx_advance_approval": lambda r: transform_advance_approval(r, tenant_no, bx_user_map, admin_id, stats),
        "bx_reimburse_apply": lambda r: transform_reimburse_apply(r, tenant_no, bx_user_map, admin_id, stats),
        "bx_reimburse_detail": lambda r: transform_reimburse_detail(r, tenant_no, stats),
        "bx_reimburse_approval": lambda r: transform_reimburse_approval(r, tenant_no, bx_user_map, admin_id, stats),
        "bx_reimburse_offset": lambda r: transform_reimburse_offset(r, tenant_no, stats),
        "bx_attachment": lambda r: transform_attachment(r, tenant_no, bx_user_map, admin_id, stats),
    }
    payloads = [transformers[table](r) for r in source_rows]

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

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


def print_report(stats: MigrationStats, dry_run: bool, phase: str) -> None:
    mode = "DRY-RUN" if dry_run else "COMMITTED"
    print(f"\n=== 报销数据迁移 [{phase}] [{mode}] ===")
    for table in ALL_STEPS:
        s = stats.step(table)
        if s.source_count or s.inserted or s.skipped:
            print(f"{table}: 源={s.source_count} 插入={s.inserted} 跳过={s.skipped}")
    if stats.warnings:
        print(f"警告 ({len(stats.warnings)}):")
        for w in stats.warnings[:25]:
            print(f"  - {w}")
        if len(stats.warnings) > 25:
            print(f"  ... 另有 {len(stats.warnings) - 25} 条")


def resolve_steps(phase: str, only_tables: Optional[List[str]]) -> List[str]:
    if phase == "all":
        steps = list(ALL_STEPS)
    else:
        steps = list(PHASE_STEPS.get(phase, ()))
        if not steps:
            raise SystemExit(f"未知 phase: {phase}")
    if only_tables:
        unknown = [t for t in only_tables if t not in ALL_STEPS]
        if unknown:
            raise SystemExit(f"未知表: {unknown}")
        steps = [t for t in steps if t in only_tables]
    return steps


def run_migration(
    source_engine: Engine,
    target_engine: Engine,
    *,
    tenant_no: str,
    user_mapping_path: Path,
    mapping_dir: Path,
    dry_run: bool,
    phase: str,
    only_tables: Optional[List[str]],
    replace_workflow: bool,
) -> MigrationStats:
    stats = MigrationStats()
    bx_user_map, admin_id = load_bx_user_mapping(user_mapping_path)
    if not admin_id:
        admin_id = fetch_portal_admin_id(target_engine)
    print(f"bx 用户映射: {len(bx_user_map)} 条; admin={admin_id or '未找到'}")

    steps = resolve_steps(phase, only_tables)
    for table in steps:
        print(f"\n>> 迁移 {table} ...")
        source_rows = migrate_table(
            source_engine,
            target_engine,
            table,
            tenant_no=tenant_no,
            bx_user_map=bx_user_map,
            admin_id=admin_id,
            stats=stats,
            dry_run=dry_run,
            replace_workflow=replace_workflow,
        )
        if not dry_run and source_rows:
            write_mapping_csv(table, source_rows, mapping_dir / f"bx_{table.replace('bx_', '')}_id_mapping.csv")

    print_report(stats, dry_run, phase)
    return stats


def main() -> int:
    parser = argparse.ArgumentParser(description="Migrate bx reimbursement data 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("--phase", default="all", choices=["base", "advance", "reimburse", "attachment", "all"])
    parser.add_argument("--tables", default="", help="逗号分隔，限定当前 phase 内的表")
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument(
        "--no-replace-workflow",
        action="store_true",
        help="不清理门户预置工作流（默认 base/all 时会清理并写入源库 id=1/2）",
    )
    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
    replace_workflow = not args.no_replace_workflow

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

    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,
        phase=args.phase,
        only_tables=only_tables,
        replace_workflow=replace_workflow,
    )
    return 0


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