|
|
#!/usr/bin/env python
|
|
|
# -*- coding: utf-8 -*-
|
|
|
# @Time : 2023/5/14 21:15
|
|
|
# @Author : old tom
|
|
|
# @File : metadata_comparator.py
|
|
|
# @Project : futool-tiny-datahub
|
|
|
# @Desc : 版本比较器
|
|
|
|
|
|
import abc
|
|
|
|
|
|
from datahub.metadata.metadata_warehouse import MetadataWareHouse
|
|
|
from datahub.metadata.constant.metadata_constant import MetaDataObjType
|
|
|
from common.futool.core.fu_collection import list_diff, list_intersection, is_not_empty, list_2_md5
|
|
|
from common.futool.core.fu_lang import str_md5
|
|
|
|
|
|
|
|
|
class MetadataComparator(metaclass=abc.ABCMeta):
|
|
|
|
|
|
def __init__(self, version_1: int, version_2: int, source_id_1, source_id_2=None):
|
|
|
"""
|
|
|
:param version_1: 版本号1
|
|
|
:param version_2: 版本号2
|
|
|
:param source_id_1: 数据源1
|
|
|
:param source_id_2: 数据源2
|
|
|
"""
|
|
|
self.version_1 = version_1
|
|
|
self.version_2 = version_2
|
|
|
self.warehouse_1 = MetadataWareHouse(source_id_1)
|
|
|
self.warehouse_2 = MetadataWareHouse(source_id_2) if source_id_2 else None
|
|
|
|
|
|
@abc.abstractmethod
|
|
|
def compare(self):
|
|
|
pass
|
|
|
|
|
|
def _do_compare(self):
|
|
|
self._compare_table()
|
|
|
self._compare_view()
|
|
|
self._compare_procedure()
|
|
|
|
|
|
def _compare_table(self):
|
|
|
"""
|
|
|
比较表
|
|
|
1.表名
|
|
|
2.字段名、字段类型、长度、是否可为空、默认值
|
|
|
:return:
|
|
|
"""
|
|
|
v1_to_v2, v2_to_v1, same_name_compare_result = self._compare_metadata(MetaDataObjType.Table.value)
|
|
|
return v1_to_v2, v2_to_v1, same_name_compare_result
|
|
|
|
|
|
def _compare_view(self):
|
|
|
"""
|
|
|
比较视图
|
|
|
:return:
|
|
|
"""
|
|
|
self._compare_metadata(MetaDataObjType.View.value)
|
|
|
|
|
|
def _compare_procedure(self):
|
|
|
"""
|
|
|
比较存储过程
|
|
|
:return:
|
|
|
"""
|
|
|
self._compare_metadata(MetaDataObjType.Procedure.value)
|
|
|
|
|
|
def _compare_metadata(self, obj_type):
|
|
|
"""
|
|
|
比较元数据名称
|
|
|
:param obj_type:
|
|
|
:return:
|
|
|
"""
|
|
|
# 版本号对应元数据名
|
|
|
obj_v1, obj_v2 = self._query_metadata_name(obj_type)
|
|
|
# v1与v2比,v2与v1比,得到新建及删除的元数据名称
|
|
|
v1_to_v2, v2_to_v1 = list_diff(obj_v1, obj_v2), list_diff(obj_v2, obj_v1)
|
|
|
# 相同名称比较
|
|
|
same_name_objs = list_intersection(obj_v1, obj_v2) if (is_not_empty(v1_to_v2) or is_not_empty(v2_to_v1)) \
|
|
|
else v1_to_v2
|
|
|
same_name_compare_result = self._compare_detail(same_name_objs, obj_type)
|
|
|
return v1_to_v2, v2_to_v1, same_name_compare_result
|
|
|
|
|
|
def _query_metadata_name(self, obj_type):
|
|
|
"""
|
|
|
查询表名称
|
|
|
:return:
|
|
|
"""
|
|
|
obj_v1_name = self.warehouse_1.query_metadata_name(obj_type=obj_type, version_code=self.version_1)
|
|
|
obj_v2_name = self.warehouse_2.query_metadata_name(obj_type=obj_type,
|
|
|
version_code=self.version_2) if self.warehouse_2 \
|
|
|
else self.warehouse_1.query_metadata_name(obj_type=obj_type, version_code=self.version_2)
|
|
|
return obj_v1_name, obj_v2_name
|
|
|
|
|
|
def _query_table_field(self, table_name):
|
|
|
"""
|
|
|
查询表字段
|
|
|
:param table_name:
|
|
|
:return:
|
|
|
"""
|
|
|
table_field_v1 = self.warehouse_1.query_table_fields(table_name, self.version_1)
|
|
|
table_field_v2 = self.warehouse_2.query_table_fields(table_name, self.version_2) if self.warehouse_2 \
|
|
|
else self.warehouse_1.query_table_fields(table_name, self.version_2)
|
|
|
|
|
|
# 包装返回为dict,key为字段名,value为 field_ch_name, field_type, nullable, default_value, field_length
|
|
|
def trans_field_to_kv(fields):
|
|
|
field_kv = {}
|
|
|
for f in fields:
|
|
|
field_name = f[0]
|
|
|
field_kv[field_name] = f[1:]
|
|
|
return field_kv
|
|
|
|
|
|
return trans_field_to_kv(table_field_v1), trans_field_to_kv(table_field_v2)
|
|
|
|
|
|
def _query_obj_create(self, obj_name):
|
|
|
"""
|
|
|
查询试图或存储过程SQL
|
|
|
:param obj_name:
|
|
|
:return:
|
|
|
"""
|
|
|
create_v1 = self.warehouse_1.query_create_by_name(obj_name, self.version_1)
|
|
|
create_v2 = self.warehouse_2.query_create_by_name(obj_name, self.version_2) if self.warehouse_2 \
|
|
|
else self.warehouse_1.query_create_by_name(obj_name, self.version_2)
|
|
|
return create_v1, create_v2
|
|
|
|
|
|
def _compare_detail(self, metadata_obj, obj_type):
|
|
|
"""
|
|
|
比较明细
|
|
|
表：字段
|
|
|
试图、存储过程：创建语句
|
|
|
:param metadata_obj:
|
|
|
:param obj_type:
|
|
|
:return:
|
|
|
"""
|
|
|
compare_result = []
|
|
|
for obj_name in metadata_obj:
|
|
|
obj_compare_result = self._compare_table_field(
|
|
|
obj_name) if obj_type is MetaDataObjType.Table.value else self._compare_create_sql(obj_name)
|
|
|
if obj_compare_result:
|
|
|
compare_result.append(obj_compare_result)
|
|
|
return compare_result
|
|
|
|
|
|
def _compare_create_sql(self, obj_name):
|
|
|
"""
|
|
|
比较试图或存储过程SQL
|
|
|
:param obj_name:
|
|
|
:return:
|
|
|
"""
|
|
|
compare_result = {}
|
|
|
create_v1, create_v2 = self._query_obj_create(obj_name)
|
|
|
if str_md5(create_v1) != str_md5(create_v2):
|
|
|
compare_result[obj_name] = {
|
|
|
'create_v1': create_v1,
|
|
|
'create_v2': create_v2
|
|
|
}
|
|
|
return compare_result
|
|
|
|
|
|
def _compare_table_field(self, table_name):
|
|
|
"""
|
|
|
比较字段
|
|
|
:param table_name: 表名
|
|
|
:return:
|
|
|
"""
|
|
|
compare_result = {}
|
|
|
table_field_v1, table_field_v2 = self._query_table_field(table_name)
|
|
|
# 得到差异字段
|
|
|
table_v1_fields, table_v2_fields = list(table_field_v1.keys()), list(table_field_v2.keys())
|
|
|
# 版本1相对版本2,版本2相对版本1
|
|
|
k1_to_k2, k2_to_k1 = list_diff(table_v1_fields, table_v2_fields), list_diff(table_v2_fields,
|
|
|
table_v1_fields)
|
|
|
# 得到相同字段
|
|
|
same_fields = list_intersection(table_v1_fields, table_v2_fields) if (
|
|
|
is_not_empty(k1_to_k2) or is_not_empty(k2_to_k1)) \
|
|
|
else k1_to_k2
|
|
|
# 比较相同字段,内容合并转为md5进行比较。如果数据量较大可以使用minHash算法，现在没必要
|
|
|
same_field_compare_result = []
|
|
|
for field_name in same_fields:
|
|
|
# 取出相同字段比较
|
|
|
f_v1, f_v2 = table_field_v1[field_name], table_field_v2[field_name]
|
|
|
if list_2_md5(f_v1) != list_2_md5(f_v2):
|
|
|
# 记录相同字段,不同属性
|
|
|
same_field_compare_result.append(field_name)
|
|
|
if is_not_empty(k1_to_k2) or is_not_empty(k2_to_k1) or is_not_empty(same_field_compare_result):
|
|
|
compare_result[table_name] = {
|
|
|
'v1_to_v2': k1_to_k2,
|
|
|
'v2_to_v1': k2_to_k1,
|
|
|
'same_compare_result': same_field_compare_result
|
|
|
}
|
|
|
return compare_result
|
|
|
|
|
|
|
|
|
class SameSourceComparator(MetadataComparator):
|
|
|
"""
|
|
|
同库比较
|
|
|
"""
|
|
|
|
|
|
def __init__(self, source_id, version_1, version_2):
|
|
|
super().__init__(version_1, version_2, source_id)
|
|
|
|
|
|
def compare(self):
|
|
|
return super()._do_compare()
|
|
|
|
|
|
|
|
|
class DiffSourceComparator(MetadataComparator):
|
|
|
"""
|
|
|
不同库比较
|
|
|
"""
|
|
|
|
|
|
def __init__(self, source_id_1, version_1, source_id_2, version_2):
|
|
|
super().__init__(version_1, version_2, source_id_1, source_id_2)
|
|
|
|
|
|
def compare(self):
|
|
|
return super()._do_compare()
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
ssc = DiffSourceComparator('834164a2d62de959c0261e6239dd1e55', 24, 'f98fede74826c709329a65d63db167df', 1)
|
|
|
ssc.compare()
|
