本文目录导读：
我来分享一个Python自动化修复同步异常的脚本方案,涵盖常见的同步问题处理。
基础同步异常修复框架
import os
import sys
import time
import shutil
import logging
import hashlib
import subprocess
from datetime import datetime
from pathlib import Path
import json
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('sync_fix.log'),
logging.StreamHandler()
]
)
class SyncRepairTool:
def __init__(self, config_path='sync_config.json'):
self.config = self.load_config(config_path)
self.fix_history = []
def load_config(self, config_path):
"""加载配置文件"""
try:
with open(config_path, 'r', encoding='utf-8') as f:
return json.load(f)
except FileNotFoundError:
# 默认配置
return {
'source_dir': '/path/to/source',
'target_dir': '/path/to/target',
'backup_dir': '/path/to/backup',
'temp_dir': '/tmp/sync_temp',
'max_retries': 3,
'timeout': 30
}
def create_backup(self, file_path):
"""创建文件备份"""
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
backup_path = Path(self.config['backup_dir']) / f"{file_path.name}.{timestamp}.bak"
try:
shutil.copy2(file_path, backup_path)
logging.info(f"备份创建成功: {backup_path}")
return backup_path
except Exception as e:
logging.error(f"备份失败 {file_path}: {str(e)}")
return None
常见异常检测与修复
class SyncExceptionHandler:
"""处理常见的同步异常"""
@staticmethod
def fix_file_locked(file_path):
"""修复文件被锁定的问题"""
try:
# 检查文件是否被其他进程占用
if os.name == 'nt': # Windows
with open(file_path, 'rb') as f:
f.read(1)
else: # Linux/Mac
import fcntl
with open(file_path, 'rb') as f:
fcntl.flock(f, fcntl.LOCK_SH | fcntl.LOCK_NB)
fcntl.flock(f, fcntl.LOCK_UN)
return True
except (IOError, BlockingIOError):
# 文件被锁定，尝试强制关闭相关进程
SyncExceptionHandler._force_close_process(file_path)
time.sleep(2)
return False
@staticmethod
def _force_close_process(file_path):
"""强制关闭占用文件的进程"""
if os.name == 'nt':
import psutil
for proc in psutil.process_iter(['pid', 'name']):
try:
if file_path in [f.path for f in proc.open_files()]:
proc.terminate()
logging.info(f"终止进程: {proc.name()} (PID: {proc.pid})")
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
else:
subprocess.run(['fuser', '-k', str(file_path)],
capture_output=True, timeout=5)
@staticmethod
def fix_permission_error(file_path):
"""修复权限错误"""
try:
if os.name == 'nt':
subprocess.run(['icacls', str(file_path), '/grant',
f'{os.getlogin()}:(F)'], capture_output=True)
else:
os.chmod(file_path, 0o644)
# 修改所有者
import pwd, grp
uid = pwd.getpwnam(os.getlogin()).pw_uid
gid = grp.getgrnam('staff').gr_gid
os.chown(file_path, uid, gid)
logging.info(f"权限修复成功: {file_path}")
return True
except Exception as e:
logging.error(f"权限修复失败: {str(e)}")
return False
@staticmethod
def fix_disk_space(target_dir):
"""检查并修复磁盘空间不足问题"""
try:
if os.name == 'nt':
import ctypes
free_bytes = ctypes.c_ulonglong(0)
ctypes.windll.kernel32.GetDiskFreeSpaceExW(
ctypes.c_wchar_p(target_dir), None, None,
ctypes.pointer(free_bytes))
free_space = free_bytes.value
else:
stat = os.statvfs(target_dir)
free_space = stat.f_frsize * stat.f_bavail
min_required = 1 * 1024 * 1024 * 1024 # 1GB最小空间
if free_space < min_required:
# 清理临时文件和旧备份
SyncExceptionHandler._cleanup_old_files(target_dir)
return False
return True
except Exception as e:
logging.error(f"磁盘空间检查失败: {str(e)}")
return False
@staticmethod
def _cleanup_old_files(directory):
"""清理旧文件释放空间"""
try:
# 删除30天前的备份文件
threshold = time.time() - (30 * 24 * 3600)
for f in Path(directory).glob('*.bak'):
if f.stat().st_mtime < threshold:
f.unlink()
logging.info(f"清理旧备份: {f}")
except Exception as e:
logging.error(f"文件清理失败: {str(e)}")
网络同步修复
class NetworkSyncRepair:
"""修复网络同步问题"""
@staticmethod
def fix_network_connection(host='8.8.8.8', port=53, timeout=3):
"""修复网络连接问题"""
import socket
try:
# 测试DNS解析
socket.gethostbyname('google.com')
except socket.gaierror:
# DNS问题，尝试刷新DNS缓存
NetworkSyncRepair._flush_dns_cache()
# 测试网络连通性
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
result = sock.connect_ex((host, port))
sock.close()
return result == 0
except:
return False
@staticmethod
def _flush_dns_cache():
"""刷新DNS缓存"""
try:
if os.name == 'nt':
subprocess.run(['ipconfig', '/flushdns'],
capture_output=True, timeout=10)
elif sys.platform == 'darwin':
subprocess.run(['dscacheutil', '-flushcache'],
capture_output=True, timeout=10)
else:
subprocess.run(['systemctl', 'restart', 'systemd-resolved'],
capture_output=True, timeout=10)
logging.info("DNS缓存已刷新")
except Exception as e:
logging.error(f"DNS刷新失败: {str(e)}")
@staticmethod
def fix_sync_service(service_name):
"""修复同步服务"""
try:
if os.name == 'nt':
subprocess.run(['net', 'stop', service_name],
capture_output=True, timeout=30)
time.sleep(2)
subprocess.run(['net', 'start', service_name],
capture_output=True, timeout=30)
else:
subprocess.run(['systemctl', 'restart', service_name],
capture_output=True, timeout=30)
logging.info(f"服务已重启: {service_name}")
return True
except Exception as e:
logging.error(f"服务重启失败: {str(e)}")
return False
自动修复流程
class AutoSyncRepair:
"""自动化同步修复主程序"""
def __init__(self, sync_tool):
self.tool = sync_tool
self.handler = SyncExceptionHandler()
self.network = NetworkSyncRepair()
def repair_sync(self, file_path, error_type='unknown'):
"""执行自动修复流程"""
logging.info(f"开始修复文件: {file_path} (错误类型: {error_type})")
# 1. 创建备份
backup_path = self.tool.create_backup(file_path)
# 2. 检查常见问题
fixes_applied = []
# 2.1 检查网络连接
if not self.network.fix_network_connection():
fixes_applied.append('network_fix')
# 2.2 检查磁盘空间
if not self.handler.fix_disk_space(self.tool.config['target_dir']):
fixes_applied.append('disk_space_fix')
# 2.3 检查文件锁定
if self.handler.fix_file_locked(file_path):
fixes_applied.append('file_lock_fix')
# 2.4 检查权限
if self.handler.fix_permission_error(file_path):
fixes_applied.append('permission_fix')
# 3. 重试同步
if fixes_applied:
logging.info(f"应用了修复: {fixes_applied}")
return self.retry_sync(file_path)
# 4. 如果常规修复无效，尝试更强力的方法
return self.force_fix(file_path)
def retry_sync(self, file_path, max_retries=3):
"""重试同步操作"""
for attempt in range(max_retries):
logging.info(f"重试同步 (第 {attempt + 1} 次)")
try:
# 模拟同步命令
result = subprocess.run(
['rsync', '-avz', str(file_path),
str(Path(self.tool.config['target_dir']))],
capture_output=True, text=True,
timeout=self.tool.config['timeout']
)
if result.returncode == 0:
logging.info(f"同步成功: {file_path}")
return True
except subprocess.TimeoutExpired:
logging.warning("同步超时")
time.sleep(2)
except Exception as e:
logging.error(f"同步错误: {str(e)}")
return False
def force_fix(self, file_path):
"""强力修复方法"""
logging.warning("尝试强力修复...")
try:
# 1. 计算文件哈希
hash_md5 = hashlib.md5()
with open(file_path, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b''):
hash_md5.update(chunk)
original_hash = hash_md5.hexdigest()
# 2. 复制到临时目录
temp_path = Path(self.tool.config['temp_dir']) / file_path.name
shutil.copy2(file_path, temp_path)
# 3. 删除原文件
file_path.unlink()
# 4. 从临时目录恢复
shutil.copy2(temp_path, file_path)
# 5. 验证完整性
hash_md5 = hashlib.md5()
with open(file_path, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b''):
hash_md5.update(chunk)
if hash_md5.hexdigest() == original_hash:
logging.info("强力修复成功")
return self.retry_sync(file_path)
else:
logging.error("文件完整性验证失败")
return False
except Exception as e:
logging.error(f"强力修复失败: {str(e)}")
return False
监控与自动执行
class SyncMonitor:
"""同步状态监控器"""
def __init__(self, repair_tool, check_interval=60):
self.repair_tool = repair_tool
self.check_interval = check_interval
self.failed_syncs = []
def monitor_and_repair(self):
"""持续监控并自动修复"""
logging.info("开始同步监控...")
while True:
try:
# 检查同步状态
sync_status = self.check_sync_status()
if sync_status['failed']:
for failed_item in sync_status['failed']:
# 自动修复
if failed_item not in self.failed_syncs:
success = self.repair_tool.repair_sync(
failed_item['path'],
failed_item['error']
)
if not success:
self.failed_syncs.append(failed_item)
logging.warning(f"需要手动干预: {failed_item['path']}")
# 清除已成功的修复记录
self.failed_syncs = [
f for f in self.failed_syncs
if not Path(f['path']).exists()
]
time.sleep(self.check_interval)
except KeyboardInterrupt:
logging.info("监控已停止")
break
except Exception as e:
logging.error(f"监控错误: {str(e)}")
time.sleep(self.check_interval)
def check_sync_status(self):
"""检查同步状态"""
# 这里应该实现实际的同步状态检查逻辑
return {
'failed': [], # 失败项列表
'success': [], # 成功项列表
'pending': [] # 等待项列表
}
# 使用示例
if __name__ == '__main__':
# 创建修复工具实例
repair_tool = AutoSyncRepair(SyncRepairTool())
# 执行单次修复
repair_tool.repair_sync(Path('/path/to/problem/file.txt'))
# 或者启动持续监控
# monitor = SyncMonitor(repair_tool, check_interval=60)
# monitor.monitor_and_repair()
这个脚本提供了完善的同步异常修复能力,包括：
- 文件锁定问题修复
- 权限错误处理
- 磁盘空间管理
- 网络连接修复
- 服务重启机制
- 自动重试逻辑
- 监控与告警
使用时请根据实际的同步环境和需求调整配置参数。
