shanxia2.0/backend/scripts/migrate_crm_product.py

#!/usr/bin/env python3
"""将 sx 库 CRM 产品主表迁入 shanxia2.0 crm_product。

数据源：仅 sx.crm_product（原 CRM），不含地磅 cargos、报销表。
ID：原样保留 str(bigint)，见 docs/data-migration/05-id-preservation-policy.md。

字段映射（sx → 门户，对照 docs/sql/sx.sql 与 shanxia2.0.sql）：
  id/category_id          bigint → VARCHAR 同值字符串
  cost                    → cost_price
  min_price               丢弃（门户无此列）
  tenant_id               → tenant_no（固定 shanxia2.0）
  create_dept             不迁（create_dept_id 置 ''；用户外键走 user_mapping）
  version                 丢弃
  is_toxic/is_explosive   门户新增，默认 0（sx 无）
  crm_product_brand/alias 门户新表，sx 无 → 本脚本不迁

前置：先跑 migrate_crm_product_category.py（分类 status 语义已修正）。

用法：
  cd backend
  SX_URL='mysql+pymysql://.../sx'
  venv/bin/python scripts/migrate_crm_product.py --dry-run --env-file .env.production --source-url "$SX_URL"
  # 正式前可只清产品表：
  venv/bin/python scripts/migrate_crm_product.py --env-file .env.production --source-url "$SX_URL" \\
    --clear-target --user-mapping ../docs/data-migration/mappings/user_mapping_result.csv \\
    --mapping-out ../docs/data-migration/mappings/crm_product_id_mapping.csv
"""
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]
if str(BACKEND_ROOT) not in sys.path:
    sys.path.insert(0, str(BACKEND_ROOT))

DEFAULT_TENANT_NO = "shanxia2.0"
SOURCE_TABLE = "crm_product"
TARGET_TABLE = "crm_product"
CRM_ADMIN_SOURCE_USER_ID = 1


@dataclass
class MigrationStats:
    source_count: int = 0
    inserted: int = 0
    updated: int = 0
    skipped: int = 0
    warnings: List[str] = field(default_factory=list)


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


def count_target_rows(engine: Engine, tenant_no: str) -> int:
    with engine.connect() as conn:
        return int(
            conn.execute(
                text(
                    f"SELECT COUNT(*) FROM `{TARGET_TABLE}` "
                    "WHERE tenant_no=:t AND del_flag=0"
                ),
                {"t": tenant_no},
            ).scalar_one()
        )


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


def transform_row(
    row: dict,
    tenant_no: str,
    crm_user_map: Dict[int, str],
    admin_id: Optional[str],
    stats: MigrationStats,
) -> dict:
    new_id = legacy_id_to_str(row["id"])
    cat_id = legacy_id_to_str(row.get("category_id"))

    status = str(row.get("status") or "on").strip()
    if status not in ("on", "off"):
        status = "on"

    unit = (row.get("unit") or "").strip() or None

    return {
        "id": new_id,
        "product_no": (row.get("product_no") or "")[:50],
        "product_name": (row.get("product_name") or "")[:200],
        "category_id": cat_id,
        "category_name": (row.get("category_name") or "")[:100] or None,
        "model": (row.get("model") or "")[:100] or None,
        "unit": (unit[:20] if unit else None),
        "barcode": (row.get("barcode") or "")[:100] or None,
        "price": _dec(row.get("price")),
        "cost_price": _dec(row.get("cost")),
        "stock": int(row.get("stock") or 0),
        "low_stock": int(row.get("low_stock") or 10),
        "tax_rate": _dec(row.get("tax_rate")) or Decimal("13.00"),
        "is_toxic": 0,
        "is_explosive": 0,
        "status": status[:3],
        "description": row.get("description"),
        "images": (row.get("images") or "")[:1000] or None,
        "sale_count": int(row.get("sale_count") or 0),
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": (row.get("remark") or "")[:512] or None,
        "create_by": resolve_portal_user_id(
            row.get("create_by"), crm_user_map, admin_id, stats, f"product#{new_id} create_by"
        ),
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": resolve_portal_user_id(
            row.get("update_by"), crm_user_map, admin_id, stats, f"product#{new_id} 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_SQL = text(
    f"""
    INSERT INTO `{TARGET_TABLE}` (
        id, product_no, product_name, category_id, category_name, model, unit, barcode,
        price, cost_price, stock, low_stock, tax_rate, is_toxic, is_explosive, status,
        description, images, sale_count, tenant_no, order_num, remark,
        create_by, create_dept_id, create_time, update_by, update_time, del_flag
    ) VALUES (
        :id, :product_no, :product_name, :category_id, :category_name, :model, :unit, :barcode,
        :price, :cost_price, :stock, :low_stock, :tax_rate, :is_toxic, :is_explosive, :status,
        :description, :images, :sale_count, :tenant_no, :order_num, :remark,
        :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
    )
    """
)

UPDATE_SQL = text(
    f"""
    UPDATE `{TARGET_TABLE}` SET
        product_no=:product_no, product_name=:product_name, category_id=:category_id,
        category_name=:category_name, model=:model, unit=:unit, barcode=:barcode,
        price=:price, cost_price=:cost_price, stock=:stock, low_stock=:low_stock,
        tax_rate=:tax_rate, status=:status, description=:description, images=:images,
        sale_count=:sale_count, tenant_no=:tenant_no, order_num=:order_num, remark=:remark,
        create_by=:create_by, update_by=:update_by, update_time=:update_time, del_flag=:del_flag
    WHERE id=:id
    """
)


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


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", "product_no", "product_name", "category_id", "migrated_at"],
        )
        w.writeheader()
        for row in rows:
            sid = legacy_id_to_str(row["id"])
            w.writerow(
                {
                    "source_id": sid,
                    "new_id": sid,
                    "product_no": row.get("product_no") or "",
                    "product_name": row.get("product_name") or "",
                    "category_id": legacy_id_to_str(row.get("category_id")) or "",
                    "migrated_at": migrated_at,
                }
            )


def print_report(stats: MigrationStats, dry_run: bool) -> None:
    mode = "DRY-RUN" if dry_run else "COMMITTED"
    print(f"\n=== CRM 产品迁移 [{mode}] ===")
    print(f"源表行数: {stats.source_count}")
    print(f"插入: {stats.inserted}  更新: {stats.updated}  跳过: {stats.skipped}")
    if stats.warnings:
        print(f"警告 ({len(stats.warnings)}):")
        for w in stats.warnings[:15]:
            print(f"  - {w}")
        if len(stats.warnings) > 15:
            print(f"  ... 另有 {len(stats.warnings) - 15} 条")


def clear_target_products(conn, tenant_no: str) -> int:
    return int(
        conn.execute(
            text(f"DELETE FROM `{TARGET_TABLE}` WHERE tenant_no=:t OR tenant_no='' OR tenant_no IS NULL"),
            {"t": tenant_no},
        ).rowcount
        or 0
    )


def run_migration(
    source_engine: Engine,
    target_engine: Engine,
    *,
    tenant_no: str,
    user_mapping_path: Path,
    mapping_out: Optional[Path],
    dry_run: bool,
    update_existing: bool,
    include_deleted: bool,
    clear_target: bool = False,
) -> 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)

    source_rows = fetch_source_rows(source_engine, include_deleted)
    stats.source_count = len(source_rows)
    payloads = [transform_row(r, tenant_no, crm_user_map, admin_id, stats) for r in source_rows]

    if dry_run:
        print("\n预览（前 5 条）:")
        for p in payloads[:5]:
            print(f"  id={p['id']} no={p['product_no']} name={p['product_name']!r} cat={p['category_id']}")
        if clear_target:
            print("（dry-run 未执行 --clear-target）")
        stats.inserted = len(payloads)
        print_report(stats, dry_run=True)
        return stats

    with target_engine.begin() as conn:
        if clear_target:
            n = clear_target_products(conn, tenant_no)
            print(f"--clear-target: deleted {n} rows from {TARGET_TABLE}")
        for payload in payloads:
            rid = payload["id"]
            if target_row_exists(conn, rid):
                if update_existing:
                    conn.execute(UPDATE_SQL, payload)
                    stats.updated += 1
                else:
                    stats.skipped += 1
            else:
                conn.execute(INSERT_SQL, payload)
                stats.inserted += 1

    if mapping_out:
        write_mapping_csv(source_rows, mapping_out)
        print(f"\nID 映射已写入: {mapping_out}")

    print_report(stats, dry_run=False)
    return stats


def main() -> int:
    parser = argparse.ArgumentParser(description="Migrate CRM products 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(
        "--clear-target",
        action="store_true",
        help="正式执行前 DELETE 目标租户 crm_product（仅本表，逐步清空）",
    )
    parser.add_argument("--mapping-out", default="")
    parser.add_argument("--env-file", default=os.environ.get("ENV_FILE", ""))
    args = parser.parse_args()

    if args.clear_target and args.dry_run:
        print("提示: --clear-target 在 dry-run 下不会删库", file=sys.stderr)

    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}")
    print("数据源: sx.crm_product（原 CRM，不含地磅 cargos）")
    print(f"目标表现有产品: {count_target_rows(create_engine(target_url), args.tenant_no)}")
    print(f"clear_target: {args.clear_target}")

    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_out=Path(args.mapping_out) if args.mapping_out else None,
        dry_run=args.dry_run,
        update_existing=args.update,
        include_deleted=args.include_deleted,
        clear_target=args.clear_target,
    )
    return 0


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