Lei PAN
163 lines
5.0 KiB
Python
163 lines
5.0 KiB
Python
import subprocess
|
|
import os
|
|
import datetime
|
|
import sys
|
|
|
|
from db.mongo_manager import MONGO_DB_URL
|
|
|
|
# ================= 配置区域 =================
|
|
# 数据库连接信息
|
|
MONGO_HOST = "mongo.lpaconsulting.fr"
|
|
MONGO_PORT = "27017"
|
|
MONGO_DB_NAME = "appointment" # 你要备份/恢复的数据库名
|
|
|
|
# Get MongoDB credentials from environment variables
|
|
MONGO_USER = os.getenv(
|
|
"MONGO_USER", "appointment"
|
|
) # Default to 'appointment' if not set
|
|
MONGO_PASS = os.getenv("MONGO_PASS", "Rdv@2022") # Default to 'Rdv@2022' if not set
|
|
|
|
# 备份存放的根目录
|
|
BACKUP_DIR_ROOT = "./mongo_backups"
|
|
|
|
# ===========================================
|
|
|
|
|
|
def get_auth_args():
|
|
"""构建认证参数列表"""
|
|
args = []
|
|
if MONGO_USER and MONGO_PASS:
|
|
args.extend(
|
|
[
|
|
"--username",
|
|
MONGO_USER,
|
|
"--password",
|
|
MONGO_PASS,
|
|
"--authenticationDatabase",
|
|
"appointment",
|
|
]
|
|
)
|
|
return args
|
|
|
|
|
|
def backup_mongo():
|
|
"""执行备份操作"""
|
|
# 1. 创建带有时间戳的备份文件夹
|
|
timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
|
backup_path = os.path.join(BACKUP_DIR_ROOT, timestamp)
|
|
|
|
if not os.path.exists(backup_path):
|
|
os.makedirs(backup_path)
|
|
|
|
print(f"[*] 开始备份数据库: {MONGO_DB_NAME} 到 {backup_path} ...")
|
|
|
|
# 2. 构建 mongodump 命令
|
|
# 命令格式: mongodump --host <host> --port <port> --db <db> --out <path> [auth]
|
|
cmd = [
|
|
"mongodump",
|
|
"--host",
|
|
MONGO_HOST,
|
|
"--port",
|
|
MONGO_PORT,
|
|
"--db",
|
|
MONGO_DB_NAME,
|
|
"--out",
|
|
backup_path,
|
|
]
|
|
|
|
# 添加认证参数
|
|
cmd.extend(get_auth_args())
|
|
|
|
try:
|
|
# 3. 执行命令
|
|
result = subprocess.run(cmd, check=True, text=True, capture_output=True)
|
|
print(f"[+] 备份成功！")
|
|
print(f" 存储路径: {backup_path}")
|
|
print(f" 日志: {result.stderr}") # mongodump 通常把进度输出到 stderr
|
|
return backup_path
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"[-] 备份失败: {e}")
|
|
print(f" 错误信息: {e.stderr}")
|
|
return None
|
|
|
|
|
|
def restore_mongo(backup_source_path):
|
|
"""
|
|
执行恢复操作
|
|
backup_source_path: 备份文件夹的路径 (例如 ./mongo_backups/2023-10-27_10-00-00)
|
|
"""
|
|
# mongodump 的输出结构通常是: backup_dir/db_name/collection.bson
|
|
# 所以我们需要指向具体的数据库文件夹，或者指向父文件夹并指定 --db
|
|
|
|
target_dir = os.path.join(backup_source_path, MONGO_DB_NAME)
|
|
|
|
if not os.path.exists(target_dir):
|
|
print(
|
|
f"[-] 错误: 在路径 {backup_source_path} 下找不到数据库 {MONGO_DB_NAME} 的备份文件。"
|
|
)
|
|
return
|
|
|
|
print(f"[*] 开始恢复数据库: {MONGO_DB_NAME} 从 {target_dir} ...")
|
|
|
|
# 构建 mongorestore 命令
|
|
# 命令格式: mongorestore --host <host> --port <port> --db <db> <path_to_bson_files> [auth]
|
|
cmd = [
|
|
"mongorestore",
|
|
"--host",
|
|
MONGO_HOST,
|
|
"--port",
|
|
MONGO_PORT,
|
|
"--db",
|
|
MONGO_DB_NAME,
|
|
"--drop", # 警告：这会在恢复前删除现有集合，确保数据干净。根据需要移除此项。
|
|
target_dir,
|
|
]
|
|
|
|
cmd.extend(get_auth_args())
|
|
|
|
try:
|
|
result = subprocess.run(cmd, check=True, text=True, capture_output=True)
|
|
print(f"[+] 恢复成功！")
|
|
print(f" 日志: {result.stderr}")
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"[-] 恢复失败: {e}")
|
|
print(f" 错误信息: {e.stderr}")
|
|
|
|
|
|
# ================= 主程序入口 =================
|
|
if __name__ == "__main__":
|
|
print("请选择操作:")
|
|
print("1. 备份数据库 (Backup)")
|
|
print("2. 恢复数据库 (Restore)")
|
|
|
|
choice = input("请输入数字 (1/2): ").strip()
|
|
|
|
if choice == "1":
|
|
backup_mongo()
|
|
elif choice == "2":
|
|
# 列出所有备份供用户选择
|
|
if not os.path.exists(BACKUP_DIR_ROOT):
|
|
print("[-] 没有找到备份目录。")
|
|
else:
|
|
backups = sorted(os.listdir(BACKUP_DIR_ROOT))
|
|
if not backups:
|
|
print("[-] 目录为空。")
|
|
else:
|
|
print("\n可用备份:")
|
|
for idx, name in enumerate(backups):
|
|
print(f"{idx + 1}. {name}")
|
|
|
|
try:
|
|
idx_choice = int(input("\n请选择要恢复的备份编号: ")) - 1
|
|
if 0 <= idx_choice < len(backups):
|
|
selected_backup = os.path.join(
|
|
BACKUP_DIR_ROOT, backups[idx_choice]
|
|
)
|
|
restore_mongo(selected_backup)
|
|
else:
|
|
print("[-] 无效的选择。")
|
|
except ValueError:
|
|
print("[-] 请输入数字。")
|
|
else:
|
|
print("[-] 无效输入，退出。")
|