windows平台下MongoDB数据库自动备份，python实现

需求环境

MongoDB >= 8.0
python >= 3.8
请自行安装好以上相关软件，并确认设置好环境变量。

创建执行文件(python文件)

在任意目录创建一个 .py文件，例如我在D盘根目录下创建 D:\backup.py，用记事本或其他编辑器打开刚刚创建的 .py文件，复制以下代码粘贴进去。代码前面几行变量可根据自己实际情况更改

import os
import shutil
from datetime import datetime

# 备份目录
backup_dir = "D:\\backup\\database\\MongoDB"

# 备份文件保留最近个数
keep_num = 5

# 数据库主机
host = "localhost"

# 数据库端口
port = 27017

# 数据库名称
db_name = "test"

backup_temp_dir = backup_dir + "\\temp"

# 执行 mongodump 命令 windows需要先安装mongodump 命令，并配置环境变量
backup_command = f"mongodump --host {host} --port {port} -d {db_name} --out {backup_temp_dir}"

def check_dir(dir_path):
   """
   检查目录是否存在，如果不存在则返回False，否则返回True
   :param dir_path:
   :return:
   """
   if not os.path.exists(dir_path):
       return False
   return True

def check_temp_dir():
   """
   检查临时目录是否存在，如果存在则清空，不存在则创建
   :return:
   """
   if check_dir(backup_temp_dir):
       for file in os.listdir(backup_temp_dir):
           os.remove(backup_temp_dir + "\\" + file)

       return True
   else:
       os.mkdir(backup_temp_dir)
       return True

def run_command(command):
   """
   运行系统命令
   :param command:
   :return:
   """
   print(f"正在执行命令：{command}")
   run_status = os.system(command)
   # 输出运行响应的字符串
   print(run_status)
   print("命令执行完毕")
   return True

def zip_backup(src_dir, output_filename):
   """
   压缩指定目录到一个 ZIP 文件。

   :param src_dir: 要压缩的目录路径
   :param output_filename: 输出的 ZIP 文件名（不包括 .zip 扩展名）
   """

   # 压缩目录
   shutil.make_archive(output_filename, 'zip', src_dir)
   print(f"目录 {src_dir} 已成功压缩为 {output_filename}.zip")

def delete_old_files(directory, keep_count):
   """
   删除指定目录中除最新文件外的所有文件。
   :param directory: 目录
   :param keep_count: 保留文件个数
   """
   files = os.listdir(directory)
   if len(files) <= keep_count:
       print("目录中文件个数小于等于保留个数，无需删除")
       return True
   files.sort(key=lambda x: os.path.getmtime(os.path.join(directory, x)))
   delete_count = len(files) - keep_count
   print(f"需要删除 {delete_count} 个文件")
   for i in range(delete_count):
       file_to_delete = files[i]
       file_path = os.path.join(directory, file_to_delete)
       print(f"正在删除 {file_path}")
       os.remove(file_path)
       print(f"{file_path} 已删除")
   return True

print("开始执行备份命令")
check = check_dir(backup_dir)
# 如果目录不存在，则递归创建
if not check:
   print("备份目录不存在，正在创建")
   os.makedirs(backup_dir)

check = check_temp_dir()
if not check:
   print("检查备份临时目录失败，请检查路径是否正确")
   exit()

run_command(backup_command)

# 检查临时目录是否存在文件
file_count = len(os.listdir(backup_temp_dir))
print(f"临时目录下共有 {file_count} 个文件")
if file_count <= 0:
   print("临时目录下还不存在文件，请确认备份命令是否执行成功！")
   exit()

print("开始压缩备份文件")
zip_name = backup_dir + "\\backup_" + db_name + "_" + datetime.now().strftime("%Y%m%d-%H%M")
zip_backup(backup_temp_dir, zip_name)
# 删除临时目录
shutil.rmtree(backup_temp_dir)
# 删除旧文件
delete_old_files(backup_dir, keep_num)
print("清理旧文件完毕")
print("备份完成!")