shanxia2.0/backend/scripts/migrate_crm_sys_oss.py

#!/usr/bin/env python3
"""将 sx 库 CRM sys_oss 迁入 shanxia2.0，并反查业务表回填 biz_type/biz_id。

旧 CRM ``sys_oss`` 无 ``biz_type``/``biz_id``；门户新增这两列。迁移分两步：
  1. 整表 INSERT（id=str(oss_id)，url 绝对路径→相对 object key）
  2. 扫描业务表 OSS ID 引用，UPDATE 回填 biz；反查不到则保持 NULL（不用 sys_oper_log）

用法：
  cd backend
  venv/bin/python scripts/migrate_crm_sys_oss.py --dry-run --env-file .env.production

  venv/bin/python scripts/migrate_crm_sys_oss.py --env-file .env.production \\
    --user-mapping ../docs/data-migration/mappings/user_mapping_result.csv \\
    --mapping-out ../docs/data-migration/mappings/crm_oss_id_mapping.csv \\
    --backfill-out ../docs/data-migration/mappings/crm_oss_biz_backfill.csv \\
    --unresolved-out ../docs/data-migration/mappings/crm_oss_unresolved.csv
"""
from __future__ import annotations

import argparse
import csv
import json
import os
import re
import sys
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Set, 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 normalize_oss_storage_url  # noqa: E402

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

OSS_ID_RE = re.compile(r"^\d{1,20}$")


@dataclass
class BackfillRule:
    table: str
    id_column: str
    field: str
    biz_type: str
    attach_type: Optional[str] = None
    value_kind: str = "csv"  # csv | json_array | bigint


BACKFILL_RULES: List[BackfillRule] = [
    BackfillRule("crm_customer", "id", "business_license_images", "crm_customer", "business_license"),
    BackfillRule("crm_customer", "id", "business_permit_images", "crm_customer", "business_permit"),
    BackfillRule("crm_customer", "id", "other_cert_images", "crm_customer", "other_cert"),
    BackfillRule("crm_customer_follow", "id", "attachments", "crm_customer_follow"),
    BackfillRule("crm_clue_follow", "id", "attachments", "crm_clue_follow"),
    BackfillRule("crm_opportunity_follow", "id", "attachments", "crm_opportunity_follow"),
    BackfillRule("crm_quote", "id", "attachments", "crm_quote"),
    BackfillRule("crm_contract", "id", "attachment_files", "crm_contract"),
    BackfillRule("crm_contract", "id", "contract_file", "crm_contract", "contract_file"),
    BackfillRule("crm_product", "id", "images", "crm_product", "image"),
    BackfillRule(
        "crm_contract_product_price_tier",
        "id",
        "letter_oss_id",
        "crm_contract_product_price_tier",
        "letter",
        value_kind="bigint",
    ),
    BackfillRule(
        "crm_contract_product_price_tier",
        "id",
        "letter_oss_ids",
        "crm_contract_product_price_tier",
        "letter",
        value_kind="json_array",
    ),
]


@dataclass
class BizRef:
    biz_type: str
    biz_id: str
    source_table: str
    source_field: str
    attach_type: Optional[str] = None


@dataclass
class MigrationStats:
    source_count: int = 0
    inserted: int = 0
    updated: int = 0
    skipped: int = 0
    backfilled: int = 0
    unresolved: int = 0
    conflicts: 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) -> List[dict]:
    sql = f"SELECT * FROM `{SOURCE_TABLE}` ORDER BY oss_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"),
                {"t": tenant_no},
            ).scalar_one()
        )


def transform_row(
    row: dict,
    tenant_no: str,
    crm_user_map: Dict[int, str],
    admin_id: Optional[str],
    stats: MigrationStats,
) -> dict:
    oss_id = legacy_id_to_str(row["oss_id"])
    relative_url = normalize_oss_storage_url(row.get("url"), row.get("file_name"))
    if not relative_url:
        stats.warnings.append(f"oss#{oss_id}: url 无法规范化为相对路径")

    return {
        "id": oss_id,
        "tenant_no": tenant_no,
        "biz_type": None,
        "biz_id": None,
        "file_name": (row.get("file_name") or "")[:255],
        "original_name": (row.get("original_name") or "")[:255],
        "file_suffix": (row.get("file_suffix") or "")[:10],
        "url": (relative_url or (row.get("file_name") or ""))[:500],
        "file_size": None,
        "file_type": None,
        "attach_type": "attachment",
        "sort_order": 0,
        "service": (row.get("service") or "minio")[:20],
        "ext1": row.get("ext1"),
        "order_num": 0,
        "remark": None,
        "create_by": resolve_portal_user_id(
            row.get("create_by"), crm_user_map, admin_id, stats, f"oss#{oss_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"oss#{oss_id} update_by"
        ),
        "update_time": row.get("update_time") or row.get("create_time") or datetime.now(),
        "del_flag": 0,
    }


INSERT_SQL = text(
    f"""
    INSERT INTO `{TARGET_TABLE}` (
        id, tenant_no, biz_type, biz_id, file_name, original_name, file_suffix, url,
        file_size, file_type, attach_type, sort_order, service, ext1, order_num, remark,
        create_by, create_dept_id, create_time, update_by, update_time, del_flag
    ) VALUES (
        :id, :tenant_no, :biz_type, :biz_id, :file_name, :original_name, :file_suffix, :url,
        :file_size, :file_type, :attach_type, :sort_order, :service, :ext1, :order_num, :remark,
        :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
    )
    """
)

UPDATE_SQL = text(
    f"""
    UPDATE `{TARGET_TABLE}` SET
        file_name=:file_name, original_name=:original_name, file_suffix=:file_suffix, url=:url,
        service=:service, ext1=:ext1, tenant_no=:tenant_no,
        create_by=:create_by, update_by=:update_by, update_time=:update_time
    WHERE id=:id
    """
)

UPDATE_BIZ_SQL = text(
    f"""
    UPDATE `{TARGET_TABLE}` SET
        biz_type=:biz_type, biz_id=:biz_id, attach_type=:attach_type, update_time=:update_time
    WHERE id=:id AND tenant_no=:tenant_no
      AND (biz_type IS NULL OR biz_type = '')
    """
)


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 parse_oss_id_tokens(raw: Any, *, value_kind: str = "csv") -> List[str]:
    if raw is None:
        return []
    if value_kind == "bigint":
        sid = legacy_id_to_str(raw)
        return [sid] if sid and OSS_ID_RE.match(sid) else []

    text_val = str(raw).strip()
    if not text_val:
        return []

    if value_kind in ("json_array", "json_oss_list") or (
        value_kind == "csv" and text_val.startswith("[")
    ):
        try:
            parsed = json.loads(text_val)
            if isinstance(parsed, list):
                tokens: List[str] = []
                for item in parsed:
                    if isinstance(item, dict):
                        sid = legacy_id_to_str(item.get("ossId") or item.get("oss_id"))
                        if sid and OSS_ID_RE.match(sid):
                            tokens.append(sid)
                    else:
                        token = _normalize_oss_token(str(item))
                        if token:
                            tokens.append(token)
                return tokens
        except json.JSONDecodeError:
            if value_kind != "csv":
                return []

    tokens: List[str] = []
    for part in text_val.split(","):
        token = _normalize_oss_token(part)
        if token:
            tokens.append(token)
    return tokens


def _normalize_oss_token(part: str) -> Optional[str]:
    token = (part or "").strip().strip('"').strip("'")
    if not token:
        return None
    if token.startswith("http://") or token.startswith("https://"):
        return None
    if "/" in token or "\\" in token:
        return None
    if not OSS_ID_RE.match(token):
        return None
    return token


def table_exists(engine: Engine, table_name: str) -> bool:
    with engine.connect() as conn:
        row = conn.execute(
            text(
                "SELECT COUNT(*) FROM information_schema.tables "
                "WHERE table_schema = DATABASE() AND table_name = :t"
            ),
            {"t": table_name},
        ).scalar_one()
        return int(row) > 0


def build_biz_index(
    source_engine: Engine,
    known_oss_ids: Set[str],
    stats: MigrationStats,
) -> Tuple[Dict[str, BizRef], List[dict]]:
    index: Dict[str, BizRef] = {}
    conflicts: List[dict] = []

    for rule in BACKFILL_RULES:
        if not table_exists(source_engine, rule.table):
            stats.warnings.append(f"回填跳过：源库无表 {rule.table}")
            continue
        sql = f"SELECT `{rule.id_column}` AS biz_row_id, `{rule.field}` AS field_value FROM `{rule.table}`"
        with source_engine.connect() as conn:
            rows = conn.execute(text(sql)).mappings().all()
        for row in rows:
            biz_id = legacy_id_to_str(row["biz_row_id"])
            if not biz_id:
                continue
            for oss_id in parse_oss_id_tokens(row["field_value"], value_kind=rule.value_kind):
                if oss_id not in known_oss_ids:
                    continue
                ref = BizRef(
                    biz_type=rule.biz_type,
                    biz_id=biz_id,
                    source_table=rule.table,
                    source_field=rule.field,
                    attach_type=rule.attach_type,
                )
                existing = index.get(oss_id)
                if existing is None:
                    index[oss_id] = ref
                    continue
                if (
                    existing.biz_type == ref.biz_type
                    and existing.biz_id == ref.biz_id
                    and existing.attach_type == ref.attach_type
                ):
                    continue
                conflicts.append(
                    {
                        "oss_id": oss_id,
                        "kept_biz_type": existing.biz_type,
                        "kept_biz_id": existing.biz_id,
                        "kept_source": f"{existing.source_table}.{existing.source_field}",
                        "ignored_biz_type": ref.biz_type,
                        "ignored_biz_id": ref.biz_id,
                        "ignored_source": f"{ref.source_table}.{ref.field}",
                    }
                )
                stats.conflicts += 1

    return index, conflicts


def apply_biz_backfill(
    target_engine: Engine,
    biz_index: Dict[str, BizRef],
    tenant_no: str,
    *,
    dry_run: bool,
) -> int:
    if dry_run or not biz_index:
        return len(biz_index)

    now = datetime.now()
    updated = 0
    with target_engine.begin() as conn:
        for oss_id, ref in biz_index.items():
            result = conn.execute(
                UPDATE_BIZ_SQL,
                {
                    "id": oss_id,
                    "tenant_no": tenant_no,
                    "biz_type": ref.biz_type,
                    "biz_id": ref.biz_id,
                    "attach_type": ref.attach_type or "attachment",
                    "update_time": now,
                },
            )
            if result.rowcount:
                updated += 1
    return updated


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_oss_id",
                "new_id",
                "file_name",
                "original_name",
                "url_relative",
                "migrated_at",
            ],
        )
        w.writeheader()
        for row in rows:
            sid = legacy_id_to_str(row["oss_id"])
            w.writerow(
                {
                    "source_oss_id": sid,
                    "new_id": sid,
                    "file_name": row.get("file_name") or "",
                    "original_name": row.get("original_name") or "",
                    "url_relative": normalize_oss_storage_url(row.get("url"), row.get("file_name")),
                    "migrated_at": migrated_at,
                }
            )


def write_backfill_csv(biz_index: Dict[str, BizRef], path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(
            f,
            fieldnames=["oss_id", "biz_type", "biz_id", "attach_type", "source_table", "source_field"],
        )
        w.writeheader()
        for oss_id in sorted(biz_index.keys(), key=lambda x: int(x)):
            ref = biz_index[oss_id]
            w.writerow(
                {
                    "oss_id": oss_id,
                    "biz_type": ref.biz_type,
                    "biz_id": ref.biz_id,
                    "attach_type": ref.attach_type or "",
                    "source_table": ref.source_table,
                    "source_field": ref.source_field,
                }
            )


def write_unresolved_csv(
    all_oss_ids: Iterable[str],
    biz_index: Dict[str, BizRef],
    path: Path,
) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    unresolved = sorted(set(all_oss_ids) - set(biz_index.keys()), key=lambda x: int(x))
    with path.open("w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=["oss_id", "reason"])
        w.writeheader()
        for oss_id in unresolved:
            w.writerow({"oss_id": oss_id, "reason": "no_business_reference"})


def print_report(stats: MigrationStats, dry_run: bool) -> None:
    mode = "DRY-RUN" if dry_run else "COMMITTED"
    print(f"\n=== CRM sys_oss 迁移 [{mode}] ===")
    print(f"源表行数: {stats.source_count}")
    print(f"插入: {stats.inserted}  更新: {stats.updated}  跳过: {stats.skipped}")
    print(f"biz 回填: {stats.backfilled}  未解析 biz: {stats.unresolved}  冲突(保留首条): {stats.conflicts}")
    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 run_migration(
    source_engine: Engine,
    target_engine: Engine,
    *,
    tenant_no: str,
    user_mapping_path: Path,
    mapping_out: Optional[Path],
    backfill_out: Optional[Path],
    unresolved_out: Optional[Path],
    dry_run: bool,
    update_existing: bool,
    skip_insert: bool,
    skip_backfill: 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)

    source_rows = fetch_source_rows(source_engine)
    stats.source_count = len(source_rows)
    known_oss_ids = {legacy_id_to_str(r["oss_id"]) for r in source_rows if legacy_id_to_str(r["oss_id"])}

    payloads = [transform_row(r, tenant_no, crm_user_map, admin_id, stats) for r in source_rows]

    if not skip_insert:
        if dry_run:
            print("\n预览（前 5 条）:")
            for p in payloads[:5]:
                print(
                    f"  id={p['id']} url={p['url']!r} file={p['file_name']!r} "
                    f"biz=({p['biz_type']},{p['biz_id']})"
                )
            stats.inserted = len(payloads)
        else:
            with target_engine.begin() as conn:
                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

    biz_index: Dict[str, BizRef] = {}
    if not skip_backfill:
        biz_index, conflicts = build_biz_index(source_engine, known_oss_ids, stats)
        if conflicts and dry_run:
            print(f"\n回填冲突预览（前 3 条，共 {len(conflicts)}）:")
            for item in conflicts[:3]:
                print(f"  oss={item['oss_id']} kept={item['kept_source']} ignored={item['ignored_source']}")
        stats.backfilled = apply_biz_backfill(target_engine, biz_index, tenant_no, dry_run=dry_run)
        stats.unresolved = len(known_oss_ids) - len(biz_index)

    if mapping_out and not dry_run:
        write_mapping_csv(source_rows, mapping_out)
        print(f"\nID 映射已写入: {mapping_out}")
    if backfill_out:
        write_backfill_csv(biz_index, backfill_out)
        print(f"biz 回填映射已写入: {backfill_out}")
    if unresolved_out:
        write_unresolved_csv(known_oss_ids, biz_index, unresolved_out)
        print(f"未回填 oss 列表已写入: {unresolved_out}")

    print_report(stats, dry_run)
    return stats


def main() -> int:
    parser = argparse.ArgumentParser(description="Migrate CRM sys_oss 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", help="目标已存在同 id 时更新元数据（不含 biz 覆盖）")
    parser.add_argument("--skip-insert", action="store_true", help="仅执行 biz 回填")
    parser.add_argument("--skip-backfill", action="store_true", help="仅 INSERT，不回填 biz_type/biz_id")
    parser.add_argument("--mapping-out", default="")
    parser.add_argument("--backfill-out", default="")
    parser.add_argument("--unresolved-out", default="")
    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}")
    print("数据源: sx.sys_oss（旧 CRM 附件元数据，共用 MinIO）")
    print(f"目标表现有 sys_oss: {count_target_rows(create_engine(target_url), args.tenant_no)}")

    default_mapping_dir = BACKEND_ROOT.parent / "docs/data-migration/mappings"
    mapping_out = Path(args.mapping_out) if args.mapping_out else default_mapping_dir / "crm_oss_id_mapping.csv"
    backfill_out = Path(args.backfill_out) if args.backfill_out else default_mapping_dir / "crm_oss_biz_backfill.csv"
    unresolved_out = (
        Path(args.unresolved_out) if args.unresolved_out else default_mapping_dir / "crm_oss_unresolved.csv"
    )

    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=mapping_out if not args.dry_run else None,
        backfill_out=backfill_out if not args.skip_backfill else None,
        unresolved_out=unresolved_out if not args.skip_backfill else None,
        dry_run=args.dry_run,
        update_existing=args.update,
        skip_insert=args.skip_insert,
        skip_backfill=args.skip_backfill,
    )
    return 0


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