2. 确认超时处理

任务目标

为确认机制增加超时处理能力：当用户发起一个需要确认的操作后，系统等待用户响应；若在指定时间内（如10秒）未收到确认或拒绝指令，则自动取消该操作，并将超时事件写入日志。

核心实现

import time
import threading
import logging
from enum import Enum

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger("confirmation")

class ConfirmResult(Enum):
    CONFIRMED = "confirmed"
    REJECTED = "rejected"
    TIMEOUT = "timeout"

class ConfirmationTimeout:
    def __init__(self, timeout_seconds=10):
        self.timeout = timeout_seconds
        self._result = None
        self._event = threading.Event()
        self._timer = None

    def _on_timeout(self):
        """超时回调：设置结果为TIMEOUT并唤醒等待线程"""
        self._result = ConfirmResult.TIMEOUT
        self._event.set()
        logger.warning(f"确认超时（{self.timeout}秒未响应），操作已自动取消")

    def wait_for_response(self):
        """启动超时计时，等待用户确认/拒绝/超时"""
        self._timer = threading.Timer(self.timeout, self._on_timeout)
        self._timer.start()
        self._event.wait()  # 阻塞直到用户响应或超时
        self._timer.cancel()
        return self._result

    def confirm(self):
        """用户点击确认"""
        self._result = ConfirmResult.CONFIRMED
        self._event.set()
        logger.info("用户已确认，操作继续执行")

    def reject(self):
        """用户点击拒绝"""
        self._result = ConfirmResult.REJECTED
        self._event.set()
        logger.info("用户已拒绝，操作已取消")

def execute_operation_with_confirmation(operation_name, timeout_seconds=10):
    """
    带超时确认的操作执行入口。
    返回 (ConfirmResult, 是否执行)
    """
    confirmation = ConfirmationTimeout(timeout_seconds)
    print(f"操作「{operation_name}」需要确认，请在{timeout_seconds}秒内输入 y（确认）/ n（拒绝）")

    # 模拟用户输入线程（实际项目中替换为GUI/API回调）
    def simulate_user_input():
        user_input = input("> ").strip().lower()
        if user_input == 'y':
            confirmation.confirm()
        elif user_input == 'n':
            confirmation.reject()
        # 其他输入忽略，等待超时

    input_thread = threading.Thread(target=simulate_user_input, daemon=True)
    input_thread.start()

    result = confirmation.wait_for_response()

    if result == ConfirmResult.CONFIRMED:
        print(f"操作「{operation_name}」已执行")
        return result, True
    else:
        print(f"操作「{operation_name}」未执行（原因：{result.value}）")
        return result, False

if __name__ == "__main__":
    # 演示：5秒超时
    result, executed = execute_operation_with_confirmation("删除文件", timeout_seconds=5)
    print(f"最终结果: {result.value}, 操作执行: {executed}")

运行与验证

运行方式：

python confirmation_timeout.py

验证路径（三种场景）：

1. 确认路径：程序启动后5秒内输入 y，观察日志输出“用户已确认”，控制台显示“操作已执行”，返回 confirmed。
2. 拒绝路径：程序启动后5秒内输入 n，观察日志输出“用户已拒绝”，控制台显示“操作未执行”，返回 rejected。
3. 超时路径：程序启动后不做任何输入，等待5秒，观察日志输出“确认超时”，控制台显示“操作未执行（原因：timeout）”，返回 timeout。

判定标准：三种场景下，wait_for_response() 均能正确返回对应 ConfirmResult 枚举值；超时场景下操作被自动取消且日志中记录了超时事件；确认/拒绝场景下计时器被正确取消，不会在用户响应后再次触发超时回调。

关键设计说明

• 使用 threading.Timer 实现非阻塞超时计时，避免 time.sleep() 阻塞主流程。
• threading.Event 用于协调用户响应与超时回调之间的竞争条件——无论哪个先发生，wait_for_response() 都会立即返回。
• 超时回调中调用 logger.warning 记录日志，实际项目中可替换为文件日志或审计系统。
• 示例中的 simulate_user_input 仅为演示控制台交互，生产环境应替换为GUI按钮回调、API接口或消息队列事件。