117 lines
4.1 KiB
Python
117 lines
4.1 KiB
Python
"""CLI：argparse 与分发模板。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
from typing import List, Optional
|
||
|
||
from service.task_service import (
|
||
cmd_config_path,
|
||
cmd_health,
|
||
cmd_log_get,
|
||
cmd_logs,
|
||
cmd_run,
|
||
cmd_version,
|
||
)
|
||
from util.config_bootstrap import bootstrap_skill_config
|
||
from util.argparse_zh import ZhArgumentParser
|
||
from util.constants import LOG_LOGGER_NAME, SKILL_SLUG
|
||
from util.logging_config import get_skill_logger, setup_skill_logging
|
||
|
||
|
||
def _cli_str_or_none(raw: Optional[str]) -> Optional[str]:
|
||
if raw is None:
|
||
return None
|
||
v = str(raw).strip()
|
||
return v or None
|
||
|
||
|
||
def _handle_run(args: argparse.Namespace) -> int:
|
||
tail = [str(x).strip() for x in (args.run_tail or []) if str(x).strip()]
|
||
if len(tail) > 2:
|
||
print("❌ 参数过多。")
|
||
print("用法：python main.py run [target [input_id]] | run [-t target] [-i input_id]")
|
||
return 1
|
||
|
||
t_target: Optional[str] = None
|
||
t_input: Optional[str] = None
|
||
if len(tail) == 2:
|
||
t_target, t_input = tail[0], tail[1]
|
||
elif len(tail) == 1:
|
||
if tail[0].isdigit():
|
||
t_input = tail[0]
|
||
else:
|
||
t_target = tail[0]
|
||
|
||
pick_t = _cli_str_or_none(getattr(args, "target", None))
|
||
pick_i = _cli_str_or_none(getattr(args, "input_id", None))
|
||
target = pick_t or t_target
|
||
input_id = pick_i or t_input
|
||
return cmd_run(target=target, input_id=input_id)
|
||
|
||
|
||
def _print_full_usage() -> None:
|
||
print("通用业务技能模板（main.py）可用命令：")
|
||
print(" python main.py run [target [input_id]] [-t target] [-i input_id]")
|
||
print(" python main.py logs [--limit N] [--status s] [--task-type t] [--target-id tid]")
|
||
print(" python main.py log-get <log_id>")
|
||
print(" python main.py health")
|
||
print(" python main.py config-path")
|
||
print(" python main.py version")
|
||
|
||
|
||
def build_parser() -> ZhArgumentParser:
|
||
p = ZhArgumentParser(
|
||
prog="main.py",
|
||
description="通用业务技能模板：任务执行命令骨架、日志查询、健康检查、版本输出。",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
sub = p.add_subparsers(dest="cmd", required=True, parser_class=ZhArgumentParser)
|
||
|
||
sp = sub.add_parser("run", help="任务执行命令骨架")
|
||
sp.add_argument("--target", "-t", default=None, metavar="目标")
|
||
sp.add_argument("--input-id", "-i", default=None, metavar="输入id", dest="input_id")
|
||
sp.add_argument("run_tail", nargs="*", metavar="位置参数")
|
||
sp.set_defaults(handler=_handle_run)
|
||
|
||
sp = sub.add_parser("logs", help="查看任务日志")
|
||
sp.add_argument("--limit", type=int, default=10)
|
||
sp.add_argument("--status", default=None)
|
||
sp.add_argument("--task-type", default=None, dest="task_type")
|
||
sp.add_argument("--target-id", default=None, dest="target_id")
|
||
sp.set_defaults(handler=lambda a: cmd_logs(
|
||
limit=a.limit, status=a.status, task_type=a.task_type, target_id=a.target_id
|
||
))
|
||
|
||
sp = sub.add_parser("log-get", help="按 log_id 查看单条任务日志(JSON)")
|
||
sp.add_argument("log_id")
|
||
sp.set_defaults(handler=lambda a: cmd_log_get(a.log_id))
|
||
|
||
sp = sub.add_parser("health", help="健康检查")
|
||
sp.set_defaults(handler=lambda _a: cmd_health())
|
||
|
||
sp = sub.add_parser("config-path", help="输出用户 .env 与模板路径")
|
||
sp.set_defaults(handler=lambda _a: cmd_config_path())
|
||
|
||
sp = sub.add_parser("version", help="版本信息(JSON)")
|
||
sp.set_defaults(handler=lambda _a: cmd_version())
|
||
return p
|
||
|
||
|
||
def main(argv: Optional[List[str]] = None) -> int:
|
||
argv = argv if argv is not None else sys.argv[1:]
|
||
bootstrap_skill_config()
|
||
setup_skill_logging(SKILL_SLUG, LOG_LOGGER_NAME)
|
||
get_skill_logger().info("cli_start argv=%s", sys.argv)
|
||
if not argv:
|
||
_print_full_usage()
|
||
return 1
|
||
if len(argv) == 2 and argv[0] not in {
|
||
"run", "logs", "log-get", "health", "config-path", "version", "-h", "--help"
|
||
}:
|
||
return cmd_run(target=argv[0], input_id=argv[1])
|
||
parser = build_parser()
|
||
args = parser.parse_args(argv)
|
||
return int(args.handler(args))
|