一句话总结：自动化脚本的幂等设计核心是”可重入”——无论脚本在哪一步崩溃、重试多少次，最终状态都一样。实现方式：运行前扫描已完成的进度 → 只处理未完成的部分 → 每完成一步就持久化记录进度。

目录

  1. 问题场景
  2. 幂等设计的核心思想
  3. 实现方式一：本地状态追踪
  4. 实现方式二：页面状态校验
  5. 组合方案：双重校验
  6. 异常处理与重试策略
  7. 总结

问题场景

需要自动批量上传 100 个章节到某小说平台。每个章节的操作流程是：

1
2
3
打开章节编辑器 → 填写标题 → 填写内容 → 点击下一步
→ 处理错别字提示 → 选择检测模式 → 确认发布
→ 验证发布成功 → 返回章节列表

整个流程需要 30~60 分钟。期间可能因为网络波动、安全软件拦截、浏览器异常等原因中断。

如果第 57 章上传到一半时崩溃，重新运行脚本应该：

  • 不重复：已成功发布的第 1~56 章不再重新上传
  • 不遗漏：第 57 章继续上传（如果是第 57 章已发布但记录没保存，就跳过）
  • 自恢复：中间状态丢失时能从最后已知状态继续

这就是幂等设计要解决的问题。

幂等设计的核心思想

幂等性（Idempotency）在 HTTP 中的定义是：一次和多次请求的效果相同。对于自动化脚本，幂等意味着：

无论脚本在哪一步失败、被中断多少次、重新运行多少次，最终达到的状态都是一样的。

实现这一目标的三步策略：

1
2
3
1. 扫描 → 启动时检查"当前已完成什么"
2. 过滤 → 跳过已完成的，只处理未完成的
3. 记录 → 每完成一项就持久化记录进度

实现方式一：本地状态追踪

持久化记录

用一个 JSON 文件追踪已处理的项目：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import os
import json

RECORD_FILE = 'progress.json'

def load_progress():
    """加载已完成的进度"""
    if os.path.exists(RECORD_FILE):
        with open(RECORD_FILE, 'r', encoding='utf-8') as f:
            return json.load(f)
    return {'published': [], 'failed': []}

def save_progress(progress):
    """持久化保存进度"""
    with open(RECORD_FILE, 'w', encoding='utf-8') as f:
        json.dump(progress, f, ensure_ascii=False, indent=2)

def mark_published(progress, chapter_id):
    """标记章节为已发布"""
    if chapter_id not in progress['published']:
        progress['published'].append(chapter_id)
        save_progress(progress)

def mark_failed(progress, chapter_id, error):
    """标记章节为发布失败"""
    progress['failed'].append({
        'id': chapter_id,
        'error': str(error),
        'time': time.strftime('%Y-%m-%d %H:%M:%S')
    })
    save_progress(progress)

启动时扫描

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def get_pending_chapters(all_chapters, progress):
    """获取待处理的章节列表"""
    published = set(progress['published'])
    failed_ids = {f['id'] for f in progress['failed']}
    
    pending = []
    for ch in all_chapters:
        if ch['id'] in published:
            continue  # 跳过已发布的
        if ch['id'] in failed_ids:
            continue  # 也可以选择重试失败的
        pending.append(ch)
    
    return pending

实现方式二：页面状态校验

记录文件可能丢失或被误删。更可靠的方式是从页面当前状态来验证：

从管理页面获取已发布列表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
def get_published_chapters(ws):
    """登录到作者后台，扫描管理页面获取已发布的章节编号"""
    # 导航到章节管理页
    cmd(ws, 'Page.navigate', {'url': 'https://mouplatform.com/author/chapters'})
    time.sleep(3)
    
    # 获取页面上显示的已发布章节编号
    chapters = js(ws, """
        (() => {
            const items = document.querySelectorAll('.chapter-item');
            return Array.from(items).map(item => {
                const num = item.querySelector('.chapter-num');
                const status = item.querySelector('.chapter-status');
                return {
                    id: parseInt(num?.innerText || '0'),
                    status: status?.innerText?.trim() || ''
                };
            }).filter(ch => ch.status === '已发布' || ch.status === 'Published');
        })()
    """)
    
    return {ch['id'] for ch in chapters}

合并使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def get_pending_list(ws, all_chapters):
    """双重验证：本地记录 + 页面校验"""
    # 方法1：本地记录
    progress = load_progress()
    local_published = set(progress['published'])
    
    # 方法2：页面校验（更权威）
    page_published = get_published_chapters(ws)
    
    # 合并——取并集（只要任意一方认为已发布，就跳过）
    already_published = local_published | page_published
    
    # 同时更新本地记录以保持同步
    progress['published'] = list(already_published)
    save_progress(progress)
    
    return [ch for ch in all_chapters if ch['id'] not in already_published]

组合方案：双重校验

最可靠的方案是组合使用本地记录和页面校验：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def run_upload_flow(ws, all_chapters):
    """完整的幂等发布流程"""
    # 第一步：双重验证获取待处理列表
    pending = get_pending_list(ws, all_chapters)
    print(f'总共 {len(all_chapters)} 章，待处理 {len(pending)} 章')
    
    if not pending:
        print('全部章节已发布完毕！')
        return
    
    # 第二步：逐章处理
    progress = load_progress()
    
    for idx, ch in enumerate(pending):
        print(f'[{idx+1}/{len(pending)}] 处理第 {ch["id"]} 章...')
        
        try:
            success = process_one_chapter(ws, ch)
            
            if success:
                mark_published(progress, ch['id'])
                print(f'  ✅ 第 {ch["id"]} 章发布成功')
            else:
                mark_failed(progress, ch['id'], '返回状态异常')
                print(f'  ❌ 第 {ch["id"]} 章发布失败（跳过）')
        
        except (ConnectionAbortedError, ConnectionResetError) as e:
            # 连接错误 → 重连 + 重试当前章节
            print(f'  ⚡ 连接断开，正在重连...')
            ws = reconnect_ws(ws)
            
            # 重试当前章节
            try:
                success = process_one_chapter(ws, ch)
                if success:
                    mark_published(progress, ch['id'])
                    print(f'  ✅ 第 {ch["id"]} 章发布成功（重试后）')
                else:
                    mark_failed(progress, ch['id'], '重试后仍失败')
            except Exception as e2:
                mark_failed(progress, ch['id'], f'重试异常: {e2}')
                print(f'  ❌ 第 {ch["id"]} 章重试失败')
        
        except Exception as e:
            # 其他错误 → 记日志，继续下一章
            mark_failed(progress, ch['id'], str(e))
            traceback.print_exc()
        
        # 章与章之间加入短暂间隔
        time.sleep(1)
    
    # 最终报告
    final_progress = load_progress()
    print(f'\n发布完成！')
    print(f'  成功: {len(final_progress["published"])} 章')
    print(f'  失败: {len(final_progress["failed"])} 章')

异常处理与重试策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def process_with_retry(ws, func, *args, max_retries=3):
    """带重试的操作执行"""
    for attempt in range(max_retries):
        try:
            result = func(ws, *args)
            if result:
                return result
        except ConnectionError as e:
            if attempt < max_retries - 1:
                ws = reconnect_ws(ws)
                time.sleep(2 ** attempt)  # 指数退避：1, 2, 4 秒
                continue
            raise
        
        except Exception as e:
            # 非连接错误，直接失败
            raise
    
    return None

幂等操作清单

每个操作应该是可重入的：

操作 是否幂等 说明
点击”下一步”按钮 ❌ 再次点击可能提交两次
填写 input 框 ✅ 多次填写效果相同
选择 radio ✅ 多次选择效果相同
点击”确认发布” ❌ 可能导致重复发布
页面导航 ✅ 多次导航到同一 URL 效果相同
读取页面数据 ✅ 只读操作自然幂等
写 JSON 记录文件 ✅ 覆盖写而非追加写

非幂等操作的防护：在执行非幂等操作前，先校验是否已经执行过。

总结

  1. 幂等设计不是事后补丁，而应该在设计脚本之初就考虑进去
  2. 三重保障：本地记录 + 页面校验 + 异常重试，层层兜底
  3. 非幂等操作要加防护：执行前先校验状态，确认需要执行再执行
  4. 连接错误单独特判：WebSocket 断线是高频问题，重连后重试当前操作
  5. 记录文件要即时保存：每完成一项就写盘，而不是等全部完成再批量写

总结：幂等设计让自动化脚本从”脆弱的一次性脚本”升级为”健壮的长期服务”。核心很朴素——运行前看”已经做了什么”而不是假设”什么都没做”。配合持久化记录和页面状态双重校验，即使面对网络波动、浏览器崩溃、安全软件拦截，脚本也能稳定运行到最后一刻。