"""CLI:argparse 与分发模板。 只负责参数解析并转交 service;业务内核在 domain / task_service,勿在此按入口复制逻辑。 """ 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 ") 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))