目的

实现远程自动化oracle一键运维、性能优化等，避免现场翻sql文件，执行sql语句、重复执行等，提高效率

输出结果

(.venv) ➜  ~ /Users/zhoushaohua/Jaja_Code/Jaja_Studio/.venv/bin/python /Users/zhoushaohua/Jaja_Code/Jaja_Studio/Python/python/oracle工具/oracle_tools.py --host 127.0.0.1 --user system --password Passw0rd --port 1521 --server xe
请输入用户名: admin
请输入密码: 
                                       认证成功                                                                                   
                                       操作菜单                                                                                   
######################################################################################
1. 查看数据库 Profile 配置信息
100. 退出
######################################################################################
[主机:127.0.0.1 用户:system 数据库:xe]: 

脚本内容

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@文件        :oracle_tools.py
@邮件        :zhoushao12@163.com
@时间        :2024/08/19 18:16:15
@作者        :Jaja
@版本        :1.0
'''


import argparse
import cx_Oracle
import sys
import os
import time
import logging
import shutil
import sqlite3
from logging.handlers import TimedRotatingFileHandler
from prettytable import from_db_cursor
from colorama import init, Fore, Style
import getpass
import hashlib

# 初始化 colorama
init(autoreset=True)

# 设置输出编码为 UTF-8
sys.stdout = open(sys.stdout.fileno(), mode='w', encoding='utf-8', buffering=1)

def setup_logging():
    """设置日志记录配置"""
    log_directory = os.path.dirname(os.path.abspath(__file__))
    log_filename = os.path.join(log_directory, "database_script.log")
    log_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    
    file_handler = TimedRotatingFileHandler(log_filename, when="D", interval=1, backupCount=7)
    file_handler.setFormatter(log_formatter)
    
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    logger.addHandler(file_handler)
    
    return logger

def hash_password(password):
    """对密码进行 SHA-256 哈希处理"""
    return hashlib.sha256(password.encode()).hexdigest()

def connect_to_database(args, logger):
    """连接到 Oracle 数据库"""
    try:
        os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.AL32UTF8'
        logger.info(f"尝试连接到 Oracle 数据库 IP: {args.host} 用户: {args.user}")
        connection = cx_Oracle.connect(f"{args.user}/{args.password}@{args.host}:{args.port}/{args.server}")
        logger.info(f"成功连接到 Oracle 数据库 IP: {args.host} 用户: {args.user}")
        return connection
    except cx_Oracle.DatabaseError as e:
        logger.error(f"连接 Oracle 数据库失败: IP: {args.host}, 用户: {args.user}, 错误: {e}")
        print(f"数据库连接失败: {e}", Fore.RED)
        sys.exit(1)

def authenticate_user(logger):
    """认证用户"""
    db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'users.db')
    connection = sqlite3.connect(db_path)
    cursor = connection.cursor()
    try:
        username = input(Fore.YELLOW + "请输入用户名: ").strip()
        password = getpass.getpass(Fore.YELLOW + "请输入密码: ")
        hashed_password = hash_password(password)
        
        cursor.execute("SELECT password FROM users WHERE username = ?", (username,))
        row = cursor.fetchone()
        if row and row[0] == hashed_password:
            print_centered("认证成功", Fore.GREEN)
            logger.info(f"用户 {username} 成功认证")
        else:
            print_centered("认证失败", Fore.RED)
            logger.warning(f"用户 {username} 认证失败")
            sys.exit(1)
    finally:
        cursor.close()
        connection.close()

def get_terminal_width():
    """获取终端宽度"""
    return shutil.get_terminal_size((80, 20)).columns

def print_centered(text, color=Fore.WHITE):
    """居中显示文本"""
    terminal_width = get_terminal_width()
    centered_text = text.center(terminal_width)
    print(color + centered_text)

def print_left_aligned(text, color=Fore.WHITE):
    """左对齐显示文本"""
    print(color + text)

def print_menu_option(option, description):
    """打印菜单选项，左对齐"""
    print(f"{option}. {description}")

def log_database_operation(logger, operation_desc, db_info):
    """记录数据库操作日志"""
    log_message = (f"数据库操作: {operation_desc}. "
                   f"IP: {db_info['host']}, 用户: {db_info['user']}, 服务: {db_info['server']}")
    logger.info(log_message)

def save_query_result(result, db_info):
    """保存查询结果到文件"""
    timestamp = time.strftime("%Y%m%d_%H%M%S", time.localtime())
    result_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'results')
    os.makedirs(result_dir, exist_ok=True)
    file_path = os.path.join(result_dir, f"{db_info['host']}_{timestamp}.txt")
    with open(file_path, 'w', encoding='utf-8') as file:
        file.write(result)
    print_left_aligned(f"结果已保存到: {file_path}", Fore.GREEN)
    logger.info(f"查询结果保存到 {file_path}")

def close_database_resources(cursor, connection, logger):
    """关闭数据库游标和连接"""
    try:
        if cursor:
            cursor.close()
            logger.info("数据库游标已关闭")
        if connection:
            connection.close()
            logger.info("数据库连接已关闭")
    except cx_Oracle.InterfaceError as e:
        logger.error(f"关闭数据库资源时发生错误: {e}")

class Things:
    """定义数据库操作类"""
    def __init__(self, connection, logger):
        self.connection = connection
        self.logger = logger

    def db_profile(self):
        """查询数据库 profile 信息"""
        cursor = None
        try:
            operation_desc = "查询数据库用户 profile 信息"
            db_info = {'host': args.host, 'user': args.user, 'server': args.server}
            log_database_operation(self.logger, operation_desc, db_info)
            
            cursor = self.connection.cursor()
            cursor.execute("SELECT profile, resource_name, resource_type, limit FROM dba_profiles order by profile")
            table = from_db_cursor(cursor)
            print_centered("数据库 Profile 配置信息", Fore.GREEN)
            print_centered(str(table), Fore.GREEN)  # 居中显示查询结果
            
            # 提示用户是否保存结果
            save_choice = input(Fore.YELLOW + "是否保存结果？(y/n): ").strip().lower()
            if save_choice == 'y':
                save_query_result(str(table), db_info)
        except cx_Oracle.DatabaseError as e:
            self.logger.error(f"检索 Profile 信息时发生错误: {e}")
            print_centered(f"获取 Profile 信息失败: {e}", Fore.RED)
        finally:
            if cursor:
                cursor.close()
                self.logger.info("数据库游标已关闭")

class Menu:
    def __init__(self, connection, logger):
        self.things = Things(connection, logger)
        self.logger = logger
        self.choices = {
            "1": {"desc": "查看数据库 Profile 配置信息", "action": self.things.db_profile},
            "100": {"desc": "退出", "action": self.quit}
        }

    def display_menu(self):
        print_centered("操作菜单", Fore.CYAN)
        print(Style.BRIGHT + Fore.CYAN + "#" * get_terminal_width())
        for key, value in self.choices.items():
            print_menu_option(key, value["desc"])
        print(Style.BRIGHT + Fore.CYAN + "#" * get_terminal_width())

    def run(self):
        while True:
            self.display_menu()
            try:
                choice = input(Fore.YELLOW + f"[主机:{args.host} 用户:{args.user} 数据库:{args.server}]: ").strip()
                self.logger.info(f"用户选择的选项: {choice}")
            except Exception as e:
                self.logger.error(f"无效输入: {e}")
                print_centered("无效输入，请重新输入！", Fore.RED)
                continue

            action = self.choices.get(choice, {}).get("action")
            if action:
                action()
            else:
                self.logger.warning(f"无效的菜单选项: {choice}")
                print_centered(f"{choice} 无效输入！！", Fore.RED)

    def quit(self):
        """退出程序"""
        self.logger.info("用户选择退出脚本")
        print_centered("大爷请再来哟！", Fore.GREEN)
        sys.exit(0)

if __name__ == '__main__':
    logger = setup_logging()
    authenticate_user(logger)

    parser = argparse.ArgumentParser(description='Oracle数据库操作脚本')
    parser.add_argument("--host", type=str, default="127.0.0.1")
    parser.add_argument("--user", type=str, default="system")
    parser.add_argument("--password", type=str, default="oracle")
    parser.add_argument("--port", type=str, default="1521")
    parser.add_argument("--server", type=str, required=True)
    args = parser.parse_args()

    connection = connect_to_database(args, logger)
    try:
        Menu(connection, logger).run()
    except KeyboardInterrupt as e1:
        logger.warning(f"脚本中断: {e1}")
        print_centered(f'其他错误，请检查：{e1}', Fore.RED)
    finally:
        close_database_resources(None, connection, logger)