标签：mysql备份   ace   Nging   %s   type   localtime   text   log   pre   

-- coding: utf-8 --

import time
import datetime
import subprocess
import requests
import json
import logging

my.cnf文件

cnf_file="/etc/my.cnf"

备份目录

backup_dir="/opt/back"

dt=time.strftime("%Y%m%d",time.localtime())
logging.basicConfig(level=logging.INFO,
format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘,
datefmt=‘%a, %d %b %Y %H:%M:%S‘,
filename=‘/tmp/test.log‘,
filemode=‘w‘)
def back(user,dt,day,lastday):
user = user
dt = dt
day = day
lastday = lastday
full_date = [‘1‘,‘4‘,‘5‘]
success = []
fail=[]

inc_date = [‘0‘,‘2‘,‘3‘,‘6‘]

if day in full_date:
    # command = ‘sudo innobackupex --defaults-file=/etc/my.cnf  --login-path=%s --no-timestamp --backup /opt/back/xtra_base_%s‘ % (user,dt)
    logging.info("开始全备。。。")
    command = ‘sudo innobackupex --defaults-file=/etc/my.cnf  --login-path={user} --no-timestamp --backup /opt/back/xtra_base_{dt}‘.format(user=user,dt=dt)
    result = subprocess.Popen(command,shell=True,stderr=subprocess.PIPE,stdout=subprocess.PIPE,encoding="utf-8")
    # out,err = result.communicate()
    # print("全out: %s" % out)
    # print("全err: %s" % err)
    aa = result.wait()
    # logging.info(‘全备错误信息：{}‘.format(result.stderr.read()))
    if aa != 0:
        logging.info(‘全备错误信息：{}‘.format(result.stderr.read()))
        fail.append("全备失败,请查看/tmp/test.log")
        return fail
    success.append("全备成功")
    logging.info("开始压缩备份。。。")
    print(success)
    command1 = ‘sudo tar -zcvPf /opt/back/full/xtra_base_{dt}_{day}.tar.gz /opt/back/xtra_base_{dt1}/‘.format(dt=dt, day=day, dt1=dt)
    print(command1)
    result = subprocess.Popen(command1,shell=True, stderr=subprocess.PIPE, stdout=subprocess.PIPE,encoding="utf-8")
    # out, err = result1.communicate()
    # print("qweq")
    # print(result.wait())
    # logging.info(‘压缩错误信息：{}‘.format(err))
    aa = result.wait()
    if aa != 0:
        logging.info(‘全备压缩错误信息：{}‘.format(result.stderr.read()))
        fail.append("全备压缩失败,请查看/tmp/test.log")
        return fail
    logging.info("开始生成md5值。。。")
    success.append("压缩成功")
    command = ‘sudo md5sum /opt/back/full/xtra_base_%s_%s.tar.gz > /opt/back/full/xtra_base_%s_%s.md5‘ % (dt, day, dt, day)
    result = subprocess.Popen(command, shell=True,stderr=subprocess.PIPE, stdout=subprocess.PIPE,encoding="utf-8")
    # out, err = result.communicate()
    # print(result.wait())
    # logging.info(‘MD5失败信息：{}‘.format(err))
    aa = result.wait()
    print(aa)
    if aa != 0:
        logging.info(‘全量MD5失败信息：{}‘.format(result.stderr.read()))
        fail.append(‘全量MD5失败,请查看/tmp/test.log‘)
        return fail
    success.append(‘压缩包MD5值生成成功‘)
else:
    command = ‘sudo innobackupex --defaults-file=/etc/my.cnf  --login-path=%s --no-timestamp  --incremental --backup --incremental-basedir=/opt/back/xtra_base_%s /opt/back/xtra_inc_%s‘ % (user, lastday, dt)
    result = subprocess.Popen(command,shell=True, stderr=subprocess.PIPE, stdout=subprocess.PIPE)
    aa = result.wait()
    if aa != 0:
        logging.info(‘增量备份失败信息：{}‘.format(result.stderr.read()))
        fail.append("增量备份失败,请查看/tmp/test.log")
        return fail
    success.append("增量备份成功")
    command = ‘sudo tar -zcvPf /opt/back/inc/xtra_inc_%s_%s.tar.gz /opt/back/xtra_inc_%s/‘ % (dt, day, dt)
    result = subprocess.Popen(command,shell=True, stderr=subprocess.PIPE, stdout=subprocess.PIPE)
    aa = result.wait()
    if aa != 0:
        logging.info(‘增量备份压缩失败信息：{}‘.format(result.stderr.read()))
        fail.append("增量备份压缩失败,请查看/tmp/test.log")
        return fail
    success.append("增量备份压缩成功")
    command = ‘sudo md5sum /opt/back/inc/xtra_inc_%s_%s.tar.gz > /opt/back/inc/xtra_inc_%s_%s.md5‘ % (dt, day, dt, day)
    result = subprocess.Popen(command,shell=True, stderr=subprocess.PIPE, stdout=subprocess.PIPE)
    aa = result.wait()
    if aa != 0:
        logging.info(‘增量备份md5失败信息：{}‘.format(result.stderr.read()))
        fail.append("增量备份md5失败,请查看/tmp/test.log")
        return fail
    success.append("增量备份MD5值计算成功")

if len(success) == 3:
    result = "备份成功"
    return result
else:
    print(fail)
    result = "备份失败"
    return result

def dinging(results):
#钉钉测试机器人地址
url = "https://oapi.dingtalk.com/robot/send?access_token=xxxxxxx"
#钉钉生产机器人地址

url = "https://oapi.dingtalk.com/robot/send?access_token=xxxxxxx"

program = {
    "msgtype": "text",
    "text": {
        #"content": Exception_interface
        "content": "mysql_backup: %s" % (results)
    },
}
# print(program)

headers = {‘Content-Type‘: ‘application/json‘}
f = requests.post(url, data=json.dumps(program), headers=headers)

if name == "main":

mysql备份账户

user = "backtest1"
dt = time.strftime("%Y%m%d", time.localtime())
now_time = datetime.datetime.now()
lastday = (now_time + datetime.timedelta(days=-1)).strftime("%Y%m%d")
day = time.strftime("%w", time.localtime())
dinging(back(user, dt, day, lastday))

mysql5.7_xtrackup_脚本

标签：mysql备份   ace   Nging   %s   type   localtime   text   log   pre   

原文地址：https://blog.51cto.com/1054054/2472906