feat: 添加用户交互超时自动继续功能

新增独立的超时输入工具类,实现Word内容生成时如果用户10秒内未输入则自动使用默认值继续流程。

主要改动:
- 新建 utils/timeout_input.py 工具类,提供 timeout_prompt 和 timeout_choice_prompt
- 在 settings.py 添加 interaction_timeout 配置项(默认10秒)
- 修改 interaction.py 的 _handle_text 和 _handle_choice 使用超时输入
- 超时后显示友好提示并自动使用默认值

设计优势:
- 独立封装,解耦业务逻辑
- 配置化,可通过 config.yaml 或环境变量自定义超时时间
- 可复用,其他交互场景也可使用
- 不影响 SILENT/PROGRAMMATIC 模式

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
sladro 2025-10-02 08:40:27 +08:00
parent 81e4947b99
commit 9e4f2379fa
4 changed files with 145 additions and 16 deletions

View File

@ -11,7 +11,11 @@ from typing import Any, Dict, List, Optional, Tuple, Union, Callable
import click
from rich.console import Console
from ..config.settings import get_settings
from ..utils.timeout_input import timeout_prompt, timeout_choice_prompt
logger = logging.getLogger(__name__)
settings = get_settings()
class InteractionMode:
@ -121,36 +125,43 @@ class InteractionHandler:
Returns:
选择的值
"""
self.console.print(f"\n📝 {prompt}:", style="blue")
# 标准化选项格式
normalized_options = []
display_options = []
for i, option in enumerate(options):
if isinstance(option, tuple):
value, label = option
normalized_options.append((value, label))
self.console.print(f"{i + 1}. {label}")
display_options.append(label)
else:
normalized_options.append((option, option))
self.console.print(f"{i + 1}. {option}")
display_options.append(option)
# 构建选择列表
choices = [str(i) for i in range(1, len(normalized_options) + 1)]
# 找到默认值对应的索引
default_index = None
if default is not None:
for i, (value, _) in enumerate(normalized_options):
if value == default:
default_index = str(i + 1)
break
try:
choice_str = click.prompt(
"请输入选择",
type=click.Choice(choices),
show_choices=False
# 使用带超时的选择输入
choice_str = timeout_choice_prompt(
prompt=prompt,
options=display_options,
default=default_index or "1",
timeout=settings.interaction_timeout
)
choice_index = int(choice_str) - 1
return normalized_options[choice_index][0]
except (click.Abort, KeyboardInterrupt):
except (KeyboardInterrupt, EOFError, ValueError):
if default is not None:
self.console.print(f"使用默认值: {default}", style="yellow")
return default
raise
# 没有默认值时返回第一个选项
return normalized_options[0][0]
def _handle_text(self,
prompt: str,
@ -167,13 +178,17 @@ class InteractionHandler:
Returns:
输入的文本
"""
# 先用console.print显示格式化的提示信息
# 显示格式化的提示信息
self.console.print(f"\n{prompt}", style="blue")
while True:
try:
# 使用简化的prompt获取输入
result = click.prompt("", default=default or "", type=str, show_default=False)
# 使用带超时的输入
result = timeout_prompt(
prompt="",
default=default or "",
timeout=settings.interaction_timeout
)
# 验证输入
if validation:
@ -184,7 +199,7 @@ class InteractionHandler:
return result
except (click.Abort, KeyboardInterrupt):
except (KeyboardInterrupt, EOFError):
if default is not None:
return default
raise

View File

@ -57,6 +57,9 @@ class Settings(BaseSettings):
max_workers: int = Field(default=4, description="最大工作线程数")
timeout: int = Field(default=300, description="超时时间(秒)")
# 交互配置
interaction_timeout: int = Field(default=10, description="用户交互超时时间(秒)")
# 日志配置
log_level: str = Field(default="INFO", description="日志级别")
log_format: str = Field(

View File

@ -0,0 +1,5 @@
"""工具模块"""
from .timeout_input import timeout_prompt
__all__ = ["timeout_prompt"]

View File

@ -0,0 +1,106 @@
"""带超时的用户输入工具
提供带超时机制的输入函数,超时后自动使用默认值
"""
import sys
import threading
from queue import Queue, Empty
from typing import Optional, List
from rich.console import Console
console = Console()
def timeout_prompt(
prompt: str,
default: Optional[str] = None,
timeout: int = 10,
choices: Optional[List[str]] = None,
) -> str:
"""带超时的输入提示
Args:
prompt: 提示文本
default: 默认值
timeout: 超时时间()
choices: 可选的选项列表(如果提供,则进行验证)
Returns:
用户输入或默认值
"""
result_queue = Queue()
def input_thread():
"""输入线程"""
try:
user_input = input(prompt)
result_queue.put(user_input)
except EOFError:
# 处理EOF情况
result_queue.put(None)
except Exception as e:
console.print(f"[red]输入错误: {e}[/red]")
result_queue.put(None)
# 启动输入线程
thread = threading.Thread(target=input_thread, daemon=True)
thread.start()
try:
# 等待输入,带超时
result = result_queue.get(timeout=timeout)
# 验证选项
if choices and result and result not in choices:
console.print(f"[yellow]输入无效,使用默认值: {default}[/yellow]")
return default or ""
# 空输入使用默认值
if not result or result.strip() == "":
return default or ""
return result.strip()
except Empty:
# 超时 - 需要打印换行以清除输入行
print() # 打印空行以清理输入提示
console.print(f"[yellow]⏰ {timeout}秒内未输入,自动使用默认值: {default}[/yellow]")
return default or ""
def timeout_choice_prompt(
prompt: str,
options: List[str],
default: Optional[str] = None,
timeout: int = 10,
) -> str:
"""带超时的选择输入
Args:
prompt: 提示文本
options: 选项列表
default: 默认选项
timeout: 超时时间()
Returns:
选择的选项索引(字符串形式,"1","2")
"""
# 显示选项
console.print(f"\n[blue]{prompt}:[/blue]")
for i, option in enumerate(options, 1):
console.print(f"{i}. {option}")
# 构建有效选项列表
valid_choices = [str(i) for i in range(1, len(options) + 1)]
# 获取输入
choice = timeout_prompt(
"请输入选择: ",
default=default,
timeout=timeout,
choices=valid_choices
)
return choice