#!/usr/bin/env python3 """将 sx 库报销/CRM 用户迁入 shanxia2.0 门户 sys_user。 规则（2026-07-12 确认）： - 跳过 admin（门户已存在） - 不迁移地磅 users - 登录账号 user_no 使用中文姓名 - 初始密码：每人随机 4 位数字，bcrypt 加密（与门户 get_password_hash 一致） - 支持 --dept-mapping-csv 按人指定部门 - 可选同步角色到 sys_role / sys_user_role 用法： cd backend && source venv/bin/activate ENV_FILE=.env.production python scripts/migrate_users.py --dry-run --check-infra ENV_FILE=.env.production python scripts/migrate_users.py \\ --source-url mysql+pymysql://root:***@121.43.228.102:3306/sx \\ --mapping-out ../docs/data-migration/mappings/user_mapping_result.json 环境变量： ENV_FILE 目标库配置（默认 backend/.env，生产用 .env.production） DATABASE_URL 目标库 shanxia2.0（由 Settings 读取） SX_SOURCE_DATABASE_URL 源库 sx（未传 --source-url 时优先使用） """ from __future__ import annotations import argparse import csv import json import os import re import secrets import sys from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Dict, 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] if str(BACKEND_ROOT) not in sys.path: sys.path.insert(0, str(BACKEND_ROOT)) from app.core.security import get_password_hash from app.core.utils import generate_uuid # 门户默认部门（销售部，见 migration_20260712_reset_sys_dept.sql） DEFAULT_DEPT_ID = "D65CAAE057914CCCAE2A81BE627A43EE" DEFAULT_TENANT_NO = "shanxia2.0" ADMIN_USER_NO = "admin" # 可选部门（供 --dept-mapping-csv 填写 dept_id 或 dept_name） DEPT_BY_ID = { "b13c2be2-4e71-11f1-9f47-00163e3bdd96": "安庆市山峡化工有限公司", "E65CAAE057914CCCAE2A81BE627A43EA": "销售部", "F65CAAE057914CCCAE2A81BE627A43EF": "财务部门", "C65CAAE057914CCCAE2A81BE627A43ED": "人力资源部", "D65CAAE057914CCCAE2A81BE627A43EE": "生产现场部", } DEPT_BY_NAME = { "安庆市山峡化工有限公司": "b13c2be2-4e71-11f1-9f47-00163e3bdd96", "销售部": "E65CAAE057914CCCAE2A81BE627A43EA", "销售部门": "E65CAAE057914CCCAE2A81BE627A43EA", "财务部门": "F65CAAE057914CCCAE2A81BE627A43EF", "财务部": "F65CAAE057914CCCAE2A81BE627A43EF", "人力资源部": "C65CAAE057914CCCAE2A81BE627A43ED", "人力资源部门": "C65CAAE057914CCCAE2A81BE627A43ED", "生产现场部": "D65CAAE057914CCCAE2A81BE627A43EE", } # 业务确认的手机号覆盖（中文名 -> 手机） PHONE_OVERRIDES: Dict[str, str] = { "吴九华": "18656008189", } # bx 角色 -> 门户角色名 BX_ROLE_MAP = { 1: "管理员", 2: "员工", 3: "财务人员", 4: "总经理", } # CRM 角色 ID -> 门户角色名（sx.sql 生产快照） CRM_ROLE_MAP = { 1975370859695071234: "销售人员", 1979711425949528066: "销售主管", 1984885915415953409: "行政", } @dataclass class SourceUser: user_no: str nick_name: str phone_no: str email: str sex: str bx_user_id: Optional[int] = None crm_user_id: Optional[int] = None bx_role_ids: Set[int] = field(default_factory=set) crm_role_ids: Set[int] = field(default_factory=set) sources: List[str] = field(default_factory=list) dept_id: str = DEFAULT_DEPT_ID @dataclass class MigrationStats: inserted_users: int = 0 skipped_users: int = 0 updated_mapping: int = 0 roles_created: int = 0 user_roles_inserted: int = 0 warnings: List[str] = field(default_factory=list) def _has_chinese(text_value: str) -> bool: return bool(re.search(r"[\u4e00-\u9fff]", text_value or "")) def _clip(value: Any, max_len: int) -> str: s = (value or "").strip() if value is not None else "" return s[:max_len] def _normalize_phone(value: Any) -> str: s = re.sub(r"\D", "", str(value or "")) return s[:11] if s else "" def generate_initial_password(used: Set[str]) -> str: """生成不重复的 4 位数字初始密码（0000–9999）。""" for _ in range(200): plain = f"{secrets.randbelow(10000):04d}" if plain not in used: used.add(plain) return plain raise RuntimeError("无法生成唯一 4 位初始密码，请扩大位数或手动指定") def bx_role_labels(role_ids: Set[int]) -> str: return "、".join(BX_ROLE_MAP[r] for r in sorted(role_ids) if r in BX_ROLE_MAP) def crm_role_labels(role_ids: Set[int]) -> str: return "、".join(CRM_ROLE_MAP[r] for r in sorted(role_ids) if r in CRM_ROLE_MAP) def load_dept_mapping_csv(path: Path, default_dept_id: str) -> Dict[str, str]: """读取 user_import_roster 填好的部门：user_no -> dept_id。""" if not path.is_file(): return {} mapping: Dict[str, str] = {} with path.open(encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) for row in reader: user_no = (row.get("user_no") or "").strip() if not user_no: continue dept_id = (row.get("dept_id") or "").strip() dept_name = (row.get("dept_name") or "").strip() if not dept_id and dept_name: dept_id = DEPT_BY_NAME.get(dept_name, "") if dept_id: if dept_id not in DEPT_BY_ID: raise ValueError(f"用户「{user_no}」部门 ID 无效: {dept_id}") mapping[user_no] = dept_id else: mapping[user_no] = default_dept_id return mapping def derive_source_url(target_url: str, source_db: str = "sx") -> str: parsed = urlparse(target_url) # /shanxia2.0 -> /sx new_path = f"/{source_db}" return urlunparse(parsed._replace(path=new_path)) def load_runtime_settings(): """在 load_dotenv 之后实例化，确保读到 ENV_FILE / .env.production。""" from app.core.config import Settings return Settings() def load_settings_env(env_file: Optional[str]) -> None: if env_file: os.environ["ENV_FILE"] = env_file candidates: List[Path] = [] if env_file: p = Path(env_file) candidates.append(p if p.is_absolute() else BACKEND_ROOT / p) candidates.append(BACKEND_ROOT / ".env.production") candidates.append(BACKEND_ROOT / ".env") from dotenv import load_dotenv for path in candidates: if path.is_file(): load_dotenv(path, override=True) break def check_mysql(engine: Engine, label: str) -> None: with engine.connect() as conn: conn.execute(text("SELECT 1")) def check_redis(runtime_settings) -> Tuple[bool, str]: try: import redis client = redis.Redis( host=runtime_settings.REDIS_HOST or "localhost", port=runtime_settings.REDIS_PORT, db=runtime_settings.REDIS_DB, password=runtime_settings.REDIS_PASSWORD or None, socket_connect_timeout=runtime_settings.REDIS_CONNECT_TIMEOUT_SEC, socket_timeout=runtime_settings.REDIS_OP_TIMEOUT_SEC, ) client.ping() return True, f"{runtime_settings.REDIS_HOST}:{runtime_settings.REDIS_PORT}/{runtime_settings.REDIS_DB}" except Exception as exc: return False, str(exc) def check_minio(runtime_settings) -> Tuple[bool, str]: try: from minio import Minio if not runtime_settings.MINIO_ENDPOINT: return False, "MINIO_ENDPOINT 未配置" client = Minio( runtime_settings.MINIO_ENDPOINT, access_key=runtime_settings.MINIO_ACCESS_KEY, secret_key=runtime_settings.MINIO_SECRET_KEY, secure=runtime_settings.MINIO_SECURE, ) client.bucket_exists(runtime_settings.MINIO_BUCKET) return True, f"{runtime_settings.MINIO_ENDPOINT} bucket={runtime_settings.MINIO_BUCKET}" except Exception as exc: return False, str(exc) def fetch_bx_users(engine: Engine) -> List[dict]: sql = """ SELECT id, username, password, real_name, phone, email, status, del_flag FROM bx_user WHERE del_flag = 0 AND username <> :admin ORDER BY id """ with engine.connect() as conn: rows = conn.execute(text(sql), {"admin": ADMIN_USER_NO}).mappings().all() return [dict(r) for r in rows] def fetch_crm_users(engine: Engine) -> List[dict]: sql = """ SELECT user_id, user_name, nick_name, phonenumber, email, password, sex, status, del_flag FROM sys_user WHERE del_flag = '0' AND status = '0' AND user_name <> :admin ORDER BY user_id """ with engine.connect() as conn: rows = conn.execute(text(sql), {"admin": ADMIN_USER_NO}).mappings().all() return [dict(r) for r in rows] def fetch_bx_role_users(engine: Engine) -> Dict[int, Set[int]]: sql = "SELECT user_id, role_id FROM bx_role_user" out: Dict[int, Set[int]] = {} with engine.connect() as conn: for row in conn.execute(text(sql)).mappings(): uid = int(row["user_id"]) out.setdefault(uid, set()).add(int(row["role_id"])) return out def fetch_crm_user_roles(engine: Engine) -> Dict[int, Set[int]]: sql = "SELECT user_id, role_id FROM sys_user_role" out: Dict[int, Set[int]] = {} with engine.connect() as conn: for row in conn.execute(text(sql)).mappings(): uid = int(row["user_id"]) out.setdefault(uid, set()).add(int(row["role_id"])) return out def build_merged_users( source_engine: Engine, stats: MigrationStats, dept_mapping: Optional[Dict[str, str]] = None, default_dept_id: str = DEFAULT_DEPT_ID, ) -> List[SourceUser]: bx_rows = fetch_bx_users(source_engine) crm_rows = fetch_crm_users(source_engine) bx_roles = fetch_bx_role_users(source_engine) crm_roles = fetch_crm_user_roles(source_engine) merged: Dict[str, SourceUser] = {} def upsert(key: str, **kwargs: Any) -> SourceUser: key = key.strip() if key not in merged: merged[key] = SourceUser( user_no=key, nick_name=key, phone_no="", email="", sex="0", ) item = merged[key] for k, v in kwargs.items(): if v is None: continue if k == "sources": values = v if isinstance(v, (list, tuple, set)) else [v] for s in values: if s and s not in item.sources: item.sources.append(s) elif k == "bx_role_ids" or k == "crm_role_ids": getattr(item, k).update(v) elif k in ("phone_no", "email") and getattr(item, k): continue else: setattr(item, k, v) return item for row in bx_rows: name = (row.get("real_name") or "").strip() if not name: stats.warnings.append(f"bx_user id={row['id']} 缺少 real_name，已跳过") continue upsert( name, phone_no=_normalize_phone(row.get("phone")), email=_clip(row.get("email"), 50), bx_user_id=int(row["id"]), bx_role_ids=bx_roles.get(int(row["id"]), set()), sources="bx", ) for row in crm_rows: nick = (row.get("nick_name") or "").strip() uname = (row.get("user_name") or "").strip() name = nick if _has_chinese(nick) else (uname if _has_chinese(uname) else nick or uname) if not name or not _has_chinese(name): continue upsert( name, phone_no=_normalize_phone(row.get("phonenumber")), email=_clip(row.get("email"), 50), sex=str(row.get("sex") or "0"), crm_user_id=int(row["user_id"]), crm_role_ids=crm_roles.get(int(row["user_id"]), set()), sources="crm", ) # 手机号业务覆盖 for name, phone in PHONE_OVERRIDES.items(): if name in merged: merged[name].phone_no = phone dept_mapping = dept_mapping or {} result = [] for item in merged.values(): item.dept_id = dept_mapping.get(item.user_no, default_dept_id) result.append(item) result.sort(key=lambda u: u.user_no) return result def target_user_exists_conn(conn, user_no: str, tenant_no: str) -> Optional[str]: sql = """ SELECT id FROM sys_user WHERE user_no = :user_no AND tenant_no = :tenant AND del_flag = 0 LIMIT 1 """ row = conn.execute(text(sql), {"user_no": user_no, "tenant": tenant_no}).first() return str(row[0]) if row else None def ensure_portal_role( conn, role_name: str, tenant_no: str, admin_id: str, dept_id: str, dry_run: bool, cache: Dict[str, str], stats: MigrationStats, ) -> str: if role_name in cache: return cache[role_name] row = conn.execute( text( """ SELECT id FROM sys_role WHERE role_name = :name AND tenant_no = :tenant AND del_flag = 0 LIMIT 1 """ ), {"name": role_name, "tenant": tenant_no}, ).first() if row: cache[role_name] = str(row[0]) return cache[role_name] role_id = generate_uuid() if not dry_run: conn.execute( text( """ INSERT INTO sys_role ( id, role_name, data_scope, status, tenant_no, order_num, remark, create_by, create_dept_id, create_time, update_by, update_time, del_flag ) VALUES ( :id, :role_name, '5', '1', :tenant_no, 10, '用户迁移脚本自动创建', :create_by, :dept_id, :now, :create_by, :now, 0 ) """ ), { "id": role_id, "role_name": role_name, "tenant_no": tenant_no, "create_by": admin_id, "dept_id": dept_id, "now": datetime.now(), }, ) cache[role_name] = role_id stats.roles_created += 1 return role_id def portal_role_names_for_user(user: SourceUser) -> Set[str]: names: Set[str] = set() for rid in user.bx_role_ids: name = BX_ROLE_MAP.get(rid) if name and name != "管理员": names.add(name) for rid in user.crm_role_ids: name = CRM_ROLE_MAP.get(rid) if name: names.add(name) if not names: names.add("员工") return names def insert_user_role( conn, user_id: str, role_id: str, tenant_no: str, admin_id: str, dept_id: str, dry_run: bool, stats: MigrationStats, ) -> None: exists = conn.execute( text( """ SELECT id FROM sys_user_role WHERE user_id = :uid AND role_id = :rid AND tenant_no = :tenant AND del_flag = 0 LIMIT 1 """ ), {"uid": user_id, "rid": role_id, "tenant": tenant_no}, ).first() if exists: return if not dry_run: conn.execute( text( """ INSERT INTO sys_user_role ( id, user_id, role_id, tenant_no, order_num, remark, create_by, create_dept_id, create_time, update_by, update_time, del_flag ) VALUES ( :id, :user_id, :role_id, :tenant_no, 0, 'migrate_users', :create_by, :dept_id, :now, :create_by, :now, 0 ) """ ), { "id": generate_uuid(), "user_id": user_id, "role_id": role_id, "tenant_no": tenant_no, "create_by": admin_id, "dept_id": dept_id, "now": datetime.now(), }, ) stats.user_roles_inserted += 1 def migrate_users( source_engine: Engine, target_engine: Engine, *, tenant_no: str, dept_id: str, dept_mapping: Optional[Dict[str, str]], dry_run: bool, with_roles: bool, stats: MigrationStats, ) -> List[dict]: users = build_merged_users( source_engine, stats, dept_mapping=dept_mapping, default_dept_id=dept_id ) mapping_rows: List[dict] = [] used_passwords: Set[str] = set() with target_engine.connect() as raw_conn: admin_row = raw_conn.execute( text("SELECT id FROM sys_user WHERE user_no = :u AND del_flag = 0 LIMIT 1"), {"u": ADMIN_USER_NO}, ).first() admin_id = str(admin_row[0]) if admin_row else "" with target_engine.begin() as conn: role_cache: Dict[str, str] = {} for user in users: existing_id = target_user_exists_conn(conn, user.user_no, tenant_no) plain_password = "" password_hash = "" if existing_id: stats.skipped_users += 1 stats.warnings.append(f"已存在，跳过: {user.user_no} (id={existing_id})") new_id = existing_id else: plain_password = generate_initial_password(used_passwords) password_hash = get_password_hash(plain_password) new_id = generate_uuid() user_dept_id = user.dept_id or dept_id if not dry_run: conn.execute( text( """ INSERT INTO sys_user ( id, dept_id, user_no, nick_name, email, phone_no, sex, avatar, password, status, tenant_no, order_num, remark, create_by, create_dept_id, create_time, update_by, update_time, del_flag ) VALUES ( :id, :dept_id, :user_no, :nick_name, :email, :phone_no, :sex, NULL, :password, '1', :tenant_no, 0, :remark, :create_by, :dept_id, :now, :create_by, :now, 0 ) """ ), { "id": new_id, "dept_id": user_dept_id, "user_no": _clip(user.user_no, 30), "nick_name": _clip(user.nick_name, 30), "email": _clip(user.email, 50), "phone_no": _clip(user.phone_no, 11), "sex": _clip(user.sex, 1) or "0", "password": password_hash, "tenant_no": tenant_no, "remark": f"migrate_users:{','.join(user.sources)}", "create_by": admin_id or None, "now": datetime.now(), }, ) stats.inserted_users += 1 user_dept_id = user.dept_id or dept_id if with_roles and admin_id: for role_name in portal_role_names_for_user(user): role_id = ensure_portal_role( conn, role_name, tenant_no, admin_id, user_dept_id, dry_run, role_cache, stats, ) insert_user_role( conn, new_id, role_id, tenant_no, admin_id, user_dept_id, dry_run, stats, ) mapping_rows.append( { "new_user_id": new_id, "user_no": user.user_no, "nick_name": user.nick_name, "phone_no": user.phone_no, "dept_id": user_dept_id, "dept_name": DEPT_BY_ID.get(user_dept_id, ""), "bx_user_id": user.bx_user_id or "", "crm_user_id": user.crm_user_id or "", "dibang_user_id": "", "initial_password": plain_password, "password_hash_algo": "bcrypt", "status": "1", "sources": ",".join(user.sources), "bx_roles": bx_role_labels(user.bx_role_ids), "crm_roles": crm_role_labels(user.crm_role_ids), "portal_roles": ",".join(sorted(portal_role_names_for_user(user))), } ) return mapping_rows def write_mapping_outputs( mapping_rows: List[dict], json_path: Optional[Path], csv_path: Optional[Path], credentials_path: Optional[Path], ) -> None: safe_rows = [] for row in mapping_rows: item = dict(row) item.pop("initial_password", None) safe_rows.append(item) if json_path: json_path.parent.mkdir(parents=True, exist_ok=True) payload = { "generated_at": datetime.now().isoformat(timespec="seconds"), "count": len(safe_rows), "users": safe_rows, "bx_id_to_user": { str(r["bx_user_id"]): r["new_user_id"] for r in safe_rows if r.get("bx_user_id") }, "crm_id_to_user": { str(r["crm_user_id"]): r["new_user_id"] for r in safe_rows if r.get("crm_user_id") }, } json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") if csv_path: csv_path.parent.mkdir(parents=True, exist_ok=True) fieldnames = [ "new_user_id", "user_no", "nick_name", "phone_no", "dept_id", "dept_name", "bx_user_id", "crm_user_id", "dibang_user_id", "password_hash_algo", "status", "sources", "bx_roles", "crm_roles", "portal_roles", ] with csv_path.open("w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(safe_rows) if credentials_path: credentials_path.parent.mkdir(parents=True, exist_ok=True) cred_fields = [ "user_no", "nick_name", "phone_no", "dept_name", "initial_password", "new_user_id", ] with credentials_path.open("w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=cred_fields, extrasaction="ignore") writer.writeheader() for row in mapping_rows: if row.get("initial_password"): writer.writerow(row) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="迁移 sx 报销/CRM 用户到 shanxia2.0") parser.add_argument("--dry-run", action="store_true", help="只演练不写库") parser.add_argument("--env-file", default=".env.production", help="如 .env.production") parser.add_argument( "--source-url", default=os.getenv("SX_SOURCE_DATABASE_URL", ""), help="源库 sx 连接串", ) parser.add_argument( "--source-db", default="sx", help="从目标 DATABASE_URL 推导源库时的库名", ) parser.add_argument( "--tenant-no", default=DEFAULT_TENANT_NO, help="目标租户编号", ) parser.add_argument( "--dept-id", default=DEFAULT_DEPT_ID, help="默认部门 ID", ) parser.add_argument( "--dept-mapping-csv", default=str(BACKEND_ROOT.parent / "docs/data-migration/mappings/user_import_roster.csv"), help="按人指定部门的 CSV（填 dept_id 或 dept_name）", ) parser.add_argument( "--credentials-out", default=str( BACKEND_ROOT.parent / "docs/data-migration/mappings/user_import_credentials.csv" ), help="初始明文密码清单（仅新建用户，勿提交版本库）", ) parser.add_argument( "--mapping-out", default=str(BACKEND_ROOT.parent / "docs/data-migration/mappings/user_mapping_result.json"), help="输出 JSON 映射", ) parser.add_argument( "--mapping-csv", default=str(BACKEND_ROOT.parent / "docs/data-migration/mappings/user_mapping_result.csv"), help="输出 CSV 映射", ) parser.add_argument( "--no-roles", action="store_true", help="不写入 sys_role / sys_user_role", ) parser.add_argument( "--check-infra", action="store_true", help="检查 MySQL/Redis/MinIO 连通性", ) return parser.parse_args() def main() -> int: args = parse_args() load_settings_env(args.env_file or None) runtime_settings = load_runtime_settings() target_url = runtime_settings.DATABASE_URL source_url = args.source_url or derive_source_url(target_url, args.source_db) print("=== 用户迁移 migrate_users ===") print(f"模式: {'DRY-RUN' if args.dry_run else 'EXECUTE'}") print(f"源库: {source_url}") print(f"目标库: {target_url}") print(f"租户: {args.tenant_no}") print("跳过: admin、地磅账号") print() source_engine = create_engine(source_url, pool_pre_ping=True) target_engine = create_engine(target_url, pool_pre_ping=True) if args.check_infra: check_mysql(source_engine, "source") check_mysql(target_engine, "target") ok, msg = check_redis(runtime_settings) print(f"Redis: {'OK' if ok else 'FAIL'} — {msg}") ok, msg = check_minio(runtime_settings) print(f"MinIO: {'OK' if ok else 'FAIL'} — {msg}") print() dept_mapping_path = Path(args.dept_mapping_csv) dept_mapping = load_dept_mapping_csv(dept_mapping_path, args.dept_id) if dept_mapping_path.is_file(): filled = sum( 1 for k, v in dept_mapping.items() if v and v != args.dept_id ) print(f"部门映射: {dept_mapping_path.name}（{filled} 人已指定非默认部门，其余默认生产现场部）") else: print(f"部门映射: 未找到 {dept_mapping_path}，全部使用默认部门") print("密码策略: 每人随机 4 位数字 + bcrypt（与门户一致）") print() stats = MigrationStats() mapping_rows = migrate_users( source_engine, target_engine, tenant_no=args.tenant_no, dept_id=args.dept_id, dept_mapping=dept_mapping or None, dry_run=args.dry_run, with_roles=not args.no_roles, stats=stats, ) if not args.dry_run: write_mapping_outputs( mapping_rows, Path(args.mapping_out), Path(args.mapping_csv), Path(args.credentials_out), ) else: print("（dry-run 不写库；映射预览见下方统计，正式执行后写入 JSON/CSV）") print(f"待迁移/处理用户数: {len(mapping_rows)}") print(f"新增用户: {stats.inserted_users}") print(f"跳过已存在: {stats.skipped_users}") print(f"新建角色: {stats.roles_created}") print(f"用户-角色关联: {stats.user_roles_inserted}") if stats.warnings: print("警告:") for w in stats.warnings: print(f" - {w}") if not args.dry_run: print(f"映射 JSON: {args.mapping_out}") print(f"映射 CSV: {args.mapping_csv}") print(f"密码清单: {args.credentials_out}") print("完成.") return 0 if __name__ == "__main__": raise SystemExit(main())