142 lines
5.2 KiB
Python
142 lines
5.2 KiB
Python
"""CLI：argparse 解析与分发（业务内核在 service，本层只做参数→service 映射）。"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
from typing import List, Optional
|
||
|
||
from service.task_service import (
|
||
cmd_add,
|
||
cmd_config_path,
|
||
cmd_delete,
|
||
cmd_done,
|
||
cmd_health,
|
||
cmd_init_db,
|
||
cmd_list,
|
||
cmd_remind,
|
||
cmd_stats,
|
||
cmd_version,
|
||
cmd_run,
|
||
)
|
||
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 _parse_ids(raw: str) -> list[int]:
|
||
"""解析 ``1`` 或 ``1,2,3`` 为 int 列表。"""
|
||
parts = [p.strip() for p in (raw or "").split(",") if p.strip()]
|
||
try:
|
||
return [int(p) for p in parts]
|
||
except ValueError as exc:
|
||
raise SystemExit(f"参数错误：ids 必须是数字或逗号分隔的数字列表，收到 {raw!r}\n{exc}")
|
||
|
||
|
||
def _print_full_usage() -> None:
|
||
print("通用业务技能模板（main.py）可用命令：")
|
||
print(" python main.py init-db # 初始化数据库与元数据")
|
||
print(" python main.py todo add <title> [-p high|normal|low] [-d <unix秒>")
|
||
print(" python main.py todo list [-s pending|done] [-n <limit>]")
|
||
print(" python main.py todo done <id>")
|
||
print(" python main.py todo delete <id>")
|
||
print(" python main.py todo stats")
|
||
print(" python main.py remind [-s <秒>]")
|
||
print(" python main.py health / version / config-path")
|
||
|
||
|
||
def _add_run_subparser(sub) -> None:
|
||
sp = sub.add_parser("todo", help="待办事项操作")
|
||
todo_sub = sp.add_subparsers(dest="todo_cmd", required=True, parser_class=ZhArgumentParser)
|
||
|
||
p = todo_sub.add_parser("add", help="新增一条待办")
|
||
p.add_argument("title", help="待办标题（必填）")
|
||
p.add_argument("-p", "--priority", default="normal",
|
||
choices=["high", "normal", "low"], help="优先级，默认 normal")
|
||
p.add_argument("-d", "--due-at", type=int, default=None,
|
||
help="截止时间（Unix 秒，可选）")
|
||
p.set_defaults(handler=lambda a: cmd_add(
|
||
title=a.title, priority=a.priority, due_at=a.due_at))
|
||
|
||
p = todo_sub.add_parser("list", help="列出待办")
|
||
p.add_argument("-s", "--status", default=None,
|
||
choices=["pending", "done"], help="按状态过滤")
|
||
p.add_argument("-n", "--limit", type=int, default=100, help="返回数量上限")
|
||
p.set_defaults(handler=lambda a: cmd_list(status=a.status, limit=a.limit))
|
||
|
||
p = todo_sub.add_parser("done", help="标记完成（可传逗号分隔的多个 id）")
|
||
p.add_argument("ids", help="待办 id 或逗号分隔的 id 列表（如 1 或 1,2,3）")
|
||
p.set_defaults(handler=lambda a: cmd_done(todo_ids=_parse_ids(a.ids)))
|
||
|
||
p = todo_sub.add_parser("delete", help="删除（可传逗号分隔的多个 id）")
|
||
p.add_argument("ids", help="待办 id 或逗号分隔的 id 列表")
|
||
p.set_defaults(handler=lambda a: cmd_delete(todo_ids=_parse_ids(a.ids)))
|
||
|
||
p = todo_sub.add_parser("stats", help="简单统计")
|
||
p.set_defaults(handler=lambda _a: cmd_stats())
|
||
|
||
|
||
|
||
def _add_remind_subparser(sub) -> None:
|
||
sp = sub.add_parser("remind", help="演示 async 提醒任务（job_context/emit/finish）")
|
||
sp.add_argument("-s", "--seconds", type=int, default=3,
|
||
help="模拟运行秒数，默认 3")
|
||
sp.set_defaults(handler=lambda a: cmd_remind(seconds=a.seconds))
|
||
|
||
|
||
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("init-db", help="幂等初始化 SQLite 与数据管理元数据")
|
||
sp.set_defaults(handler=lambda _a: cmd_init_db())
|
||
|
||
_add_run_subparser(sub)
|
||
_add_remind_subparser(sub)
|
||
|
||
sp = sub.add_parser("health", help="健康检查")
|
||
sp.set_defaults(handler=lambda _a: cmd_health())
|
||
|
||
sp = sub.add_parser("version", help="版本信息(JSON)")
|
||
sp.set_defaults(handler=lambda _a: cmd_version())
|
||
|
||
sp = sub.add_parser("config-path", help="输出 .env 与模板路径")
|
||
sp.set_defaults(handler=lambda _a: cmd_config_path())
|
||
|
||
sp = sub.add_parser("run", help="运行占位符")
|
||
sp.set_defaults(handler=lambda _a: cmd_run())
|
||
|
||
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", "health", "config-path", "version", "init-db",
|
||
"-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))
|
||
|