394 lines
13 KiB
Python
394 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""将 sx 库 CRM 回款/开票相关表迁入 shanxia2.0。
|
|
|
|
门户侧当前仅有 crm_invoice_info 表；源库回款/发票主表及明细均为空，跳过。
|
|
实际迁移：crm_invoice_info（客户开票信息子表）。
|
|
|
|
用法：
|
|
cd backend
|
|
venv/bin/python scripts/migrate_crm_payment_invoice.py --dry-run --env-file .env.production
|
|
venv/bin/python scripts/migrate_crm_payment_invoice.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 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]
|
|
if str(BACKEND_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(BACKEND_ROOT))
|
|
|
|
DEFAULT_TENANT_NO = "shanxia2.0"
|
|
CRM_ADMIN_SOURCE_USER_ID = 1
|
|
|
|
# 源库存在但门户暂无对应业务表，或源库无数据
|
|
SKIPPED_TABLES = (
|
|
"crm_payment",
|
|
"crm_payment_plan",
|
|
"crm_payment_collection",
|
|
"crm_invoice",
|
|
"crm_invoice_item",
|
|
)
|
|
|
|
|
|
@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_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,
|
|
) -> Optional[str]:
|
|
if source_user_id is None or str(source_user_id).strip() == "":
|
|
return None
|
|
try:
|
|
uid = int(source_user_id)
|
|
except (TypeError, ValueError):
|
|
stats.warnings.append(f"{label}: 非法用户 id {source_user_id!r}")
|
|
return 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 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 fetch_source_rows(engine: Engine, table: str, include_deleted: bool) -> List[dict]:
|
|
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 count_source_rows(engine: Engine, table: str) -> int:
|
|
with engine.connect() as conn:
|
|
try:
|
|
return int(conn.execute(text(f"SELECT COUNT(*) FROM `{table}` WHERE del_flag=0")).scalar_one())
|
|
except Exception:
|
|
return int(conn.execute(text(f"SELECT COUNT(*) FROM `{table}`")).scalar_one())
|
|
|
|
|
|
def target_table_exists(engine: Engine, table: str) -> bool:
|
|
with engine.connect() as conn:
|
|
row = conn.execute(text("SHOW TABLES LIKE :t"), {"t": table}).first()
|
|
return row is not None
|
|
|
|
|
|
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_invoice_info(
|
|
row: dict,
|
|
tenant_no: str,
|
|
crm_user_map: Dict[int, str],
|
|
admin_id: Optional[str],
|
|
stats: MigrationStats,
|
|
) -> Optional[dict]:
|
|
rid = legacy_id_to_str(row["id"])
|
|
label = f"invoice_info#{rid}"
|
|
customer_id = legacy_id_to_str(row.get("customer_id"))
|
|
if not customer_id:
|
|
stats.warnings.append(f"{label}: 缺少 customer_id，已跳过")
|
|
return None
|
|
|
|
company_name = _str_or_none(row.get("company_name"), 200)
|
|
if not company_name:
|
|
stats.warnings.append(f"{label}: company_name 为空，已跳过")
|
|
return None
|
|
|
|
def map_user(field_val: Any) -> Optional[str]:
|
|
return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label)
|
|
|
|
is_default = (str(row.get("is_default") or "0").strip() or "0")[:1]
|
|
if is_default not in ("0", "1"):
|
|
is_default = "1" if is_default in ("true", "True", "Y", "y") else "0"
|
|
|
|
return {
|
|
"id": rid,
|
|
"customer_id": customer_id,
|
|
"company_name": company_name,
|
|
"tax_no": None,
|
|
"company_address": _str_or_none(row.get("company_address"), 500),
|
|
"legal_representative": _str_or_none(row.get("legal_representative"), 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),
|
|
"is_default": is_default,
|
|
"tenant_no": tenant_no,
|
|
"order_num": 0,
|
|
"remark": _str_or_none(row.get("remark"), 512),
|
|
"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),
|
|
}
|
|
|
|
|
|
INSERT_INVOICE_INFO_SQL = text(
|
|
"""
|
|
INSERT INTO crm_invoice_info (
|
|
id, customer_id, company_name, tax_no, company_address, legal_representative,
|
|
agent, phone_fax, bank_name, bank_account, is_default,
|
|
tenant_no, order_num, remark, create_by, create_dept_id,
|
|
create_time, update_by, update_time, del_flag
|
|
) VALUES (
|
|
:id, :customer_id, :company_name, :tax_no, :company_address, :legal_representative,
|
|
:agent, :phone_fax, :bank_name, :bank_account, :is_default,
|
|
:tenant_no, :order_num, :remark, :create_by, :create_dept_id,
|
|
:create_time, :update_by, :update_time, :del_flag
|
|
)
|
|
"""
|
|
)
|
|
|
|
|
|
def write_mapping_csv(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 report_skipped_tables(source_engine: Engine, target_engine: Engine, stats: MigrationStats) -> None:
|
|
print("\n>> 跳过表（门户无表或源库无数据）:")
|
|
for table in SKIPPED_TABLES:
|
|
source_count = count_source_rows(source_engine, table)
|
|
target_exists = target_table_exists(target_engine, table)
|
|
step = stats.step(table)
|
|
step.source_count = source_count
|
|
if source_count == 0:
|
|
print(f" {table}: 源库 0 行，跳过")
|
|
elif not target_exists:
|
|
print(f" {table}: 源库 {source_count} 行，门户无对应表，跳过")
|
|
stats.warnings.append(f"{table}: 源库有 {source_count} 行但门户无表，未迁移")
|
|
else:
|
|
print(f" {table}: 源={source_count} 目标表存在（本脚本未实现，需单独处理）")
|
|
|
|
|
|
def migrate_invoice_info(
|
|
source_engine: Engine,
|
|
target_engine: Engine,
|
|
*,
|
|
tenant_no: str,
|
|
crm_user_map: Dict[int, str],
|
|
admin_id: Optional[str],
|
|
stats: MigrationStats,
|
|
dry_run: bool,
|
|
mapping_dir: Path,
|
|
) -> None:
|
|
table = "crm_invoice_info"
|
|
step = stats.step(table)
|
|
source_rows = fetch_source_rows(source_engine, table, include_deleted=False)
|
|
step.source_count = len(source_rows)
|
|
|
|
payloads: List[dict] = []
|
|
for row in source_rows:
|
|
payload = transform_invoice_info(row, tenant_no, crm_user_map, admin_id, stats)
|
|
if payload:
|
|
payloads.append(payload)
|
|
else:
|
|
step.skipped += 1
|
|
|
|
if dry_run:
|
|
step.inserted = len(payloads)
|
|
print(f"{table}: 源={step.source_count} 可插入={step.inserted} 跳过={step.skipped}")
|
|
return
|
|
|
|
with target_engine.begin() as conn:
|
|
for payload in payloads:
|
|
rid = payload["id"]
|
|
if target_row_exists(conn, table, rid):
|
|
step.skipped += 1
|
|
else:
|
|
conn.execute(INSERT_INVOICE_INFO_SQL, payload)
|
|
step.inserted += 1
|
|
|
|
write_mapping_csv(source_rows, mapping_dir / "crm_invoice_info_id_mapping.csv")
|
|
print(f"{table}: 源={step.source_count} 插入={step.inserted} 跳过={step.skipped}")
|
|
|
|
|
|
def print_report(stats: MigrationStats, dry_run: bool) -> None:
|
|
mode = "DRY-RUN" if dry_run else "COMMITTED"
|
|
print(f"\n=== CRM 回款/开票迁移 [{mode}] ===")
|
|
for name in (*SKIPPED_TABLES, "crm_invoice_info"):
|
|
s = stats.step(name)
|
|
if s.source_count or s.inserted or s.skipped:
|
|
print(f"{name}: 源={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,
|
|
) -> 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)
|
|
|
|
report_skipped_tables(source_engine, target_engine, stats)
|
|
print("\n>> 迁移 crm_invoice_info ...")
|
|
migrate_invoice_info(
|
|
source_engine,
|
|
target_engine,
|
|
tenant_no=tenant_no,
|
|
crm_user_map=crm_user_map,
|
|
admin_id=admin_id,
|
|
stats=stats,
|
|
dry_run=dry_run,
|
|
mapping_dir=mapping_dir,
|
|
)
|
|
print_report(stats, dry_run)
|
|
return stats
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Migrate CRM payment/invoice bundle 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("--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)
|
|
|
|
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,
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|