448 lines
17 KiB
Python
448 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
多数据库交互脚本
|
|
支持 MySQL, MongoDB, Redis, SQLite, PostgreSQL
|
|
跨平台支持: Windows, Linux, Termux 等
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
|
|
# 导入数据库模块（处理可能的导入错误）
|
|
try:
|
|
from mysql_db import MySQLDatabase
|
|
from mongodb_db import MongoDatabase
|
|
from redis_db import RedisDatabase
|
|
from sqlite_db import SQLiteDatabase
|
|
from postgres_db import PostgreSQLDatabase
|
|
except ImportError as e:
|
|
print(f"错误: 导入数据库模块失败")
|
|
print(f"详细信息: {e}")
|
|
print("\n请确保已安装所需的依赖包:")
|
|
print(" pip install -r requirements.txt")
|
|
sys.exit(1)
|
|
|
|
|
|
def print_banner():
|
|
"""打印欢迎信息"""
|
|
banner = """
|
|
╔══════════════════════════════════════════╗
|
|
║ 多数据库交互工具 v1.0 ║
|
|
║ 支持: MySQL, MongoDB, Redis, ║
|
|
║ SQLite, PostgreSQL ║
|
|
╚══════════════════════════════════════════╝
|
|
"""
|
|
print(banner)
|
|
|
|
|
|
def print_help():
|
|
"""打印帮助信息"""
|
|
help_text = """
|
|
========================================================================
|
|
脚本通用命令:
|
|
- help : 显示帮助信息
|
|
- connect : 连接到数据库
|
|
- disconnect : 断开数据库连接
|
|
- status : 查看连接状态
|
|
- switch <db> : 切换数据库类型 (mysql/mongodb/redis/sqlite/postgres)
|
|
- show databases: 列出所有数据库 (仅MySQL/PostgreSQL)
|
|
- use <database>: 切换到指定数据库 (仅MySQL/PostgreSQL)
|
|
- quit/exit : 退出程序
|
|
========================================================================
|
|
MySQL/PostgreSQL/SQLite 通用SQL语句:
|
|
SELECT * FROM users;
|
|
INSERT INTO users (name, age) VALUES ('张三', 25);
|
|
UPDATE users SET age = 26 WHERE name = '张三';
|
|
DELETE FROM users WHERE name = '张三';
|
|
========================================================================
|
|
SQLite 点命令:
|
|
.tables - 列出所有表
|
|
.schema - 显示所有表的结构
|
|
.schema <table> - 显示特定表的结构
|
|
.databases - 列出数据库信息
|
|
.indexes - 列出所有索引
|
|
.indexes <table> - 列出特定表的索引
|
|
========================================================================
|
|
MongoDB Shell 命令:
|
|
show dbs - 列出所有数据库
|
|
show collections - 列出当前数据库的集合
|
|
use <database> - 切换数据库
|
|
========================================================================
|
|
MongoDB JSON 命令:
|
|
{"collection": "users", "operation": "find", "query": {}}
|
|
{"collection": "users", "operation": "insert", "document": {"name": "张三", "age": 25}}
|
|
{"collection": "users", "operation": "update", "query": {"name": "张三"}, "update": {"$set": {"age": 26}}}
|
|
{"collection": "users", "operation": "delete", "query": {"name": "张三"}}
|
|
========================================================================
|
|
Redis 键值命令:
|
|
SET name 张三
|
|
GET name
|
|
HSET user:1 name 张三 age 25
|
|
HGETALL user:1
|
|
KEYS *
|
|
========================================================================
|
|
"""
|
|
print(help_text)
|
|
|
|
|
|
def get_database_config(db_type):
|
|
"""
|
|
获取数据库配置
|
|
:param db_type: 数据库类型
|
|
:return: 配置字典
|
|
"""
|
|
config = {}
|
|
|
|
if db_type == 'mysql':
|
|
print("\n请输入 MySQL 配置信息:")
|
|
config['host'] = input("主机地址 [localhost]: ").strip() or 'localhost'
|
|
config['port'] = int(input("端口 [3306]: ").strip() or '3306')
|
|
config['user'] = input("用户名 [root]: ").strip() or 'root'
|
|
config['password'] = input("密码: ").strip()
|
|
|
|
# 默认先列出数据库，然后让用户选择
|
|
config['database'] = None # 先不指定数据库
|
|
config['_list_databases'] = True # 标记需要列出数据库
|
|
|
|
elif db_type == 'mongodb':
|
|
print("\n请输入 MongoDB 配置信息:")
|
|
config['host'] = input("主机地址 [localhost]: ").strip() or 'localhost'
|
|
config['port'] = int(input("端口 [27017]: ").strip() or '27017')
|
|
config['database'] = input("数据库名 [test]: ").strip() or 'test'
|
|
|
|
# 询问是否需要认证
|
|
need_auth = input("是否需要用户认证? (y/n) [n]: ").strip().lower()
|
|
if need_auth == 'y':
|
|
config['username'] = input("用户名: ").strip()
|
|
config['password'] = input("密码: ").strip()
|
|
else:
|
|
config['username'] = None
|
|
config['password'] = None
|
|
|
|
elif db_type == 'redis':
|
|
print("\n请输入 Redis 配置信息:")
|
|
config['host'] = input("主机地址 [localhost]: ").strip() or 'localhost'
|
|
config['port'] = int(input("端口 [6379]: ").strip() or '6379')
|
|
config['db'] = int(input("数据库编号 [0]: ").strip() or '0')
|
|
password = input("密码 (留空表示无密码): ").strip()
|
|
config['password'] = password if password else None
|
|
|
|
elif db_type == 'sqlite':
|
|
print("\n请输入 SQLite 配置信息:")
|
|
config['database'] = input("数据库文件路径 [database.db]: ").strip() or 'database.db'
|
|
|
|
elif db_type == 'postgres':
|
|
print("\n请输入 PostgreSQL 配置信息:")
|
|
config['host'] = input("主机地址 [localhost]: ").strip() or 'localhost'
|
|
config['port'] = int(input("端口 [5432]: ").strip() or '5432')
|
|
config['user'] = input("用户名 [postgres]: ").strip() or 'postgres'
|
|
config['password'] = input("密码: ").strip()
|
|
|
|
# 默认先列出数据库，然后让用户选择
|
|
config['database'] = 'postgres' # 先连接到默认数据库
|
|
config['_list_databases'] = True # 标记需要列出数据库
|
|
|
|
return config
|
|
|
|
|
|
def create_database_instance(db_type, config):
|
|
"""
|
|
创建数据库实例
|
|
:param db_type: 数据库类型
|
|
:param config: 配置字典
|
|
:return: 数据库实例
|
|
"""
|
|
if db_type == 'mysql':
|
|
return MySQLDatabase(**config)
|
|
elif db_type == 'mongodb':
|
|
return MongoDatabase(**config)
|
|
elif db_type == 'redis':
|
|
return RedisDatabase(**config)
|
|
elif db_type == 'sqlite':
|
|
return SQLiteDatabase(**config)
|
|
elif db_type == 'postgres':
|
|
return PostgreSQLDatabase(**config)
|
|
else:
|
|
return None
|
|
|
|
|
|
def format_output(data):
|
|
"""
|
|
格式化输出结果
|
|
:param data: 数据
|
|
:return: 格式化后的字符串
|
|
"""
|
|
if isinstance(data, list):
|
|
# 检查是否是字符串列表（如 .schema 的输出）
|
|
if data and all(isinstance(item, str) for item in data):
|
|
# 直接输出，保留换行符
|
|
return '\n\n'.join(data)
|
|
else:
|
|
# JSON 格式化
|
|
return json.dumps(data, ensure_ascii=False, indent=2)
|
|
elif isinstance(data, dict):
|
|
return json.dumps(data, ensure_ascii=False, indent=2)
|
|
else:
|
|
return str(data)
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print_banner()
|
|
|
|
# 选择数据库类型
|
|
print("请选择数据库类型:")
|
|
print("1. MySQL")
|
|
print("2. MongoDB")
|
|
print("3. Redis")
|
|
print("4. SQLite")
|
|
print("5. PostgreSQL")
|
|
|
|
choice = input("\n请输入选项 (1-5): ").strip()
|
|
|
|
db_types = {
|
|
'1': 'mysql',
|
|
'2': 'mongodb',
|
|
'3': 'redis',
|
|
'4': 'sqlite',
|
|
'5': 'postgres'
|
|
}
|
|
|
|
if choice not in db_types:
|
|
print("无效的选项!")
|
|
return
|
|
|
|
current_db_type = db_types[choice]
|
|
print(f"\n已选择: {current_db_type.upper()}")
|
|
|
|
# 获取配置并创建数据库实例
|
|
config = get_database_config(current_db_type)
|
|
list_databases = config.pop('_list_databases', False)
|
|
db = create_database_instance(current_db_type, config)
|
|
|
|
if not db:
|
|
print("创建数据库实例失败!")
|
|
return
|
|
|
|
# 是否自动连接
|
|
#auto_connect = input("\n是否立即连接? (y/n) [y]: ").strip().lower()
|
|
#if auto_connect != 'n':
|
|
success, message = db.connect()
|
|
print(f"\n{message}")
|
|
connected = success
|
|
|
|
# 如果需要列出数据库
|
|
if connected and list_databases and hasattr(db, 'list_databases'):
|
|
success_list, result = db.list_databases()
|
|
if success_list:
|
|
print("\n可用的数据库:")
|
|
for i, db_name in enumerate(result, 1):
|
|
print(f" {i}. {db_name}")
|
|
|
|
# 让用户选择数据库
|
|
if current_db_type in ['mysql', 'postgres']:
|
|
while True:
|
|
db_choice = input("\n请输入数据库名或编号 (直接回车跳过): ").strip()
|
|
if not db_choice:
|
|
break
|
|
|
|
# 判断是数字还是数据库名
|
|
selected_db = None
|
|
if db_choice.isdigit():
|
|
index = int(db_choice) - 1
|
|
if 0 <= index < len(result):
|
|
selected_db = result[index]
|
|
else:
|
|
print(f"错误: 编号超出范围，请输入 1-{len(result)} 之间的数字")
|
|
retry = input("是否重新输入? (y/n) [y]: ").strip().lower()
|
|
if retry == 'n':
|
|
break
|
|
continue
|
|
else:
|
|
selected_db = db_choice
|
|
|
|
# 尝试切换数据库
|
|
use_success, use_msg = db.use_database(selected_db)
|
|
print(use_msg)
|
|
|
|
if use_success:
|
|
break
|
|
else:
|
|
retry = input("是否重新输入? (y/n) [y]: ").strip().lower()
|
|
if retry == 'n':
|
|
break
|
|
else:
|
|
print(f"\n{result}")
|
|
|
|
|
|
print("\n输入 'help' 查看帮助信息")
|
|
print("=" * 50)
|
|
|
|
# 主循环
|
|
while True:
|
|
try:
|
|
# 显示提示符
|
|
prompt = f"\n[{current_db_type.upper()}{'*' if connected else ''}]> "
|
|
command = input(prompt).strip()
|
|
|
|
if not command:
|
|
continue
|
|
|
|
# 处理特殊命令
|
|
cmd_lower = command.lower()
|
|
|
|
if cmd_lower in ['quit', 'exit']:
|
|
if connected:
|
|
print(db.close())
|
|
print("\n再见!")
|
|
break
|
|
|
|
elif cmd_lower == 'help':
|
|
print_help()
|
|
continue
|
|
|
|
elif cmd_lower == 'connect':
|
|
if connected:
|
|
print("已经连接到数据库")
|
|
else:
|
|
success, message = db.connect()
|
|
print(message)
|
|
connected = success
|
|
continue
|
|
|
|
elif cmd_lower == 'disconnect':
|
|
if not connected:
|
|
print("未连接到数据库")
|
|
else:
|
|
print(db.close())
|
|
connected = False
|
|
continue
|
|
|
|
elif cmd_lower == 'status':
|
|
status = "已连接" if connected else "未连接"
|
|
print(f"数据库类型: {current_db_type.upper()}")
|
|
print(f"连接状态: {status}")
|
|
if connected and hasattr(db, 'database') and db.database:
|
|
print(f"当前数据库: {db.database}")
|
|
continue
|
|
|
|
elif cmd_lower == 'show databases':
|
|
if not connected:
|
|
print("请先连接到数据库 (使用 'connect' 命令)")
|
|
elif not hasattr(db, 'list_databases'):
|
|
print("当前数据库类型不支持此命令")
|
|
else:
|
|
success, result = db.list_databases()
|
|
if success:
|
|
print("\n可用的数据库:")
|
|
for i, db_name in enumerate(result, 1):
|
|
print(f" {i}. {db_name}")
|
|
else:
|
|
print(f"\n{result}")
|
|
continue
|
|
|
|
elif cmd_lower.startswith('use '):
|
|
if not connected:
|
|
print("请先连接到数据库 (使用 'connect' 命令)")
|
|
elif not hasattr(db, 'use_database'):
|
|
print("当前数据库类型不支持此命令")
|
|
else:
|
|
db_input = command.split(maxsplit=1)[1].strip()
|
|
|
|
# 先获取数据库列表
|
|
list_success, db_list = db.list_databases()
|
|
|
|
# 判断是数字还是数据库名
|
|
selected_db = None
|
|
if db_input.isdigit() and list_success:
|
|
index = int(db_input) - 1
|
|
if 0 <= index < len(db_list):
|
|
selected_db = db_list[index]
|
|
else:
|
|
print(f"错误: 编号超出范围，请使用 'show databases' 查看可用数据库")
|
|
continue
|
|
else:
|
|
selected_db = db_input
|
|
|
|
# 尝试切换数据库
|
|
success, message = db.use_database(selected_db)
|
|
print(message)
|
|
|
|
# 如果失败，询问是否重试
|
|
if not success:
|
|
retry = input("是否重新输入数据库名? (y/n) [y]: ").strip().lower()
|
|
if retry != 'n':
|
|
while True:
|
|
new_db = input("请输入数据库名或编号: ").strip()
|
|
if not new_db:
|
|
break
|
|
|
|
# 再次判断数字或名称
|
|
if new_db.isdigit() and list_success:
|
|
idx = int(new_db) - 1
|
|
if 0 <= idx < len(db_list):
|
|
new_db = db_list[idx]
|
|
else:
|
|
print(f"错误: 编号超出范围")
|
|
continue
|
|
|
|
success2, message2 = db.use_database(new_db)
|
|
print(message2)
|
|
if success2:
|
|
break
|
|
|
|
retry_again = input("是否继续重试? (y/n) [y]: ").strip().lower()
|
|
if retry_again == 'n':
|
|
break
|
|
continue
|
|
|
|
elif cmd_lower.startswith('switch '):
|
|
new_type = cmd_lower.split()[1]
|
|
if new_type not in ['mysql', 'mongodb', 'redis', 'sqlite', 'postgres']:
|
|
print("无效的数据库类型!")
|
|
continue
|
|
|
|
# 关闭当前连接
|
|
if connected:
|
|
print(db.close())
|
|
connected = False
|
|
|
|
# 切换数据库
|
|
current_db_type = new_type
|
|
config = get_database_config(current_db_type)
|
|
db = create_database_instance(current_db_type, config)
|
|
print(f"已切换到 {current_db_type.upper()}")
|
|
continue
|
|
|
|
# 执行数据库命令
|
|
if not connected:
|
|
print("请先连接到数据库 (使用 'connect' 命令)")
|
|
continue
|
|
|
|
success, result = db.execute(command)
|
|
|
|
if success:
|
|
print(f"\n执行成功:")
|
|
print(format_output(result))
|
|
else:
|
|
print(f"\n执行失败:")
|
|
print(result)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\n操作已取消")
|
|
continue
|
|
except EOFError:
|
|
print("\n\n再见!")
|
|
break
|
|
except Exception as e:
|
|
print(f"\n发生错误: {str(e)}")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
print("\n\n程序已退出")
|
|
sys.exit(0)
|