feat: 新增将两种加密方式合并一个脚本、新增批量修改邮箱功能

hyb

2026-01-30 7657e1b2fa251a2ea372710ad75cb395a3c0e374

 测试组/脚本/Change_password/数据库密码批量修改工具（合并版）.py

@@ -198,14 +198,15 @@
    error_signal = pyqtSignal(str)
    export_finished_signal = pyqtSignal(str)

    def __init__(self, excel_path, new_password, mode, hash_mode, user_filter=None, selected_projects=None):
    def __init__(self, excel_path, new_value, mode, hash_mode, user_filter=None, selected_projects=None, modify_type='password'):
        super().__init__()
        self.excel_path = excel_path
        self.new_password = new_password
        self.new_value = new_value  # 新密码或新邮箱
        self.mode = mode  # 'all' 或 'specific'
        self.hash_mode = hash_mode  # 'bcrypt' 或 'argon2id'
        self.user_filter = user_filter
        self.selected_projects = selected_projects
        self.modify_type = modify_type  # 'password' 或 'email'
        self.updaters = []
        self.log_records = []
        self.is_running = True
@@ -250,7 +251,7 @@
            for _, row in df_filtered.iterrows():
                project, host, port, database, user, password = row[:6]
                self.log_signal.emit(f"准备处理项目：{project} | 数据库：{database} | 端口：{port}")
                updater = ProjectUpdater(project, host, database, user, password, port, self.hash_mode)
                updater = ProjectUpdater(project, host, database, user, password, port, self.hash_mode, self.modify_type)
                self.updaters.append(updater)

            # 执行更新任务
@@ -262,10 +263,16 @@
                self.log_signal.emit(f"正在处理项目: {updater.project}")

                # 运行更新
                success = updater.update_database(
                    user_filter=self.user_filter if self.mode == 'specific' else None,
                    new_password=self.new_password
                )
                if self.modify_type == 'password':
                    success = updater.update_database(
                        user_filter=self.user_filter if self.mode == 'specific' else None,
                        new_password=self.new_value
                    )
                else:
                    success = updater.update_email(
                        user_filter=self.user_filter if self.mode == 'specific' else None,
                        new_email=self.new_value
                    )

                if success:
                    self.log_records.extend(updater.log_records)
@@ -332,7 +339,7 @@
            conn = updater.connect()
            if conn:
                cursor = conn.cursor()
                success = updater.export_user_info(cursor, self.new_password)
                success = updater.export_user_info(cursor, self.new_value)
                cursor.close()
                conn.close()

@@ -350,7 +357,7 @@
class ProjectUpdater:
    """单个项目更新器"""

    def __init__(self, project, host, database, user, password, port=3306, hash_mode='bcrypt'):
    def __init__(self, project, host, database, user, password, port=3306, hash_mode='bcrypt', modify_type='password'):
        self.project = project
        self.host = host
        self.database = database
@@ -358,6 +365,7 @@
        self.db_password = password
        self.port = port
        self.hash_mode = hash_mode  # 'bcrypt' 或 'argon2id'
        self.modify_type = modify_type  # 'password' 或 'email'
        self.updated_users = []
        self.log_records = []

@@ -448,7 +456,7 @@
        except Error:
            return "获取失败"

    def export_user_info(self, cursor, new_password):
    def export_user_info(self, cursor, new_value):
        """导出用户信息到Excel"""
        try:
            if not self.updated_users:
@@ -469,21 +477,32 @@
            # 准备导出数据
            export_data = []
            for user in self.updated_users:
                export_data.append({
                    '用户名': user['login_name'],
                    '密码': new_password,
                    '姓名': user['name'],
                    '角色名称': role_name_map.get(user['role_id'], '无角色')
                })
                if self.modify_type == 'password':
                    export_data.append({
                        '用户名': user['login_name'],
                        '密码': new_value,
                        '姓名': user['name'],
                        '角色名称': role_name_map.get(user['role_id'], '无角色')
                    })
                else:  # 修改邮箱
                    export_data.append({
                        '用户名': user['login_name'],
                        '邮箱': new_value,
                        '姓名': user['name'],
                        '角色名称': role_name_map.get(user['role_id'], '无角色')
                    })

            # 创建DataFrame并导出
            df_export = pd.DataFrame(export_data)

            # 按照要求格式化文件名：项目名称测试环境账号密码_年月日
            safe_project_name = re.sub(r'[\\/*?:"<>|]', "", self.project)
            # 按照要求格式化文件名
            safe_project_name = re.sub(r'[\/*?:"<>|]', "", self.project)
            safe_project_name = safe_project_name.replace(" ", "_")
            date_str = datetime.now().strftime("%Y%m%d")
            filename = f"{safe_project_name}测试环境账号密码_{date_str}.xlsx"
            if self.modify_type == 'password':
                filename = f"{safe_project_name}测试环境账号密码_{date_str}.xlsx"
            else:  # 修改邮箱
                filename = f"{safe_project_name}测试环境账号邮箱_{date_str}.xlsx"
            filepath = os.path.join(DEFAULT_EXPORT_DIR, filename)

            df_export.to_excel(filepath, index=False)
@@ -555,6 +574,72 @@
            if conn:
                conn.close()

    def update_email(self, user_filter=None, new_email=''):
        """更新数据库中的用户邮箱"""
        conn = None
        cursor = None
        try:
            conn = self.connect()
            if not conn:
                return False

            # 使用同步方式创建游标
            cursor = conn.cursor()

            # 获取需要更新的用户
            users = self.get_users_to_update(cursor, user_filter)

            if not users:
                return False

            # 更新每个用户的邮箱
            for uid, login_name, name, current_role_id in users:
                success = self.update_user_email(cursor, uid, login_name, new_email)

                if success:
                    # 记录更新成功的用户信息
                    self.updated_users.append({
                        'id': uid,
                        'login_name': login_name,
                        'name': name,
                        'role_id': current_role_id
                    })

                    # 写入日志
                    self.log_records.append({
                        '项目': self.project,
                        '数据库': self.database,
                        '用户ID': uid,
                        '用户名': login_name,
                        '新邮箱': new_email
                    })

            # 提交事务
            conn.commit()
            return True

        except Error:
            return False
        finally:
            if cursor:
                cursor.close()
            if conn:
                conn.close()

    def update_user_email(self, cursor, user_id, login_name, new_email):
        """更新用户邮箱"""
        try:
            # 构造更新 SQL
            update_sql = """
                UPDATE sys_user
                SET email = %s
                WHERE id = %s
            """
            cursor.execute(update_sql, (new_email, user_id))
            return True
        except Error:
            return False


class PasswordUpdaterApp(QMainWindow):
    """主应用程序窗口"""
@@ -593,6 +678,24 @@
        excel_layout.addWidget(excel_browse_btn)
        config_layout.addWidget(excel_group)

        # 修改类型选择
        modify_type_group = QGroupBox("修改类型")
        modify_type_layout = QVBoxLayout(modify_type_group)
        
        self.modify_type_group = QButtonGroup()
        self.modify_password_radio = QRadioButton("修改密码")
        self.modify_email_radio = QRadioButton("修改邮箱")
        
        # 默认选择修改密码
        self.modify_password_radio.setChecked(True)
        
        self.modify_type_group.addButton(self.modify_password_radio)
        self.modify_type_group.addButton(self.modify_email_radio)
        
        modify_type_layout.addWidget(self.modify_password_radio)
        modify_type_layout.addWidget(self.modify_email_radio)
        config_layout.addWidget(modify_type_group)

        # 新密码设置
        password_group = QGroupBox("新密码设置")
        password_layout = QHBoxLayout(password_group)
@@ -601,6 +704,16 @@
        password_layout.addWidget(QLabel("新密码:"))
        password_layout.addWidget(self.password_edit)
        config_layout.addWidget(password_group)

        # 新邮箱设置
        email_group = QGroupBox("新邮箱设置")
        email_layout = QHBoxLayout(email_group)
        self.email_edit = QLineEdit()
        self.email_edit.setPlaceholderText("请输入新邮箱")
        self.email_edit.setEnabled(False)  # 默认禁用
        email_layout.addWidget(QLabel("新邮箱:"))
        email_layout.addWidget(self.email_edit)
        config_layout.addWidget(email_group)

        # 加密方式选择
        hash_group = QGroupBox("加密方式选择")
@@ -647,6 +760,10 @@
            lambda: self.usernames_edit.setEnabled(not self.all_users_radio.isChecked()))
        self.specific_users_radio.toggled.connect(
            lambda: self.usernames_edit.setEnabled(self.specific_users_radio.isChecked()))
        
        # 修改类型选择信号
        self.modify_password_radio.toggled.connect(self.on_modify_type_changed)
        self.modify_email_radio.toggled.connect(self.on_modify_type_changed)

        mode_layout.addWidget(self.all_users_radio)
        mode_layout.addWidget(self.specific_users_radio)
@@ -742,20 +859,31 @@
                return
            user_filter = [name.strip() for name in usernames.replace('，', ',').split(',')]

        # 获取新密码
        new_password = self.password_edit.text()
        if not new_password:
            QMessageBox.warning(self, "错误", "请输入新密码")
            return
        # 获取修改类型
        modify_type = 'password' if self.modify_password_radio.isChecked() else 'email'
        
        # 获取新密码或新邮箱
        new_value = ''
        if modify_type == 'password':
            new_value = self.password_edit.text()
            if not new_value:
                QMessageBox.warning(self, "错误", "请输入新密码")
                return
        else:
            new_value = self.email_edit.text()
            if not new_value:
                QMessageBox.warning(self, "错误", "请输入新邮箱")
                return

        # 创建并启动更新线程
        self.updater_thread = DatabaseUpdater(
            self.excel_path_edit.text(),
            new_password,
            new_value,
            mode,
            hash_mode,
            user_filter,
            selected_projects
            selected_projects,
            modify_type
        )

        # 连接信号
@@ -801,6 +929,17 @@
            self.log_text.append(f"开始导出项目 {project_name} 的用户信息...")
            self.updater_thread.export_user_info(project_name)

    def on_modify_type_changed(self):
        """修改类型变更时的处理"""
        is_modify_password = self.modify_password_radio.isChecked()
        # 启用或禁用密码相关控件
        self.password_edit.setEnabled(is_modify_password)
        # 启用或禁用邮箱相关控件
        self.email_edit.setEnabled(not is_modify_password)
        # 启用或禁用加密方式选择
        self.bcrypt_radio.setEnabled(is_modify_password)
        self.argon2id_radio.setEnabled(is_modify_password)

    def show_error(self, error_msg):
        """显示错误信息"""
        QMessageBox.critical(self, "错误", error_msg)