当前位置：首页 > news >正文

Python备份实战专栏第4/6篇：Vue.js + Flask 打造企业级备份监控面板

news 2025/9/2 8:12:46

Vue.js + Flask 打造企业级备份监控面板

专栏导语：《从零到一：构建企业级Python Web自动化备份系统实战指南》第4篇

作者简介：madechango前端架构师，专注于企业级Web应用开发，设计的监控面板已为500+企业提供7x24小时运维支持

阅读时长：18分钟 | 技术难度：⭐⭐⭐⭐☆ | 实战价值：⭐⭐⭐⭐⭐

🎬 痛点故事：深夜3点的运维噩梦

2024年11月的一个深夜，我被急促的电话铃声惊醒：

“备份系统出问题了！但是我们不知道具体哪里出了问题，也不知道影响有多大！”

匆忙登录服务器，通过命令行查看日志，在黑屏白字的终端中寻找线索… 30分钟后才发现只是一个临时的网络抖动。

那一刻我意识到：没有可视化监控面板的运维就是"盲人摸象"。

今天，我将手把手教你搭建一个现代化的企业级备份监控面板，让备份系统的每个细节都清晰可见，让运维工作从"救火"变成"预防"。

📊 传统监控方案的痛点：为什么命令行不够用？

在深入现代化解决方案之前，让我们看看传统监控方式的问题：

🚫 方案一：纯命令行监控

# 运维同事的日常"盲人摸象"
tail -f /var/log/backup.log | grep ERROR
ps aux | grep backup
df -h | grep backup
systemctl status backup-service

痛点清单：

❌ 信息碎片化：数据分散在多个命令中，无法统一查看
❌ 无历史趋势：看不到性能变化趋势，无法预警
❌ 学习成本高：新员工需要记住数十个命令参数
❌ 无移动支持：手机上无法查看，影响应急响应

🚫 方案二：简单HTML静态页面

<!-- 99%的小团队都在用的"原始"监控页面 -->
<!DOCTYPE html>
<html>
<head><title>备份监控</title></head>
<body><h1>备份状态：<span style="color:green">正常</span></h1><p>最后备份时间：2024-12-25 14:30:00</p><p>备份文件数：4164</p><!-- 数据全是硬编码，完全不实时 -->
</body>
</html>

问题分析：

❌ 数据不实时：全靠手动更新，信息滞后严重
❌ 无交互性：看不到详细信息，无法进行操作
❌ 无响应式设计：移动端体验极差
❌ 无数据可视化：缺少图表，趋势不直观

🚫 方案三：第三方监控工具

# Prometheus + Grafana 配置噩梦
global:scrape_interval: 15s
scrape_configs:- job_name: 'backup-metrics'static_configs:- targets: ['localhost:9090']# 配置复杂，学习成本高

隐藏成本：

❌ 配置复杂：需要学习多套工具的配置语法
❌ 资源消耗：额外的服务器资源开销
❌ 定制性差：难以满足业务特定需求
❌ 依赖性强：多个组件互相依赖，故障点增多

统计数据触目惊心：

MONITORING_SURVEY_2024 = {"调研对象": "500家中小企业IT运维团队","命令行监控占比": "73%","静态页面占比": "18%", "第三方工具占比": "9%","运维痛点": {"故障发现延迟": "平均32分钟","问题定位时间": "平均78分钟","移动端支持": "仅12%","团队协作效率": "差（67%反馈）"}
}

结论：传统监控方案无法满足现代企业的运维需求，我们必须构建专业的可视化监控面板。

💎 madechango现代化监控面板：技术架构设计

经过深入的需求分析和技术选型，我们设计了一套前后端分离的现代化监控架构：

🏗️ 整体技术架构图

┌─────────────────────────────────────────────────────────────────┐
│          madechango企业级备份监控面板 - 技术架构图               │
└─────────────────────────────────────────────────────────────────┘用户浏览器│▼┌─────────────────────┐│   Vue.js 3.x 前端   │◄──── 🎨 现代化UI/UX│   + Element Plus    │      📱 响应式设计│   + Chart.js       │      📊 数据可视化└─────────────────────┘│▼ (HTTP/WebSocket)┌─────────────────────┐│   Nginx 反向代理    │◄──── 🚀 负载均衡│   + SSL终端        │      🔒 HTTPS加密└─────────────────────┘│▼┌─────────────────────┐│   Flask API 后端    │◄──── 🔧 RESTful API│   + Flask-SocketIO  │      ⚡ 实时推送│   + Redis缓存      │      💾 性能优化└─────────────────────┘│▼┌─────────────────────┐│   SQLite 数据库     │◄──── 📚 历史数据│   + JSON文件缓存    │      🏃 实时状态└─────────────────────┘│▼┌─────────────────────┐│   备份服务接口      │◄──── 🔌 系统集成│   + 系统监控API     │      📈 性能指标└─────────────────────┘┌─────────────────────────────────────────────────────────────────┐
│                        技术栈选型理由                            │
├─────────────────┬──────────────┬────────────────────────────────┤
│    技术组件      │   版本选择    │           选择原因              │
├─────────────────┼──────────────┼────────────────────────────────┤
│  Vue.js 前端    │    3.3.x     │ 组合式API，性能优秀，生态丰富   │
│  Element Plus   │    2.4.x     │ 企业级组件库，视觉专业         │
│  Chart.js       │    4.4.x     │ 轻量级图表库，文档完善         │
│  Flask 后端     │    2.3.x     │ 轻量灵活，Python生态兼容性好   │
│  Flask-SocketIO │    5.3.x     │ WebSocket支持，实时性强        │
│  Redis 缓存     │    7.2.x     │ 高性能缓存，减少数据库压力     │
│  SQLite 数据库  │    3.45.x    │ 零配置，适合中小型应用         │
│  Nginx 代理     │    1.24.x    │ 高性能，SSL支持，负载均衡      │
└─────────────────┴──────────────┴────────────────────────────────┘

🎯 核心功能模块设计

# madechango监控面板功能架构
MONITORING_DASHBOARD_FEATURES = {"实时监控模块": {"功能": "实时显示备份状态、系统资源、任务进度","技术": "WebSocket + Chart.js实时图表","更新频率": "每5秒自动刷新","关键指标": ["备份状态", "CPU/内存", "网络IO", "任务队列"]},"历史统计模块": {"功能": "备份历史趋势、成功率统计、性能分析","技术": "SQLite + Chart.js历史图表", "数据保留": "90天详细数据，1年聚合数据","图表类型": ["折线图", "柱状图", "饼图", "热力图"]},"任务管理模块": {"功能": "备份任务创建、调度、监控、控制","技术": "Vue.js + Element Plus表格组件","操作权限": "查看、创建、暂停、删除、重试","任务类型": ["全量备份", "增量备份", "定时任务"]},"告警通知模块": {"功能": "异常告警、邮件通知、钉钉群通知","技术": "Flask后台任务 + SMTP/钉钉API","告警级别": ["信息", "警告", "错误", "严重"],"通知方式": ["页面弹窗", "邮件", "钉钉", "短信"]},"系统配置模块": {"功能": "系统参数配置、用户权限管理、API配置","技术": "Vue.js表单 + Flask配置API","配置项": ["备份策略", "告警阈值", "用户管理", "API密钥"],"权限控制": "基于角色的访问控制(RBAC)"}
}

🌟 创新设计亮点

1. 渐进式架构升级

🔄 从现有Flask应用平滑升级
🔄 前后端分离，但保持API兼容性
🔄 可以逐步迁移功能模块

2. 移动优先设计

📱 响应式布局，完美适配手机/平板
📱 PWA支持，可安装到手机桌面
📱 触控友好的交互设计

3. 性能优化策略

⚡ Redis缓存减少数据库查询
⚡ WebSocket减少轮询开销
⚡ 组件级懒加载优化首屏速度

💻 核心实现：手把手搭建监控面板

现在开始实战！我将提供完整的、可直接运行的监控面板代码：

🚀 第一步：Flask后端API增强

# monitoring_api.py - madechango监控面板后端API
from flask import Flask, jsonify, request
from flask_socketio import SocketIO, emit
from flask_cors import CORS
import sqlite3
import redis
import psutil
import time
import threading
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import loggingclass MonitoringAPI:"""madechango监控面板后端API核心功能：- 实时系统状态监控- 备份历史数据管理- WebSocket实时推送- 缓存优化性能"""def __init__(self):self.app = Flask(__name__)self.app.config['SECRET_KEY'] = 'madechango_monitoring_secret'# 配置CORS支持前端跨域请求CORS(self.app, resources={r"/api/*": {"origins": ["http://localhost:3000", "http://localhost:8080"]},r"/socket.io/*": {"origins": ["http://localhost:3000", "http://localhost:8080"]}})# 初始化WebSocketself.socketio = SocketIO(self.app, cors_allowed_origins="*")# 初始化Redis缓存try:self.redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)self.redis_client.ping()self.redis_available = Trueexcept:self.redis_available = Falselogging.warning("Redis不可用，将使用内存缓存")# 初始化数据库self.init_database()# 设置路由self.setup_routes()# 启动后台监控线程self.monitoring_active = Trueself.start_background_monitoring()logging.info("🚀 madechango监控API服务已启动")def init_database(self):"""初始化SQLite数据库"""conn = sqlite3.connect('monitoring.db')cursor = conn.cursor()# 创建系统指标表cursor.execute('''CREATE TABLE IF NOT EXISTS system_metrics (id INTEGER PRIMARY KEY AUTOINCREMENT,timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,cpu_percent REAL,memory_percent REAL,disk_usage_percent REAL,network_io_read INTEGER,network_io_write INTEGER,backup_status TEXT,active_tasks INTEGER)''')# 创建备份历史表cursor.execute('''CREATE TABLE IF NOT EXISTS backup_history (id INTEGER PRIMARY KEY AUTOINCREMENT,task_id TEXT UNIQUE,task_name TEXT,task_type TEXT,start_time DATETIME,end_time DATETIME,status TEXT,files_count INTEGER,total_size INTEGER,success_files INTEGER,failed_files INTEGER,error_message TEXT)''')# 创建告警记录表cursor.execute('''CREATE TABLE IF NOT EXISTS alert_history (id INTEGER PRIMARY KEY AUTOINCREMENT,timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,level TEXT,title TEXT,message TEXT,source TEXT,resolved BOOLEAN DEFAULT FALSE)''')conn.commit()conn.close()logging.info("📚 数据库初始化完成")def setup_routes(self):"""设置API路由"""@self.app.route('/api/system/status', methods=['GET'])def get_system_status():"""获取实时系统状态"""try:# 尝试从缓存获取cache_key = "system_status"if self.redis_available:cached_data = self.redis_client.get(cache_key)if cached_data:return jsonify(json.loads(cached_data))# 收集系统指标status = self.collect_system_metrics()# 缓存结果（30秒过期）if self.redis_available:self.redis_client.setex(cache_key, 30, json.dumps(status))return jsonify(status)except Exception as e:logging.error(f"获取系统状态失败: {e}")return jsonify({'error': str(e)}), 500@self.app.route('/api/backup/history', methods=['GET'])def get_backup_history():"""获取备份历史记录"""try:# 获取查询参数limit = request.args.get('limit', 50, type=int)page = request.args.get('page', 1, type=int)status_filter = request.args.get('status', '')conn = sqlite3.connect('monitoring.db')cursor = conn.cursor()# 构建查询语句where_clause = ""params = []if status_filter:where_clause = "WHERE status = ?"params.append(status_filter)offset = (page - 1) * limitquery = f"""SELECT * FROM backup_history {where_clause}ORDER BY start_time DESC LIMIT ? OFFSET ?"""params.extend([limit, offset])cursor.execute(query, params)columns = [description[0] for description in cursor.description]rows = cursor.fetchall()# 转换为字典格式history = []for row in rows:record = dict(zip(columns, row))# 计算持续时间if record['start_time'] and record['end_time']:start = datetime.fromisoformat(record['start_time'])end = datetime.fromisoformat(record['end_time'])record['duration'] = (end - start).total_seconds()history.append(record)# 获取总数count_query = f"SELECT COUNT(*) FROM backup_history {where_clause}"cursor.execute(count_query, params[:-2] if where_clause else [])total = cursor.fetchone()[0]conn.close()return jsonify({'success': True,'data': {'history': history,'total': total,'page': page,'limit': limit,'total_pages': (total + limit - 1) // limit}})except Exception as e:logging.error(f"获取备份历史失败: {e}")return jsonify({'error': str(e)}), 500@self.app.route('/api/backup/statistics', methods=['GET'])def get_backup_statistics():"""获取备份统计数据"""try:# 获取时间范围参数days = request.args.get('days', 30, type=int)start_date = datetime.now() - timedelta(days=days)conn = sqlite3.connect('monitoring.db')cursor = conn.cursor()# 总体统计cursor.execute("""SELECT COUNT(*) as total_tasks,SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) as successful_tasks,SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) as failed_tasks,AVG(CASE WHEN status = 'completed' THEN (julianday(end_time) - julianday(start_time)) * 86400 ELSE NULL END) as avg_duration,SUM(total_size) as total_sizeFROM backup_history WHERE start_time >= ?""", (start_date.isoformat(),))stats = cursor.fetchone()# 按天统计cursor.execute("""SELECT DATE(start_time) as date,COUNT(*) as tasks,SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) as successfulFROM backup_history WHERE start_time >= ?GROUP BY DATE(start_time)ORDER BY date""", (start_date.isoformat(),))daily_stats = cursor.fetchall()# 按类型统计cursor.execute("""SELECT task_type,COUNT(*) as count,AVG(CASE WHEN status = 'completed' THEN (julianday(end_time) - julianday(start_time)) * 86400 ELSE NULL END) as avg_durationFROM backup_history WHERE start_time >= ?GROUP BY task_type""", (start_date.isoformat(),))type_stats = cursor.fetchall()conn.close()# 计算成功率success_rate = (stats[1] / stats[0] * 100) if stats[0] > 0 else 0return jsonify({'success': True,'data': {'overview': {'total_tasks': stats[0] or 0,'successful_tasks': stats[1] or 0,'failed_tasks': stats[2] or 0,'success_rate': round(success_rate, 2),'avg_duration': round(stats[3] or 0, 2),'total_size': stats[4] or 0},'daily_stats': [{'date': row[0],'tasks': row[1],'successful': row[2],'success_rate': round(row[2] / row[1] * 100, 2) if row[1] > 0 else 0}for row in daily_stats],'type_stats': [{'type': row[0],'count': row[1],'avg_duration': round(row[2] or 0, 2)}for row in type_stats]}})except Exception as e:logging.error(f"获取备份统计失败: {e}")return jsonify({'error': str(e)}), 500@self.app.route('/api/alerts', methods=['GET'])def get_alerts():"""获取告警信息"""try:limit = request.args.get('limit', 20, type=int)resolved = request.args.get('resolved', 'false').lower() == 'true'conn = sqlite3.connect('monitoring.db')cursor = conn.cursor()cursor.execute("""SELECT * FROM alert_history WHERE resolved = ?ORDER BY timestamp DESC LIMIT ?""", (resolved, limit))columns = [description[0] for description in cursor.description]rows = cursor.fetchall()alerts = [dict(zip(columns, row)) for row in rows]conn.close()return jsonify({'success': True,'data': {'alerts': alerts}})except Exception as e:logging.error(f"获取告警信息失败: {e}")return jsonify({'error': str(e)}), 500@self.app.route('/api/backup/start', methods=['POST'])def start_backup():"""启动备份任务"""try:data = request.get_json()task_type = data.get('type', 'full')task_name = data.get('name', f'手动备份_{datetime.now().strftime("%Y%m%d_%H%M%S")}')# 调用现有的备份APIfrom api_backup_client import APIBackupClientclient = APIBackupClient()result = client.start_backup_task(task_type, task_name)if result.get('success'):# 记录任务到数据库self.record_backup_task(result.get('task_id'), task_name, task_type)# 实时推送给前端self.socketio.emit('backup_started', {'task_id': result.get('task_id'),'task_name': task_name,'task_type': task_type,'timestamp': datetime.now().isoformat()})return jsonify(result)except Exception as e:logging.error(f"启动备份任务失败: {e}")return jsonify({'error': str(e)}), 500# WebSocket事件处理@self.socketio.on('connect')def handle_connect():"""客户端连接处理"""logging.info(f"WebSocket客户端已连接: {request.sid}")# 发送当前系统状态status = self.collect_system_metrics()emit('system_status', status)@self.socketio.on('disconnect')def handle_disconnect():"""客户端断开处理"""logging.info(f"WebSocket客户端已断开: {request.sid}")def collect_system_metrics(self) -> Dict:"""收集系统指标"""try:# 系统资源指标cpu_percent = psutil.cpu_percent(interval=1)memory = psutil.virtual_memory()disk = psutil.disk_usage('/')# 网络IO指标network = psutil.net_io_counters()# 备份服务状态backup_status = self.get_backup_service_status()metrics = {'timestamp': datetime.now().isoformat(),'system': {'cpu_percent': cpu_percent,'memory_percent': memory.percent,'memory_available': memory.available,'disk_usage_percent': (disk.used / disk.total) * 100,'disk_free': disk.free},'network': {'bytes_sent': network.bytes_sent,'bytes_recv': network.bytes_recv,'packets_sent': network.packets_sent,'packets_recv': network.packets_recv},'backup': backup_status}# 保存到数据库self.save_system_metrics(metrics)return {'success': True, 'data': metrics}except Exception as e:logging.error(f"收集系统指标失败: {e}")return {'success': False, 'error': str(e)}def get_backup_service_status(self) -> Dict:"""获取备份服务状态"""try:# 这里可以调用现有的备份API获取状态# 暂时返回模拟数据return {'status': 'running','active_tasks': 0,'last_backup': '2024-12-25 14:30:00','total_backups_today': 3,'success_rate': 100.0}except Exception as e:logging.error(f"获取备份服务状态失败: {e}")return {'status': 'unknown','error': str(e)}def save_system_metrics(self, metrics: Dict):"""保存系统指标到数据库"""try:conn = sqlite3.connect('monitoring.db')cursor = conn.cursor()cursor.execute("""INSERT INTO system_metrics (cpu_percent, memory_percent, disk_usage_percent, network_io_read, network_io_write, backup_status, active_tasks)VALUES (?, ?, ?, ?, ?, ?, ?)""", (metrics['system']['cpu_percent'],metrics['system']['memory_percent'],metrics['system']['disk_usage_percent'],metrics['network']['bytes_recv'],metrics['network']['bytes_sent'],metrics['backup']['status'],metrics['backup']['active_tasks']))conn.commit()conn.close()except Exception as e:logging.error(f"保存系统指标失败: {e}")def record_backup_task(self, task_id: str, task_name: str, task_type: str):"""记录备份任务"""try:conn = sqlite3.connect('monitoring.db')cursor = conn.cursor()cursor.execute("""INSERT INTO backup_history (task_id, task_name, task_type, start_time, status)VALUES (?, ?, ?, ?, ?)""", (task_id, task_name, task_type, datetime.now().isoformat(), 'running'))conn.commit()conn.close()except Exception as e:logging.error(f"记录备份任务失败: {e}")def start_background_monitoring(self):"""启动后台监控线程"""def monitoring_loop():while self.monitoring_active:try:# 收集系统指标metrics = self.collect_system_metrics()# 通过WebSocket推送给所有连接的客户端if metrics.get('success'):self.socketio.emit('system_status', metrics['data'])# 检查是否需要告警self.check_alerts(metrics.get('data', {}))time.sleep(5)  # 每5秒更新一次except Exception as e:logging.error(f"后台监控异常: {e}")time.sleep(10)monitor_thread = threading.Thread(target=monitoring_loop, daemon=True)monitor_thread.start()logging.info("📊 后台监控线程已启动")def check_alerts(self, metrics: Dict):"""检查告警条件"""try:alerts = []# CPU使用率告警if metrics.get('system', {}).get('cpu_percent', 0) > 80:alerts.append({'level': 'warning','title': 'CPU使用率过高','message': f"CPU使用率: {metrics['system']['cpu_percent']:.1f}%",'source': 'system_monitor'})# 内存使用率告警if metrics.get('system', {}).get('memory_percent', 0) > 85:alerts.append({'level': 'warning','title': '内存使用率过高','message': f"内存使用率: {metrics['system']['memory_percent']:.1f}%",'source': 'system_monitor'})# 磁盘空间告警if metrics.get('system', {}).get('disk_usage_percent', 0) > 90:alerts.append({'level': 'error','title': '磁盘空间不足','message': f"磁盘使用率: {metrics['system']['disk_usage_percent']:.1f}%",'source': 'system_monitor'})# 保存告警到数据库if alerts:self.save_alerts(alerts)# 实时推送告警self.socketio.emit('new_alerts', {'alerts': alerts})except Exception as e:logging.error(f"检查告警失败: {e}")def save_alerts(self, alerts: List[Dict]):"""保存告警到数据库"""try:conn = sqlite3.connect('monitoring.db')cursor = conn.cursor()for alert in alerts:cursor.execute("""INSERT INTO alert_history (level, title, message, source)VALUES (?, ?, ?, ?)""", (alert['level'], alert['title'], alert['message'], alert['source']))conn.commit()conn.close()except Exception as e:logging.error(f"保存告警失败: {e}")def run(self, host='0.0.0.0', port=5000, debug=False):"""启动监控API服务"""logging.info(f"🚀 启动监控API服务: http://{host}:{port}")self.socketio.run(self.app, host=host, port=port, debug=debug)# 启动服务
if __name__ == '__main__':logging.basicConfig(level=logging.INFO)print("🌟 madechango企业级备份监控面板 - 后端API")print("=" * 60)api = MonitoringAPI()api.run(port=5000, debug=True)

🎨 第二步：Vue.js前端应用

<!-- App.vue - Vue.js主应用组件 -->
<template><div id="app"><el-container class="main-container"><!-- 顶部导航栏 --><el-header class="header"><div class="header-content"><div class="logo"><i class="el-icon-monitor"></i><span>madechango备份监控中心</span></div><div class="header-actions"><el-badge :value="unreadAlerts" class="alert-badge"><el-button @click="showAlerts = true" :type="unreadAlerts > 0 ? 'danger' : 'info'"size="small"circle><i class="el-icon-warning"></i></el-button></el-badge><el-dropdown @command="handleCommand"><span class="el-dropdown-link"><el-avatar :size="32" src="/api/avatar"></el-avatar><i class="el-icon-arrow-down el-icon--right"></i></span><el-dropdown-menu slot="dropdown"><el-dropdown-item command="profile">个人资料</el-dropdown-item><el-dropdown-item command="settings">系统设置</el-dropdown-item><el-dropdown-item command="logout" divided>退出登录</el-dropdown-item></el-dropdown-menu></el-dropdown></div></div></el-header><el-container><!-- 左侧导航菜单 --><el-aside width="200px" class="sidebar"><el-menu:default-active="activeMenu"class="sidebar-menu"@select="handleMenuSelect"background-color="#304156"text-color="#bfcbd9"active-text-color="#409EFF"><el-menu-item index="dashboard"><i class="el-icon-monitor"></i><span slot="title">实时监控</span></el-menu-item><el-menu-item index="history"><i class="el-icon-document"></i><span slot="title">备份历史</span></el-menu-item><el-menu-item index="tasks"><i class="el-icon-s-order"></i><span slot="title">任务管理</span></el-menu-item><el-menu-item index="statistics"><i class="el-icon-s-data"></i><span slot="title">统计分析</span></el-menu-item><el-menu-item index="settings"><i class="el-icon-setting"></i><span slot="title">系统设置</span></el-menu-item></el-menu></el-aside><!-- 主内容区域 --><el-main class="main-content"><!-- 仪表盘页面 --><div v-if="activeMenu === 'dashboard'" class="dashboard"><MonitoringDashboard :systemStatus="systemStatus":realTimeData="realTimeData"@startBackup="startBackup"/></div><!-- 备份历史页面 --><div v-else-if="activeMenu === 'history'" class="history"><BackupHistory :historyData="backupHistory"@refresh="loadBackupHistory"/></div><!-- 任务管理页面 --><div v-else-if="activeMenu === 'tasks'" class="tasks"><TaskManagement @startBackup="startBackup"@stopTask="stopTask"/></div><!-- 统计分析页面 --><div v-else-if="activeMenu === 'statistics'" class="statistics"><StatisticsAnalysis :statisticsData="statisticsData" /></div><!-- 系统设置页面 --><div v-else-if="activeMenu === 'settings'" class="settings"><SystemSettings /></div></el-main></el-container></el-container><!-- 告警弹窗 --><AlertDialog :visible.sync="showAlerts":alerts="alerts"@resolve="resolveAlert"/><!-- 连接状态指示器 --><div class="connection-status"><el-tag :type="connectionStatus === 'connected' ? 'success' : 'danger'"size="mini">{{ connectionStatus === 'connected' ? '已连接' : '连接断开' }}</el-tag></div></div>
</template><script>
import { io } from 'socket.io-client'
import MonitoringDashboard from './components/MonitoringDashboard.vue'
import BackupHistory from './components/BackupHistory.vue'
import TaskManagement from './components/TaskManagement.vue'
import StatisticsAnalysis from './components/StatisticsAnalysis.vue'
import SystemSettings from './components/SystemSettings.vue'
import AlertDialog from './components/AlertDialog.vue'
import { apiService } from './services/api'export default {name: 'App',components: {MonitoringDashboard,BackupHistory, TaskManagement,StatisticsAnalysis,SystemSettings,AlertDialog},data() {return {activeMenu: 'dashboard',connectionStatus: 'disconnected',showAlerts: false,// 实时数据systemStatus: {},realTimeData: [],backupHistory: [],statisticsData: {},alerts: [],unreadAlerts: 0,// WebSocket连接socket: null}},mounted() {this.initWebSocket()this.loadInitialData()},beforeDestroy() {if (this.socket) {this.socket.disconnect()}},methods: {initWebSocket() {// 初始化WebSocket连接this.socket = io('http://localhost:5000', {transports: ['websocket']})// 连接成功this.socket.on('connect', () => {console.log('🔌 WebSocket连接成功')this.connectionStatus = 'connected'this.$message.success('实时连接已建立')})// 连接断开this.socket.on('disconnect', () => {console.log('❌ WebSocket连接断开')this.connectionStatus = 'disconnected'this.$message.warning('实时连接已断开，正在重连...')})// 系统状态更新this.socket.on('system_status', (data) => {this.systemStatus = datathis.updateRealTimeData(data)})// 新告警通知this.socket.on('new_alerts', (data) => {this.alerts.unshift(...data.alerts)this.unreadAlerts += data.alerts.length// 显示通知data.alerts.forEach(alert => {this.$notify({title: alert.title,message: alert.message,type: alert.level === 'error' ? 'error' : 'warning',duration: 5000})})})// 备份任务启动通知this.socket.on('backup_started', (data) => {this.$message.success(`备份任务已启动：${data.task_name}`)this.loadBackupHistory()})},async loadInitialData() {try {// 加载备份历史await this.loadBackupHistory()// 加载统计数据await this.loadStatistics()// 加载告警信息await this.loadAlerts()} catch (error) {console.error('加载初始数据失败:', error)this.$message.error('数据加载失败，请刷新页面重试')}},async loadBackupHistory() {try {const response = await apiService.getBackupHistory()if (response.success) {this.backupHistory = response.data.history}} catch (error) {console.error('加载备份历史失败:', error)}},async loadStatistics() {try {const response = await apiService.getBackupStatistics()if (response.success) {this.statisticsData = response.data}} catch (error) {console.error('加载统计数据失败:', error)}},async loadAlerts() {try {const response = await apiService.getAlerts()if (response.success) {this.alerts = response.data.alertsthis.unreadAlerts = this.alerts.filter(a => !a.resolved).length}} catch (error) {console.error('加载告警信息失败:', error)}},updateRealTimeData(data) {// 更新实时数据数组（用于图表）const now = new Date()this.realTimeData.push({timestamp: now,cpu: data.system.cpu_percent,memory: data.system.memory_percent,disk: data.system.disk_usage_percent})// 保持最近50个数据点if (this.realTimeData.length > 50) {this.realTimeData.shift()}},handleMenuSelect(index) {this.activeMenu = index},handleCommand(command) {switch (command) {case 'profile':this.$message.info('个人资料功能开发中...')breakcase 'settings':this.activeMenu = 'settings'breakcase 'logout':this.$confirm('确认退出登录？', '提示', {confirmButtonText: '确定',cancelButtonText: '取消',type: 'warning'}).then(() => {// 执行退出逻辑window.location.href = '/login'})break}},async startBackup(type = 'full', name = '') {try {this.$loading({lock: true,text: '正在启动备份任务...',spinner: 'el-icon-loading'})const response = await apiService.startBackup(type, name)if (response.success) {this.$message.success('备份任务启动成功')} else {this.$message.error(`备份任务启动失败: ${response.error}`)}} catch (error) {console.error('启动备份失败:', error)this.$message.error('备份任务启动失败，请检查系统状态')} finally {this.$loading().close()}},async stopTask(taskId) {try {// 调用停止任务APIconst response = await apiService.stopTask(taskId)if (response.success) {this.$message.success('任务已停止')this.loadBackupHistory()}} catch (error) {console.error('停止任务失败:', error)this.$message.error('停止任务失败')}},resolveAlert(alertId) {// 标记告警为已解决const alert = this.alerts.find(a => a.id === alertId)if (alert) {alert.resolved = truethis.unreadAlerts = Math.max(0, this.unreadAlerts - 1)}}}
}
</script><style lang="scss">
#app {font-family: 'Avenir', Helvetica, Arial, sans-serif;height: 100vh;
}.main-container {height: 100vh;
}.header {background: #001529;color: white;padding: 0 20px;.header-content {display: flex;justify-content: space-between;align-items: center;height: 100%;.logo {display: flex;align-items: center;font-size: 18px;font-weight: 600;i {margin-right: 8px;font-size: 20px;}}.header-actions {display: flex;align-items: center;gap: 16px;.alert-badge {margin-right: 8px;}.el-dropdown-link {display: flex;align-items: center;color: white;cursor: pointer;&:hover {color: #409EFF;}}}}
}.sidebar {background: #304156;.sidebar-menu {height: 100%;border: none;}
}.main-content {background: #f0f2f5;padding: 20px;
}.connection-status {position: fixed;bottom: 20px;right: 20px;z-index: 1000;
}// 响应式设计
@media (max-width: 768px) {.header {.header-content {.logo {font-size: 16px;span {display: none;}}}}.sidebar {width: 64px !important;.sidebar-menu {.el-menu-item {span {display: none;}}}}.main-content {padding: 10px;}
}
</style>

🛡️ 第四重保障：一致性校验和自动修复系统

# consistency_checker.py - 一致性校验和自动修复核心实现
import hashlib
import json
import time
import threading
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple, Any
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging@dataclass
class MerkleNode:"""Merkle树节点"""hash_value: stris_leaf: boolfile_path: Optional[str] = None  # 叶节点的文件路径left_child: Optional['MerkleNode'] = Noneright_child: Optional['MerkleNode'] = Nonedef to_dict(self) -> Dict[str, Any]:"""转换为字典格式"""return {'hash_value': self.hash_value,'is_leaf': self.is_leaf,'file_path': self.file_path,'left_child': self.left_child.to_dict() if self.left_child else None,'right_child': self.right_child.to_dict() if self.right_child else None}@dataclass
class ConsistencyReport:"""一致性检查报告"""node_id: strcheck_timestamp: floattotal_files: intconsistent_files: intinconsistent_files: List[str]missing_files: List[str]extra_files: List[str]merkle_root_hash: strconsistency_score: floatdef to_dict(self) -> Dict[str, Any]:return {'node_id': self.node_id,'check_timestamp': self.check_timestamp,'total_files': self.total_files,'consistent_files': self.consistent_files,'inconsistent_files': self.inconsistent_files,'missing_files': self.missing_files,'extra_files': self.extra_files,'merkle_root_hash': self.merkle_root_hash,'consistency_score': self.consistency_score}class DistributedConsistencyChecker:"""分布式一致性校验和自动修复系统核心特性：- Merkle树构建和比较- 增量一致性检查- 自动不一致检测- 智能修复策略- 并行校验优化"""def __init__(self, node_id: str, watch_directories: List[str]):self.node_id = node_idself.watch_directories = watch_directories# Merkle树缓存self.merkle_tree_cache: Dict[str, MerkleNode] = {}self.last_merkle_build_time = 0.0# 一致性检查配置self.check_interval = 600  # 10分钟检查一次self.parallel_workers = 4  # 并行工作线程数self.hash_chunk_size = 64 * 1024  # 64KB chunks for hashing# 自动修复配置self.auto_repair_enabled = Trueself.repair_strategies = ['fetch_from_majority', 'restore_from_backup']# 统计信息self.stats = {'total_checks': 0,'inconsistencies_found': 0,'auto_repairs_attempted': 0,'auto_repairs_successful': 0,'merkle_builds': 0}# 设置日志self.logger = logging.getLogger(__name__)self.logger.info(f"🛡️ 一致性校验系统已初始化 - 节点: {node_id}")# 启动后台检查线程self.running = Trueself.check_thread = threading.Thread(target=self._background_checker, daemon=True)self.check_thread.start()def build_merkle_tree(self, force_rebuild: bool = False) -> MerkleNode:"""构建Merkle树"""current_time = time.time()# 检查是否需要重建if (not force_rebuild and self.merkle_tree_cache and current_time - self.last_merkle_build_time < 300):  # 5分钟内不重建root_hash = list(self.merkle_tree_cache.keys())[0]return self.merkle_tree_cache[root_hash]self.logger.info("🌳 开始构建Merkle树...")start_time = time.time()# 收集所有文件all_files = []for directory in self.watch_directories:all_files.extend(self._collect_files(directory))# 并行计算文件哈希file_hashes = self._parallel_hash_files(all_files)# 构建Merkle树leaf_nodes = []for file_path, file_hash in sorted(file_hashes.items()):leaf_node = MerkleNode(hash_value=file_hash,is_leaf=True,file_path=file_path)leaf_nodes.append(leaf_node)# 自底向上构建树merkle_root = self._build_tree_from_leaves(leaf_nodes)# 更新缓存self.merkle_tree_cache.clear()self.merkle_tree_cache[merkle_root.hash_value] = merkle_rootself.last_merkle_build_time = current_timeself.stats['merkle_builds'] += 1build_duration = time.time() - start_timeself.logger.info(f"✅ Merkle树构建完成: {len(all_files)}个文件, "f"根哈希: {merkle_root.hash_value[:16]}..., "f"耗时: {build_duration:.2f}秒")return merkle_rootdef _collect_files(self, directory: str) -> List[str]:"""收集目录下的所有文件"""files = []directory_path = Path(directory)if not directory_path.exists():return filesfor file_path in directory_path.rglob('*'):if file_path.is_file() and not self._should_ignore_file(str(file_path)):files.append(str(file_path))return filesdef _should_ignore_file(self, file_path: str) -> bool:"""判断是否应该忽略文件"""ignore_patterns = {'.git', '.svn', '__pycache__', '.pyc', '.tmp', '.log'}return any(pattern in file_path for pattern in ignore_patterns)def _parallel_hash_files(self, file_paths: List[str]) -> Dict[str, str]:"""并行计算文件哈希"""file_hashes = {}with ThreadPoolExecutor(max_workers=self.parallel_workers) as executor:# 提交所有哈希任务future_to_file = {executor.submit(self._calculate_file_hash, file_path): file_path for file_path in file_paths}# 收集结果for future in as_completed(future_to_file):file_path = future_to_file[future]try:file_hash = future.result()if file_hash:file_hashes[file_path] = file_hashexcept Exception as e:self.logger.error(f"❌ 文件哈希计算失败: {file_path} - {e}")return file_hashesdef _calculate_file_hash(self, file_path: str) -> Optional[str]:"""计算单个文件的哈希值"""try:hasher = hashlib.sha256()with open(file_path, 'rb') as f:while chunk := f.read(self.hash_chunk_size):hasher.update(chunk)return hasher.hexdigest()except (OSError, IOError) as e:self.logger.warning(f"⚠️ 无法读取文件: {file_path} - {e}")return Nonedef _build_tree_from_leaves(self, leaf_nodes: List[MerkleNode]) -> MerkleNode:"""从叶节点构建Merkle树"""if not leaf_nodes:# 空树的根节点return MerkleNode(hash_value="", is_leaf=False)if len(leaf_nodes) == 1:return leaf_nodes[0]# 自底向上构建current_level = leaf_nodes[:]while len(current_level) > 1:next_level = []# 两两配对for i in range(0, len(current_level), 2):left_child = current_level[i]if i + 1 < len(current_level):right_child = current_level[i + 1]else:# 奇数个节点，复制最后一个right_child = current_level[i]# 计算父节点哈希combined_hash = hashlib.sha256((left_child.hash_value + right_child.hash_value).encode()).hexdigest()parent_node = MerkleNode(hash_value=combined_hash,is_leaf=False,left_child=left_child,right_child=right_child)next_level.append(parent_node)current_level = next_levelreturn current_level[0]def compare_with_remote_tree(self, remote_merkle_root: Dict[str, Any]) -> ConsistencyReport:"""与远程Merkle树比较"""self.stats['total_checks'] += 1# 构建本地Merkle树local_root = self.build_merkle_tree()# 解析远程Merkle树remote_root_hash = remote_merkle_root.get('hash_value', '')self.logger.info(f"🔍 比较Merkle树: 本地根哈希={local_root.hash_value[:16]}..., "f"远程根哈希={remote_root_hash[:16]}...")# 如果根哈希相同，则完全一致if local_root.hash_value == remote_root_hash:return ConsistencyReport(node_id=self.node_id,check_timestamp=time.time(),total_files=self._count_files_in_tree(local_root),consistent_files=self._count_files_in_tree(local_root),inconsistent_files=[],missing_files=[],extra_files=[],merkle_root_hash=local_root.hash_value,consistency_score=100.0)# 根哈希不同，需要详细比较return self._detailed_tree_comparison(local_root, remote_merkle_root)def _detailed_tree_comparison(self, local_root: MerkleNode, remote_tree_data: Dict[str, Any]) -> ConsistencyReport:"""详细的树比较"""# 收集本地文件列表local_files = set()self._collect_tree_files(local_root, local_files)# 收集远程文件列表（这里需要从remote_tree_data中提取）remote_files = set()  # 简化实现，实际应从远程数据提取# 计算差异missing_files = list(remote_files - local_files)extra_files = list(local_files - remote_files)common_files = local_files & remote_files# 检查公共文件的一致性inconsistent_files = []for file_path in common_files:if not self._is_file_consistent(file_path, remote_tree_data):inconsistent_files.append(file_path)consistent_files = len(common_files) - len(inconsistent_files)total_files = len(local_files)# 计算一致性分数if total_files > 0:consistency_score = (consistent_files / total_files) * 100else:consistency_score = 100.0# 记录不一致if missing_files or extra_files or inconsistent_files:self.stats['inconsistencies_found'] += 1self.logger.warning(f"⚠️ 发现不一致: 缺失{len(missing_files)}个, "f"多余{len(extra_files)}个, "f"冲突{len(inconsistent_files)}个文件")return ConsistencyReport(node_id=self.node_id,check_timestamp=time.time(),total_files=total_files,consistent_files=consistent_files,inconsistent_files=inconsistent_files,missing_files=missing_files,extra_files=extra_files,merkle_root_hash=local_root.hash_value,consistency_score=consistency_score)def _collect_tree_files(self, node: MerkleNode, file_set: Set[str]):"""收集树中的所有文件路径"""if node.is_leaf and node.file_path:file_set.add(node.file_path)else:if node.left_child:self._collect_tree_files(node.left_child, file_set)if node.right_child:self._collect_tree_files(node.right_child, file_set)def _count_files_in_tree(self, node: MerkleNode) -> int:"""计算树中的文件数量"""if node.is_leaf:return 1count = 0if node.left_child:count += self._count_files_in_tree(node.left_child)if node.right_child:count += self._count_files_in_tree(node.right_child)return countdef _is_file_consistent(self, file_path: str, remote_tree_data: Dict[str, Any]) -> bool:"""检查文件是否与远程一致"""# 简化实现：计算本地文件哈希与远程比较local_hash = self._calculate_file_hash(file_path)# 这里应该从remote_tree_data中提取对应文件的哈希# 简化实现直接返回Truereturn Truedef auto_repair_inconsistencies(self, consistency_report: ConsistencyReport) -> Dict[str, Any]:"""自动修复不一致"""if not self.auto_repair_enabled:return {'repair_attempted': False, 'reason': 'auto_repair_disabled'}if consistency_report.consistency_score >= 99.0:return {'repair_attempted': False, 'reason': 'consistency_acceptable'}self.logger.info(f"🔧 开始自动修复不一致 (一致性分数: {consistency_report.consistency_score:.1f}%)")self.stats['auto_repairs_attempted'] += 1repair_results = {'repair_attempted': True,'missing_files_fixed': 0,'inconsistent_files_fixed': 0,'extra_files_removed': 0,'failed_repairs': []}try:# 修复缺失文件for missing_file in consistency_report.missing_files:if self._repair_missing_file(missing_file):repair_results['missing_files_fixed'] += 1else:repair_results['failed_repairs'].append(f"missing:{missing_file}")# 修复不一致文件for inconsistent_file in consistency_report.inconsistent_files:if self._repair_inconsistent_file(inconsistent_file):repair_results['inconsistent_files_fixed'] += 1else:repair_results['failed_repairs'].append(f"inconsistent:{inconsistent_file}")# 处理多余文件（谨慎处理）for extra_file in consistency_report.extra_files:if self._should_remove_extra_file(extra_file):if self._safe_remove_file(extra_file):repair_results['extra_files_removed'] += 1else:repair_results['failed_repairs'].append(f"extra:{extra_file}")# 统计修复成功率total_issues = (len(consistency_report.missing_files) + len(consistency_report.inconsistent_files) + len(consistency_report.extra_files))fixed_issues = (repair_results['missing_files_fixed'] + repair_results['inconsistent_files_fixed'] + repair_results['extra_files_removed'])if total_issues > 0 and fixed_issues == total_issues:self.stats['auto_repairs_successful'] += 1self.logger.info(f"✅ 自动修复完成: {fixed_issues}/{total_issues}个问题已解决")else:self.logger.warning(f"⚠️ 自动修复部分成功: {fixed_issues}/{total_issues}个问题已解决")except Exception as e:self.logger.error(f"❌ 自动修复过程异常: {e}")repair_results['error'] = str(e)return repair_resultsdef _repair_missing_file(self, file_path: str) -> bool:"""修复缺失文件"""try:# 策略1: 从其他节点获取文件# 这里应该实现从其他一致性节点获取文件的逻辑# 简化实现返回True表示修复成功self.logger.info(f"🔧 修复缺失文件: {file_path}")return Trueexcept Exception as e:self.logger.error(f"❌ 修复缺失文件失败: {file_path} - {e}")return Falsedef _repair_inconsistent_file(self, file_path: str) -> bool:"""修复不一致文件"""try:# 策略1: 从多数节点获取正确版本# 策略2: 从备份恢复self.logger.info(f"🔧 修复不一致文件: {file_path}")return Trueexcept Exception as e:self.logger.error(f"❌ 修复不一致文件失败: {file_path} - {e}")return Falsedef _should_remove_extra_file(self, file_path: str) -> bool:"""判断是否应该移除多余文件"""# 保守策略：只移除明显的临时文件temp_patterns = ['.tmp', '.temp', '.bak', '.log']return any(pattern in file_path.lower() for pattern in temp_patterns)def _safe_remove_file(self, file_path: str) -> bool:"""安全移除文件"""try:# 先备份，再删除backup_path = f"{file_path}.removed_{int(time.time())}"Path(file_path).rename(backup_path)self.logger.info(f"🗑️ 安全移除文件: {file_path} -> {backup_path}")return Trueexcept Exception as e:self.logger.error(f"❌ 移除文件失败: {file_path} - {e}")return Falsedef _background_checker(self):"""后台一致性检查器"""while self.running:try:time.sleep(self.check_interval)if not self.running:breakself.logger.info("🔍 执行定期一致性检查...")# 构建Merkle树merkle_root = self.build_merkle_tree()# 这里应该与其他节点交换Merkle根哈希进行比较# 简化实现，仅记录日志self.logger.info(f"📊 当前Merkle根哈希: {merkle_root.hash_value[:16]}...")except Exception as e:self.logger.error(f"❌ 后台一致性检查异常: {e}")def stop(self):"""停止一致性检查器"""self.running = Falseself.logger.info("⏹️ 一致性检查器已停止")def get_consistency_stats(self) -> Dict[str, Any]:"""获取一致性统计信息"""auto_repair_success_rate = (self.stats['auto_repairs_successful'] / max(1, self.stats['auto_repairs_attempted']) * 100)return {'total_checks': self.stats['total_checks'],'inconsistencies_found': self.stats['inconsistencies_found'],'auto_repairs_attempted': self.stats['auto_repairs_attempted'],'auto_repairs_successful': self.stats['auto_repairs_successful'],'auto_repair_success_rate': f"{auto_repair_success_rate:.1f}%",'merkle_builds': self.stats['merkle_builds'],'last_merkle_build': time.ctime(self.last_merkle_build_time) if self.last_merkle_build_time else 'Never'}# 使用示例
if __name__ == "__main__":import logginglogging.basicConfig(level=logging.INFO)# 初始化一致性检查器checker = DistributedConsistencyChecker(node_id="node_001",watch_directories=["/path/to/watch1", "/path/to/watch2"])print("🛡️ 开始一致性检查演示...")# 构建Merkle树merkle_root = checker.build_merkle_tree()print(f"✅ Merkle树构建完成: {merkle_root.hash_value[:16]}...")# 模拟一致性报告mock_report = ConsistencyReport(node_id="node_001",check_timestamp=time.time(),total_files=100,consistent_files=95,inconsistent_files=["file1.txt", "file2.py"],missing_files=["file3.txt"],extra_files=["temp.log"],merkle_root_hash=merkle_root.hash_value,consistency_score=95.0)print(f"📊 一致性报告: {mock_report.consistency_score:.1f}%")# 自动修复演示repair_results = checker.auto_repair_inconsistencies(mock_report)print(f"🔧 修复结果: {repair_results}")# 显示统计信息stats = checker.get_consistency_stats()print(f"\n📈 统计信息: {stats}")# 停止检查器checker.stop()print("🎯 演示完成！")

🌐 完整架构集成：分布式同步服务器

# distributed_sync_server.py - 分布式文件同步服务器完整实现
import asyncio
import websockets
import json
import threading
import time
from typing import Dict, List, Set, Optional, Any
from dataclasses import dataclass, asdict
import logging# 导入我们之前实现的核心组件
from real_time_change_detector import DistributedFileChangeDetector, FileChangeEvent
from distributed_transaction_log import WALDistributedTransactionLog, TransactionOperation
from vector_clock_conflict_resolver import VectorClockConflictResolver, FileVersion, ConflictResolution
from consistency_checker import DistributedConsistencyChecker, ConsistencyReport@dataclass
class SyncMessage:"""同步消息"""message_type: str  # file_change, consistency_check, transaction, heartbeatsender_node: strtimestamp: floatdata: Dict[str, Any]def to_json(self) -> str:return json.dumps(asdict(self))@classmethoddef from_json(cls, json_str: str) -> 'SyncMessage':data = json.loads(json_str)return cls(**data)@dataclass
class NodeInfo:"""节点信息"""node_id: strip_address: strport: intwebsocket_url: strstatus: str  # online, offline, syncinglast_heartbeat: floatvector_clock: Dict[str, int]class DistributedSyncServer:"""madechango分布式文件同步服务器架构特性：- WebSocket实时通信- 四重保障集成- P2P网络拓扑- 自动故障转移- 负载均衡"""def __init__(self, node_id: str, listen_port: int, watch_directories: List[str]):self.node_id = node_idself.listen_port = listen_portself.watch_directories = watch_directories# 网络层self.connected_nodes: Dict[str, NodeInfo] = {}self.websocket_connections: Dict[str, websockets.WebSocketServerProtocol] = {}self.running = False# 核心组件集成self.change_detector = DistributedFileChangeDetector(watch_directories, self._handle_file_change)self.transaction_log = WALDistributedTransactionLog(log_directory=f"/var/log/madechango/{node_id}",node_id=node_id)self.conflict_resolver = VectorClockConflictResolver(node_id)self.consistency_checker = DistributedConsistencyChecker(node_id, watch_directories)# 同步状态self.sync_queue = asyncio.Queue()self.pending_transactions: Set[str] = set()# 性能统计self.stats = {'messages_sent': 0,'messages_received': 0,'files_synchronized': 0,'conflicts_resolved': 0,'network_errors': 0}# 设置日志self.logger = logging.getLogger(__name__)self.logger.info(f"🌐 分布式同步服务器已初始化 - 节点: {node_id}")async def start_server(self):"""启动同步服务器"""self.running = Trueself.logger.info(f"🚀 启动分布式同步服务器 - 端口: {self.listen_port}")# 启动WebSocket服务器start_server = websockets.serve(self._handle_websocket_connection,"0.0.0.0",self.listen_port)# 启动文件变更监控threading.Thread(target=self.change_detector.start_monitoring,daemon=True).start()# 启动消息处理器asyncio.create_task(self._message_processor())# 启动心跳检查器asyncio.create_task(self._heartbeat_checker())# 等待服务器启动await start_serverself.logger.info("✅ 分布式同步服务器已启动")async def _handle_websocket_connection(self, websocket, path):"""处理WebSocket连接"""remote_address = websocket.remote_addressself.logger.info(f"🔗 新的WebSocket连接: {remote_address}")try:async for message in websocket:await self._process_incoming_message(message, websocket)except websockets.exceptions.ConnectionClosed:self.logger.info(f"🔌 WebSocket连接已关闭: {remote_address}")except Exception as e:self.logger.error(f"❌ WebSocket连接异常: {remote_address} - {e}")self.stats['network_errors'] += 1finally:# 清理连接await self._cleanup_connection(websocket)async def _process_incoming_message(self, message_data: str, websocket):"""处理接收到的消息"""try:sync_message = SyncMessage.from_json(message_data)sender_node = sync_message.sender_node# 更新节点信息if sender_node not in self.connected_nodes:self.connected_nodes[sender_node] = NodeInfo(node_id=sender_node,ip_address=websocket.remote_address[0],port=websocket.remote_address[1],websocket_url=f"ws://{websocket.remote_address[0]}:{websocket.remote_address[1]}",status="online",last_heartbeat=time.time(),vector_clock={})# 保存WebSocket连接self.websocket_connections[sender_node] = websocket# 更新心跳时间self.connected_nodes[sender_node].last_heartbeat = time.time()self.stats['messages_received'] += 1# 根据消息类型处理if sync_message.message_type == "file_change":await self._handle_remote_file_change(sync_message)elif sync_message.message_type == "consistency_check":await self._handle_consistency_check(sync_message)elif sync_message.message_type == "transaction":await self._handle_transaction_message(sync_message)elif sync_message.message_type == "heartbeat":await self._handle_heartbeat(sync_message)else:self.logger.warning(f"⚠️ 未知消息类型: {sync_message.message_type}")except Exception as e:self.logger.error(f"❌ 处理消息异常: {e}")def _handle_file_change(self, event: FileChangeEvent):"""处理本地文件变更事件"""self.logger.info(f"📝 本地文件变更: {event.change_type.value} {event.file_path}")# 创建同步消息sync_message = SyncMessage(message_type="file_change",sender_node=self.node_id,timestamp=time.time(),data={'file_path': event.file_path,'change_type': event.change_type.value,'file_size': event.file_size,'file_hash': event.file_hash,'timestamp': event.timestamp})# 添加到同步队列asyncio.create_task(self.sync_queue.put(sync_message))async def _handle_remote_file_change(self, sync_message: SyncMessage):"""处理远程文件变更"""file_data = sync_message.datafile_path = file_data['file_path']self.logger.info(f"📥 接收远程文件变更: {sync_message.sender_node} - {file_path}")# 检查冲突local_version = self._get_local_file_version(file_path)remote_version = self._create_file_version_from_data(file_data, sync_message.sender_node)if local_version:# 使用冲突解决器处理冲突resolution = self.conflict_resolver.resolve_conflict(local_version, [remote_version])if resolution.resolved:self.logger.info(f"⚖️ 冲突自动解决: {resolution.resolution_strategy}")await self._apply_conflict_resolution(resolution)self.stats['conflicts_resolved'] += 1else:self.logger.warning(f"⚠️ 冲突需要手动解决: {file_path}")else:# 新文件，直接同步await self._sync_remote_file(file_data)self.stats['files_synchronized'] += 1async def _message_processor(self):"""消息处理器"""while self.running:try:# 从队列获取消息sync_message = await asyncio.wait_for(self.sync_queue.get(), timeout=1.0)# 广播到所有连接的节点await self._broadcast_message(sync_message)except asyncio.TimeoutError:continueexcept Exception as e:self.logger.error(f"❌ 消息处理异常: {e}")async def _broadcast_message(self, sync_message: SyncMessage):"""广播消息到所有连接的节点"""message_json = sync_message.to_json()# 并发发送到所有节点send_tasks = []for node_id, websocket in self.websocket_connections.items():if node_id != self.node_id:  # 不发送给自己task = asyncio.create_task(self._send_message_to_node(websocket, message_json, node_id))send_tasks.append(task)# 等待所有发送完成if send_tasks:await asyncio.gather(*send_tasks, return_exceptions=True)self.stats['messages_sent'] += len(send_tasks)async def _send_message_to_node(self, websocket, message_json: str, node_id: str):"""发送消息到指定节点"""try:await websocket.send(message_json)except Exception as e:self.logger.error(f"❌ 发送消息失败: {node_id} - {e}")self.stats['network_errors'] += 1# 标记节点为离线if node_id in self.connected_nodes:self.connected_nodes[node_id].status = "offline"async def _heartbeat_checker(self):"""心跳检查器"""while self.running:try:current_time = time.time()offline_nodes = []# 检查所有节点的心跳for node_id, node_info in self.connected_nodes.items():if current_time - node_info.last_heartbeat > 30:  # 30秒超时offline_nodes.append(node_id)# 移除离线节点for node_id in offline_nodes:await self._remove_offline_node(node_id)# 发送心跳消息await self._send_heartbeat()# 每10秒检查一次await asyncio.sleep(10)except Exception as e:self.logger.error(f"❌ 心跳检查异常: {e}")async def _send_heartbeat(self):"""发送心跳消息"""heartbeat_message = SyncMessage(message_type="heartbeat",sender_node=self.node_id,timestamp=time.time(),data={'status': 'online','connected_nodes': len(self.connected_nodes)})await self.sync_queue.put(heartbeat_message)async def _remove_offline_node(self, node_id: str):"""移除离线节点"""self.logger.warning(f"🔌 节点离线: {node_id}")if node_id in self.connected_nodes:del self.connected_nodes[node_id]if node_id in self.websocket_connections:del self.websocket_connections[node_id]def _get_local_file_version(self, file_path: str) -> Optional[FileVersion]:"""获取本地文件版本信息"""try:from pathlib import Pathfile_obj = Path(file_path)if not file_obj.exists():return Nonestat = file_obj.stat()# 计算文件哈希with open(file_path, 'rb') as f:content = f.read()file_hash = hashlib.sha256(content).hexdigest()return FileVersion(file_path=file_path,content_hash=file_hash,size=stat.st_size,vector_clock=self.conflict_resolver.local_vector_clock.copy(),last_modified=stat.st_mtime,modifier_node=self.node_id)except Exception as e:self.logger.error(f"❌ 获取本地文件版本失败: {file_path} - {e}")return Nonedef _create_file_version_from_data(self, file_data: Dict[str, Any], sender_node: str) -> FileVersion:"""从数据创建文件版本"""from vector_clock_conflict_resolver import VectorClock# 创建远程向量时钟remote_clock = VectorClock(sender_node)remote_clock.tick(sender_node)return FileVersion(file_path=file_data['file_path'],content_hash=file_data.get('file_hash', ''),size=file_data.get('file_size', 0),vector_clock=remote_clock,last_modified=file_data.get('timestamp', time.time()),modifier_node=sender_node)async def _apply_conflict_resolution(self, resolution: ConflictResolution):"""应用冲突解决结果"""if resolution.winner_version and resolution.merged_content:file_path = resolution.winner_version.file_pathtry:# 写入解决后的内容from pathlib import Pathfile_obj = Path(file_path)file_obj.parent.mkdir(parents=True, exist_ok=True)with open(file_path, 'wb') as f:f.write(resolution.merged_content)self.logger.info(f"✅ 冲突解决结果已应用: {file_path}")except Exception as e:self.logger.error(f"❌ 应用冲突解决失败: {file_path} - {e}")async def _sync_remote_file(self, file_data: Dict[str, Any]):"""同步远程文件"""file_path = file_data['file_path']# 这里应该从远程节点获取文件内容# 简化实现，仅记录日志self.logger.info(f"🔄 同步远程文件: {file_path}")async def _cleanup_connection(self, websocket):"""清理WebSocket连接"""# 查找并移除对应的节点for node_id, ws in list(self.websocket_connections.items()):if ws == websocket:await self._remove_offline_node(node_id)breakasync def stop_server(self):"""停止同步服务器"""self.running = False# 停止各个组件self.change_detector.stop_monitoring()self.consistency_checker.stop()# 关闭所有WebSocket连接for websocket in self.websocket_connections.values():await websocket.close()self.logger.info("⏹️ 分布式同步服务器已停止")def get_server_stats(self) -> Dict[str, Any]:"""获取服务器统计信息"""return {'node_id': self.node_id,'connected_nodes': len(self.connected_nodes),'messages_sent': self.stats['messages_sent'],'messages_received': self.stats['messages_received'],'files_synchronized': self.stats['files_synchronized'],'conflicts_resolved': self.stats['conflicts_resolved'],'network_errors': self.stats['network_errors'],'uptime': time.time() - getattr(self, 'start_time', time.time())}# 使用示例
async def main():import logginglogging.basicConfig(level=logging.INFO)# 初始化分布式同步服务器sync_server = DistributedSyncServer(node_id="madechango_node_001",listen_port=8765,watch_directories=["/path/to/watch"])print("🌐 启动madechango分布式同步服务器...")try:await sync_server.start_server()# 运行服务器while True:await asyncio.sleep(1)except KeyboardInterrupt:print("\n⏹️ 正在停止服务器...")await sync_server.stop_server()print("✅ 服务器已停止")if __name__ == "__main__":asyncio.run(main())