Python脚本-mysqldump命令备份

# -*- coding: utf-8 -*-
import datetime
import os
import django
import time
DINGDING_TOKEN = "xxxxx"


def nvwa_error_warning(title, info, error, token=DINGDING_TOKEN, number=''):
    handler = dingding.DingDing(token)
    title = "
 **%s**
" % title
    message = "
* 相关信息：%s
" % info
    error = "
* 错误详情：%s
" % error
    at_numbers = []
    if number and not DEBUG:
        for item in number.split(","):
            at_numbers.append(item)
    at_numbers = list(set(at_numbers))
    at_message = '@' + '@'.join(at_numbers) if len(at_numbers) else ''
    content = "  
".join(
        [title, message, error, at_message]
    )
    handler.send_markdown(content, title, at_numbers)


def main():
    try:
        # 创建临时文件夹
        dir_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "temp")
        if not os.path.exists(dir_path):
            os.mkdir(dir_path)
        # 备份数据库
        default_db = DATABASES["default"]
        date_time = datetime.datetime.now().strftime("%Y%m%d")
        file_name = "_".join([default_db["NAME"], date_time, ".sql"]) if not DEBUG else 
            "_".join(["test", default_db["NAME"], date_time, ".sql"])
        sql_file_name = os.path.join(dir_path, file_name)
        shell = "mysqldump -h %s -u %s -p'%s' %s > %s" % (default_db["HOST"], default_db["USER"],
                                                          default_db["PASSWORD"], default_db["NAME"],
                                                          sql_file_name)
        os.system(shell)
        # 检查备份
        with open(sql_file_name, "r") as f:
            while True:
                line_str = f.readline()
                if not line_str:
                    raise Exception("备份异常")
                if "Dump completed" in line_str:
                    break

        # 上传备份
        my_oss = OSS()
        remote_path = '/'.join(["NvWa", "mysql_data"])
        retry_count = 3
        res = False
        while retry_count > 0:
            res = my_oss.slice_upload(dir_path, remote_path, file_name)
            if res:
                break
            else:
                retry_count -= 1
        if not res:
            raise Exception("上传备份失败！")
        else:
            title = "nvwa正式数据库备份" if not DEBUG else "nvwa测试数据库备份"
            nvwa_error_warning(title, "success", "")
    except Exception as e:
        title = "nvwa正式数据库备份" if not DEBUG else "nvwa测试数据库备份"
        nvwa_error_warning(title, "fail", e.message)


if __name__ == "__main__":
    # 引入django配置文件
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "nvwa.settings")

    # 启动django
    django.setup()

    # 引入django模块
    from nvwa.settings import DATABASES, DEBUG
    from notification import dingding
    from nvwa.oss import OSS

    # 执行任务
    main()