bump version to 0.2.6

1.  移除本地setenvs函数，改用封装好的SETENV任务
2.  替换hf下载命令为modelscope下载命令
3.  优化参数命名和默认下载目录逻辑
4.  简化任务编排代码

pyproject.toml

@@ -20,12 +20,12 @@ license = { text = "MIT" }
 name = "pyflowx"
 readme = "README.md"
 requires-python = ">=3.8"
-version = "0.2.5"
+version = "0.2.6"
 
 [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"

src/pyflowx/__init__.py

@@ -84,7 +84,7 @@ from .runner import CliExitCode, CliRunner
 from .storage import JSONBackend, MemoryBackend, StateBackend
 from .task import TaskCmd, TaskEvent, TaskResult, TaskSpec, TaskStatus
 
-__version__ = "0.2.5"
+__version__ = "0.2.6"
 
 __all__ = [
     "IS_LINUX",

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")

src/pyflowx/cli/hfdownload.py

@@ -1,82 +1,50 @@
 import argparse
-import os
 from pathlib import Path
 from typing import Literal, get_args
 
 import pyflowx as px
+from pyflowx.tasks.system import SETENV
 
 HFDownloadType = Literal["model", "dataset", "space"]
 
 
-def setenvs():
-    """设置 HuggingFace mirror 环境变量."""
-    os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
-
-
 def main():
     parser = argparse.ArgumentParser(description="Download a model from HuggingFace.")
-    parser.add_argument("dataset_name", type=str, help="HuggingFace dataset name.")
+    parser.add_argument("name", help="Target name.")
     parser.add_argument(
-        "--type",
-        type=str,
-        nargs="?",
-        default="dataset",
-        choices=get_args(HFDownloadType),
-        help="HuggingFace dataset type.",
+        "--type", "-t", nargs="?", default="model", choices=get_args(HFDownloadType), help="Target type."
     )
-    parser.add_argument("--use-hfd", action="store_true", help="Use HFD tool to download dataset.")
+    parser.add_argument("--dir", default=None, help="Download directory.")
     args = parser.parse_args()
 
-    if not args.dataset_name:
-        parser.error("dataset_name is required")
+    if not args.name:
+        parser.error("name is required")
 
-    dataset_name = args.dataset_name
+    target_name = args.name
 
     # 创建下载目录
-    download_dir = Path.cwd() / dataset_name
+    if args.dir:
+        download_dir = Path(args.dir)
+    else:
+        download_dir = Path.home() / ".models" / target_name.split("/")[-1]
     download_dir.mkdir(parents=True, exist_ok=True)
 
-    if args.use_hfd:
-        graph = px.Graph.from_specs([
-            px.TaskSpec(name="setenvs", fn=setenvs, verbose=True),
-            px.TaskSpec(
-                name="download_hfd",
-                cmd=["wget", "https://hf-mirror.com/hfd/hfd.sh"],
-                depends_on=("setenvs",),
-                verbose=True,
-            ),
-            px.TaskSpec(
-                name="chmod_hfd",
-                cmd=["chmod", "a+x", "hfd.sh"],
-                depends_on=("download_hfd",),
-                verbose=True,
-            ),
-            px.TaskSpec(
-                name="run_hfd",
-                cmd=["./hfd.sh", dataset_name, args.type],
-                depends_on=("chmod_hfd",),
-                verbose=True,
-            ),
-        ])
-    else:
-        graph = px.Graph.from_specs([
-            px.TaskSpec(name="setenvs", fn=setenvs, verbose=True),
-            px.TaskSpec(
-                name="download",
-                cmd=[
-                    "uvx",
-                    "hf",
-                    "download",
-                    "--repo-type",
-                    args.type,
-                    "--force-download",
-                    dataset_name,
-                    "--local-dir",
-                    str(Path.cwd() / dataset_name),
-                ],
-                depends_on=("setenvs",),
-                verbose=True,
-            ),
-        ])
+    graph = px.Graph.from_specs([
+        SETENV("HF_ENDPOINT", "https://hf-mirror.com"),
+        px.TaskSpec(
+            name="download",
+            cmd=[
+                "uvx",
+                "modelscope",
+                "download",
+                f"--{args.type}",
+                target_name,
+                "--local_dir",
+                str(download_dir),
+            ],
+            depends_on=("setenv_hf_endpoint",),
+            verbose=True,
+        ),
+    ])
 
     px.run(graph, strategy="thread", verbose=True)

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> [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")

src/pyflowx/storage.py

@@ -17,11 +17,15 @@
 from __future__ import annotations
 
 import json
+import sys
 from abc import ABC, abstractmethod
 from pathlib import Path
 from typing import Any, Mapping
 
-from typing_extensions import override
+if sys.version_info >= (3, 12):
+    from typing import override
+else:
+    from typing_extensions import override
 
 from .errors import StorageError
 

src/pyflowx/task.py

@@ -15,6 +15,7 @@
 * ``TaskStatus`` 是封闭枚举；执行器绝不发明临时字符串。
 """
 
+import sys
 from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum
@@ -32,7 +33,10 @@ from typing import (
     cast,
 )
 
-from typing_extensions import TypeVar
+if sys.version_info >= (3, 13):
+    from typing import TypeVar
+else:
+    from typing_extensions import TypeVar
 
 T = TypeVar("T", default=Any)
 

src/pyflowx/tasks/__init__.py

@@ -0,0 +1 @@
+

src/pyflowx/tasks/system.py

@@ -0,0 +1,51 @@
+"""系统操作任务模块.
+
+提供常用的系统操作任务封装, 包括清屏、环境变量设置、命令查找等.
+遵循实用主义原则, 仅提供核心功能, 无过度设计.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+import pyflowx as px
+from pyflowx.conditions import Constants
+
+
+def CLR():
+    """清屏任务."""
+    cmd = ["cls"] if Constants.IS_WINDOWS else ["clear"]
+    return px.TaskSpec("clear_screen", fn=lambda: subprocess.run(cmd, check=False))
+
+
+def SETENV(name: str, value: str, default: bool = False):
+    """设置环境变量任务."""
+
+    def set_env():
+        if default:
+            os.environ.setdefault(name, value)
+        else:
+            os.environ[name] = value
+
+    return px.TaskSpec(f"setenv_{name.lower()}", fn=set_env, verbose=True)
+
+
+def WHICH(cmd: str):
+    """查找命令路径任务."""
+    which_cmd = "where" if Constants.IS_WINDOWS else "which"
+
+    def find_command():
+        result = subprocess.run([which_cmd, cmd], capture_output=True, text=True, check=False)
+
+        if result.returncode == 0:
+            # Windows 的 where 可能返回多行, 取第一个
+            path = result.stdout.strip().split("\n")[0].strip()
+            print(f"{cmd} -> {path}")
+        else:
+            print(f"{cmd} -> 未找到")
+
+    return px.TaskSpec(f"which_{cmd}", fn=find_command)
+
+
+__all__ = ["CLR", "SETENV", "WHICH"]

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 '匹配路径: - <path>' 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 '<command>: 未找到' on failure."""
-        with patch.object(shutil, "which", return_value=None):
-            _ = which.which_command("missing")
-            captured = capsys.readouterr()
-            assert "missing: 未找到" in captured.out
-
-
 # ---------------------------------------------------------------------- #
 # main function
 # ---------------------------------------------------------------------- #

uv.lock

@@ -2184,7 +2184,7 @@ wheels = [
 
 [[package]]
 name = "pyflowx"
-version = "0.2.3"
+version = "0.2.5"
 source = { editable = "." }
 dependencies = [
     { name = "graphlib-backport", marker = "python_full_version < '3.9'" },