zlwuandCursor
10c3c5470c
Bundle pending backend/web-admin feature updates, data-migration scripts, and related docs so the V1.5 branch on the Tencent Cloud Gogs matches the current workspace (excluding local secrets). Co-authored-by: Cursor <cursoragent@cursor.com>
828 lines
32 KiB
Python
828 lines
32 KiB
Python
#!/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())
|