#!/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 == "json_array": try: parsed = json.loads(text_val) if isinstance(parsed, list): return [t for item in parsed if (t := _normalize_oss_token(str(item)))] except json.JSONDecodeError: pass 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())