From 10bbc071187e7a902010b0651b903f0b3d5b6bb3 Mon Sep 17 00:00:00 2001 From: gooker_young Date: Fri, 26 Jun 2026 23:27:45 +0800 Subject: [PATCH] =?UTF-8?q?refactor(cli):=20=E9=87=8D=E6=9E=84=E6=B8=85?= =?UTF-8?q?=E5=B1=8F=E5=92=8Cwhich=E5=91=BD=E4=BB=A4=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 提取清屏、设置环境变量、命令查找逻辑到system任务模块 2. 统一命令行工具的任务实现方式，减少重复代码 3. 修正pyproject.toml中的cli命令名拼写错误 4. 移除过时的测试用例代码 --- pyproject.toml | 2 +- src/pyflowx/cli/clearscreen.py | 8 ++----- src/pyflowx/cli/which.py | 40 +++++----------------------------- src/pyflowx/tasks/__init__.py | 0 src/pyflowx/tasks/system.py | 38 ++++++++++++++++++++++++++++++++ tests/cli/test_which.py | 40 ---------------------------------- 6 files changed, 46 insertions(+), 82 deletions(-) create mode 100644 src/pyflowx/tasks/__init__.py create mode 100644 src/pyflowx/tasks/system.py diff --git a/pyproject.toml b/pyproject.toml index 6f52d7a..4c90c7e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,7 +25,7 @@ version = "0.2.5" [project.scripts] autofmt = "pyflowx.cli.autofmt:main" bumpversion = "pyflowx.cli.bumpversion:main" -cls = "pyflowx.cli.clearscreen:main" +clr = "pyflowx.cli.clearscreen:main" emlman = "pyflowx.cli.emlmanager:main" envdev = "pyflowx.cli.envdev:main" envpy = "pyflowx.cli.envpy:main" diff --git a/src/pyflowx/cli/clearscreen.py b/src/pyflowx/cli/clearscreen.py index be241ad..34a274d 100644 --- a/src/pyflowx/cli/clearscreen.py +++ b/src/pyflowx/cli/clearscreen.py @@ -6,14 +6,10 @@ from __future__ import annotations import pyflowx as px -from pyflowx.conditions import Constants +from pyflowx.tasks.system import CLR def main() -> None: """清屏工具主函数.""" - graph = px.Graph.from_specs([ - px.TaskSpec("cls_win", cmd=["cmd", "/c", "cls"], conditions=(lambda: Constants.IS_WINDOWS,)), - px.TaskSpec("cls_unix", cmd=["clear"], conditions=(lambda: not Constants.IS_WINDOWS,)), - px.TaskSpec("cls_ascii", fn=lambda: print("\033[2J\033[H", end="")), - ]) + graph = px.Graph.from_specs([CLR()]) px.run(graph, strategy="thread") diff --git a/src/pyflowx/cli/which.py b/src/pyflowx/cli/which.py index 7a2d260..ff40fb3 100644 --- a/src/pyflowx/cli/which.py +++ b/src/pyflowx/cli/which.py @@ -6,46 +6,16 @@ from __future__ import annotations import argparse -import shutil -from pathlib import Path import pyflowx as px - - -def which_command(command: str) -> Path | None: - """查找命令路径. - - Parameters - ---------- - command : str - 命令名称 - - Returns - ------- - Path | None - 命令路径, 如果未找到则返回 None - """ - cmd_path = shutil.which(command) - if cmd_path: - print(f"匹配路径: - {cmd_path}") - return Path(cmd_path) - else: - print(f"{command}: 未找到") - return None +from pyflowx.tasks.system import WHICH def main() -> None: """命令查找工具主函数.""" - parser = argparse.ArgumentParser( - description="Which - 命令查找工具", - usage="which [command ...]", - ) - parser.add_argument( - "commands", - type=str, - nargs="+", - help="要查找的命令名称 (如: python pip node npm git uv rustc cargo)", - ) + parser = argparse.ArgumentParser(description="Which - 命令查找工具") + parser.add_argument("commands", nargs="+", help="要查找的命令名称, 如: python ls ps gcc...") args = parser.parse_args() - graph = px.Graph.from_specs([px.TaskSpec(f"which_{cmd}", fn=which_command, args=(cmd,)) for cmd in args.commands]) + + graph = px.Graph.from_specs([WHICH(cmd) for cmd in args.commands]) px.run(graph, strategy="thread") diff --git a/src/pyflowx/tasks/__init__.py b/src/pyflowx/tasks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pyflowx/tasks/system.py b/src/pyflowx/tasks/system.py new file mode 100644 index 0000000..b966a29 --- /dev/null +++ b/src/pyflowx/tasks/system.py @@ -0,0 +1,38 @@ +import pyflowx as px + + +def CLR(): + """清屏.""" + import subprocess + + from pyflowx.conditions import Constants + + return px.TaskSpec( + "clear_screen", fn=lambda: subprocess.run(["cls"] if Constants.IS_WINDOWS else ["clear"], check=True) + ) + + +def SETENV(name: str, value: str): + """设置环境变量.""" + import os + + return px.TaskSpec("set_env", fn=lambda: os.environ.setdefault(name, value)) + + +def WHICH(cmd: str): + """查找命令路径.""" + import subprocess + + def find_command() -> str | None: + """查找命令并返回路径, 找不到时返回 None.""" + result = subprocess.run(["which", cmd], capture_output=True, text=True, check=False) + + if result.returncode == 0: + path = result.stdout.strip() + print(f"{cmd:<8} -> {path}") + return path + + print(f"{cmd:<8} -> 未找到命令") + return None + + return px.TaskSpec(f"which_{cmd}", fn=find_command) diff --git a/tests/cli/test_which.py b/tests/cli/test_which.py index 5ee4577..71954a1 100644 --- a/tests/cli/test_which.py +++ b/tests/cli/test_which.py @@ -3,7 +3,6 @@ from __future__ import annotations import shutil -from pathlib import Path from unittest.mock import patch import pytest @@ -12,45 +11,6 @@ import pyflowx as px from pyflowx.cli import which -# ---------------------------------------------------------------------- # -# which_command -# ---------------------------------------------------------------------- # -class TestWhichCommand: - """Test which_command function.""" - - def test_returns_path_when_command_found(self, capsys: pytest.CaptureFixture[str]) -> None: - """Should return Path when command is found.""" - with patch.object(shutil, "which", return_value="/usr/bin/python"): - result = which.which_command("python") - assert result == Path("/usr/bin/python") - captured = capsys.readouterr() - assert "匹配路径" in captured.out - assert "/usr/bin/python" in captured.out - - def test_returns_none_when_command_not_found(self, capsys: pytest.CaptureFixture[str]) -> None: - """Should return None when command is not found.""" - with patch.object(shutil, "which", return_value=None): - result = which.which_command("nonexistent_cmd") - assert result is None - captured = capsys.readouterr() - assert "未找到" in captured.out - assert "nonexistent_cmd" in captured.out - - def test_prints_match_path_on_success(self, capsys: pytest.CaptureFixture[str]) -> None: - """Should print '匹配路径: - ' on success.""" - with patch.object(shutil, "which", return_value="C:\\Python\\python.exe"): - _ = which.which_command("python") - captured = capsys.readouterr() - assert "匹配路径: - C:\\Python\\python.exe" in captured.out - - def test_prints_not_found_on_failure(self, capsys: pytest.CaptureFixture[str]) -> None: - """Should print ': 未找到' on failure.""" - with patch.object(shutil, "which", return_value=None): - _ = which.which_command("missing") - captured = capsys.readouterr() - assert "missing: 未找到" in captured.out - - # ---------------------------------------------------------------------- # # main function # ---------------------------------------------------------------------- #