SeleniumBase脚本中断或完成时无法生成/保存输出文件

背景 #

我正在开发SeleniumBase脚本，希望脚本无论正常完成还是中途中断，都能生成并保存输出文件。这个功能在之前的纯Selenium脚本里正常，但换成SeleniumBase后出了问题。

核心问题 #

脚本负责网页操作并把进度保存到CSV，正常运行时功能没问题，但脚本中断或执行完成时，输出文件经常无法生成或保存。

脚本流程 #

• 用凭据登录网站
• 从bowhunter.csv读取URL，导航到在线课程设置页面
• 开启编辑模式、展开所有课程分段，提取含视频页面的当前要求时长
• 将要求时长更新为视频实际时长
• 在多个节点调用保存方法，把进度写入CSV
• 用pandas处理数据，datetime生成时间戳

问题代码 #

问题出在save_progress方法，它本该把DataFrame存为CSV，但中断或完成时无法稳定工作：

def save_progress(self, df, suffix=""):
    try:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_file = f'C:\\Users\\salim\\OneDrive\\Documents\\bowhunter_updated_{suffix}_{timestamp}.csv'
        print(f"Attempting to save file to {output_file}...")
        df.to_csv(output_file, index=False)
        print(f"Progress saved to {output_file}")
        return output_file
    except Exception as e:
        print(f"Error saving file: {e}")

已尝试的排查步骤 #

• 确认文件路径和命名规则无误
• 验证保存前DataFrame非空且数据正常
• 添加打印语句，确认save_progress方法确实被调用过

疑问 #

怎么确保save_progress在脚本中断或成功完成时都能稳定保存文件？有哪些实用的修改方案或最佳实践？

1. 用try/finally覆盖所有执行场景 #

把核心脚本逻辑包裹在try/finally块中，确保无论脚本是正常完成还是抛出异常中断，保存逻辑都会执行：

def run_script(self):
    df = self.load_initial_data()  # 从bowhunter.csv加载初始数据
    try:
        # 核心业务逻辑：登录、页面操作等
        self.login()
        for url in df['url']:
            self.navigate_to_course(url)
            self.update_video_duration()
            self.save_progress(df, suffix="partial")  # 中途阶段性保存
    except Exception as e:
        print(f"脚本执行中断: {e}")
        self.save_progress(df, suffix="interrupted")  # 中断时紧急保存
        raise  # 可选：保留错误栈，便于排查问题
    finally:
        self.save_progress(df, suffix="final")  # 最终统一保存

2. 处理系统中断信号 #

SeleniumBase可能会捕获Ctrl+C这类系统中断信号，需要手动注册信号处理器，确保触发保存：

import signal

# 假设是类方法，用类变量self.df存储数据
def handle_interrupt(self, signal_num, frame):
    print("检测到中断信号，正在保存进度...")
    self.save_progress(self.df, suffix="interrupted")
    exit(1)

# 脚本初始化时注册信号监听
def __init__(self):
    self.df = self.load_initial_data()
    signal.signal(signal.SIGINT, self.handle_interrupt)  # 处理Ctrl+C
    signal.signal(signal.SIGTERM, self.handle_interrupt)  # 处理进程终止信号

3. 优化保存方法的可靠性 #

• 用pathlib处理路径：避免转义字符错误，提升路径兼容性

  from pathlib import Path
  def save_progress(self, df, suffix=""):
      try:
          timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
          output_dir = Path('C:/Users/salim/OneDrive/Documents')
          output_file = output_dir / f'bowhunter_updated_{suffix}_{timestamp}.csv'
          print(f"Attempting to save file to {output_file}...")
          df.to_csv(output_file, index=False)
          # 强制刷新缓冲区，防止数据滞留内存
          import os
          with open(output_file, 'a') as f:
              os.fsync(f.fileno())
          print(f"Progress saved to {output_file}")
          return output_file
      except Exception as e:
          print(f"Error saving file: {e}")
  

• 替换打印为日志：用logging模块记录过程，即使脚本中断也能留存保存日志

  import logging
  logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
  
  # 在save_progress中替换print为logging.info
  logging.info(f"Attempting to save file to {output_file}...")
  

4. 增加定期自动保存 #

每处理一定数量的任务就自动保存一次，降低中断时的数据丢失风险：

counter = 0
for url in df['url']:
    self.navigate_to_course(url)
    self.update_video_duration()
    counter += 1
    if counter % 5 == 0:  # 每处理5条数据保存一次
        self.save_progress(df, suffix=f"partial_{counter}")

5. 排查OneDrive同步问题 #

保存路径在OneDrive目录下，可能存在同步锁导致写入失败。可以先保存到本地非同步目录测试，或者临时关闭OneDrive实时同步，确认是否是同步机制干扰了文件写入。

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