466 lines
18 KiB
Python
466 lines
18 KiB
Python
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 = '_SHIERFENG_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,
|
||
{},
|
||
'shierfeng',
|
||
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()
|