项目文件夹

文件
wehub-resource-sync 4b6817381b
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
CI / coverage-report (push) Has been cancelled
CI / test-kubernetes (push) Has been cancelled
CI / should-run-thorough (push) Has been cancelled
CI / test-thorough (cloudwatch-demo) (push) Has been cancelled
CI / test-thorough (flink-ecs) (push) Has been cancelled
CI / test-thorough (upstream-lambda) (push) Has been cancelled
CI / test-thorough (prefect-ecs-fargate) (push) Has been cancelled
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Has been cancelled
Benchmark image — build + push to ECR (any adapter) / build + push (push) Has been cancelled
CI / quality (ubuntu-latest) (push) Has been cancelled
CI / test (tools-runtime) (push) Has been cancelled
CI / test (e2e-general) (push) Has been cancelled
CI / test (cli-runtime) (push) Has been cancelled
CI / test (e2e-provider-and-openclaw) (push) Has been cancelled
CI / test (integrations-and-misc) (push) Has been cancelled
Release / verify (push) Has been cancelled
Release / build-python-dist (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Has been cancelled
Release / publish-release (push) Has been cancelled
Release / publish-main-release (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Has been cancelled
Release / prepare (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Has been cancelled
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:10:45 +08:00

286 行
8.7 KiB
Python

此文件含有模棱两可的 Unicode 字符
此文件含有可能会与其他字符混淆的 Unicode 字符。 如果您是想特意这样的,可以安全地忽略该警告。 使用 Escape 按钮显示他们。
"""Interactive choice helpers for TTY-first REPL flows.
Inline menus render in the terminal scrollback (below the submitted command),
not as a separate prompt-toolkit full-screen dialog — important when the REPL
already runs under asyncio.
Each menu erases itself on exit (selection or Esc) so nested menus never
pile up — only the result output and the next level appear on screen.
"""
from __future__ import annotations
import os
import shutil
import sys
from typing import Literal
from rich.console import Console
from rich.markup import escape
import platform.terminal.theme as ui_theme
from surfaces.interactive_shell.ui.components.key_reader import read_key_unix, read_key_windows
_HINT = "↑↓/j/k/Tab Enter/Space Esc/q"
CRUMB_SEP = " "
# Blank line after the submitted slash line before the menu header (all pickers).
_MENU_LEADING_LINES = 1
_TERMINAL_NEWLINE = "\r\n"
MenuAction = Literal["up", "down", "enter", "cancel", "eof", "ignore"]
def repl_tty_interactive() -> bool:
"""Return True when stdin/stdout support an interactive picker UI."""
return bool(sys.stdin.isatty() and sys.stdout.isatty())
def ensure_tty_column_zero() -> None:
"""Reset the cursor column before Rich output when a TTY is active."""
if repl_tty_interactive():
reset_tty_column()
def prepare_repl_output_line() -> None:
"""Begin Rich output on a new line after inline menu I/O."""
if repl_tty_interactive():
sys.stdout.write(_TERMINAL_NEWLINE)
reset_tty_column()
def repl_section_break(console: Console) -> None:
"""Blank line + dim rule between an inline menu step and Rich output."""
prepare_repl_output_line()
console.print()
console.rule(characters="─", style=str(ui_theme.DIM))
console.print()
# ── raw key reader ───────────────────────────────────────────────────────────
def _read_action() -> MenuAction:
"""Map a raw keypress to a menu action.
Delegates terminal I/O to :mod:`key_reader` and applies
choice_menu-specific overrides: Tab → ``"down"``,
right-arrow → ``"enter"``, left-arrow → ``"ignore"``.
"""
key = read_key_windows() if os.name == "nt" else read_key_unix()
if key == "tab":
return "down"
if key == "right":
return "enter"
if key == "left":
return "ignore"
return key # type: ignore[return-value]
def read_menu_action() -> MenuAction:
"""Read one normalized inline-menu action from stdin."""
return _read_action()
# ── rendering helpers ────────────────────────────────────────────────────────
def _cols() -> int:
return max(40, shutil.get_terminal_size(fallback=(80, 24)).columns)
def menu_columns() -> int:
"""Return the current terminal width floor used by inline menus."""
return _cols()
def _rule(width: int) -> str:
return "─" * width
def _pad(sym: str, label: str, width: int) -> str:
content = f" {sym} {label}"
pad = width - len(content)
return content + (" " * pad if pad > 0 else "")
def _menu_height(crumb: str, labels: list[str]) -> int:
# leading, title, [crumb], rule, blank, choices, blank, hint
return _MENU_LEADING_LINES + 1 + (1 if crumb else 0) + 1 + 1 + len(labels) + 1 + 1
def write_menu_line(text: str = "") -> None:
"""Write one inline-menu line at column zero even while the terminal is in raw mode."""
if text:
sys.stdout.write(f"\r{text}{_TERMINAL_NEWLINE}")
return
sys.stdout.write(_TERMINAL_NEWLINE)
def _erase_menu_block(height: int) -> None:
if height:
sys.stdout.write(f"\r\x1b[{height}A\r\x1b[J")
reset_tty_column()
def reset_tty_column() -> None:
"""Return the cursor to column zero after inline menu I/O.
Menu rows are padded to the terminal width, so the cursor often ends on a
high column. Rich output that follows must start at column zero or tables
render as a diagonal block of leading whitespace.
"""
sys.stdout.write("\r")
sys.stdout.flush()
def erase_menu_lines(height: int) -> None:
"""Erase a previously-rendered inline menu block."""
_erase_menu_block(height)
def _draw_menu(
*,
title: str,
crumb: str,
labels: list[str],
index: int,
erase_lines: int,
) -> None:
out = sys.stdout
w = _cols()
if erase_lines:
_erase_menu_block(erase_lines)
for _ in range(_MENU_LEADING_LINES):
write_menu_line()
# title
write_menu_line(f"{ui_theme.PROMPT_ACCENT_ANSI}{title}{ui_theme.ANSI_RESET}")
# breadcrumb path
if crumb:
write_menu_line(f"{ui_theme.DIM_COUNTER_ANSI}{crumb}{ui_theme.ANSI_RESET}")
# separator below header
write_menu_line(f"{ui_theme.DIM_COUNTER_ANSI}{_rule(w)}{ui_theme.ANSI_RESET}")
write_menu_line()
# choices
for i, label in enumerate(labels):
here = i == index
sym = ">" if here else " "
padded = _pad(sym, label, w)
if here:
write_menu_line(f"{ui_theme.MENU_SELECTION_ROW_ANSI}{padded}{ui_theme.ANSI_RESET}")
else:
write_menu_line(f"{ui_theme.DIM_COUNTER_ANSI}{padded}{ui_theme.ANSI_RESET}")
write_menu_line()
write_menu_line(f"{ui_theme.DIM_COUNTER_ANSI}{_HINT}{ui_theme.ANSI_RESET}")
out.flush()
def _erase_menu(crumb: str, labels: list[str]) -> None:
"""Move cursor up to the start of this menu block and wipe it."""
height = _menu_height(crumb, labels)
_erase_menu_block(height)
sys.stdout.flush()
# ── picker loop ──────────────────────────────────────────────────────────────
def _pick(
*,
title: str,
crumb: str,
labels: list[str],
initial_index: int = 0,
) -> int | None:
"""Draw an inline menu, let user navigate, erase on exit. Returns index or None."""
if not labels:
return None
idx = initial_index % len(labels)
height = _menu_height(crumb, labels)
first = True
while True:
_draw_menu(
title=title,
crumb=crumb,
labels=labels,
index=idx,
erase_lines=0 if first else height,
)
first = False
action = _read_action()
if action == "enter":
_erase_menu(crumb, labels)
return idx
if action in ("cancel", "eof"):
_erase_menu(crumb, labels)
return None
if action == "ignore":
continue
if action == "up":
idx = (idx - 1) % len(labels)
elif action == "down":
idx = (idx + 1) % len(labels)
# ── public API ───────────────────────────────────────────────────────────────
def repl_choose_one(
*,
title: str,
choices: list[tuple[str, str]],
breadcrumb: str = "",
initial_value: str | None = None,
) -> str | None:
"""Show an inline erasing arrow-key menu; return selected value or None on Esc.
``breadcrumb`` is a slash-separated path shown dimly below the title, e.g.
``/model set``. Only call when :func:`repl_tty_interactive` is True.
"""
from surfaces.interactive_shell.ui.components.cpr_stdin import drain_stale_cpr_bytes
if not choices or not repl_tty_interactive():
return None
drain_stale_cpr_bytes()
crumb = breadcrumb
labels = [label for _value, label in choices]
initial_index = 0
if initial_value is not None:
for index, (value, _label) in enumerate(choices):
if value == initial_value:
initial_index = index
break
picked = _pick(title=title, crumb=crumb, labels=labels, initial_index=initial_index)
if picked is None:
return None
value = choices[picked][0]
return value if isinstance(value, str) else None
def print_valid_choice_list(
console: Console,
*,
title: str,
choices: list[str],
) -> None:
"""Print one choice per line for scan-friendly fallback/error messaging."""
if not choices:
return
console.print(f"[{ui_theme.SECONDARY}]{title}[/]")
for choice in choices:
console.print(f"[{ui_theme.SECONDARY}] - {escape(choice)}[/]")
__all__ = [
"CRUMB_SEP",
"erase_menu_lines",
"menu_columns",
"print_valid_choice_list",
"read_menu_action",
"repl_choose_one",
"ensure_tty_column_zero",
"prepare_repl_output_line",
"repl_section_break",
"repl_tty_interactive",
"reset_tty_column",
"write_menu_line",
]