refactor: 用 YamlCliRunner/PfApp class 封装 CLI 入口逻辑
yaml_loader.py 新增 YamlCliRunner class, 将 run_cli 的 90 行 流程拆为 run/_load_config/_add_global_options/_extract_variables/ _handle_list 等单一职责方法; run_cli 保留为薄包装 (公共 API). pf.py 新增 PfApp class, 将 main 的路由逻辑拆为 run/_list_tools/ _resolve_tool/_run_legacy/_run_yaml 等方法; main 仅调用 PfApp().run(). 附带修复 run_yaml 的 jobs=None 误报 bug: 原代码 jobs is None 时 raise ValueError, 但 None 本意是执行全部任务, 导致 pf folderzip 报 "jobs 不能为空".
This commit is contained in:
+149
-130
@@ -13,154 +13,173 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Sequence
|
||||
|
||||
import pyflowx as px
|
||||
|
||||
_CONFIGS_DIR = Path(__file__).parent.parent / "configs"
|
||||
|
||||
|
||||
# 工具名到 YAML 配置文件的映射 (支持短别名)
|
||||
_TOOL_ALIASES: dict[str, str] = {
|
||||
"autofmt": "autofmt",
|
||||
"af": "autofmt",
|
||||
"bump": "bumpversion",
|
||||
"bumpversion": "bumpversion",
|
||||
"bv": "bumpversion",
|
||||
"filedate": "filedate",
|
||||
"fd": "filedate",
|
||||
"filelevel": "filelevel",
|
||||
"fl": "filelevel",
|
||||
"folderback": "folderback",
|
||||
"foldback": "folderback",
|
||||
"fb": "folderback",
|
||||
"folderzip": "folderzip",
|
||||
"foldzip": "folderzip",
|
||||
"fz": "folderzip",
|
||||
"git": "gittool",
|
||||
"gitt": "gittool",
|
||||
"gittool": "gittool",
|
||||
"gt": "gittool",
|
||||
"ls": "lscalc",
|
||||
"lscalc": "lscalc",
|
||||
"pack": "packtool",
|
||||
"packtool": "packtool",
|
||||
"pk": "packtool",
|
||||
"pdf": "pdftool",
|
||||
"pdftool": "pdftool",
|
||||
"pt": "pdftool",
|
||||
"pip": "piptool",
|
||||
"piptool": "piptool",
|
||||
"pp": "piptool",
|
||||
"screenshot": "screenshot",
|
||||
"scrcap": "screenshot",
|
||||
"ss": "screenshot",
|
||||
"ssh": "sshcopyid",
|
||||
"sshcopy": "sshcopyid",
|
||||
"sshcopyid": "sshcopyid",
|
||||
"sc": "sshcopyid",
|
||||
}
|
||||
class PfApp:
|
||||
"""pf 统一入口应用.
|
||||
|
||||
# 特殊工具: 有自己的 main() 函数 (尚未完全 YAML 化)
|
||||
_LEGACY_TOOLS: dict[str, str] = {
|
||||
"pymake": "pyflowx.cli.pymake:main",
|
||||
"emlman": "pyflowx.cli.emlmanager:main",
|
||||
"profiler": "pyflowx.cli.profiler:main",
|
||||
"pxp": "pyflowx.cli.profiler:main",
|
||||
"reseticon": "pyflowx.cli.reseticoncache:main",
|
||||
"yamlrun": "pyflowx.cli.yamlrun:main",
|
||||
}
|
||||
|
||||
|
||||
def _list_tools() -> None:
|
||||
"""列出所有可用工具."""
|
||||
print("PyFlowX 工具列表:")
|
||||
print()
|
||||
print("YAML 配置工具:")
|
||||
yaml_tools = sorted(set(_TOOL_ALIASES.values()))
|
||||
for tool in yaml_tools:
|
||||
print(f" pf {tool:<15} - {_tool_description(tool)}")
|
||||
print()
|
||||
print("传统工具:")
|
||||
for tool in sorted(_LEGACY_TOOLS):
|
||||
print(f" pf {tool:<15}")
|
||||
print()
|
||||
print("示例:")
|
||||
print(" pf filedate add a.txt")
|
||||
print(" pf pymake b")
|
||||
|
||||
|
||||
def _tool_description(tool_name: str) -> str:
|
||||
"""获取工具描述 (从 YAML cli.description)."""
|
||||
config_path = _CONFIGS_DIR / f"{tool_name}.yaml"
|
||||
if not config_path.exists():
|
||||
return ""
|
||||
try:
|
||||
import yaml
|
||||
|
||||
data = yaml.safe_load(config_path.read_text(encoding="utf-8"))
|
||||
if isinstance(data, dict) and isinstance(data.get("cli"), dict):
|
||||
return str(data["cli"].get("description", ""))
|
||||
except Exception:
|
||||
pass
|
||||
return ""
|
||||
|
||||
|
||||
def _resolve_tool(name: str) -> tuple[str, str] | None:
|
||||
"""解析工具名, 返回 (类型, 目标).
|
||||
|
||||
类型: "yaml" 或 "legacy"
|
||||
目标: YAML 文件名 (不含 .yaml) 或 legacy 模块路径
|
||||
路由 ``pf <tool> [command]`` 到 YAML 配置工具或传统 Python 工具.
|
||||
"""
|
||||
if name in _TOOL_ALIASES:
|
||||
return ("yaml", _TOOL_ALIASES[name])
|
||||
if name in _LEGACY_TOOLS:
|
||||
return ("legacy", _LEGACY_TOOLS[name])
|
||||
return None
|
||||
|
||||
_CONFIGS_DIR = _CONFIGS_DIR
|
||||
|
||||
def _run_legacy(module_path: str, argv: list[str]) -> int:
|
||||
"""运行传统工具的 main() 函数."""
|
||||
import importlib
|
||||
# 工具名到 YAML 配置文件的映射 (支持短别名)
|
||||
_TOOL_ALIASES: dict[str, str] = {
|
||||
"autofmt": "autofmt",
|
||||
"af": "autofmt",
|
||||
"bump": "bumpversion",
|
||||
"bumpversion": "bumpversion",
|
||||
"bv": "bumpversion",
|
||||
"filedate": "filedate",
|
||||
"fd": "filedate",
|
||||
"filelevel": "filelevel",
|
||||
"fl": "filelevel",
|
||||
"folderback": "folderback",
|
||||
"foldback": "folderback",
|
||||
"fb": "folderback",
|
||||
"folderzip": "folderzip",
|
||||
"foldzip": "folderzip",
|
||||
"fz": "folderzip",
|
||||
"git": "gittool",
|
||||
"gitt": "gittool",
|
||||
"gittool": "gittool",
|
||||
"gt": "gittool",
|
||||
"ls": "lscalc",
|
||||
"lscalc": "lscalc",
|
||||
"pack": "packtool",
|
||||
"packtool": "packtool",
|
||||
"pk": "packtool",
|
||||
"pdf": "pdftool",
|
||||
"pdftool": "pdftool",
|
||||
"pt": "pdftool",
|
||||
"pip": "piptool",
|
||||
"piptool": "piptool",
|
||||
"pp": "piptool",
|
||||
"screenshot": "screenshot",
|
||||
"scrcap": "screenshot",
|
||||
"ss": "screenshot",
|
||||
"ssh": "sshcopyid",
|
||||
"sshcopy": "sshcopyid",
|
||||
"sshcopyid": "sshcopyid",
|
||||
"sc": "sshcopyid",
|
||||
}
|
||||
|
||||
module_name, func_name = module_path.split(":", 1)
|
||||
module = importlib.import_module(module_name)
|
||||
func = getattr(module, func_name)
|
||||
# 传统工具: 有自己的 main() 函数 (尚未完全 YAML 化)
|
||||
_LEGACY_TOOLS: dict[str, str] = {
|
||||
"pymake": "pyflowx.cli.pymake:main",
|
||||
"emlman": "pyflowx.cli.emlmanager:main",
|
||||
"profiler": "pyflowx.cli.profiler:main",
|
||||
"pxp": "pyflowx.cli.profiler:main",
|
||||
"reseticon": "pyflowx.cli.reseticoncache:main",
|
||||
"yamlrun": "pyflowx.cli.yamlrun:main",
|
||||
}
|
||||
|
||||
original_argv = sys.argv
|
||||
sys.argv = [f"pf {module_name.split('.')[-1]}", *argv]
|
||||
try:
|
||||
func()
|
||||
return 0
|
||||
except SystemExit as e:
|
||||
return int(e.code) if e.code is not None else 0
|
||||
finally:
|
||||
sys.argv = original_argv
|
||||
def __init__(self, argv: Sequence[str] | None = None) -> None:
|
||||
self._argv = list(argv) if argv is not None else sys.argv[1:]
|
||||
|
||||
def run(self) -> int:
|
||||
"""主入口, 返回退出码."""
|
||||
if not self._argv:
|
||||
self._list_tools()
|
||||
return 0
|
||||
|
||||
tool_name = self._argv[0]
|
||||
rest_argv = self._argv[1:]
|
||||
|
||||
resolved = self._resolve_tool(tool_name)
|
||||
if resolved is None:
|
||||
print(f"错误: 未知工具 '{tool_name}'", file=sys.stderr)
|
||||
print("运行 'pf' 查看可用工具列表", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
tool_type, target = resolved
|
||||
if tool_type == "legacy":
|
||||
return self._run_legacy(target, rest_argv)
|
||||
return self._run_yaml(target, rest_argv)
|
||||
|
||||
def _list_tools(self) -> None:
|
||||
"""列出所有可用工具."""
|
||||
print("PyFlowX 工具列表:")
|
||||
print()
|
||||
print("YAML 配置工具:")
|
||||
yaml_tools = sorted(set(self._TOOL_ALIASES.values()))
|
||||
for tool in yaml_tools:
|
||||
print(f" pf {tool:<15} - {self._tool_description(tool)}")
|
||||
print()
|
||||
print("传统工具:")
|
||||
for tool in sorted(self._LEGACY_TOOLS):
|
||||
print(f" pf {tool:<15}")
|
||||
print()
|
||||
print("示例:")
|
||||
print(" pf filedate add a.txt")
|
||||
print(" pf pymake b")
|
||||
|
||||
def _tool_description(self, tool_name: str) -> str:
|
||||
"""获取工具描述 (从 YAML cli.description)."""
|
||||
config_path = self._CONFIGS_DIR / f"{tool_name}.yaml"
|
||||
if not config_path.exists():
|
||||
return ""
|
||||
try:
|
||||
import yaml
|
||||
|
||||
data = yaml.safe_load(config_path.read_text(encoding="utf-8"))
|
||||
if isinstance(data, dict) and isinstance(data.get("cli"), dict):
|
||||
return str(data["cli"].get("description", ""))
|
||||
except Exception:
|
||||
pass
|
||||
return ""
|
||||
|
||||
def _resolve_tool(self, name: str) -> tuple[str, str] | None:
|
||||
"""解析工具名, 返回 (类型, 目标).
|
||||
|
||||
类型: "yaml" 或 "legacy"
|
||||
目标: YAML 文件名 (不含 .yaml) 或 legacy 模块路径
|
||||
"""
|
||||
if name in self._TOOL_ALIASES:
|
||||
return ("yaml", self._TOOL_ALIASES[name])
|
||||
if name in self._LEGACY_TOOLS:
|
||||
return ("legacy", self._LEGACY_TOOLS[name])
|
||||
return None
|
||||
|
||||
def _run_legacy(self, module_path: str, argv: list[str]) -> int:
|
||||
"""运行传统工具的 main() 函数."""
|
||||
module_name, func_name = module_path.split(":", 1)
|
||||
module = importlib.import_module(module_name)
|
||||
func = getattr(module, func_name)
|
||||
|
||||
original_argv = sys.argv
|
||||
sys.argv = [f"pf {module_name.split('.')[-1]}", *argv]
|
||||
try:
|
||||
func()
|
||||
return 0
|
||||
except SystemExit as e:
|
||||
return int(e.code) if e.code is not None else 0
|
||||
finally:
|
||||
sys.argv = original_argv
|
||||
|
||||
def _run_yaml(self, target: str, argv: list[str]) -> int:
|
||||
"""运行 YAML 配置工具."""
|
||||
config_path = self._CONFIGS_DIR / f"{target}.yaml"
|
||||
if not config_path.exists():
|
||||
print(f"错误: 未找到配置文件 '{config_path}'", file=sys.stderr)
|
||||
print("运行 'pf' 查看可用工具列表", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
print(f"运行配置文件 '{config_path}'")
|
||||
return px.run_cli(config_path, argv)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""pf 统一入口主函数."""
|
||||
if len(sys.argv) < 2:
|
||||
_list_tools()
|
||||
return
|
||||
|
||||
tool_name = sys.argv[1]
|
||||
rest_argv = sys.argv[2:]
|
||||
|
||||
resolved = _resolve_tool(tool_name)
|
||||
if resolved is None:
|
||||
print(f"错误: 未知工具 '{tool_name}'", file=sys.stderr)
|
||||
print("运行 'pf' 查看可用工具列表", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
tool_type, target = resolved
|
||||
|
||||
if tool_type == "legacy":
|
||||
sys.exit(_run_legacy(target, rest_argv))
|
||||
else:
|
||||
config_path = _CONFIGS_DIR / f"{target}.yaml"
|
||||
sys.exit(px.run_cli(config_path, rest_argv))
|
||||
sys.exit(PfApp().run())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -12,6 +12,7 @@ cli:
|
||||
- name: CWD
|
||||
flag: "--cwd"
|
||||
type: str
|
||||
required: false
|
||||
default: "."
|
||||
help: "工作目录 (默认: 当前目录)"
|
||||
jobs:
|
||||
|
||||
+158
-132
@@ -258,8 +258,6 @@ def run_yaml(
|
||||
``jobs`` 指定了不存在的 job 名
|
||||
"""
|
||||
graph = load_yaml(path, variables=variables)
|
||||
if jobs is None:
|
||||
raise ValueError("jobs 不能为空")
|
||||
|
||||
if jobs is not None:
|
||||
job_names = [jobs] if isinstance(jobs, str) else list(jobs)
|
||||
@@ -908,65 +906,167 @@ def _add_args_to_parser(parser: Any, config: Mapping[str, Any]) -> None: # noqa
|
||||
parser.add_argument(flag, dest=name.lower().replace("-", "_"), **kwargs)
|
||||
|
||||
|
||||
def _extract_variables(args_namespace: Any, cli_config: Mapping[str, Any]) -> dict[str, Any]:
|
||||
"""从解析后的 namespace 提取变量字典 (大写 key)."""
|
||||
variables: dict[str, Any] = {}
|
||||
# 收集所有位置参数和选项参数的值
|
||||
arg_names: set[str] = set()
|
||||
subcommands = cli_config.get("subcommands")
|
||||
if subcommands is not None and isinstance(subcommands, Mapping):
|
||||
cmd = getattr(args_namespace, "command", None)
|
||||
if cmd and cmd in subcommands:
|
||||
sub_cfg = subcommands[cmd]
|
||||
if isinstance(sub_cfg, Mapping):
|
||||
_collect_arg_names(arg_names, sub_cfg)
|
||||
else:
|
||||
_collect_arg_names(arg_names, cli_config)
|
||||
|
||||
for name in arg_names:
|
||||
attr_name = name.lower().replace("-", "_")
|
||||
if hasattr(args_namespace, attr_name):
|
||||
variables[name] = getattr(args_namespace, attr_name)
|
||||
|
||||
return variables
|
||||
|
||||
|
||||
def _collect_arg_names(names: set[str], config: Mapping[str, Any]) -> None:
|
||||
"""收集配置中的参数名."""
|
||||
for arg_cfg in config.get("positional", []) or []:
|
||||
if isinstance(arg_cfg, Mapping) and "name" in arg_cfg:
|
||||
names.add(str(arg_cfg["name"]))
|
||||
for opt_cfg in config.get("options", []) or []:
|
||||
if isinstance(opt_cfg, Mapping) and "name" in opt_cfg:
|
||||
names.add(str(opt_cfg["name"]))
|
||||
|
||||
|
||||
def _add_global_options(parser: Any) -> None:
|
||||
"""向解析器添加全局选项.
|
||||
|
||||
verbose 默认开启 (显示执行过程), 用 --quiet 关闭.
|
||||
"""
|
||||
parser.add_argument("--dry-run", action="store_true", help="仅打印执行计划, 不执行")
|
||||
parser.add_argument("--quiet", "-q", action="store_false", dest="verbose", help="减少输出, 不显示执行过程")
|
||||
parser.add_argument("--strategy", type=str, default=None, help="执行策略 (sequential/thread/dependency)")
|
||||
parser.add_argument("--list", action="store_true", dest="list_jobs", help="列出所有任务名后退出")
|
||||
|
||||
# 如果有 subparsers, 也给每个子命令添加
|
||||
for action in parser._actions:
|
||||
if action.dest == "command" and hasattr(action, "choices"):
|
||||
for sub in action.choices.values():
|
||||
sub.add_argument("--dry-run", action="store_true", help="仅打印执行计划, 不执行")
|
||||
sub.add_argument("--quiet", "-q", action="store_false", dest="verbose", help="减少输出")
|
||||
sub.add_argument("--strategy", type=str, default=None, help="执行策略")
|
||||
sub.add_argument("--list", action="store_true", dest="list_jobs", help="列出所有任务名后退出")
|
||||
|
||||
|
||||
def run_cli(path: str | Path, argv: Sequence[str] | None = None) -> int: # noqa: PLR0911, PLR0912
|
||||
"""从 YAML 配置构建 CLI 并执行.
|
||||
class YamlCliRunner:
|
||||
"""从 YAML 配置构建并执行 CLI.
|
||||
|
||||
读取 YAML 的 ``cli:`` 段构建 argparse 解析器, 解析命令行参数,
|
||||
然后调用 :func:`run_yaml` 执行对应的 job.
|
||||
|
||||
全局选项 ``--dry-run`` / ``--quiet`` / ``--strategy`` / ``--list``
|
||||
自动添加到所有工具.
|
||||
"""
|
||||
|
||||
def __init__(self, path: str | Path, argv: Sequence[str] | None = None) -> None:
|
||||
self._path = Path(path)
|
||||
self._argv = list(argv) if argv is not None else sys.argv[1:]
|
||||
self._data: Mapping[str, Any] = {}
|
||||
self._cli_config: Mapping[str, Any] = {}
|
||||
|
||||
def run(self) -> int: # noqa: PLR0911
|
||||
"""执行 CLI, 返回退出码 (0 成功, 1 失败)."""
|
||||
from pyflowx.errors import TaskFailedError
|
||||
|
||||
if not self._path.exists():
|
||||
print(f"错误: YAML 文件不存在: {self._path}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
if not self._load_config():
|
||||
return 1
|
||||
|
||||
try:
|
||||
parser = build_cli_parser(self._data)
|
||||
except YamlLoadError as e:
|
||||
print(f"错误: CLI 配置错误: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
self._add_global_options(parser)
|
||||
args = parser.parse_args(self._argv)
|
||||
|
||||
has_subcommands = bool(self._cli_config.get("subcommands"))
|
||||
if has_subcommands:
|
||||
cmd = getattr(args, "command", None)
|
||||
if cmd is None:
|
||||
parser.print_help()
|
||||
return 0
|
||||
job_name: str | None = cmd
|
||||
else:
|
||||
job_name = None
|
||||
|
||||
if getattr(args, "list_jobs", False):
|
||||
return self._handle_list()
|
||||
|
||||
variables = self._extract_variables(args)
|
||||
job_display = job_name if job_name else "全部任务"
|
||||
print(f"[{self._path.stem}] 执行: {job_display}", flush=True)
|
||||
|
||||
try:
|
||||
run_yaml(
|
||||
self._path,
|
||||
jobs=job_name,
|
||||
variables=variables if variables else None,
|
||||
strategy=getattr(args, "strategy", None),
|
||||
dry_run=getattr(args, "dry_run", False),
|
||||
verbose=getattr(args, "verbose", False),
|
||||
)
|
||||
return 0
|
||||
except TaskFailedError as e:
|
||||
print(f"错误: {e}", file=sys.stderr)
|
||||
return 1
|
||||
except YamlLoadError as e:
|
||||
print(f"错误: {e}", file=sys.stderr)
|
||||
return 1
|
||||
except KeyboardInterrupt:
|
||||
print("\n已中断", file=sys.stderr)
|
||||
return 130
|
||||
|
||||
def _load_config(self) -> bool:
|
||||
"""加载并校验 YAML 配置, 返回是否成功."""
|
||||
try:
|
||||
content = self._path.read_text(encoding="utf-8")
|
||||
data = _parse_yaml(content)
|
||||
except YamlLoadError as e:
|
||||
print(f"错误: YAML 加载失败: {e}", file=sys.stderr)
|
||||
return False
|
||||
|
||||
if not isinstance(data, Mapping):
|
||||
print("错误: YAML 根节点必须是 mapping", file=sys.stderr)
|
||||
return False
|
||||
|
||||
cli_raw = data.get("cli") or {}
|
||||
if not isinstance(cli_raw, Mapping):
|
||||
print("错误: cli 段格式错误", file=sys.stderr)
|
||||
return False
|
||||
|
||||
self._data = data
|
||||
self._cli_config = cli_raw
|
||||
return True
|
||||
|
||||
def _add_global_options(self, parser: Any) -> None:
|
||||
"""向解析器添加全局选项.
|
||||
|
||||
verbose 默认开启 (显示执行过程), 用 --quiet 关闭.
|
||||
"""
|
||||
parser.add_argument("--dry-run", action="store_true", help="仅打印执行计划, 不执行")
|
||||
parser.add_argument("--quiet", "-q", action="store_false", dest="verbose", help="减少输出, 不显示执行过程")
|
||||
parser.add_argument("--strategy", type=str, default=None, help="执行策略 (sequential/thread/dependency)")
|
||||
parser.add_argument("--list", action="store_true", dest="list_jobs", help="列出所有任务名后退出")
|
||||
|
||||
# 给子命令也添加全局选项
|
||||
for action in parser._actions:
|
||||
if action.dest == "command" and hasattr(action, "choices"):
|
||||
for sub in action.choices.values():
|
||||
sub.add_argument("--dry-run", action="store_true", help="仅打印执行计划, 不执行")
|
||||
sub.add_argument("--quiet", "-q", action="store_false", dest="verbose", help="减少输出")
|
||||
sub.add_argument("--strategy", type=str, default=None, help="执行策略")
|
||||
sub.add_argument("--list", action="store_true", dest="list_jobs", help="列出所有任务名后退出")
|
||||
|
||||
def _extract_variables(self, args: Any) -> dict[str, Any]:
|
||||
"""从解析后的 namespace 提取变量字典 (大写 key)."""
|
||||
variables: dict[str, Any] = {}
|
||||
arg_names: set[str] = set()
|
||||
subcommands = self._cli_config.get("subcommands")
|
||||
if subcommands is not None and isinstance(subcommands, Mapping):
|
||||
cmd = getattr(args, "command", None)
|
||||
if cmd and cmd in subcommands:
|
||||
sub_cfg = subcommands[cmd]
|
||||
if isinstance(sub_cfg, Mapping):
|
||||
self._collect_arg_names(arg_names, sub_cfg)
|
||||
else:
|
||||
self._collect_arg_names(arg_names, self._cli_config)
|
||||
|
||||
for name in arg_names:
|
||||
attr_name = name.lower().replace("-", "_")
|
||||
if hasattr(args, attr_name):
|
||||
variables[name] = getattr(args, attr_name)
|
||||
|
||||
return variables
|
||||
|
||||
@staticmethod
|
||||
def _collect_arg_names(names: set[str], config: Mapping[str, Any]) -> None:
|
||||
"""收集配置中的参数名."""
|
||||
for arg_cfg in config.get("positional", []) or []:
|
||||
if isinstance(arg_cfg, Mapping) and "name" in arg_cfg:
|
||||
names.add(str(arg_cfg["name"]))
|
||||
for opt_cfg in config.get("options", []) or []:
|
||||
if isinstance(opt_cfg, Mapping) and "name" in opt_cfg:
|
||||
names.add(str(opt_cfg["name"]))
|
||||
|
||||
def _handle_list(self) -> int:
|
||||
"""处理 --list 选项, 打印任务列表."""
|
||||
graph = load_yaml(self._path)
|
||||
print("任务列表:")
|
||||
for name in graph.names:
|
||||
spec = graph.spec(name)
|
||||
deps = ", ".join(spec.depends_on) if spec.depends_on else "(无依赖)"
|
||||
print(f" - {name} (依赖: {deps})")
|
||||
return 0
|
||||
|
||||
|
||||
def run_cli(path: str | Path, argv: Sequence[str] | None = None) -> int:
|
||||
"""从 YAML 配置构建 CLI 并执行.
|
||||
|
||||
:class:`YamlCliRunner` 的便捷包装.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
path : str | Path
|
||||
@@ -979,78 +1079,4 @@ def run_cli(path: str | Path, argv: Sequence[str] | None = None) -> int: # noqa
|
||||
int
|
||||
退出码 (0 成功, 1 失败)
|
||||
"""
|
||||
from pyflowx.errors import TaskFailedError
|
||||
|
||||
p = Path(path)
|
||||
if not p.exists():
|
||||
print(f"错误: YAML 文件不存在: {p}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
try:
|
||||
content = p.read_text(encoding="utf-8")
|
||||
data = _parse_yaml(content)
|
||||
except YamlLoadError as e:
|
||||
print(f"错误: YAML 加载失败: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
if not isinstance(data, Mapping):
|
||||
print("错误: YAML 根节点必须是 mapping", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
cli_config = data.get("cli") or {}
|
||||
if not isinstance(cli_config, Mapping):
|
||||
print("错误: cli 段格式错误", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
try:
|
||||
parser = build_cli_parser(data)
|
||||
except YamlLoadError as e:
|
||||
print(f"错误: CLI 配置错误: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
_add_global_options(parser)
|
||||
args = parser.parse_args(list(argv) if argv is not None else sys.argv[1:])
|
||||
|
||||
has_subcommands = bool(cli_config.get("subcommands"))
|
||||
if has_subcommands:
|
||||
cmd = getattr(args, "command", None)
|
||||
if cmd is None:
|
||||
parser.print_help()
|
||||
return 0
|
||||
job_name = cmd
|
||||
else:
|
||||
job_name = None
|
||||
|
||||
if getattr(args, "list_jobs", False):
|
||||
graph = load_yaml(p)
|
||||
print("任务列表:")
|
||||
for name in graph.names:
|
||||
spec = graph.spec(name)
|
||||
deps = ", ".join(spec.depends_on) if spec.depends_on else "(无依赖)"
|
||||
print(f" - {name} (依赖: {deps})")
|
||||
return 0
|
||||
|
||||
variables = _extract_variables(args, cli_config)
|
||||
|
||||
job_display = job_name if job_name else "全部任务"
|
||||
print(f"[{p.stem}] 执行: {job_display}", flush=True)
|
||||
|
||||
try:
|
||||
run_yaml(
|
||||
p,
|
||||
jobs=job_name,
|
||||
variables=variables if variables else None,
|
||||
strategy=getattr(args, "strategy", None),
|
||||
dry_run=getattr(args, "dry_run", False),
|
||||
verbose=getattr(args, "verbose", False),
|
||||
)
|
||||
return 0
|
||||
except TaskFailedError as e:
|
||||
print(f"错误: {e}", file=sys.stderr)
|
||||
return 1
|
||||
except YamlLoadError as e:
|
||||
print(f"错误: {e}", file=sys.stderr)
|
||||
return 1
|
||||
except KeyboardInterrupt:
|
||||
print("\n已中断", file=sys.stderr)
|
||||
return 130
|
||||
return YamlCliRunner(path, argv).run()
|
||||
|
||||
Reference in New Issue
Block a user