基于Python编写MySQL数据库备份脚本

import subprocess
import gzip
import os
import logging
import argparse
from datetime import datetime, timedelta

# 配置日志记录
logging.basicconfig(
    level=logging.info,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.filehandler('db_backup.log'),
        logging.streamhandler()
    ]
)

def backup_mysql(host, user, password, database, backup_dir, retain_days):
    try:
        # 创建备份目录
        os.makedirs(backup_dir, exist_ok=true)
        
        # 生成备份文件名
        timestamp = datetime.now().strftime('%y%m%d%h%m%s')
        backup_file = os.path.join(backup_dir, f"{database}_{timestamp}.sql")
        compressed_file = f"{backup_file}.gz"

        # 执行mysqldump命令
        logging.info(f"starting backup for database: {database}")
        with open(backup_file, 'w') as f:
            cmd = [
                'mysqldump',
                f'-h{host}',
                f'-u{user}',
                f'-p{password}',
                '--single-transaction',
                '--routines',
                '--triggers',
                database
            ]
            result = subprocess.run(cmd, stdout=f, stderr=subprocess.pipe, text=true)

        if result.returncode != 0:
            raise exception(f"mysqldump failed: {result.stderr}")

        # 压缩备份文件
        with open(backup_file, 'rb') as f_in:
            with gzip.open(compressed_file, 'wb') as f_out:
                f_out.writelines(f_in)
        os.remove(backup_file)
        logging.info(f"backup compressed: {compressed_file}")

        # 清理旧备份
        cutoff = datetime.now() - timedelta(days=retain_days)
        for filename in os.listdir(backup_dir):
            filepath = os.path.join(backup_dir, filename)
            if os.path.isfile(filepath) and filename.endswith('.sql.gz'):
                file_time = datetime.fromtimestamp(os.path.getmtime(filepath))
                if file_time < cutoff:
                    os.remove(filepath)
                    logging.info(f"deleted old backup: {filename}")

    except exception as e:
        logging.error(f"backup failed: {str(e)}")
        raise

if __name__ == "__main__":
    parser = argparse.argumentparser(description='mysql database backup tool')
    parser.add_argument('--host', required=true, help='mysql host')
    parser.add_argument('--user', required=true, help='mysql username')
    parser.add_argument('--password', required=true, help='mysql password')
    parser.add_argument('--database', required=true, help='database name')
    parser.add_argument('--backup-dir', default='./backups', help='backup directory')
    parser.add_argument('--retain-days', type=int, default=7, 
                       help='number of days to retain backups')
    
    args = parser.parse_args()
    
    try:
        backup_mysql(
            args.host,
            args.user,
            args.password,
            args.database,
            args.backup_dir,
            args.retain_days
        )
        logging.info("backup completed successfully")
    except:
        logging.error("backup process terminated with errors")

#-*- coding:utf-8 -*-
import os
import time
import tarfile
import zipfile
'''
mysqldump
usage: mysqldump [options] database [tables]
or mysqldump [options] --databases [options] db1 [db2 db3...]
or mysqldump [options] --all-databases [options]
for more options, use mysqldump --help
'''
db_host="localhost"
db_user="root"
db_passwd="root"
db_name="crm"
db_charset="utf8"
db_backup_name=r"c:\crm_%s.sql" %(time.strftime("%y%m%d%h%m"))
zip_src = db_backup_name
zip_dest = zip_src + ".zip"
def zip_files():
f = zipfile.zipfile(zip_dest, 'w' ,zipfile.zip_deflated)
f.write(zip_src)
f.close()
if __name__ == "__main__":
print("begin to dump mysql database crm...");
os.system("mysqldump -h%s -u%s -p%s %s --default_character-set=%s > %s" %(db_host, db_user, db_passwd, db_name, db_charset, db_backup_name))
print("begin zip files...")
zip_files()
print("done, pyhon is great!")

import os
import sys  
# import glob 
import filecmp
import time
import datetime
import subprocess 
import sys 

databackup_dir_path = r'e:\databackup'
mysqldump_path = r'e:\program files\mariadb 10.5\bin\mysqldump'
mysql_ip = '127.0.0.1'
mysql_username='root'
mysql_password='root'
mysql_port=3306

backdatabasenames =  ["kwjl5","kwjl5c2","kwjl5ga" ]

def deleteoldfile(path,day):
    for eachfile in os.listdir(path):
        filename = os.path.join(path, eachfile)
        # backup_ 开头的文件才会被删除
        if os.path.isfile(filename) :  
            lastmodifytime = os.path.getmtime(filename)
            # 设置删除多久之前的文件
            endfiletime = time.time() - 3600 * 24 * day
            if endfiletime > lastmodifytime:
                if filename[0:7] == "backup_" and filename[-4:] == ".sql":
                    os.remove(filename)
                    print ("删除文件 %s 成功" % filename)
        # 如果是目录则递归调用当前函数
        elif os.path.isdir(filename):  
            deleteoldfile(filename,day)


def backupmysql(dbname): 
  time1_str = datetime.datetime.now().strftime('%y_%m_%d')
  cmd =[]
  cmd.append("\"" + mysqldump_path + "\"")  
  cmd.append("--opt")
  cmd.append("--single-transaction=true")
  cmd.append("--user="+ mysql_username)
  cmd.append("--password="+ mysql_password)
  cmd.append("--host="+ mysql_ip)
  cmd.append("--protocol=tcp")
  cmd.append("--port="+ str(mysql_port))
  cmd.append("--default-character-set=utf8")
  cmd.append("--single-transaction=true")
  cmd.append("--routines")
  cmd.append("--events \"%s\" > \"%s\\backup_%s_%s.sql\" " % (dbname, databackup_dir_path, dbname,time1_str) )
  #cmd.append(" \n pause")
  strcmd = str.join(" ",cmd )
  print("cmd=>" + strcmd)
  # os.system(strcmd) 
  try: 
      retcode = subprocess.call(strcmd, shell=true) 
      if retcode < 0: 
        print(sys.stderr, "child was terminated by signal", -retcode )
      else:
        print(sys.stderr, "child returned", retcode )
  except oserror as e: 
      print(sys.stderr, "execution failed:", e )


# 删除老文件 15天以前的文件会被删掉
deleteoldfile(databackup_dir_path,15)

#逐个数据库备份
for f in backdatabasenames: 
  backupmysql(f)

print("执行完成")