Python open()初始运行正常，后续出现文件路径找不到问题求助

脚本初始多次运行正常，但一段时间后调用open()时提示找不到指定路径，实际文件确在对应路径中存在；调试模式下始终运行正常，难以复现问题。

整体流程 #

1. 从外部SFTP站点下载文件
2. 解密下载的文件
3. 解密完成后将文件移至归档目录
4. 调用open()时使用绝对路径，但后续无法识别新下载的文件

错误发生在with open(file_path, 'rb') as efile:代码行，相关代码如下：

try:
    # Open a transport
    transport = paramiko.Transport((hostname,port))
    transport.connect(None, username, password)

    # Connect to the SFTP site  
    sftp = paramiko.SFTPClient.from_transport(transport)
    sftp.chdir(host_path)
    
    # find the files to download
    to_download = sftp.listdir()
    
    # download all the files
    for file in to_download:
        # check if the file is a directory, we don't download directories
        if not stat.S_ISDIR(sftp.lstat(file).st_mode):
            sftp.get(file, destination_path + "\\" + file)
    sftp.close()
    transport.close() 
except Exception:
    raise

decrypt_creds = get_credentials(x)
# grab the key pair to decrypt
gpg = gnupg.GPG(credentials_path)    
try:
    # decrypt each file and then move it to the output path
    for f in os.listdir(encrypted_path):
        file_path = os.path.join(encrypted_path, f)
        with open(file_path, 'rb') as efile:
            gpg.decrypt_file(efile, passphrase=password, output=output_path + "\\" + f)
except FileNotFoundError as e:
    logger.error("The target import files are not found: {error}".format(error=e))
    raise
except Exception:
    raise

堆栈跟踪信息 #

File "C:\Users\user\Programs\Python\Python310\lib\runpy.py", line 289, in run_path
    return _run_module_code(code, init_globals, run_name,
  File "C:\Users\user\Programs\Python\Python310\lib\runpy.py", line 96, in _run_module_code
    _run_code(code, mod_globals, init_globals,
  File "C:\Users\user\Programs\Python\Python310\lib\runpy.py", line 86, in _run_code
    exec(code, run_globals)
  File "c:\DecryptAndTransfer\DecryptData.py", line 381, in <module>
    main()
  File "c:\DecryptData\DecryptData.py", line 53, in main
    with open(file_path, 'rb') as efile:
FileNotFoundError: [Errno 2] No such file or directory: 'encrypted/file/path/file.pgp'

• 校验下载文件的完整性与存在性：
  SFTP下载后立即检查本地文件是否存在，可选校验文件大小与远程一致，避免下载不完整或静默失败：

  downloaded_files = []
  for file in to_download:
      if not stat.S_ISDIR(sftp.lstat(file).st_mode):
          local_file = os.path.join(destination_path, file)
          sftp.get(file, local_file)
          # 检查文件是否存在
          if not os.path.exists(local_file):
              raise FileNotFoundError(f"下载失败：{local_file} 未创建")
          # 校验文件大小
          remote_size = sftp.stat(file).st_size
          local_size = os.path.getsize(local_file)
          if remote_size != local_size:
              os.remove(local_file)
              raise IOError(f"文件大小不匹配，已删除损坏文件：{local_file}")
          downloaded_files.append(local_file)
  

• 避免目录遍历的竞态条件：
  不要用os.listdir()事后扫描目录，直接使用下载阶段记录的文件路径列表进行解密。防止遍历后到打开前，文件被其他进程移动/删除：

  # 解密时直接遍历下载阶段确认的文件路径
  for file_path in downloaded_files:
      with open(file_path, 'rb') as efile:
          output_file = os.path.join(output_path, os.path.basename(file_path))
          gpg.decrypt_file(efile, passphrase=password, output=output_file)
  

• 添加文件访问的重试机制：
  针对临时的文件锁定（如杀毒软件扫描、系统索引），添加重试逻辑：

  import time
  
  max_retries = 3
  retry_delay = 2  # 秒
  for file_path in downloaded_files:
      for attempt in range(max_retries):
          try:
              with open(file_path, 'rb') as efile:
                  output_file = os.path.join(output_path, os.path.basename(file_path))
                  gpg.decrypt_file(efile, passphrase=password, output=output_file)
              break
          except FileNotFoundError:
              if attempt == max_retries - 1:
                  raise
              time.sleep(retry_delay)
              logger.warning(f"第{attempt+1}次尝试打开文件失败，重试中：{file_path}")
  

• 检查路径与文件名的特殊字符：
  打印路径的原始字符串和编码，确认是否存在非ASCII字符或转义问题：

  for file_path in downloaded_files:
      logger.info(f"待处理文件路径：{repr(file_path)}")
      logger.info(f"路径UTF-8编码：{file_path.encode('utf-8')}")
  

• 排查系统级文件锁定：
  错误发生时用Process Explorer等工具检查是否有其他进程占用目标文件；临时关闭杀毒软件的实时扫描，测试是否是杀毒软件抢先锁定文件扫描。

• 完善日志记录：
  在关键步骤添加详细日志，便于定位问题：

  logger.info(f"开始下载，远程文件列表：{to_download}")
  logger.info(f"下载完成，本地文件列表：{downloaded_files}")
  for file_path in downloaded_files:
      if os.path.exists(file_path):
          logger.info(f"文件存在，大小：{os.path.getsize(file_path)} 字节")
      else:
          logger.error(f"下载后文件消失：{file_path}")
  

内容的提问来源于stack exchange，提问作者fourten