如何编写批量用户导入脚本

wen 实用脚本 2026-07-27 1

本文目录导读：

如何编写批量用户导入脚本

Python批量导入脚本（推荐）
Shell脚本版本
使用方法
扩展功能建议
注意事项

我来帮您编写一个批量用户导入脚本,这里提供Python和Shell两种常见实现方式。

Python批量导入脚本（推荐）

基础版本 - 从CSV文件导入

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import csv
import hashlib
import json
import sys
from datetime import datetime
class UserImporter:
    def __init__(self, csv_file, output_format='json'):
        self.csv_file = csv_file
        self.output_format = output_format
        self.errors = []
        self.imported_users = []
    def validate_user(self, user):
        """验证用户数据"""
        errors = []
        # 必填字段检查
        required_fields = ['username', 'email', 'password']
        for field in required_fields:
            if not user.get(field):
                errors.append(f"缺少必填字段: {field}")
        # 用户名格式检查（示例：只允许字母数字和下划线）
        if user.get('username') and not user['username'].isalnum() and '_' not in user['username']:
            errors.append(f"用户名格式无效: {user['username']}")
        # 邮箱格式检查
        if user.get('email') and '@' not in user['email']:
            errors.append(f"邮箱格式无效: {user['email']}")
        # 密码强度检查（示例：至少6位）
        if user.get('password') and len(user['password']) < 6:
            errors.append(f"密码长度不足6位: {user['username']}")
        return errors
    def hash_password(self, password):
        """密码加密"""
        return hashlib.sha256(password.encode()).hexdigest()
    def import_from_csv(self):
        """从CSV文件导入用户"""
        try:
            with open(self.csv_file, 'r', encoding='utf-8') as f:
                reader = csv.DictReader(f)
                for row_num, user in enumerate(reader, start=2):
                    # 验证用户
                    validation_errors = self.validate_user(user)
                    if validation_errors:
                        self.errors.append({
                            'row': row_num,
                            'user': user.get('username', 'unknown'),
                            'errors': validation_errors
                        })
                        continue
                    # 处理密码
                    user['password_hash'] = self.hash_password(user['password'])
                    del user['password']
                    # 添加时间戳
                    user['created_at'] = datetime.now().isoformat()
                    user['status'] = 'active'
                    self.imported_users.append(user)
                return len(self.imported_users), len(self.errors)
        except FileNotFoundError:
            print(f"错误: 文件 {self.csv_file} 未找到")
            sys.exit(1)
        except Exception as e:
            print(f"错误: 读取CSV文件失败 - {str(e)}")
            sys.exit(1)
    def generate_password_report(self):
        """生成密码报告（仅显示密码强度等非敏感信息）"""
        report = []
        for user in self.imported_users:
            report.append({
                'username': user['username'],
                'email': user['email'],
                'password_length': 8,  # 示例：不显示真实密码长度
                'status': 'strong' if user['username'] in self.imported_users else 'weak'
            })
        return report
    def export_users(self, output_file):
        """导出用户数据"""
        if self.output_format == 'json':
            with open(output_file, 'w', encoding='utf-8') as f:
                json.dump({
                    'imported_users': self.imported_users,
                    'report': self.generate_password_report()
                }, f, indent=2, ensure_ascii=False)
        elif self.output_format == 'csv':
            # 导出为CSV
            pass
    def print_summary(self):
        """打印导入总结"""
        print("\n" + "="*50)
        print("批
量用户导入总结")
        print("="*50)
        print(f"成功导入: {len(self.imported_users)} 个用户")
        print(f"失败: {len(self.errors)} 个用户")
        if self.errors:
            print("\n错误详情:")
            for err in self.errors:
                print(f"  行{err['row']}: 用户 '{err['user']}' - {', '.join(err['errors'])}")
        print("="*50)
def main():
    import argparse
    parser = argparse.ArgumentParser(description='批量用户导入脚本')
    parser.add_argument('csv_file', help='CSV文件路径')
    parser.add_argument('-o', '--output', default='users_import_result.json', 
                       help='输出文件路径 (默认: users_import_result.json)')
    parser.add_argument('-f', '--format', choices=['json', 'csv'], default='json',
                       help='输出格式 (默认: json)')
    args = parser.parse_args()
    # 创建导入器实例
    importer = UserImporter(args.csv_file, args.format)
    # 执行导入
    success, failed = importer.import_from_csv()
    # 导出结果
    importer.export_users(args.output)
    # 打印总结
    importer.print_summary()
    print(f"结果已保存到: {args.output}")
if __name__ == "__main__":
    main()

CSV文件格式示例

username,email,password,department,role
zhangsan,zhangsan@example.com,password123,技术部,员工
lisi,lisi@example.com,pass456,市场部,经理
wangwu,wangwu@example.com,welcome789,财务部,主管

Shell脚本版本

#!/bin/bash
# 批量用户导入脚本 (Shell版本)
# 适用场景: Linux环境，需要批量创建系统用户
# 配置
CSV_FILE="users.csv"
DATE=$(date +%Y%m%d_%H%M%S)
LOG_FILE="user_import_${DATE}.log"
# 日志函数
log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a "$LOG_FILE"
}
# 检查文件是否存在
if [ ! -f "$CSV_FILE" ]; then
    log "错误: CSV文件 $CSV_FILE 不存在"
    exit 1
fi
# 初始化计数器
success=0
failed=0
log "开始批量导入用户..."
# 读取CSV文件（跳过第一行标题）
tail -n +2 "$CSV_FILE" | while IFS=',' read -r username email password role; do
    # 跳过空行
    [ -z "$username" ] && continue
    # 检查用户是否已存在
    if id "$username" &>/dev/null; then
        log "警告: 用户 $username 已存在，跳过"
        continue
    fi
    # 创建用户
    if useradd -m -s /bin/bash "$username" 2>>"$LOG_FILE"; then
        # 设置密码
        echo "$username:$password" | chpasswd 2>>"$LOG_FILE"
        if [ $? -eq 0 ]; then
            log "成功: 创建用户 $username"
            ((success++))
        else
            log "失败: 设置密码失败 - $username"
            ((failed++))
        fi
    else
        log "失败: 创建用户失败 - $username"
        ((failed++))
    fi
done
log "导入完成！成功: $success, 失败: $failed"

使用方法

Python版本使用

# 1. 保存脚本为 import_users.py
# 2. 准备CSV文件
# 3. 运行脚本
python import_users.py users.csv -o result.json -f json
# 查看帮助
python import_users.py -h

Shell版本使用

# 1. 保存脚本为 import_users.sh
# 2. 添加执行权限
chmod +x import_users.sh
# 3. 准备CSV文件（格式: username,email,password,role）
# 4. 运行脚本（需要root权限）
sudo ./import_users.sh

扩展功能建议

数据库集成 - 直接将用户导入数据库
API集成 - 通过REST API批量添加用户
密码策略检查 - 密码强度验证
批量发送欢迎邮件
验证码/邮箱确认
错误重试机制
日志记录增强
多格式支持 (Excel, XML, YAML等)

注意事项

安全性: 永远不要在日志中打印密码
性能: 大批量用户建议分批处理
验证: 导入前进行数据完整性检查
备份: 重要系统操作前做好备份

您可以根据具体需求选择适合的版本或进行定制开发。