"""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 DEFAULT_DATE_FROM, DEFAULT_DATE_TO, 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: return cmd_run( date_from=_cli_str_or_none(getattr(args, "date_from", None)), date_to=_cli_str_or_none(getattr(args, "date_to", None)), seller_code=_cli_str_or_none(getattr(args, "seller_code", None)), ) def _print_full_usage() -> None: print("Amazon 结算报表自动下载(main.py)可用命令:") print(" python main.py run --date-from YYYY-MM-DD --date-to YYYY-MM-DD [--seller-code CODE]") 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="Amazon 结算报表自动下载:请求并下载 Settlement xlsx,写入任务日志。", formatter_class=argparse.RawDescriptionHelpFormatter, ) sub = p.add_subparsers(dest="cmd", required=True, parser_class=ZhArgumentParser) sp = sub.add_parser("run", help="下载指定日期范围的结算报表") sp.add_argument( "--date-from", default=DEFAULT_DATE_FROM, metavar="起始日期", help=f"起始日期 YYYY-MM-DD(默认 {DEFAULT_DATE_FROM})", ) sp.add_argument( "--date-to", default=DEFAULT_DATE_TO, metavar="结束日期", help=f"结束日期 YYYY-MM-DD(默认 {DEFAULT_DATE_TO})", ) sp.add_argument( "--seller-code", default=None, metavar="店铺代码", dest="seller_code", help="可选;默认使用页面第一个 seller", ) 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 parser = build_parser() args = parser.parse_args(argv) return int(args.handler(args))