375 lines
15 KiB
Python
375 lines
15 KiB
Python
import re
|
||
import os
|
||
|
||
# 解析SQL文件，提取表结构
|
||
def parse_sql_file(file_path, ignore_prefix=None):
|
||
tables = {}
|
||
|
||
# 读取文件内容
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
content = f.read()
|
||
|
||
# 匹配CREATE TABLE语句
|
||
table_pattern = re.compile(r'CREATE TABLE\s+`?([^`\s]+)`?\s*\(([^;]+)\)\s*(?:[^;]+);', re.DOTALL | re.IGNORECASE)
|
||
|
||
matches = table_pattern.findall(content)
|
||
|
||
for full_table_name, table_def in matches:
|
||
# 处理表名，忽略前缀
|
||
table_name = full_table_name
|
||
if ignore_prefix and table_name.startswith(ignore_prefix):
|
||
table_name = table_name[len(ignore_prefix):]
|
||
|
||
# 提取列定义
|
||
columns = []
|
||
# 匹配列定义，包括列名、类型、约束等
|
||
column_pattern = re.compile(r'\s*`?([^`\s,]+)`?\s+([^\s,]+)\s*([^,]+)(?:,|$)', re.DOTALL)
|
||
column_matches = column_pattern.findall(table_def)
|
||
|
||
for col_name, col_type, col_constraints in column_matches:
|
||
# 清理约束中的换行符和多余空格
|
||
col_constraints = re.sub(r'\s+', ' ', col_constraints.strip())
|
||
columns.append((col_name, col_type, col_constraints))
|
||
|
||
# 提取主键
|
||
primary_key = None
|
||
pk_pattern = re.compile(r'PRIMARY\s+KEY\s*\(`?([^`\s,]+)`?\)', re.IGNORECASE)
|
||
pk_match = pk_pattern.search(table_def)
|
||
if pk_match:
|
||
primary_key = pk_match.group(1)
|
||
|
||
# 提取索引
|
||
indexes = []
|
||
index_pattern = re.compile(r'INDEX\s+`?([^`\s]+)`?\s*\(([^)]+)\)', re.IGNORECASE)
|
||
index_matches = index_pattern.findall(table_def)
|
||
for index_name, index_cols in index_matches:
|
||
indexes.append((index_name, index_cols.strip()))
|
||
|
||
# 提取唯一约束
|
||
unique_constraints = []
|
||
unique_pattern = re.compile(r'UNIQUE\s+KEY\s*`?([^`\s]+)`?\s*\(([^)]+)\)', re.IGNORECASE)
|
||
unique_matches = unique_pattern.findall(table_def)
|
||
for unique_name, unique_cols in unique_matches:
|
||
unique_constraints.append((unique_name, unique_cols.strip()))
|
||
|
||
tables[table_name] = {
|
||
'full_name': full_table_name,
|
||
'columns': columns,
|
||
'primary_key': primary_key,
|
||
'indexes': indexes,
|
||
'unique_constraints': unique_constraints
|
||
}
|
||
|
||
return tables
|
||
|
||
# 比较两个数据库表结构
|
||
def compare_databases(db1, db2, db1_name, db2_name):
|
||
diffs = {
|
||
'only_in_db1': [],
|
||
'only_in_db2': [],
|
||
'column_diffs': {},
|
||
'primary_key_diffs': {},
|
||
'index_diffs': {},
|
||
'unique_constraint_diffs': {}
|
||
}
|
||
|
||
# 找出只在db1中存在的表
|
||
for table_name in db1:
|
||
if table_name not in db2:
|
||
diffs['only_in_db1'].append(table_name)
|
||
|
||
# 找出只在db2中存在的表
|
||
for table_name in db2:
|
||
if table_name not in db1:
|
||
diffs['only_in_db2'].append(table_name)
|
||
|
||
# 比较共同存在的表
|
||
common_tables = set(db1.keys()) & set(db2.keys())
|
||
for table_name in common_tables:
|
||
table1 = db1[table_name]
|
||
table2 = db2[table_name]
|
||
|
||
# 比较列
|
||
col_diffs = {
|
||
'only_in_db1': [],
|
||
'only_in_db2': [],
|
||
'type_diffs': [],
|
||
'constraint_diffs': []
|
||
}
|
||
|
||
# 列名集合
|
||
cols1 = {col[0]: col for col in table1['columns']}
|
||
cols2 = {col[0]: col for col in table2['columns']}
|
||
|
||
# 只在db1中的列
|
||
for col_name in cols1:
|
||
if col_name not in cols2:
|
||
col_diffs['only_in_db1'].append(col_name)
|
||
|
||
# 只在db2中的列
|
||
for col_name in cols2:
|
||
if col_name not in cols1:
|
||
col_diffs['only_in_db2'].append(col_name)
|
||
|
||
# 比较列类型和约束
|
||
common_cols = set(cols1.keys()) & set(cols2.keys())
|
||
for col_name in common_cols:
|
||
col1 = cols1[col_name]
|
||
col2 = cols2[col_name]
|
||
|
||
# 类型差异
|
||
if col1[1] != col2[1]:
|
||
col_diffs['type_diffs'].append((col_name, col1[1], col2[1]))
|
||
|
||
# 约束差异
|
||
if col1[2] != col2[2]:
|
||
col_diffs['constraint_diffs'].append((col_name, col1[2], col2[2]))
|
||
|
||
if any(col_diffs.values()):
|
||
diffs['column_diffs'][table_name] = col_diffs
|
||
|
||
# 比较主键
|
||
if table1['primary_key'] != table2['primary_key']:
|
||
diffs['primary_key_diffs'][table_name] = (table1['primary_key'], table2['primary_key'])
|
||
|
||
# 比较索引
|
||
index_diffs = {
|
||
'only_in_db1': [],
|
||
'only_in_db2': [],
|
||
'definition_diffs': []
|
||
}
|
||
|
||
indexes1 = {idx[0]: idx[1] for idx in table1['indexes']}
|
||
indexes2 = {idx[0]: idx[1] for idx in table2['indexes']}
|
||
|
||
# 只在db1中的索引
|
||
for idx_name in indexes1:
|
||
if idx_name not in indexes2:
|
||
index_diffs['only_in_db1'].append((idx_name, indexes1[idx_name]))
|
||
|
||
# 只在db2中的索引
|
||
for idx_name in indexes2:
|
||
if idx_name not in indexes1:
|
||
index_diffs['only_in_db2'].append((idx_name, indexes2[idx_name]))
|
||
|
||
# 比较索引定义
|
||
common_indexes = set(indexes1.keys()) & set(indexes2.keys())
|
||
for idx_name in common_indexes:
|
||
if indexes1[idx_name] != indexes2[idx_name]:
|
||
index_diffs['definition_diffs'].append((idx_name, indexes1[idx_name], indexes2[idx_name]))
|
||
|
||
if any(index_diffs.values()):
|
||
diffs['index_diffs'][table_name] = index_diffs
|
||
|
||
# 比较唯一约束
|
||
unique_diffs = {
|
||
'only_in_db1': [],
|
||
'only_in_db2': [],
|
||
'definition_diffs': []
|
||
}
|
||
|
||
unique1 = {uc[0]: uc[1] for uc in table1['unique_constraints']}
|
||
unique2 = {uc[0]: uc[1] for uc in table2['unique_constraints']}
|
||
|
||
# 只在db1中的唯一约束
|
||
for uc_name in unique1:
|
||
if uc_name not in unique2:
|
||
unique_diffs['only_in_db1'].append((uc_name, unique1[uc_name]))
|
||
|
||
# 只在db2中的唯一约束
|
||
for uc_name in unique2:
|
||
if uc_name not in unique1:
|
||
unique_diffs['only_in_db2'].append((uc_name, unique2[uc_name]))
|
||
|
||
# 比较唯一约束定义
|
||
common_unique = set(unique1.keys()) & set(unique2.keys())
|
||
for uc_name in common_unique:
|
||
if unique1[uc_name] != unique2[uc_name]:
|
||
unique_diffs['definition_diffs'].append((uc_name, unique1[uc_name], unique2[uc_name]))
|
||
|
||
if any(unique_diffs.values()):
|
||
diffs['unique_constraint_diffs'][table_name] = unique_diffs
|
||
|
||
return diffs
|
||
|
||
# 打印差异报告
|
||
# 生成Markdown格式的差异报告
|
||
def generate_markdown_report(diffs, db1_name, db2_name, db1_table_count, db2_table_count):
|
||
report = []
|
||
|
||
# 报告标题
|
||
report.append(f"# 数据库差异报告: {db1_name} vs {db2_name}")
|
||
report.append("\n## 1. 表数量统计")
|
||
report.append("| 数据库文件 | 表数量 |")
|
||
report.append("|------------|--------|")
|
||
report.append(f"| {db1_name} | {db1_table_count} |")
|
||
report.append(f"| {db2_name} | {db2_table_count} |")
|
||
|
||
# 表存在性差异
|
||
report.append("\n## 2. 表存在性差异")
|
||
|
||
# 仅在db1中的表
|
||
if diffs['only_in_db1']:
|
||
report.append(f"\n### 2.1 仅在 {db1_name} 中存在的表 ({len(diffs['only_in_db1'])} 个)")
|
||
report.append("| 表名 |")
|
||
report.append("|------|")
|
||
for table in sorted(diffs['only_in_db1']):
|
||
report.append(f"| {table} |")
|
||
|
||
# 仅在db2中的表
|
||
if diffs['only_in_db2']:
|
||
report.append(f"\n### 2.2 仅在 {db2_name} 中存在的表 ({len(diffs['only_in_db2'])} 个)")
|
||
report.append("| 表名 |")
|
||
report.append("|------|")
|
||
for table in sorted(diffs['only_in_db2']):
|
||
report.append(f"| {table} |")
|
||
|
||
# 列结构差异
|
||
if diffs['column_diffs']:
|
||
report.append(f"\n## 3. 列结构差异的表 ({len(diffs['column_diffs'])} 个)")
|
||
|
||
for table, col_diffs in diffs['column_diffs'].items():
|
||
report.append(f"\n### 3.1 表: {table}")
|
||
|
||
# 仅在db1中的列
|
||
if col_diffs['only_in_db1']:
|
||
report.append(f"\n#### 3.1.1 仅在 {db1_name} 中存在的列")
|
||
report.append("| 列名 |")
|
||
report.append("|------|")
|
||
for col in col_diffs['only_in_db1']:
|
||
report.append(f"| {col} |")
|
||
|
||
# 仅在db2中的列
|
||
if col_diffs['only_in_db2']:
|
||
report.append(f"\n#### 3.1.2 仅在 {db2_name} 中存在的列")
|
||
report.append("| 列名 |")
|
||
report.append("|------|")
|
||
for col in col_diffs['only_in_db2']:
|
||
report.append(f"| {col} |")
|
||
|
||
# 列类型差异
|
||
if col_diffs['type_diffs']:
|
||
report.append(f"\n#### 3.1.3 列类型差异")
|
||
report.append(f"| 列名 | {db1_name} | {db2_name} |")
|
||
report.append("|------|------------|------------|")
|
||
for col_name, type1, type2 in col_diffs['type_diffs']:
|
||
report.append(f"| {col_name} | {type1} | {type2} |")
|
||
|
||
# 列约束差异
|
||
if col_diffs['constraint_diffs']:
|
||
report.append(f"\n#### 3.1.4 列约束差异")
|
||
report.append(f"| 列名 | {db1_name} | {db2_name} |")
|
||
report.append("|------|------------|------------|")
|
||
for col_name, constraint1, constraint2 in col_diffs['constraint_diffs']:
|
||
report.append(f"| {col_name} | {constraint1} | {constraint2} |")
|
||
|
||
# 主键差异
|
||
if diffs['primary_key_diffs']:
|
||
report.append(f"\n## 4. 主键差异的表 ({len(diffs['primary_key_diffs'])} 个)")
|
||
report.append(f"| 表名 | {db1_name} | {db2_name} |")
|
||
report.append("|------|------------|------------|")
|
||
for table, (pk1, pk2) in diffs['primary_key_diffs'].items():
|
||
report.append(f"| {table} | {pk1} | {pk2} |")
|
||
|
||
# 索引差异
|
||
if diffs['index_diffs']:
|
||
report.append(f"\n## 5. 索引差异的表 ({len(diffs['index_diffs'])} 个)")
|
||
|
||
for table, idx_diffs in diffs['index_diffs'].items():
|
||
report.append(f"\n### 5.1 表: {table}")
|
||
|
||
# 仅在db1中的索引
|
||
if idx_diffs['only_in_db1']:
|
||
report.append(f"\n#### 5.1.1 仅在 {db1_name} 中存在的索引")
|
||
report.append("| 索引名 | 索引列 |")
|
||
report.append("|--------|--------|")
|
||
for idx_name, idx_cols in idx_diffs['only_in_db1']:
|
||
report.append(f"| {idx_name} | {idx_cols} |")
|
||
|
||
# 仅在db2中的索引
|
||
if idx_diffs['only_in_db2']:
|
||
report.append(f"\n#### 5.1.2 仅在 {db2_name} 中存在的索引")
|
||
report.append("| 索引名 | 索引列 |")
|
||
report.append("|--------|--------|")
|
||
for idx_name, idx_cols in idx_diffs['only_in_db2']:
|
||
report.append(f"| {idx_name} | {idx_cols} |")
|
||
|
||
# 索引定义差异
|
||
if idx_diffs['definition_diffs']:
|
||
report.append(f"\n#### 5.1.3 索引定义差异")
|
||
report.append(f"| 索引名 | {db1_name} | {db2_name} |")
|
||
report.append("|--------|------------|------------|")
|
||
for idx_name, idx1, idx2 in idx_diffs['definition_diffs']:
|
||
report.append(f"| {idx_name} | {idx1} | {idx2} |")
|
||
|
||
# 唯一约束差异
|
||
if diffs['unique_constraint_diffs']:
|
||
report.append(f"\n## 6. 唯一约束差异的表 ({len(diffs['unique_constraint_diffs'])} 个)")
|
||
|
||
for table, uc_diffs in diffs['unique_constraint_diffs'].items():
|
||
report.append(f"\n### 6.1 表: {table}")
|
||
|
||
# 仅在db1中的唯一约束
|
||
if uc_diffs['only_in_db1']:
|
||
report.append(f"\n#### 6.1.1 仅在 {db1_name} 中存在的唯一约束")
|
||
report.append("| 约束名 | 约束列 |")
|
||
report.append("|--------|--------|")
|
||
for uc_name, uc_cols in uc_diffs['only_in_db1']:
|
||
report.append(f"| {uc_name} | {uc_cols} |")
|
||
|
||
# 仅在db2中的唯一约束
|
||
if uc_diffs['only_in_db2']:
|
||
report.append(f"\n#### 6.1.2 仅在 {db2_name} 中存在的唯一约束")
|
||
report.append("| 约束名 | 约束列 |")
|
||
report.append("|--------|--------|")
|
||
for uc_name, uc_cols in uc_diffs['only_in_db2']:
|
||
report.append(f"| {uc_name} | {uc_cols} |")
|
||
|
||
# 唯一约束定义差异
|
||
if uc_diffs['definition_diffs']:
|
||
report.append(f"\n#### 6.1.3 唯一约束定义差异")
|
||
report.append(f"| 约束名 | {db1_name} | {db2_name} |")
|
||
report.append("|--------|------------|------------|")
|
||
for uc_name, uc1, uc2 in uc_diffs['definition_diffs']:
|
||
report.append(f"| {uc_name} | {uc1} | {uc2} |")
|
||
|
||
report.append("\n## 7. 总结")
|
||
report.append("差异比较完成!")
|
||
|
||
return '\n'.join(report)
|
||
|
||
# 主函数
|
||
def main():
|
||
# 文件路径
|
||
db1_path = r'D:\projects\shop-projects\backend\docs\db\niushop_database.sql'
|
||
db2_path = r'D:\projects\shop-projects\backend\docs\db\init_v2.0_with_data.sql'
|
||
report_path = r'D:\projects\shop-projects\backend\docs\db\database_diff_report.md'
|
||
|
||
# 解析数据库结构
|
||
print(f"正在解析 {db1_path}...")
|
||
db1 = parse_sql_file(db1_path)
|
||
db1_table_count = len(db1)
|
||
print(f"解析完成，共 {db1_table_count} 个表")
|
||
|
||
print(f"\n正在解析 {db2_path}...")
|
||
db2 = parse_sql_file(db2_path, ignore_prefix='lucky_')
|
||
db2_table_count = len(db2)
|
||
print(f"解析完成，共 {db2_table_count} 个表")
|
||
|
||
# 比较差异
|
||
print("\n正在比较数据库差异...")
|
||
diffs = compare_databases(db1, db2, 'niushop_database.sql', 'init_v2.0_with_data.sql')
|
||
|
||
# 生成Markdown差异报告
|
||
print("\n正在生成Markdown差异报告...")
|
||
report = generate_markdown_report(diffs, 'niushop_database.sql', 'init_v2.0_with_data.sql', db1_table_count, db2_table_count)
|
||
|
||
# 保存报告到文件
|
||
with open(report_path, 'w', encoding='utf-8') as f:
|
||
f.write(report)
|
||
|
||
print(f"\n差异报告已生成: {report_path}")
|
||
print("差异比较完成!")
|
||
|
||
if __name__ == "__main__":
|
||
main() |