GPEN 断点续传功能设计与实现思路

中断风险点识别

批量处理时的中断影响最大，因为用户可能已等待较长时间。

断点续传技术方案设计

核心思想是：记录处理状态，支持从中断点恢复。

状态记录机制

对于 Web 应用，状态存储有几种选择：

• 服务器端文件记录：无需额外数据库依赖，适合轻量级部署。
• 数据库记录：查询管理方便，适合已有 DB 架构。
• 客户端本地存储：适用于单次会话中断恢复。

考虑到 GPEN 的轻量特性，服务器端文件记录最为合适。示例结构如下：

{
  "task_id": "batch_20250115_143022",
  "total_images": 100,
  "processed_images": [0, 1, 2, 3],
  "failed_images": [],
  "current_index": 4,
  "parameters": {
    "enhance_strength": 80,
    "mode": "powerful"
  },
  "start_time": "2025-01-15T14:30:22",
  "last_update": "2025-01-15T14:35:47",
  "status": "processing"
}

处理流程改造

改造后的批量处理流程需支持任务恢复与容错：

def batch_process_with_checkpoint(images, params, task_id=None):
    # 1. 创建或恢复任务
    if task_id and os.path.exists(f"tasks/{task_id}.json"):
        task_state = load_task_state(task_id)
        start_index = task_state["current_index"]
        processed = set(task_state["processed_images"])
    else:
        task_id = generate_task_id()
        start_index = 0
        processed = set()
        save_task_state(task_id, {
            "total_images": len(images),
            "processed_images": [],
            "current_index": 0,
            "parameters": params,
            "status": "processing"
        })

    # 2. 从断点开始处理
    for i in range(start_index, len(images)):
        if i in processed:
            continue
        try:
            update_task_state(task_id, {"current_index": i})
            img_data = load_image(images[i])
            processed_img = preprocess(img_data, params)
            enhanced_img = gpen_model(processed_img)
            final_img = postprocess(enhanced_img, params)
            save_image(final_img, f"outputs/{task_id}_{i}.png")
            processed.add(i)
            update_task_state(task_id, {
                "processed_images": list(processed),
                "last_update": get_current_time()
            })
        except Exception as e:
            update_task_state(task_id, {
                "failed_images": task_state.get("failed_images", []) + [i],
                "last_update": get_current_time()
            })
            continue

    # 3. 标记任务完成
    update_task_state(task_id, {
        "status": "completed",
        "completed_time": get_current_time()
    })
    return task_id

关键改进包括任务状态持久化、支持任务恢复以及单张失败不影响整体任务。

用户界面增强

后端逻辑改造后，前端也需配合提供任务列表、详情及恢复按钮。例如任务状态显示组件：

<div>
  <h3>批量处理任务 #batch_20250115_143022</h3>
  <div>
    <p>进度：45/100 (45%)</p>
    <p>状态：<span>处理中</span></p>
    <p>已处理：45 张 | 失败：2 张 | 剩余：53 张</p>
    <button>暂停</button>
    <button>继续</button>
    <button>取消</button>
  </div>
</div>

实现挑战与解决方案

状态一致性问题

在高并发场景下，多进程同时访问同一任务状态可能导致不一致。单机部署可使用文件锁保障一致性：

import fcntl

def update_task_state_safely(task_id, updates):
    state_file = f"tasks/{task_id}.json"
    with open(state_file, "r+") as f:
        fcntl.flock(f, fcntl.LOCK_EX)
        current_state = json.load(f)
        current_state.update(updates)
        current_state["last_update"] = get_current_time()
        f.seek(0)
        json.dump(current_state, f, indent=2)
        f.truncate()
        fcntl.flock(f, fcntl.LOCK_UN)
    return current_state

资源清理与维护

中断任务可能占用磁盘空间，需定期清理。可设置任务最大存活时间（如 24 小时），超时自动清理已完成或失败的任务：

def cleanup_old_tasks(max_age_hours=24, max_tasks=100):
    task_dir = "tasks/"
    task_files = sorted(glob.glob(os.path.join(task_dir, "*.json")), key=os.path.getmtime)
    
    if len(task_files) > max_tasks:
        files_to_remove = task_files[:len(task_files) - max_tasks]
        for task_file in files_to_remove:
            task_id = os.path.splitext(os.path.basename(task_file))[0]
            remove_task(task_id)

用户体验考虑

让功能对用户透明且易用是关键。建议实现自动恢复机制，页面加载时检测未完成任务并提示恢复：

function checkUnfinishedTasks() {
    const lastTaskId = localStorage.getItem('last_batch_task');
    if (lastTaskId) {
        fetch(`/api/task/status/${lastTaskId}`)
            .then(response => response.json())
            .then(taskState => {
                if (taskState.status === 'processing' || taskState.status === 'paused') {
                    showRecoveryPrompt(taskState);
                }
            });
    }
}
document.addEventListener('DOMContentLoaded', checkUnfinishedTasks);

性能影响评估

频繁读写状态文件会影响性能。优化建议包括批量更新、内存缓存及异步保存。例如使用批量更新类：

class TaskStateManager:
    def __init__(self, task_id, batch_size=5):
        self.task_id = task_id
        self.batch_size = batch_size
        self.pending_updates = []
        self.last_save_time = time.time()

    def update(self, updates):
        self.pending_updates.append(updates)
        if (len(self.pending_updates) >= self.batch_size or 
            time.time() - self.last_save_time > 30):
            self._save_batch()

    def _save_batch(self):
        if not self.pending_updates:
            return
        merged_updates = {}
        for update in self.pending_updates:
            merged_updates.update(update)
        save_task_state(self.task_id, merged_updates)
        self.pending_updates = []
        self.last_save_time = time.time()

实际应用场景与价值

个人用户场景

修复家庭老照片集或旅行照片批量增强时，断点续传能节省重处理时间。例如处理 100 张照片中途停电，恢复后可直接从断点继续，避免前功尽弃。

小型工作室场景

电商商品图片批量处理或摄影精修时，业务连续性至关重要。即使夜间维护中断，次日也可从断点继续，确保交付时效。

技术价值延伸

该功能为分布式处理和任务调度打下基础，使 GPEN 从单一工具向图像处理平台演进。

总结

技术可行性

从技术角度看，添加断点续传功能完全可行。核心在于状态持久化、状态恢复及 UI 支持，主要涉及工程细节而非高深难题。

实现建议

建议分阶段实施：先实现基本状态记录和恢复，再添加自动恢复机制和 UI 支持，最后优化性能。状态存储建议使用 JSON 文件，同步使用文件锁，恢复基于任务 ID 和图片索引。

实际价值

断点续传虽看似锦上添花，实则显著提升可靠性与灵活性。它减少了意外中断的时间浪费，让用户更放心地处理重要任务。

开始行动

若计划实施，可从理解现有代码、设计状态结构、实现状态保存与恢复、测试验证中断场景入手。渐进式实现往往效果更好。