mirror of
https://github.com/insistence/RuoYi-Vue3-FastAPI.git
synced 2026-09-21 12:52:27 +00:00
feat: 新增cli系统 (#103)
* feat: 新增cli系统 * feat: ruoyi completion支持powershell * perf: 优化tui显示 * fix: 修复前端构建异常
This commit is contained in:
insistence
@@ -0,0 +1,155 @@
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
import typer
|
||||
|
||||
from cli.completion.controller import CompletionCommandController
|
||||
from cli.completion.providers import COMPLETION_PROVIDER_GATEWAY, CompletionProviderGateway
|
||||
from cli.context import OutputOption
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CompletionCommandRegistration:
|
||||
"""
|
||||
completion 子命令注册描述。
|
||||
|
||||
:param name: 子命令名称
|
||||
:param help_text: 子命令帮助文本
|
||||
"""
|
||||
|
||||
name: str
|
||||
help_text: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class CompletionSubcommandRegistrar:
|
||||
"""
|
||||
completion 子命令注册器。
|
||||
|
||||
该对象负责将 `show/install/doctor` 子命令挂载到 completion 应用，
|
||||
将 Typer 命令声明细节从构建器主体中拆出。
|
||||
|
||||
:param completion_provider_gateway: completion 提供器对外网关
|
||||
"""
|
||||
|
||||
completion_provider_gateway: CompletionProviderGateway = field(default_factory=lambda: COMPLETION_PROVIDER_GATEWAY)
|
||||
|
||||
@staticmethod
|
||||
def build_registrations() -> tuple[CompletionCommandRegistration, ...]:
|
||||
"""
|
||||
返回 completion 子命令注册描述列表。
|
||||
|
||||
:return: 子命令注册描述元组
|
||||
"""
|
||||
return (
|
||||
CompletionCommandRegistration(name='show', help_text='输出指定 shell 的 completion 脚本'),
|
||||
CompletionCommandRegistration(name='install', help_text='安装指定 shell 的 completion 脚本'),
|
||||
CompletionCommandRegistration(name='doctor', help_text='检查当前 completion 配置状态'),
|
||||
)
|
||||
|
||||
def register(self, app: typer.Typer, *, controller: CompletionCommandController) -> None:
|
||||
"""
|
||||
向 completion 应用注册全部子命令。
|
||||
|
||||
:param app: completion 子应用
|
||||
:param controller: completion 命令控制器
|
||||
:return: None
|
||||
"""
|
||||
registrations = {registration.name: registration for registration in self.build_registrations()}
|
||||
|
||||
@app.command(registrations['show'].name, help=registrations['show'].help_text)
|
||||
def show(
|
||||
shell: str = typer.Argument(
|
||||
...,
|
||||
help='shell 名称，如 bash、zsh、fish',
|
||||
autocompletion=self.completion_provider_gateway.complete_shell_names,
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
输出指定 shell 的 completion 脚本。
|
||||
|
||||
:param shell: shell 名称
|
||||
:return: None
|
||||
"""
|
||||
controller.show(shell)
|
||||
|
||||
@app.command(registrations['install'].name, help=registrations['install'].help_text)
|
||||
def install(
|
||||
shell: str | None = typer.Option(
|
||||
None,
|
||||
'--shell',
|
||||
help='shell 名称，默认自动识别当前 shell',
|
||||
autocompletion=self.completion_provider_gateway.complete_shell_names,
|
||||
),
|
||||
output: OutputOption = 'text',
|
||||
target_file: Path | None = typer.Option(None, '--target-file', help='completion 脚本目标文件路径'),
|
||||
activate: bool = typer.Option(False, '--activate', help='将激活命令写入 shell rc 文件'),
|
||||
rc_file: Path | None = typer.Option(None, '--rc-file', help='自定义 shell rc 文件路径'),
|
||||
force: bool = typer.Option(False, '--force', help='覆盖已存在且内容不同的目标文件'),
|
||||
) -> None:
|
||||
"""
|
||||
安装指定 shell 的 completion 脚本。
|
||||
|
||||
:param shell: shell 名称，默认自动识别当前 shell
|
||||
:param output: 输出格式
|
||||
:param target_file: completion 脚本目标文件路径
|
||||
:param activate: 是否写入 shell rc 文件激活命令
|
||||
:param rc_file: 自定义 shell rc 文件路径
|
||||
:param force: 是否强制覆盖已存在文件
|
||||
:return: None
|
||||
"""
|
||||
controller.install(
|
||||
output,
|
||||
shell=shell,
|
||||
target_file=target_file,
|
||||
activate=activate,
|
||||
rc_file=rc_file,
|
||||
force=force,
|
||||
)
|
||||
|
||||
@app.command(registrations['doctor'].name, help=registrations['doctor'].help_text)
|
||||
def doctor(
|
||||
output: OutputOption = 'text',
|
||||
) -> None:
|
||||
"""
|
||||
检查当前 completion 配置状态。
|
||||
|
||||
:param output: 输出格式
|
||||
:return: None
|
||||
"""
|
||||
controller.doctor(output)
|
||||
|
||||
|
||||
@dataclass
|
||||
class CompletionCommandBuilder:
|
||||
"""
|
||||
completion 子应用构建器。
|
||||
|
||||
该构建器负责装配 `completion` 子应用，并将控制器实例和子命令注册
|
||||
细节收口到类式对象协作中。
|
||||
|
||||
:param completion_subcommand_registrar: completion 子命令注册器
|
||||
"""
|
||||
|
||||
completion_subcommand_registrar: CompletionSubcommandRegistrar = field(
|
||||
default_factory=CompletionSubcommandRegistrar
|
||||
)
|
||||
|
||||
def build(self, root_cli: typer.Typer) -> typer.Typer:
|
||||
"""
|
||||
构建 completion 命令组。
|
||||
|
||||
:param root_cli: 根 Typer 应用
|
||||
:return: completion 子应用
|
||||
"""
|
||||
app = typer.Typer(
|
||||
help='shell completion 相关命令',
|
||||
no_args_is_help=True,
|
||||
context_settings={'help_option_names': ['-h', '--help']},
|
||||
)
|
||||
completion_command_controller = CompletionCommandController(root_cli)
|
||||
self.completion_subcommand_registrar.register(app, controller=completion_command_controller)
|
||||
return app
|
||||
|
||||
|
||||
COMPLETION_COMMAND_BUILDER = CompletionCommandBuilder()
|
||||
@@ -0,0 +1,116 @@
|
||||
from pathlib import Path
|
||||
|
||||
import typer
|
||||
|
||||
from cli.completion.doctor import COMPLETION_DOCTOR, CompletionDoctorService
|
||||
from cli.completion.installers import COMPLETION_INSTALLER, CompletionInstallerService
|
||||
from cli.completion.presenter import CompletionCommandPresenter
|
||||
from cli.core import (
|
||||
DEFAULT_CORE_SERVICES,
|
||||
CliContextFactory,
|
||||
CliExecutionService,
|
||||
)
|
||||
|
||||
|
||||
class CompletionCommandController:
|
||||
"""
|
||||
completion 命令控制器。
|
||||
|
||||
该控制器负责组织 `completion` 命令组的上下文准备、payload 生成、
|
||||
文本渲染与脚本输出收口。
|
||||
|
||||
:param root_cli: 根 Typer 应用
|
||||
:param context_factory: CLI 上下文工厂
|
||||
:param execution_service: CLI 执行服务
|
||||
:param presenter: completion 命令文本渲染器
|
||||
:param installer_service: completion 安装与脚本生成服务
|
||||
:param doctor_service: completion 诊断服务
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
root_cli: typer.Typer,
|
||||
*,
|
||||
context_factory: CliContextFactory | None = None,
|
||||
execution_service: CliExecutionService | None = None,
|
||||
presenter: CompletionCommandPresenter | None = None,
|
||||
installer_service: CompletionInstallerService | None = None,
|
||||
doctor_service: CompletionDoctorService | None = None,
|
||||
) -> None:
|
||||
"""
|
||||
初始化 completion 命令控制器。
|
||||
|
||||
:param root_cli: 根 Typer 应用
|
||||
:param context_factory: CLI 上下文工厂
|
||||
:param execution_service: CLI 执行服务
|
||||
:param presenter: completion 命令文本渲染器
|
||||
:param installer_service: completion 安装与脚本生成服务
|
||||
:param doctor_service: completion 诊断服务
|
||||
:return: None
|
||||
"""
|
||||
self.root_cli = root_cli
|
||||
self.context_factory = context_factory or DEFAULT_CORE_SERVICES.context_factory
|
||||
self.execution_service = execution_service or DEFAULT_CORE_SERVICES.execution_service
|
||||
self.presenter = presenter or CompletionCommandPresenter()
|
||||
self.installer_service = installer_service or COMPLETION_INSTALLER
|
||||
self.doctor_service = doctor_service or COMPLETION_DOCTOR
|
||||
|
||||
def show(self, shell: str) -> None:
|
||||
"""
|
||||
输出指定 shell 的 completion 脚本。
|
||||
|
||||
:param shell: shell 名称
|
||||
:return: None
|
||||
"""
|
||||
typer.echo(self.installer_service.render_completion_script(self.root_cli, shell), nl=False)
|
||||
|
||||
def install(
|
||||
self,
|
||||
output: str,
|
||||
*,
|
||||
shell: str | None,
|
||||
target_file: Path | None,
|
||||
activate: bool,
|
||||
rc_file: Path | None,
|
||||
force: bool,
|
||||
) -> None:
|
||||
"""
|
||||
安装指定 shell 的 completion 脚本。
|
||||
|
||||
:param output: 输出格式
|
||||
:param shell: shell 名称
|
||||
:param target_file: completion 脚本目标文件路径
|
||||
:param activate: 是否写入 shell rc 文件激活命令
|
||||
:param rc_file: 自定义 shell rc 文件路径
|
||||
:param force: 是否强制覆盖已存在文件
|
||||
:return: None
|
||||
"""
|
||||
ctx = self.context_factory.build_readonly('dev', output)
|
||||
payload = self.installer_service.install_completion_script(
|
||||
self.root_cli,
|
||||
shell,
|
||||
target_file=target_file,
|
||||
activate=activate,
|
||||
rc_file=rc_file,
|
||||
force=force,
|
||||
)
|
||||
self.execution_service.complete_payload_result(
|
||||
ctx,
|
||||
payload,
|
||||
text_builder=self.presenter.build_completion_install_text,
|
||||
)
|
||||
|
||||
def doctor(self, output: str) -> None:
|
||||
"""
|
||||
检查当前 completion 配置状态。
|
||||
|
||||
:param output: 输出格式
|
||||
:return: None
|
||||
"""
|
||||
ctx = self.context_factory.build_readonly('dev', output)
|
||||
payload = self.doctor_service.build_completion_doctor_payload()
|
||||
self.execution_service.complete_payload_result(
|
||||
ctx,
|
||||
payload,
|
||||
text_builder=self.presenter.build_completion_doctor_text,
|
||||
)
|
||||
@@ -0,0 +1,95 @@
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from cli.completion.installers import (
|
||||
CLICK_COMPLETE_ENV_VAR,
|
||||
COMPLETION_INSTALLER,
|
||||
CompletionInstallerService,
|
||||
)
|
||||
from cli.metadata import (
|
||||
COMPLETION_SHELL_SPEC_REGISTRY,
|
||||
ENVIRONMENT_OPTION_SERVICE,
|
||||
CompletionShellSpecRegistry,
|
||||
EnvironmentOptionService,
|
||||
)
|
||||
|
||||
|
||||
class CompletionDoctorService:
|
||||
"""
|
||||
completion 诊断服务。
|
||||
|
||||
该服务负责汇总活跃 shell、目标脚本路径、rc 文件状态、推荐安装命令
|
||||
与环境候选列表，生成 `completion doctor` 所需的结构化结果。
|
||||
|
||||
:param installer_service: completion 安装服务
|
||||
:param shell_spec_registry: shell 元数据注册表
|
||||
:param environment_option_service: 环境选项服务
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
installer_service: CompletionInstallerService | None = None,
|
||||
shell_spec_registry: CompletionShellSpecRegistry | None = None,
|
||||
environment_option_service: EnvironmentOptionService | None = None,
|
||||
) -> None:
|
||||
"""
|
||||
初始化 completion 诊断服务。
|
||||
|
||||
:param installer_service: completion 安装服务
|
||||
:param shell_spec_registry: shell 元数据注册表
|
||||
:param environment_option_service: 环境选项服务
|
||||
:return: None
|
||||
"""
|
||||
self.installer_service = installer_service or COMPLETION_INSTALLER
|
||||
self.shell_spec_registry = shell_spec_registry or COMPLETION_SHELL_SPEC_REGISTRY
|
||||
self.environment_option_service = environment_option_service or ENVIRONMENT_OPTION_SERVICE
|
||||
|
||||
def build_completion_doctor_payload(self) -> dict[str, Any]:
|
||||
"""
|
||||
构建 completion 诊断结果。
|
||||
|
||||
:return: 诊断结果字典
|
||||
"""
|
||||
active_shell = self.installer_service.detect_active_shell()
|
||||
shells: dict[str, dict[str, Any]] = {}
|
||||
for shell_name, shell_spec in self.shell_spec_registry.specs.items():
|
||||
target_file = self.installer_service.resolve_completion_target(shell_name)
|
||||
rc_file = self.installer_service.resolve_completion_rc_file(shell_name)
|
||||
source_command = None
|
||||
if shell_spec.supported and shell_spec.generator in {'click', 'custom'}:
|
||||
source_command = self.installer_service.build_source_command(target_file, shell_name)
|
||||
recommended_install_command = f'ruoyi completion install --shell={shell_name}'
|
||||
if shell_spec.supported and not shell_spec.auto_discovery:
|
||||
recommended_install_command = f'{recommended_install_command} --activate'
|
||||
shells[shell_name] = {
|
||||
'supported': shell_spec.supported,
|
||||
'detected': shell_name == active_shell,
|
||||
'description': shell_spec.description,
|
||||
'targetFile': str(target_file),
|
||||
'targetFileExists': target_file.exists(),
|
||||
'rcFile': str(rc_file) if rc_file is not None else None,
|
||||
'rcFileExists': rc_file.exists() if rc_file is not None else None,
|
||||
'autoDiscovery': shell_spec.auto_discovery,
|
||||
'sourceCommand': source_command,
|
||||
'recommendedInstallCommand': recommended_install_command,
|
||||
}
|
||||
|
||||
recommended_shell = active_shell if self.shell_spec_registry.get_spec(active_shell) is not None else None
|
||||
recommended_install_command = None
|
||||
if recommended_shell is not None:
|
||||
recommended_install_command = shells[recommended_shell]['recommendedInstallCommand']
|
||||
|
||||
return {
|
||||
'ok': True,
|
||||
'message': 'completion 诊断信息已生成',
|
||||
'activeShell': active_shell or None,
|
||||
'projectDir': str(Path.cwd().resolve()),
|
||||
'envChoices': self.environment_option_service.discover_env_names(),
|
||||
'completeEnvVar': CLICK_COMPLETE_ENV_VAR,
|
||||
'recommendedInstallCommand': recommended_install_command,
|
||||
'shells': shells,
|
||||
}
|
||||
|
||||
|
||||
COMPLETION_DOCTOR = CompletionDoctorService()
|
||||
@@ -0,0 +1,465 @@
|
||||
import io
|
||||
import os
|
||||
from contextlib import redirect_stderr
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import click
|
||||
import typer
|
||||
import typer.main
|
||||
from click.shell_completion import BashComplete, FishComplete, ZshComplete
|
||||
|
||||
from cli.completion.providers import COMPLETION_PROVIDER_GATEWAY, CompletionProviderGateway
|
||||
from cli.completion.shells import PowerShellComplete, ensure_custom_completion_classes_registered
|
||||
from cli.exit_codes import ARGUMENT_ERROR, RUNTIME_ERROR
|
||||
from cli.metadata import COMPLETION_SHELL_SPEC_REGISTRY, CompletionShellSpec, CompletionShellSpecRegistry
|
||||
|
||||
CLICK_COMPLETE_ENV_VAR = '_RUOYI_COMPLETE'
|
||||
|
||||
|
||||
class CompletionInstallerShellSupport:
|
||||
"""
|
||||
completion shell 差异支持对象。
|
||||
|
||||
该对象负责封装不同 shell 在脚本兼容处理和 source 命令构建上的差异，
|
||||
供安装服务通过轻量 strategy 协作对象统一复用。
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def keep_script_text(script_text: str) -> str:
|
||||
"""
|
||||
保持脚本文本原样返回。
|
||||
|
||||
:param script_text: 原始脚本文本
|
||||
:return: 原样脚本文本
|
||||
"""
|
||||
return script_text
|
||||
|
||||
@staticmethod
|
||||
def make_bash_completion_script_compatible(script_text: str) -> str:
|
||||
"""
|
||||
对 Click 生成的 Bash completion 脚本做旧版本兼容处理。
|
||||
|
||||
兼容目标主要包括 macOS 默认 Bash 3.2：
|
||||
|
||||
- `complete -o nosort` 在旧 Bash 上会在 `source` 阶段直接报错
|
||||
- `compopt` 在旧 Bash 上可能不存在
|
||||
|
||||
:param script_text: Click 原始 Bash completion 脚本文本
|
||||
:return: 兼容处理后的脚本文本
|
||||
"""
|
||||
compatible_lines: list[str] = []
|
||||
for line in script_text.splitlines():
|
||||
stripped_line = line.strip()
|
||||
if stripped_line == 'compopt -o dirnames':
|
||||
indent = line[: len(line) - len(line.lstrip(' '))]
|
||||
compatible_lines.extend(
|
||||
[
|
||||
f'{indent}if command -v compopt >/dev/null 2>&1; then',
|
||||
f'{indent} compopt -o dirnames',
|
||||
f'{indent}fi',
|
||||
]
|
||||
)
|
||||
continue
|
||||
if stripped_line == 'compopt -o default':
|
||||
indent = line[: len(line) - len(line.lstrip(' '))]
|
||||
compatible_lines.extend(
|
||||
[
|
||||
f'{indent}if command -v compopt >/dev/null 2>&1; then',
|
||||
f'{indent} compopt -o default',
|
||||
f'{indent}fi',
|
||||
]
|
||||
)
|
||||
continue
|
||||
if stripped_line.startswith('complete -o nosort -F '):
|
||||
compatibility_line = stripped_line.replace('complete -o nosort -F ', '', 1)
|
||||
function_name, command_name = compatibility_line.split(' ', 1)
|
||||
compatible_lines.extend(
|
||||
[
|
||||
f'if complete -o nosort -F {function_name} {command_name} 2>/dev/null; then',
|
||||
' :',
|
||||
'else',
|
||||
f' complete -F {function_name} {command_name}',
|
||||
'fi',
|
||||
]
|
||||
)
|
||||
continue
|
||||
compatible_lines.append(line)
|
||||
return '\n'.join(compatible_lines) + '\n'
|
||||
|
||||
@staticmethod
|
||||
def build_posix_source_command(target_file: Path) -> str:
|
||||
"""
|
||||
构建 POSIX shell 使用的 source 命令。
|
||||
|
||||
:param target_file: completion 脚本文件路径
|
||||
:return: source 命令
|
||||
"""
|
||||
return f'source {target_file}'
|
||||
|
||||
@staticmethod
|
||||
def build_fish_source_command(target_file: Path) -> str:
|
||||
"""
|
||||
构建 Fish shell 使用的 source 命令。
|
||||
|
||||
:param target_file: completion 脚本文件路径
|
||||
:return: source 命令
|
||||
"""
|
||||
return f'status --is-interactive; and source {target_file}'
|
||||
|
||||
@staticmethod
|
||||
def build_powershell_source_command(target_file: Path) -> str:
|
||||
"""
|
||||
构建 PowerShell 使用的 source 命令。
|
||||
|
||||
:param target_file: completion 脚本文件路径
|
||||
:return: source 命令
|
||||
"""
|
||||
return f'. "{target_file}"'
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CompletionShellRuntimePolicy:
|
||||
"""
|
||||
completion shell 运行时策略定义。
|
||||
|
||||
:param name: shell 名称
|
||||
:param click_completion_class: Click completion 生成器类型
|
||||
:param script_transformer: 脚本文本后处理函数
|
||||
:param source_command_builder: 激活命令构建函数
|
||||
"""
|
||||
|
||||
name: str
|
||||
click_completion_class: type[Any]
|
||||
script_transformer: Any
|
||||
source_command_builder: Any
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CompletionShellRuntimePolicyRegistry:
|
||||
"""
|
||||
completion shell 运行时策略注册表。
|
||||
|
||||
:param policies: 按 shell 名称索引的运行时策略映射
|
||||
"""
|
||||
|
||||
policies: dict[str, CompletionShellRuntimePolicy]
|
||||
|
||||
def get(self, shell_name: str) -> CompletionShellRuntimePolicy | None:
|
||||
"""
|
||||
获取指定 shell 的运行时策略。
|
||||
|
||||
:param shell_name: shell 名称
|
||||
:return: 运行时策略，不存在时返回 None
|
||||
"""
|
||||
return self.policies.get(shell_name)
|
||||
|
||||
|
||||
DEFAULT_COMPLETION_SHELL_RUNTIME_POLICIES = CompletionShellRuntimePolicyRegistry(
|
||||
policies={
|
||||
'bash': CompletionShellRuntimePolicy(
|
||||
name='bash',
|
||||
click_completion_class=BashComplete,
|
||||
script_transformer=CompletionInstallerShellSupport.make_bash_completion_script_compatible,
|
||||
source_command_builder=CompletionInstallerShellSupport.build_posix_source_command,
|
||||
),
|
||||
'zsh': CompletionShellRuntimePolicy(
|
||||
name='zsh',
|
||||
click_completion_class=ZshComplete,
|
||||
script_transformer=CompletionInstallerShellSupport.keep_script_text,
|
||||
source_command_builder=CompletionInstallerShellSupport.build_posix_source_command,
|
||||
),
|
||||
'fish': CompletionShellRuntimePolicy(
|
||||
name='fish',
|
||||
click_completion_class=FishComplete,
|
||||
script_transformer=CompletionInstallerShellSupport.keep_script_text,
|
||||
source_command_builder=CompletionInstallerShellSupport.build_fish_source_command,
|
||||
),
|
||||
'powershell': CompletionShellRuntimePolicy(
|
||||
name='powershell',
|
||||
click_completion_class=PowerShellComplete,
|
||||
script_transformer=CompletionInstallerShellSupport.keep_script_text,
|
||||
source_command_builder=CompletionInstallerShellSupport.build_powershell_source_command,
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class CompletionInstallerService:
|
||||
"""
|
||||
completion 安装与脚本生成服务。
|
||||
|
||||
该服务负责 shell 元数据解析、脚本生成、目标路径解析、激活命令构建、
|
||||
rc 文件写入以及安装结果收口。
|
||||
|
||||
:param completion_provider_gateway: completion 提供器对外网关
|
||||
:param shell_spec_registry: shell 元数据注册表
|
||||
:param shell_runtime_policy_registry: shell 运行时策略注册表
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
completion_provider_gateway: CompletionProviderGateway | None = None,
|
||||
shell_spec_registry: CompletionShellSpecRegistry | None = None,
|
||||
shell_runtime_policy_registry: CompletionShellRuntimePolicyRegistry | None = None,
|
||||
) -> None:
|
||||
"""
|
||||
初始化 completion 安装服务。
|
||||
|
||||
:param completion_provider_gateway: completion 提供器对外网关
|
||||
:param shell_spec_registry: shell 元数据注册表
|
||||
:param shell_runtime_policy_registry: shell 运行时策略注册表
|
||||
:return: None
|
||||
"""
|
||||
self.completion_provider_gateway = completion_provider_gateway or COMPLETION_PROVIDER_GATEWAY
|
||||
self.shell_spec_registry = shell_spec_registry or COMPLETION_SHELL_SPEC_REGISTRY
|
||||
self.shell_runtime_policy_registry = shell_runtime_policy_registry or DEFAULT_COMPLETION_SHELL_RUNTIME_POLICIES
|
||||
ensure_custom_completion_classes_registered()
|
||||
|
||||
def resolve_completion_shell_spec(self, shell: str) -> CompletionShellSpec:
|
||||
"""
|
||||
获取指定 shell 的 completion 元数据。
|
||||
|
||||
:param shell: shell 名称
|
||||
:return: shell 元数据
|
||||
:raises typer.BadParameter: shell 不存在时抛出异常
|
||||
"""
|
||||
normalized_shell = shell.strip().lower()
|
||||
shell_spec = self.shell_spec_registry.get_spec(normalized_shell)
|
||||
if shell_spec is None:
|
||||
supported_shells = ', '.join(self.completion_provider_gateway.list_completion_shells())
|
||||
raise typer.BadParameter(f'不支持的 shell：{shell}，可选值为 {supported_shells}')
|
||||
return shell_spec
|
||||
|
||||
def resolve_shell_runtime_policy(self, shell: str) -> CompletionShellRuntimePolicy:
|
||||
"""
|
||||
获取指定 shell 的运行时策略。
|
||||
|
||||
:param shell: shell 名称
|
||||
:return: shell 运行时策略
|
||||
:raises typer.BadParameter: 当前 shell 未实现运行时策略时抛出异常
|
||||
"""
|
||||
shell_spec = self.resolve_completion_shell_spec(shell)
|
||||
runtime_policy = self.shell_runtime_policy_registry.get(shell_spec.name)
|
||||
if runtime_policy is None:
|
||||
raise typer.BadParameter(f'{shell_spec.name} completion 当前版本未实现')
|
||||
return runtime_policy
|
||||
|
||||
@staticmethod
|
||||
def build_completion_click_command(root_cli: typer.Typer) -> click.Command:
|
||||
"""
|
||||
将 Typer 根应用转换为 Click 命令对象，供 shell completion 生成使用。
|
||||
|
||||
:param root_cli: Typer 根应用
|
||||
:return: Click 命令对象
|
||||
"""
|
||||
return typer.main.get_command(root_cli)
|
||||
|
||||
def render_completion_script(self, root_cli: typer.Typer, shell: str) -> str:
|
||||
"""
|
||||
生成指定 shell 的 completion 脚本文本。
|
||||
|
||||
:param root_cli: Typer 根应用
|
||||
:param shell: shell 名称
|
||||
:return: completion 脚本文本
|
||||
:raises typer.BadParameter: 当前 shell 未实现脚本生成时抛出异常
|
||||
"""
|
||||
shell_spec = self.resolve_completion_shell_spec(shell)
|
||||
if not shell_spec.supported or shell_spec.generator not in {'click', 'custom'}:
|
||||
raise typer.BadParameter(f'{shell_spec.name} completion 当前版本未实现')
|
||||
|
||||
runtime_policy = self.resolve_shell_runtime_policy(shell_spec.name)
|
||||
click_command = self.build_completion_click_command(root_cli)
|
||||
stderr_buffer = io.StringIO()
|
||||
with redirect_stderr(stderr_buffer):
|
||||
script_text = runtime_policy.click_completion_class(
|
||||
click_command,
|
||||
{},
|
||||
'ruoyi',
|
||||
CLICK_COMPLETE_ENV_VAR,
|
||||
).source()
|
||||
return runtime_policy.script_transformer(script_text)
|
||||
|
||||
def resolve_completion_target(self, shell: str, target_file: Path | None = None) -> Path:
|
||||
"""
|
||||
解析 completion 脚本目标文件路径。
|
||||
|
||||
:param shell: shell 名称
|
||||
:param target_file: 用户显式指定的目标文件
|
||||
:return: 目标文件绝对路径
|
||||
"""
|
||||
if target_file is not None:
|
||||
return target_file.expanduser().resolve()
|
||||
|
||||
shell_spec = self.resolve_completion_shell_spec(shell)
|
||||
return (Path.home() / shell_spec.default_target).expanduser().resolve()
|
||||
|
||||
def resolve_completion_rc_file(self, shell: str, rc_file: Path | None = None) -> Path | None:
|
||||
"""
|
||||
解析 completion 激活所使用的 rc 文件路径。
|
||||
|
||||
:param shell: shell 名称
|
||||
:param rc_file: 用户显式指定的 rc 文件
|
||||
:return: rc 文件绝对路径，若当前 shell 无 rc 文件则返回 None
|
||||
"""
|
||||
if rc_file is not None:
|
||||
return rc_file.expanduser().resolve()
|
||||
|
||||
shell_spec = self.resolve_completion_shell_spec(shell)
|
||||
if shell_spec.default_rc_file is None:
|
||||
return None
|
||||
return (Path.home() / shell_spec.default_rc_file).expanduser().resolve()
|
||||
|
||||
def build_source_command(self, target_file: Path, shell: str) -> str:
|
||||
"""
|
||||
构建当前 shell 对应的激活命令。
|
||||
|
||||
:param target_file: completion 脚本文件路径
|
||||
:param shell: shell 名称
|
||||
:return: 激活命令文本
|
||||
"""
|
||||
runtime_policy = self.resolve_shell_runtime_policy(shell)
|
||||
return runtime_policy.source_command_builder(target_file)
|
||||
|
||||
@staticmethod
|
||||
def detect_active_shell() -> str:
|
||||
"""
|
||||
检测当前进程环境下的活跃 shell 名称。
|
||||
|
||||
:return: shell 名称，未知时返回空字符串
|
||||
"""
|
||||
shell_path = os.environ.get('SHELL', '').strip()
|
||||
if not shell_path:
|
||||
return ''
|
||||
return Path(shell_path).name.lower()
|
||||
|
||||
def resolve_install_shell(self, shell: str | None) -> str:
|
||||
"""
|
||||
解析安装命令实际使用的 shell。
|
||||
|
||||
:param shell: 用户显式指定的 shell，允许为空
|
||||
:return: 实际使用的 shell 名称
|
||||
:raises typer.BadParameter: 无法推断或不支持时抛出异常
|
||||
"""
|
||||
if shell and shell.strip():
|
||||
return self.resolve_completion_shell_spec(shell).name
|
||||
|
||||
active_shell = self.detect_active_shell()
|
||||
if not active_shell:
|
||||
raise typer.BadParameter('未检测到当前 shell，请显式传入 --shell')
|
||||
|
||||
shell_spec = self.shell_spec_registry.get_spec(active_shell)
|
||||
if shell_spec is None:
|
||||
supported_shells = ', '.join(self.completion_provider_gateway.list_completion_shells())
|
||||
raise typer.BadParameter(
|
||||
f'当前 shell `{active_shell}` 不在支持列表中，请显式传入 --shell，可选值为 {supported_shells}'
|
||||
)
|
||||
return shell_spec.name
|
||||
|
||||
@staticmethod
|
||||
def append_activation_line(rc_file: Path, source_command: str) -> bool:
|
||||
"""
|
||||
将激活命令追加到 rc 文件，若已存在则不重复写入。
|
||||
|
||||
:param rc_file: rc 文件路径
|
||||
:param source_command: 激活命令文本
|
||||
:return: 本次是否发生写入
|
||||
"""
|
||||
existing_text = ''
|
||||
if rc_file.exists():
|
||||
existing_text = rc_file.read_text(encoding='utf-8')
|
||||
if source_command in existing_text:
|
||||
return False
|
||||
|
||||
rc_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
with rc_file.open('a', encoding='utf-8') as file_object:
|
||||
if existing_text and not existing_text.endswith('\n'):
|
||||
file_object.write('\n')
|
||||
file_object.write(f'{source_command}\n')
|
||||
return True
|
||||
|
||||
def install_completion_script(
|
||||
self,
|
||||
root_cli: typer.Typer,
|
||||
shell: str | None,
|
||||
*,
|
||||
target_file: Path | None = None,
|
||||
activate: bool = False,
|
||||
rc_file: Path | None = None,
|
||||
force: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
安装指定 shell 的 completion 脚本文件。
|
||||
|
||||
:param root_cli: Typer 根应用
|
||||
:param shell: shell 名称
|
||||
:param target_file: completion 脚本目标文件
|
||||
:param activate: 是否写入 rc 文件激活命令
|
||||
:param rc_file: 自定义 rc 文件路径
|
||||
:param force: 目标文件已存在时是否强制覆盖
|
||||
:return: 安装结果字典
|
||||
"""
|
||||
resolved_shell = self.resolve_install_shell(shell)
|
||||
shell_spec = self.resolve_completion_shell_spec(resolved_shell)
|
||||
if not shell_spec.supported:
|
||||
return {
|
||||
'ok': False,
|
||||
'message': f'{shell_spec.name} completion 当前版本未实现',
|
||||
'shell': shell_spec.name,
|
||||
'exit_code': ARGUMENT_ERROR,
|
||||
}
|
||||
|
||||
resolved_target_file = self.resolve_completion_target(shell_spec.name, target_file)
|
||||
script_text = self.render_completion_script(root_cli, shell_spec.name)
|
||||
existing_text = resolved_target_file.read_text(encoding='utf-8') if resolved_target_file.exists() else None
|
||||
|
||||
if existing_text is not None and existing_text != script_text and not force:
|
||||
return {
|
||||
'ok': False,
|
||||
'message': '目标文件已存在且内容不同，请传入 --force 覆盖',
|
||||
'shell': shell_spec.name,
|
||||
'targetFile': str(resolved_target_file),
|
||||
'exit_code': RUNTIME_ERROR,
|
||||
}
|
||||
|
||||
resolved_target_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
resolved_target_file.write_text(script_text, encoding='utf-8')
|
||||
|
||||
source_command = self.build_source_command(resolved_target_file, shell_spec.name)
|
||||
activated = shell_spec.auto_discovery
|
||||
rc_file_path = self.resolve_completion_rc_file(shell_spec.name, rc_file)
|
||||
rc_file_updated = False
|
||||
activation_required = not shell_spec.auto_discovery
|
||||
detected_shell = self.detect_active_shell() or None
|
||||
|
||||
if activate and rc_file_path is not None:
|
||||
rc_file_updated = self.append_activation_line(rc_file_path, source_command)
|
||||
activated = True
|
||||
|
||||
next_step = '当前 shell 会自动发现 completion 脚本，无需额外 source 命令'
|
||||
if activation_required and not activate:
|
||||
next_step = f'请执行 `{source_command}`，或重新运行并传入 --activate'
|
||||
elif activation_required and activate:
|
||||
next_step = '请重启当前 shell，或手动执行 rc 文件中的 source 命令使其立即生效'
|
||||
|
||||
return {
|
||||
'ok': True,
|
||||
'message': 'completion 脚本已安装',
|
||||
'shell': shell_spec.name,
|
||||
'detectedShell': detected_shell,
|
||||
'targetFile': str(resolved_target_file),
|
||||
'activated': activated,
|
||||
'activateRequested': activate,
|
||||
'rcFile': str(rc_file_path) if rc_file_path is not None else None,
|
||||
'rcFileUpdated': rc_file_updated,
|
||||
'sourceCommand': source_command,
|
||||
'autoDiscovery': shell_spec.auto_discovery,
|
||||
'activationRequired': activation_required,
|
||||
'nextStep': next_step,
|
||||
'completeEnvVar': CLICK_COMPLETE_ENV_VAR,
|
||||
}
|
||||
|
||||
|
||||
COMPLETION_INSTALLER = CompletionInstallerService()
|
||||
@@ -0,0 +1,83 @@
|
||||
from typing import Any
|
||||
|
||||
|
||||
class CompletionCommandPresenter:
|
||||
"""
|
||||
completion 命令文本渲染器。
|
||||
|
||||
该渲染器负责将 `completion` 命令组产生的结构化 payload 转换为稳定的文本摘要，
|
||||
同时保持 JSON 输出仍由控制器直接返回，不在此处做契约变形。
|
||||
"""
|
||||
|
||||
def build_completion_doctor_text(self, payload: dict[str, Any]) -> str:
|
||||
"""
|
||||
将 completion 诊断结果渲染为文本摘要。
|
||||
|
||||
:param payload: completion 诊断结果字典
|
||||
:return: 文本摘要
|
||||
"""
|
||||
lines = [
|
||||
f'ok: {str(payload.get("ok", False)).lower()}',
|
||||
f'message: {payload.get("message", "-")}',
|
||||
f'active_shell: {payload.get("activeShell") or "-"}',
|
||||
f'project_dir: {payload.get("projectDir", "-")}',
|
||||
f'complete_env_var: {payload.get("completeEnvVar", "-")}',
|
||||
]
|
||||
if payload.get('recommendedInstallCommand'):
|
||||
lines.append(f'recommended_install_command: {payload.get("recommendedInstallCommand")}')
|
||||
env_choices = payload.get('envChoices')
|
||||
if isinstance(env_choices, list) and env_choices:
|
||||
lines.append('env_choices:')
|
||||
lines.extend(f' - {env_name}' for env_name in env_choices)
|
||||
shells = payload.get('shells')
|
||||
if isinstance(shells, dict) and shells:
|
||||
lines.append('shells:')
|
||||
for shell_name, shell_payload in shells.items():
|
||||
if not isinstance(shell_payload, dict):
|
||||
continue
|
||||
lines.extend(
|
||||
[
|
||||
f' {shell_name}:',
|
||||
f' supported: {str(shell_payload.get("supported", False)).lower()}',
|
||||
f' detected: {str(shell_payload.get("detected", False)).lower()}',
|
||||
f' target_file: {shell_payload.get("targetFile", "-")}',
|
||||
f' target_file_exists: {str(shell_payload.get("targetFileExists", False)).lower()}',
|
||||
f' rc_file: {shell_payload.get("rcFile", "-") or "-"}',
|
||||
f' rc_file_exists: {str(shell_payload.get("rcFileExists", False)).lower()}',
|
||||
f' auto_discovery: {str(shell_payload.get("autoDiscovery", False)).lower()}',
|
||||
f' source_command: {shell_payload.get("sourceCommand", "-")}',
|
||||
f' recommended_install_command: {shell_payload.get("recommendedInstallCommand", "-")}',
|
||||
]
|
||||
)
|
||||
return '\n'.join(lines)
|
||||
|
||||
@staticmethod
|
||||
def build_completion_install_text(payload: dict[str, Any]) -> str:
|
||||
"""
|
||||
将 completion 安装结果渲染为文本摘要。
|
||||
|
||||
:param payload: completion 安装结果字典
|
||||
:return: 文本摘要
|
||||
"""
|
||||
lines = [
|
||||
f'ok: {str(payload.get("ok", False)).lower()}',
|
||||
f'message: {payload.get("message", "-")}',
|
||||
f'shell: {payload.get("shell", "-")}',
|
||||
]
|
||||
field_label_mapping = {
|
||||
'detectedShell': 'detected_shell',
|
||||
'targetFile': 'target_file',
|
||||
'activated': 'activated',
|
||||
'activateRequested': 'activate_requested',
|
||||
'rcFile': 'rc_file',
|
||||
'rcFileUpdated': 'rc_file_updated',
|
||||
'sourceCommand': 'source_command',
|
||||
'autoDiscovery': 'auto_discovery',
|
||||
'activationRequired': 'activation_required',
|
||||
'nextStep': 'next_step',
|
||||
'completeEnvVar': 'complete_env_var',
|
||||
}
|
||||
for field_name, field_label in field_label_mapping.items():
|
||||
if field_name in payload:
|
||||
lines.append(f'{field_label}: {payload.get(field_name)}')
|
||||
return '\n'.join(lines)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,113 @@
|
||||
import os
|
||||
|
||||
from click.shell_completion import (
|
||||
CompletionItem,
|
||||
ShellComplete,
|
||||
add_completion_class,
|
||||
get_completion_class,
|
||||
split_arg_string,
|
||||
)
|
||||
|
||||
_SOURCE_POWERSHELL = """\
|
||||
$%(complete_func)s = {
|
||||
param($wordToComplete, $commandAst, $cursorPosition)
|
||||
|
||||
$previousCompWords = $env:COMP_WORDS
|
||||
$previousCompCword = $env:COMP_CWORD
|
||||
$previousCompleteInstruction = $env:%(complete_var)s
|
||||
|
||||
$commandLine = $commandAst.ToString()
|
||||
if ($cursorPosition -lt $commandLine.Length) {
|
||||
$commandLine = $commandLine.Substring(0, $cursorPosition)
|
||||
}
|
||||
|
||||
$env:COMP_WORDS = $commandLine
|
||||
$env:COMP_CWORD = $wordToComplete
|
||||
$env:%(complete_var)s = "powershell_complete"
|
||||
|
||||
try {
|
||||
%(prog_name)s | ForEach-Object {
|
||||
$line = $_.ToString()
|
||||
if ([string]::IsNullOrWhiteSpace($line)) {
|
||||
return
|
||||
}
|
||||
$parts = $line -split "`t", 3
|
||||
$completionValue = if ($parts.Length -ge 2) { $parts[1] } else { "" }
|
||||
$completionHelp = if ($parts.Length -ge 3 -and $parts[2]) { $parts[2] } else { $completionValue }
|
||||
[System.Management.Automation.CompletionResult]::new(
|
||||
$completionValue,
|
||||
$completionValue,
|
||||
[System.Management.Automation.CompletionResultType]::ParameterValue,
|
||||
$completionHelp
|
||||
)
|
||||
}
|
||||
} finally {
|
||||
if ($null -ne $previousCompWords) {
|
||||
$env:COMP_WORDS = $previousCompWords
|
||||
} else {
|
||||
Remove-Item Env:\\COMP_WORDS -ErrorAction SilentlyContinue
|
||||
}
|
||||
if ($null -ne $previousCompCword) {
|
||||
$env:COMP_CWORD = $previousCompCword
|
||||
} else {
|
||||
Remove-Item Env:\\COMP_CWORD -ErrorAction SilentlyContinue
|
||||
}
|
||||
if ($null -ne $previousCompleteInstruction) {
|
||||
$env:%(complete_var)s = $previousCompleteInstruction
|
||||
} else {
|
||||
Remove-Item Env:\\%(complete_var)s -ErrorAction SilentlyContinue
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Register-ArgumentCompleter -Native -CommandName %(prog_name)s -ScriptBlock $%(complete_func)s
|
||||
"""
|
||||
|
||||
|
||||
class PowerShellComplete(ShellComplete):
|
||||
"""
|
||||
PowerShell shell completion 支持。
|
||||
|
||||
基于 PowerShell `Register-ArgumentCompleter -Native` 协议，将当前
|
||||
命令行和光标位置传回 Click completion 分发器，再把返回的候选项转换为
|
||||
`CompletionResult` 对象。
|
||||
"""
|
||||
|
||||
name = 'powershell'
|
||||
source_template = _SOURCE_POWERSHELL
|
||||
|
||||
def get_completion_args(self) -> tuple[list[str], str]:
|
||||
"""
|
||||
从 PowerShell 注入的环境变量中恢复 CLI 上下文。
|
||||
|
||||
:return: 已解析的完整参数与当前不完整输入
|
||||
"""
|
||||
cwords = split_arg_string(os.environ.get('COMP_WORDS', ''))
|
||||
incomplete = os.environ.get('COMP_CWORD', '')
|
||||
if incomplete:
|
||||
incomplete_parts = split_arg_string(incomplete)
|
||||
incomplete = incomplete_parts[0] if incomplete_parts else incomplete
|
||||
args = cwords[1:]
|
||||
if incomplete and args and args[-1] == incomplete:
|
||||
args.pop()
|
||||
return args, incomplete
|
||||
|
||||
def format_completion(self, item: CompletionItem) -> str:
|
||||
"""
|
||||
将候选项格式化为 PowerShell 脚本可解析的文本行。
|
||||
|
||||
:param item: Click completion 候选项
|
||||
:return: 格式化后的文本
|
||||
"""
|
||||
help_text = item.help or item.value
|
||||
return f'{item.type}\t{item.value}\t{help_text}'
|
||||
|
||||
|
||||
def ensure_custom_completion_classes_registered() -> None:
|
||||
"""
|
||||
确保自定义 shell completion 类已注册到 Click。
|
||||
|
||||
:return: None
|
||||
"""
|
||||
if get_completion_class(PowerShellComplete.name) is None:
|
||||
add_completion_class(PowerShellComplete)
|
||||
Reference in New Issue
Block a user