cykj/shanxia2.0
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

提交

Browse Source
This commit is contained in:
zlwu committed 2026-07-26 21:06:09 +08:00
1 parent 70bdd820ce
commit bede140d12
988 files changed
+692310 -775910
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/scripts/migrate_users.py
-844
View File
@@ -1,844 +0,0 @@
#!/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())