shanxia2.0/backend/scripts/migrate_users.py

#!/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())