实现对mysql中两个数据库的表、表结构、表数据的对比功能, 初始化界面效果如下图
基础环境请参考
代码文件结构变化
api/ # 后端相关
├── daos/
│ ├── __init__.py
│ └── db_compare_dao.py # 新增
├── routes/
│ ├── __init__.py
│ └── db_compare_routes.py # 新增
├── run.py # 原有代码增加
ui/ # 前端相关
├── src/
│ ├── views/
│ └── DbCompare.vue # 新增
│ ├── router/
│ └── index.js # 原有代码增加
后端相关
api/daos/db_compare_dao.py
import pymysql
import logging
from datetime import datetime
from typing import Dict, List, Tuple, Optional
# 配置日志系统
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler()]
)
logger = logging.getLogger('db_comparison')
def get_db_connection(db_config: Dict):
"""获取数据库连接"""
return pymysql.connect(
host=db_config['host'],
port=db_config['port'],
user=db_config['user'],
password=db_config['password'],
database=db_config['database'],
cursorclass=pymysql.cursors.DictCursor,
charset='utf8mb4'
)
def compare_columns(col1: Dict, col2: Dict) -> bool:
"""比较两个列的属性是否一致"""
return (
col1['COLUMN_NAME'] == col2['COLUMN_NAME'] and
col1['DATA_TYPE'] == col2['DATA_TYPE'] and
col1['IS_NULLABLE'] == col2['IS_NULLABLE'] and
str(col1.get('COLUMN_DEFAULT')) == str(col2.get('COLUMN_DEFAULT')) and
col1.get('CHARACTER_MAXIMUM_LENGTH') == col2.get('CHARACTER_MAXIMUM_LENGTH')
)
def get_column_diff_details(col1: Dict, col2: Dict) -> List[str]:
"""获取列属性的具体差异"""
differences = []
if col1['DATA_TYPE'] != col2['DATA_TYPE']:
differences.append(f"类型不同({col1['DATA_TYPE']} vs {col2['DATA_TYPE']})")
if col1['IS_NULLABLE'] != col2['IS_NULLABLE']:
differences.append(f"可空性不同({col1['IS_NULLABLE']} vs {col2['IS_NULLABLE']})")
if str(col1.get('COLUMN_DEFAULT')) != str(col2.get('COLUMN_DEFAULT')):
differences.append(f"默认值不同({col1.get('COLUMN_DEFAULT')} vs {col2.get('COLUMN_DEFAULT')})")
if col1.get('CHARACTER_MAXIMUM_LENGTH') != col2.get('CHARACTER_MAXIMUM_LENGTH'):
differences.append(
f"长度不同({col1.get('CHARACTER_MAXIMUM_LENGTH')} vs {col2.get('CHARACTER_MAXIMUM_LENGTH')})")
return differences
def get_table_structure(cursor, db_name: str) -> List[Dict]:
"""获取表结构信息"""
cursor.execute(f"""
SELECT
TABLE_NAME,
COLUMN_NAME,
DATA_TYPE,
IS_NULLABLE,
COLUMN_DEFAULT,
CHARACTER_MAXIMUM_LENGTH,
COLUMN_TYPE,
EXTRA
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = '{db_name}'
ORDER BY TABLE_NAME, ORDINAL_POSITION
""")
return cursor.fetchall()
def get_table_hash(cursor, table_name: str) -> Tuple[Optional[str], int]:
"""获取表的哈希值和行数"""
try:
# 1. 获取表的列信息
cursor.execute(f"SHOW COLUMNS FROM `{table_name}`")
columns = [row['Field'] for row in cursor.fetchall()]
if not columns:
logger.warning(f"表 {table_name} 没有列")
return None, 0
# 2. 获取行数
cursor.execute(f"SELECT COUNT(*) AS count FROM `{table_name}`")
row_count = cursor.fetchone()['count']
if row_count == 0:
logger.info(f"表 {table_name} 是空表")
return None, 0
# 3. 准备安全的列名列表
safe_columns = [f"`{col}`" for col in columns]
columns_concat = ", ".join(safe_columns)
# 4. 增加 GROUP_CONCAT 最大长度限制
cursor.execute("SET SESSION group_concat_max_len = 1000000")
# 5. 构建并执行哈希查询
query = f"""
SELECT MD5(
GROUP_CONCAT(
CONCAT_WS('|', {columns_concat})
ORDER BY `{columns[0]}`
SEPARATOR '||'
)
) AS hash,
COUNT(*) AS count
FROM `{table_name}`
"""
cursor.execute(query)
result = cursor.fetchone()
return (result['hash'], row_count)
except Exception as e:
logger.error(f"获取表 {table_name} 哈希失败: {e}")
return None, 0
def compare_databases(db1_config: Dict, db2_config: Dict) -> Dict:
"""比较两个数据库的结构和数据"""
result = {
"status": "success",
"message": "",
"details": {
"structure": [],
"data": [],
"tables": [],
"identical_tables": [], # 新增：完全相同的表列表
"summary": {
"total_tables": 0,
"tables_with_structure_diff": 0,
"tables_with_data_diff": 0,
"tables_missing": 0,
"tables_identical": 0, # 完全相同表的数量
"tables_with_differences": 0, # 有差异表的数量
"tables_only_in_db1": 0, # 仅在db1中存在的表
"tables_only_in_db2": 0, # 仅在db2中存在的表
"tables_compared": 0, # 实际比较的表数量
"start_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"end_time": "",
"duration": ""
}
},
"logs": []
}
def add_log(message: str, level: str = "info"):
"""添加日志到结果和控制台"""
timestamp = datetime.now().strftime('%H:%M:%S')
log_msg = f"[{timestamp}] {message}"
result["logs"].append(log_msg)
if level == "error":
logger.error(log_msg)
elif level == "warning":
logger.warning(log_msg)
else:
logger.info(log_msg)
return log_msg
db1 = db2 = None
start_time = datetime.now()
try:
add_log("开始数据库对比任务")
add_log(f"数据库1配置: {db1_config['host']}:{db1_config['port']}/{db1_config['database']}")
add_log(f"数据库2配置: {db2_config['host']}:{db2_config['port']}/{db2_config['database']}")
# 连接数据库
add_log("正在连接数据库...")
db1 = get_db_connection(db1_config)
db2 = get_db_connection(db2_config)
add_log("数据库连接成功")
with db1.cursor() as cursor1, db2.cursor() as cursor2:
# 获取所有表名
cursor1.execute("SHOW TABLES")
tables1 = [list(t.values())[0].lower() for t in cursor1.fetchall()]
cursor2.execute("SHOW TABLES")
tables2 = [list(t.values())[0].lower() for t in cursor2.fetchall()]
all_tables = set(tables1 + tables2)
result["details"]["summary"]["total_tables"] = len(all_tables)
result["details"]["summary"]["tables_only_in_db1"] = len(set(tables1) - set(tables2))
result["details"]["summary"]["tables_only_in_db2"] = len(set(tables2) - set(tables1))
add_log(f"数据库1有 {len(tables1)} 个表，数据库2有 {len(tables2)} 个表，共 {len(all_tables)} 个唯一表")
# 表存在性检查
for table in set(tables1) - set(tables2):
issue_msg = f"表 {table} 不存在于数据库2"
result["details"]["tables"].append({
"table": table,
"issues": [{
"type": "table",
"message": issue_msg,
"severity": "high"
}]
})
add_log(issue_msg, "warning")
for table in set(tables2) - set(tables1):
issue_msg = f"表 {table} 不存在于数据库1"
result["details"]["tables"].append({
"table": table,
"issues": [{
"type": "table",
"message": issue_msg,
"severity": "high"
}]
})
add_log(issue_msg, "warning")
# 获取表结构
add_log("正在获取数据库表结构...")
structure1 = get_table_structure(cursor1, db1_config['database'])
structure2 = get_table_structure(cursor2, db2_config['database'])
# 转换为字典便于查找
db1_fields = {
(row['TABLE_NAME'].lower(), row['COLUMN_NAME'].lower()): row
for row in structure1
}
db2_fields = {
(row['TABLE_NAME'].lower(), row['COLUMN_NAME'].lower()): row
for row in structure2
}
# 比较共有表
common_tables = set(tables1) & set(tables2)
result["details"]["summary"]["tables_compared"] = len(common_tables)
add_log(f"共 {len(common_tables)} 个表需要比较结构和数据")
for table_idx, table in enumerate(common_tables, 1):
table_result = {
"table": table,
"structure_issues": [],
"data_issues": [],
"is_identical": True
}
add_log(f"正在比较表 ({table_idx}/{len(common_tables)}): {table}")
# 比较表结构
db1_table_fields = {k[1]: v for k, v in db1_fields.items() if k[0] == table}
db2_table_fields = {k[1]: v for k, v in db2_fields.items() if k[0] == table}
# 检查字段存在性
for field in set(db1_table_fields.keys()) - set(db2_table_fields.keys()):
issue_msg = f"表 {table} 字段 {field} 不存在于数据库2"
table_result["structure_issues"].append({
"type": "structure",
"field": field,
"message": issue_msg,
"db1_type": db1_table_fields[field]['DATA_TYPE'],
"db2_type": None
})
table_result["is_identical"] = False
add_log(issue_msg, "warning")
for field in set(db2_table_fields.keys()) - set(db1_table_fields.keys()):
issue_msg = f"表 {table} 字段 {field} 不存在于数据库1"
table_result["structure_issues"].append({
"type": "structure",
"field": field,
"message": issue_msg,
"db1_type": None,
"db2_type": db2_table_fields[field]['DATA_TYPE']
})
table_result["is_identical"] = False
add_log(issue_msg, "warning")
# 比较共有字段
common_fields = set(db1_table_fields.keys()) & set(db2_table_fields.keys())
for field in common_fields:
col1 = db1_table_fields[field]
col2 = db2_table_fields[field]
if not compare_columns(col1, col2):
differences = get_column_diff_details(col1, col2)
issue_msg = f"表 {table} 字段 {field} 属性不一致: {', '.join(differences)}"
table_result["structure_issues"].append({
"type": "structure",
"field": field,
"message": issue_msg,
"db1_type": col1['DATA_TYPE'],
"db2_type": col2['DATA_TYPE'],
"details": differences
})
table_result["is_identical"] = False
add_log(issue_msg, "warning")
# 比较表数据
hash1, row_count1 = get_table_hash(cursor1, table)
hash2, row_count2 = get_table_hash(cursor2, table)
if hash1 != hash2 or row_count1 != row_count2:
issue_details = {
"type": "data",
"message": f"表 {table} 数据不一致",
"db1_row_count": row_count1,
"db2_row_count": row_count2,
"db1_hash": hash1,
"db2_hash": hash2,
"severity": "medium"
}
if row_count1 != row_count2:
issue_details["message"] += f" (行数不同: {row_count1} vs {row_count2})"
else:
issue_details["message"] += " (行数相同但内容不同)"
table_result["data_issues"].append(issue_details)
table_result["is_identical"] = False
add_log(issue_details["message"], "warning")
else:
add_log(f"表 {table} 数据一致 (行数: {row_count1})")
# 记录结果
if table_result["structure_issues"]:
result["details"]["structure"].append({
"table": table,
"issues": table_result["structure_issues"]
})
result["details"]["summary"]["tables_with_structure_diff"] += 1
if table_result["data_issues"]:
result["details"]["data"].append({
"table": table,
"issues": table_result["data_issues"]
})
result["details"]["summary"]["tables_with_data_diff"] += 1
if table_result["is_identical"]:
result["details"]["identical_tables"].append(table)
result["details"]["summary"]["tables_identical"] += 1
else:
result["details"]["summary"]["tables_with_differences"] += 1
# 计算耗时
end_time = datetime.now()
duration = end_time - start_time
result["details"]["summary"]["end_time"] = end_time.strftime('%Y-%m-%d %H:%M:%S')
result["details"]["summary"]["duration"] = str(duration)
# 汇总结果消息
has_differences = (
result["details"]["summary"]["tables_with_structure_diff"] > 0 or
result["details"]["summary"]["tables_with_data_diff"] > 0 or
result["details"]["summary"]["tables_missing"] > 0
)
if has_differences:
diff_details = []
if result["details"]["summary"]["tables_missing"] > 0:
diff_details.append(f"{result['details']['summary']['tables_missing']}个表缺失")
if result["details"]["summary"]["tables_with_structure_diff"] > 0:
diff_details.append(f"{result['details']['summary']['tables_with_structure_diff']}个表结构不同")
if result["details"]["summary"]["tables_with_data_diff"] > 0:
diff_details.append(f"{result['details']['summary']['tables_with_data_diff']}个表数据不同")
final_msg = (
f"对比完成，共比较 {result['details']['summary']['tables_compared']} 个表。"
f"发现差异: {', '.join(diff_details)}。"
f"完全相同表: {result['details']['summary']['tables_identical']} 个。"
f"耗时: {duration}"
)
else:
final_msg = (
f"数据库完全一致，共 {result['details']['summary']['tables_identical']} 个表完全相同。"
f"耗时: {duration}"
)
add_log(final_msg)
result["message"] = final_msg
except pymysql.Error as e:
error_msg = f"数据库错误: {str(e)}"
add_log(error_msg, "error")
result["status"] = "error"
result["message"] = error_msg
except Exception as e:
error_msg = f"系统错误: {str(e)}"
add_log(error_msg, "error")
result["status"] = "error"
result["message"] = error_msg
finally:
if db1:
db1.close()
add_log("数据库1连接已关闭")
if db2:
db2.close()
add_log("数据库2连接已关闭")
result["details"]["summary"]["end_time"] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
duration = datetime.now() - start_time
result["details"]["summary"]["duration"] = str(duration)
add_log("对比任务结束")
print(result)
return result
# 示例用法
if __name__ == "__main__":
db1_config = {
'host': 'localhost',
'port': 3306,
'user': 'root',
'password': 'password',
'database': 'database1'
}
db2_config = {
'host': 'localhost',
'port': 3306,
'user': 'root',
'password': 'password',
'database': 'database2'
}
comparison_result = compare_databases(db1_config, db2_config)
print("\n对比结果:", comparison_result["message"])
print("\n详细摘要:")
for k, v in comparison_result["details"]["summary"].items():
print(f"{k}: {v}")
if comparison_result["details"]["identical_tables"]:
print("\n完全相同表:", comparison_result["details"]["identical_tables"])
api/routes/db_compare_routes.py
from flask import Blueprint, request, jsonify
from api.daos.db_compare_dao import compare_databases
db_compare_bp = Blueprint('db_compare', __name__)
@db_compare_bp.route('/compare', methods=['POST'])
def compare():
print('jinlaile')
try:
data = request.get_json()
if not data:
return jsonify({
"status": "error",
"message": "No JSON data provided"
}), 400
# 提取并验证必要参数
required_fields = [
'db1_host', 'db1_user', 'db1_password', 'db1_database',
'db2_host', 'db2_user', 'db2_password', 'db2_database'
]
missing_fields = [field for field in required_fields if field not in data]
if missing_fields:
return jsonify({
"status": "error",
"message": f"Missing required fields: {', '.join(missing_fields)}"
}), 400
# 构建配置字典
db1_config = {
"host": data["db1_host"],
"user": data["db1_user"],
"password": data["db1_password"],
"database": data["db1_database"],
"port": data.get("db1_port", 3306),
"charset": data.get("db1_charset", "utf8mb4")
}
db2_config = {
"host": data["db2_host"],
"user": data["db2_user"],
"password": data["db2_password"],
"database": data["db2_database"],
"port": data.get("db2_port", 3306),
"charset": data.get("db2_charset", "utf8mb4")
}
# 执行比较
result = compare_databases(db1_config, db2_config)
return jsonify(result)
except Exception as e:
return jsonify({
"status": "error",
"message": f"Internal server error: {str(e)}"
}), 500
api/run.py
from flask import Flask
from flask_cors import CORS
from utils.sqlite_util import init_db
from routes.user_route import bp as user_bp
from routes.db_compare_routes import db_compare_bp # 新增
app = Flask(__name__)
CORS(app, resources={r"/*": {"origins": "*"}})
app.config.from_object('config.Config')
app.register_blueprint(user_bp, url_prefix='/api')
app.register_blueprint(db_compare_bp, url_prefix='/db') # 新增
if __name__ == '__main__':
init_db()
app.run(debug=True, host='0.0.0.0')
后端启动
前端相关
ui/views/DbCompare.vue
该代码块中的第244行的后端地址，需与上图中后端启动的ip:port保持一致
<template>
<div class="container">
<div class="header">
<h2>数据库对比工具</h2>
<el-button class="back-button" type="info" @click="goBack" size="small">返回</el-button>
</div>
<el-row :gutter="20" class="db-input-row">
<el-col :span="12">
<el-card class="db-card">
<div slot="header" class="clearfix">
<span>数据库1 (源数据库)</span>
</div>
<el-form label-position="top">
<el-row :gutter="10">
<el-col :span="16">
<el-input v-model="db1.host" placeholder="主机地址，例如: 192.168.1.100"></el-input>
</el-col>
<el-col :span="8">
<el-input v-model.number="db1.port" placeholder="端口，默认: 3306" type="number"></el-input>
</el-col>
</el-row>
<el-row :gutter="10" class="form-row">
<el-col :span="8">
<el-input v-model="db1.user" placeholder="用户名"></el-input>
</el-col>
<el-col :span="8">
<el-input v-model="db1.password" type="password" placeholder="密码"></el-input>
</el-col>
<el-col :span="8">
<el-input v-model="db1.database" placeholder="数据库名"></el-input>
</el-col>
</el-row>
</el-form>
</el-card>
</el-col>
<el-col :span="12">
<el-card class="db-card">
<div slot="header" class="clearfix">
<span>数据库2 (目标数据库)</span>
</div>
<el-form label-position="top">
<el-row :gutter="10">
<el-col :span="16">
<el-input v-model="db2.host" placeholder="主机地址，例如: 192.168.1.101"></el-input>
</el-col>
<el-col :span="8">
<el-input v-model.number="db2.port" placeholder="端口，默认: 3306" type="number"></el-input>
</el-col>
</el-row>
<el-row :gutter="10" class="form-row">
<el-col :span="8">
<el-input v-model="db2.user" placeholder="用户名"></el-input>
</el-col>
<el-col :span="8">
<el-input v-model="db2.password" type="password" placeholder="密码"></el-input>
</el-col>
<el-col :span="8">
<el-input v-model="db2.database" placeholder="数据库名"></el-input>
</el-col>
</el-row>
</el-form>
</el-card>
</el-col>
</el-row>
<div class="button-container">
<el-button
type="primary"
@click="compareDatabases"
:loading="loading"
size="large"
>
开始对比
</el-button>
<el-button
@click="resetForm"
size="large"
>
重置
</el-button>
</div>
<!-- 结果展示部分保持不变 -->
<el-card class="result-card">
<h2>对比结果</h2>
<div v-if="errorMessage" class="error-message">
<el-alert :title="errorMessage" type="error" :closable="false" />
</div>
<div v-else-if="rawResponse" class="stats-container">
<el-row :gutter="20">
<el-col :span="6">
<el-statistic title="总表数" :value="rawResponse.details.summary.total_tables" />
</el-col>
<el-col :span="6">
<el-statistic title="完全相同表" :value="rawResponse.details.summary.tables_identical" />
</el-col>
<el-col :span="6">
<el-statistic title="结构差异表" :value="rawResponse.details.summary.tables_with_structure_diff" />
</el-col>
<el-col :span="6">
<el-statistic title="数据差异表" :value="rawResponse.details.summary.tables_with_data_diff" />
</el-col>
</el-row>
<div class="time-info">
<span>开始时间: {{ rawResponse.details.summary.start_time }}</span>
<span>结束时间: {{ rawResponse.details.summary.end_time }}</span>
<span>耗时: {{ rawResponse.details.summary.duration }}</span>
</div>
</div>
<el-tabs v-if="rawResponse" type="border-card" class="result-tabs">
<el-tab-pane label="摘要">
<div class="summary-message"></div>
<div v-if="rawResponse.details.identical_tables.length > 0" class="identical-tables">
<h3>完全相同表 ({{ rawResponse.details.identical_tables.length }}个):</h3>
<el-tag
v-for="table in rawResponse.details.identical_tables"
:key="table"
type="success"
class="table-tag"
>
{{ table }}
</el-tag>
</div>
</el-tab-pane>
<el-tab-pane label="结构差异" v-if="rawResponse.details.structure.length > 0">
<el-collapse v-model="activeCollapse">
<el-collapse-item
v-for="table in rawResponse.details.structure"
:key="table.table"
:title="`表: ${table.table}`"
:name="table.table"
>
<div v-for="issue in table.issues" :key="issue.field" class="issue-item">
<el-alert
:title="`字段 ${issue.field} 属性不一致: ${issue.message}`"
type="warning"
:closable="false"
/>
<div v-if="issue.details" class="issue-details">
<p>【{{ db1.database }}】库的表【 {{table.table}}】字段【 {{issue.field}}】属性默认值为: 【{{issue.message.split('(')[1].split(' vs ')[0]}}】</p>
<p>【{{ db2.database }}】库的表【{{table.table}}】字段【 {{issue.field}}】属性默认值为: 【{{issue.message.split('(')[1].split(' vs ')[1].split(')')[0]}}】</p>
</div>
</div>
</el-collapse-item>
</el-collapse>
</el-tab-pane>
<el-tab-pane label="数据差异" v-if="rawResponse.details.data.length > 0">
<el-collapse v-model="activeCollapse">
<el-collapse-item
v-for="table in rawResponse.details.data"
:key="table.table"
:title="`表: ${table.table}`"
:name="table.table"
>
<div v-for="issue in table.issues" :key="issue.message" class="issue-item">
<el-alert :title="issue.message" type="error" :closable="false" />
<div class="data-details">
<p>{{ db1.database }} 行数: {{ issue.db1_row_count }}</p>
<p>{{ db2.database }} 行数: {{ issue.db2_row_count }}</p>
</div>
</div>
</el-collapse-item>
</el-collapse>
</el-tab-pane>
<el-tab-pane label="完整日志">
<div class="full-log-container">
<pre class="full-log-content">{{ rawResponse.logs.join('\n') }}</pre>
</div>
</el-tab-pane>
</el-tabs>
<div v-else class="empty-result">
<el-empty description="暂无对比结果" />
</div>
</el-card>
</div>
</template>
<script setup>
import { ref, computed } from 'vue';
import { useRouter } from 'vue-router';
import axios from 'axios';
import { ElMessage } from 'element-plus';
const router = useRouter();
const goBack = () => {
router.push('/dashboard');
};
// 数据库连接信息
const db1 = ref({
host: '',
port: 3306,
user: 'root',
password: '',
database: ''
});
const db2 = ref({
host: '',
port: 3306,
user: 'root',
password: '',
database: ''
});
const loading = ref(false);
const rawResponse = ref(null);
const errorMessage = ref('');
const activeCollapse = ref([]);
const hasDiscrepancy = computed(() => {
if (!rawResponse.value) return false;
return (
rawResponse.value.details.summary.tables_with_structure_diff > 0 ||
rawResponse.value.details.summary.tables_with_data_diff > 0 ||
rawResponse.value.details.summary.tables_missing > 0
);
});
const compareDatabases = async () => {
// 验证必填字段
if (!db1.value.host || !db1.value.database || !db2.value.host || !db2.value.database) {
ElMessage.error('请填写完整的数据库连接信息');
return;
}
// 确保端口是整数
const db1Port = parseInt(db1.value.port) || 3306;
const db2Port = parseInt(db2.value.port) || 3306;
loading.value = true;
try {
const response = await axios.post('http://192.168.1.138:5000/db/compare', {
db1_host: db1.value.host,
db1_port: db1Port,
db1_user: db1.value.user,
db1_password: db1.value.password,
db1_database: db1.value.database,
db2_host: db2.value.host,
db2_port: db2Port,
db2_user: db2.value.user,
db2_password: db2.value.password,
db2_database: db2.value.database
});
if (response.data.status === "error") {
errorMessage.value = response.data.message;
ElMessage.error('请求失败：' + response.data.message);
} else {
rawResponse.value = response.data;
errorMessage.value = '';
if (hasDiscrepancy.value) {
ElMessage.error('数据库对比发现不一致！');
} else {
ElMessage.success('数据库对比完成，所有内容一致');
}
}
} catch (error) {
ElMessage.error('请求失败：' + error.message);
errorMessage.value = error.message;
} finally {
loading.value = false;
}
};
const resetForm = () => {
db1.value = {
host: '',
port: 3306,
user: 'root',
password: '',
database: ''
};
db2.value = {
host: '',
port: 3306,
user: 'root',
password: '',
database: ''
};
rawResponse.value = null;
errorMessage.value = '';
};
</script>
<style scoped>
.container {
width: 90%;
max-width: 1200px;
margin: 0 auto;
padding: 20px;
}
.header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 20px;
}
.header h2 {
margin: 0;
}
.back-button {
background: #42b983 !important;
color: white !important;
border: none !important;
padding: 8px 12px;
border-radius: 4px;
cursor: pointer;
transition: background 0.2s;
}
.back-button:hover {
background: #3aa876 !important;
}
.db-input-row {
margin-bottom: 20px;
}
.db-card {
height: 100%;
}
.form-row {
margin-top: 10px;
}
.button-container {
display: flex;
justify-content: center;
margin: 20px 0;
gap: 20px;
}
.result-card {
margin-top: 20px;
min-height: 600px;
}
.stats-container {
margin-bottom: 20px;
padding: 15px;
background-color: #f5f7fa;
border-radius: 4px;
}
.time-info {
margin-top: 15px;
display: flex;
justify-content: space-between;
font-size: 12px;
color: #909399;
}
.result-tabs {
margin-top: 20px;
}
.summary-message {
margin-bottom: 20px;
}
.identical-tables {
margin-top: 20px;
}
.table-tag {
margin: 5px;
}
.issue-item {
margin-bottom: 10px;
}
.issue-details {
margin-top: 10px;
}
.data-details {
margin-top: 10px;
padding-left: 20px;
}
.empty-result {
display: flex;
justify-content: center;
align-items: center;
height: 300px;
}
.full-log-container {
max-height: 500px;
overflow-y: auto;
background-color: #f5f5f5;
border-radius: 4px;
padding: 10px;
}
.full-log-content {
font-family: monospace;
white-space: pre-wrap;
margin: 0;
line-height: 1.5;
color: #333;
}
.error-message {
margin-bottom: 20px;
}
.clearfix {
clear: both;
}
:deep(.el-input__inner) {
text-align: left !important;
}
</style>
ui/router/index.js
import DbCompare from '../views/DbCompare.vue';
{ path: '/db_compare', component: DbCompare },
前端启动
cd ui
npm run dev
执行对比
对比结果汇总
对比结果 - 结构差异
对比结果 - 数据差异
对比结果 - 完整日志
声明：
本文中的代码示例由 DeepSeek 生成，经过人工整理和核对后符合工作实际需求。
DeepSeek 是一款强大的AI编程助手，能高效解决技术问题，推荐大家使用！