自动监控MySQL表结构变更脚本

如何监控MySQL数据库表结构和表变更，并通知到相关的联系人、实现报警或通知？

由于平台采用django系统实现，因此通过如下代码实现（代码low，也可以写成python文件，传参数执行）：
简单思路：
对用户指定库的所有列值进行md5，并存储到本地数据库，每次定时执行，校对md5，并找出不匹配的进行判断
会自动找出新增、删除、变更表结构的表

# models.py

class MonitorSchema(models.Model):
    table_schema = models.CharField(null=False, max_length=512)
    table_name = models.CharField(null=False, max_length=512)
    table_stru = models.TextField(null=False, default='')
    md5_sum = models.CharField(null=False, max_length=256)
 
    class Meta:
        verbose_name = u'监控表结构变更表'
        verbose_name_plural = verbose_name
        permissions = ()
        db_table = "dbaudit_monitor_schema"

# tasks.py

import datetime
import hashlib
import difflib
 
import mysql.connector as mdb
from celery import shared_task
from django.core.mail import EmailMessage
from django.template.loader import render_to_string
 
from auditdb.settings import EMAIL_FROM
 
 
@shared_task
def schema_modify_monitor(**kwargs):
    check_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    conn = connect_db(**kwargs)
    cursor = conn.cursor(dictionary=True)
 
    query_info = "select table_schema,table_name,group_concat(COLUMN_NAME) as column_name," \
                 "group_concat(COLUMN_DEFAULT) as column_default,group_concat(IS_NULLABLE) as is_nullable," \
                 "group_concat(DATA_TYPE) as data_type,group_concat(CHARACTER_MAXIMUM_LENGTH) as char_length," \
                 "group_concat(COLUMN_TYPE) as column_type,group_concat(COLUMN_COMMENT) as column_comment " \
                 "from columns where table_schema='{schema}' " \
                 "group by table_schema,table_name".format(schema=kwargs['schema'])
 
    cursor.execute(query_info)
 
    source_info = []
    table_list = []
    diff_old_data = ''
    diff_new_data = ''
    table_change_data = []
 
    for row in cursor.fetchall():
        table_schema = row['table_schema']
        table_name = row['table_name']
 
        md5_source = ''.join(str(row.values()))
        md5_sum = hashlib.md5(md5_source.encode('utf8')).hexdigest()
        source_info.append({'table_schema': table_schema, 'table_name': table_name, 'md5_sum': md5_sum})
        table_list.append(table_name)
 
    # 如果当前库没有记录，则进行初始化全量同步
    if MonitorSchema.objects.filter(table_schema=kwargs['schema']).first() is None:
        for row in source_info:
            table_schema = row['table_schema']
            table_name = row['table_name']
 
            query_table_stru = "show create table {}".format('.'.join((table_schema, table_name)))
            cursor.execute(query_table_stru)
            for i in cursor:
                table_stru = i['Create Table']
                row['table_stru'] = str(table_stru)
                MonitorSchema.objects.create(**row)
    else:
        # 如果存在，开始核验数据
        old_data = list(MonitorSchema.objects.filter(table_schema=kwargs['schema']).values_list('table_name', flat=True))
        new_data = table_list
 
        # 找出已删除的表，并处理
        table_remove = list(set(old_data).difference(set(new_data)))
        if table_remove:
            table_change_data.append({'remove': table_remove})
            # 从本地库中删除该表的记录
            MonitorSchema.objects.filter(table_schema=kwargs['schema']).filter(table_name__in=table_remove).delete()
 
        # 找出新增的表，并处理
        table_add = list(set(new_data).difference(set(old_data)))
        if table_add:
            for i in table_add:
                for j in source_info:
                    if i in j.values():
                        table_change_data.append({'add': j})
                        table_schema = j['table_schema']
                        table_name = j['table_name']
                        query_table_stru = "show create table {}".format('.'.join((table_schema, table_name)))
                        cursor.execute(query_table_stru)
                        for x in cursor:
                            table_stru = x['Create Table']
                            j['table_stru'] = str(table_stru)
                            MonitorSchema.objects.create(**j)
 
        # 找出相同的表，并核验表结构
        table_intersection = list(set(old_data).intersection(set(new_data)))
        for row in source_info:
            table_schema = row['table_schema']
            table_name = row['table_name']
            new_md5_sum = row['md5_sum']
 
            if table_name in table_intersection:
                old_table = MonitorSchema.objects.get(table_schema=table_schema, table_name=table_name)
                if new_md5_sum != old_table.md5_sum:
                    query_table_stru = "show create table {}".format('.'.join((table_schema, table_name)))
                    cursor.execute(query_table_stru)
                    for i in cursor:
                        table_stru = i['Create Table']
                        diff_old_data += old_table.table_stru + '\n'*3
                        diff_new_data += table_stru + '\n'*3
                        # 更新新表表结构到本地
                        MonitorSchema.objects.update_or_create(table_schema=table_schema, table_name=table_name,
                                                               defaults={'table_stru': table_stru,
                                                                         'md5_sum': new_md5_sum})
 
    if (diff_old_data and diff_new_data) or table_change_data:
        html_data = ''
        if diff_old_data and diff_new_data:
            diff_data = difflib.HtmlDiff(tabsize=2)
            old_table_stru = list(diff_old_data.split('\n'))
            new_table_stru = list(diff_new_data.split('\n'))
            html_data = diff_data.make_file(old_table_stru, new_table_stru, '旧表-表结构', '新表-表结构', context=False,
                                            numlines=5)
 
        email_html_body = render_to_string('_monitor_table.html', {'html_data': html_data, 'table_change_data': table_change_data})
        title = '{db}库表变更[来自:{host},检测时间:{check_time}]'.format(db=kwargs['schema'], host=kwargs['describle'], check_time=check_time)
        msg = EmailMessage(subject=title,
                           body=email_html_body,
                           from_email=EMAIL_FROM,
                           to=kwargs['receiver'].split(','),
                           )
        msg.content_subtype = "html"
        msg.send()
    cursor.close()
    conn.close()

对应的html文件：

# _monitor_table.html

<html>
<head>
    <meta charset="utf-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <style>
        body {
            font-family: Monaco, Menlo, Consolas, "Courier New", monospace;
            font-size: 12px;
            line-height: 1.42857143;
            color: #333;
        }
 
        .box.box-primary {
            border-top-color: #3c8dbc;
        }
 
        .box {
            position: relative;
            border-radius: 3px;
            background: #ffffff;
            border-top: 3px solid #d2d6de;
            margin-bottom: 20px;
            width: 100%;
            box-shadow: 0 1px 1px rgba(0, 0, 0, 0.1);
        }
 
        .panel-danger > .panel-heading {
            color: #a94442;
            background-color: #f2dede;
            border-color: #ebccd1;
        }
 
        .panel-info > .panel-heading {
            color: #31708f;
            background-color: #d9edf7;
            border-color: #bce8f1;
        }
 
        .panel-success > .panel-heading {
            color: #3c763d;
            background-color: #dff0d8;
            border-color: #d6e9c6;
        }
 
        .panel-heading {
            padding: 6px 8px;
            border-bottom: 1px solid transparent;
            border-top-left-radius: 3px;
            border-top-right-radius: 3px;
        }
 
        .panel-body {
            padding: 6px;
            color: #3c763d;
            background-color: #f5f5f5;
        }
    </style>
</head>
<body>
<div class="box box-primary">
    <p>各位同仁好：</p>
    <p>  表结构变更如下，请查阅，谢谢。</p>
 
    {% if table_change_data %}
        {% for row in table_change_data %}
            {% if row.remove %}
                <div class="panel panel-danger">
                    <div class="panel-heading">删除的表</div>
                    <div class="panel-body">
                        {% for j in row.remove %}
                            {{ j }}
                        {% endfor %}
                    </div>
                </div>
            {% endif %}
        {% endfor %}
 
        {% for row in table_change_data %}
            {% if row.add %}
                <div class="panel panel-info">
                    <div class="panel-heading">新增的表：{{ row.add.table_name }}_[表结构]</div>
                    <div class="panel-body">
                        <pre>{{ row.add.table_stru }}</pre>
                    </div>
                </div>
            {% endif %}
        {% endfor %}
    {% endif %}
 
    {% if html_data %}
        <div class="panel panel-success">
            <div class="panel-heading">变更的表结构[左侧为变更前表结构、右侧为变更后表结构、标色部分为差异]</div>
            <div class="panel-body">
                {{ html_data|safe }}
            </div>
        </div>
    {% endif %}
 
</div>
</body>
</html>

最后在django后台添加定时任务或者轮询任务

邮件输出结果：