116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
|
|
import click
|
|
import typer
|
|
from click.exceptions import NoArgsIsHelpError
|
|
|
|
from . import commands_analysis, commands_data, commands_project # noqa: F401
|
|
from .apps import app
|
|
from .core import CLIError, DEFAULT_SERVER, DEFAULT_TIMEOUT, emit_failure
|
|
from .helping import (
|
|
apply_typer_help_metadata,
|
|
build_error_guidance,
|
|
classify_click_error,
|
|
emit_help_payload,
|
|
merge_error_data,
|
|
merge_next_commands,
|
|
register_group_help_commands,
|
|
resolve_help_payload,
|
|
)
|
|
|
|
|
|
@app.callback()
|
|
def root_callback(
|
|
ctx: typer.Context,
|
|
server: Annotated[str | None, typer.Option("--server", help=f"服务端地址，默认 {DEFAULT_SERVER}")] = None,
|
|
auth_context: Annotated[Path | None, typer.Option("--auth-context", help="认证上下文 JSON 文件")] = None,
|
|
scheme: Annotated[str | None, typer.Option("--scheme", help="全局方案标识")] = None,
|
|
timeout: Annotated[int, typer.Option("--timeout", help="请求超时秒数")] = DEFAULT_TIMEOUT,
|
|
request_id: Annotated[str | None, typer.Option("--request-id", help="显式请求 ID")] = None,
|
|
) -> None:
|
|
ctx.obj = {
|
|
"server": server,
|
|
"auth_context": auth_context,
|
|
"scheme": scheme,
|
|
"timeout": timeout,
|
|
"request_id": request_id,
|
|
}
|
|
|
|
|
|
register_group_help_commands()
|
|
|
|
|
|
@app.command("help", context_settings={"allow_extra_args": True, "ignore_unknown_options": True})
|
|
def help_command(
|
|
ctx: typer.Context,
|
|
json_output: Annotated[bool, typer.Option("--json", help="输出 JSON")] = False,
|
|
) -> None:
|
|
command_path = list(ctx.args)
|
|
payload, is_index = resolve_help_payload(tuple(command_path))
|
|
if payload is None:
|
|
emit_failure(
|
|
summary="未找到命令",
|
|
code="COMMAND_NOT_FOUND",
|
|
message=f"unknown command path: {' '.join(command_path)}",
|
|
exit_code=2,
|
|
retryable=False,
|
|
server=None,
|
|
request_id=None,
|
|
data={
|
|
"usage": "tjwater help <command-path>",
|
|
"examples": ["tjwater help simulation run", "tjwater simulation help"],
|
|
},
|
|
next_commands=["tjwater help", "tjwater help simulation"],
|
|
)
|
|
raise typer.Exit(code=2)
|
|
emit_help_payload(payload, json_output=json_output, is_index=is_index)
|
|
|
|
|
|
# Must run at import time because tests call runner.invoke(app, ...) directly.
|
|
apply_typer_help_metadata()
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
try:
|
|
app(args=argv if argv is not None else sys.argv[1:], prog_name="tjwater", standalone_mode=False)
|
|
return 0
|
|
except CLIError as exc:
|
|
click_ctx = click.get_current_context(silent=True)
|
|
error_data, next_commands = build_error_guidance(click_ctx)
|
|
return emit_failure(
|
|
summary=exc.summary,
|
|
code=exc.code,
|
|
message=exc.message,
|
|
exit_code=exc.exit_code,
|
|
retryable=exc.retryable,
|
|
server=None,
|
|
request_id=None,
|
|
next_commands=merge_next_commands(exc.next_commands, next_commands),
|
|
data=merge_error_data(exc.data, error_data),
|
|
)
|
|
except NoArgsIsHelpError:
|
|
return 0
|
|
except click.ClickException as exc:
|
|
click_ctx = click.get_current_context(silent=True) or exc.ctx
|
|
error_data, next_commands = build_error_guidance(click_ctx)
|
|
summary, code = classify_click_error(exc)
|
|
return emit_failure(
|
|
summary=summary,
|
|
code=code,
|
|
message=exc.format_message(),
|
|
exit_code=2,
|
|
retryable=False,
|
|
server=None,
|
|
request_id=None,
|
|
next_commands=next_commands,
|
|
data=error_data,
|
|
)
|
|
|
|
|
|
def console_entry() -> None:
|
|
raise SystemExit(main())
|