Python中使用shutil.move()传输文件时如何处理服务器连接错误?
解决shutil.move()服务器连接错误的方案
针对偶尔出现的服务器无响应导致的传输失败,你可以从以下几个维度优化:
1. 给传输逻辑加重试机制
临时网络波动是这类问题的常见原因,通过重试可以覆盖大部分偶发故障。自己实现一个带重试的move函数:
import shutil import time def retry_move(src, dst, max_retries=3, delay=2): for attempt in range(max_retries): try: shutil.move(src, dst) return except (ConnectionError, OSError) as e: if attempt == max_retries - 1: raise # 最后一次重试失败,抛出原错误便于排查 print(f"传输失败,{delay}秒后重试 ({attempt+1}/{max_retries}):{str(e)}") time.sleep(delay) # 调用示例 retry_move("/local/data/file.csv", "/server/share/backup/file.csv")
注意:实际捕获的异常类型可能需要根据你的服务器挂载方式调整(比如SMB挂载可能抛出PermissionError或特定的OSError子类)。
2. 提前校验服务器可达性
在执行传输前先检查目标路径是否可访问,避免无效的move操作:
import os def is_server_reachable(target_path): try: # 尝试访问目标路径的父目录,验证连接 os.access(os.path.dirname(target_path), os.R_OK) return True except OSError: return False # 先校验再传输 dst_path = "/server/share/backup/file.csv" if is_server_reachable(dst_path): retry_move("/local/data/file.csv", dst_path) else: print("服务器路径不可达,跳过本次传输")
3. 替换为更健壮的传输工具
shutil的底层实现对网络文件系统的容错性有限,改用系统级传输工具会更可靠:
- Linux/macOS:用
rsync,自带增量传输、断点续传和重试逻辑,加上--remove-source-files参数实现move效果:
import subprocess def rsync_move(src, dst): try: subprocess.run( ["rsync", "-av", "--remove-source-files", src, dst], check=True, capture_output=True, text=True ) except subprocess.CalledProcessError as e: print(f"rsync传输失败:{e.stderr}") raise
- Windows:用
robocopy,同样支持重试和断点续传,/MOV参数实现移动:
import subprocess def robocopy_move(src, dst): try: subprocess.run( ["robocopy", os.path.dirname(src), dst, os.path.basename(src), "/MOV", "/R:3", "/W:2"], check=True, capture_output=True, text=True ) except subprocess.CalledProcessError as e: # robocopy返回0/1表示成功,其他为失败 if e.returncode not in (0, 1): print(f"robocopy传输失败:{e.stderr}") raise
4. 完善错误日志与告警
把失败信息记录到日志,同时添加告警机制(比如邮件、企业微信消息),方便及时发现问题:
import logging logging.basicConfig( filename='file_transfer.log', level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s' ) try: retry_move("/local/data/file.csv", "/server/share/backup/file.csv") except Exception as e: error_msg = f"文件传输最终失败:{str(e)}" logging.error(error_msg) # 这里可以添加告警逻辑,比如发送邮件 # send_alert(error_msg)
5. 处理挂载点假死(针对网络挂载场景)
如果是通过挂载方式访问服务器(比如NFS/SMB),挂载点可能出现假死,可尝试重新挂载(需要对应权限):
import subprocess def remount_smb(mount_point, smb_url): try: subprocess.run(["umount", mount_point], check=True) subprocess.run(["mount", "-t", "cifs", smb_url, mount_point], check=True) return True except subprocess.CalledProcessError: print("重新挂载失败,请手动检查") return False
内容的提问来源于stack exchange,提问作者user20052976
相关产品推荐
相关产品推荐

