diff --git a/src/pyflowx/conditions.py b/src/pyflowx/conditions.py index 178239b..a094402 100644 --- a/src/pyflowx/conditions.py +++ b/src/pyflowx/conditions.py @@ -8,6 +8,7 @@ from __future__ import annotations import os import shutil +import subprocess import sys from pathlib import Path from typing import Callable @@ -66,6 +67,43 @@ class BuiltinConditions: """ return sys.version_info >= (major, minor) + @staticmethod + def IS_RUNNING(app_name: str) -> Condition: + """检查指定应用是否正在运行. + + Parameters + ---------- + app_name : str + 应用名称 (如 "explorer", "chrome", "python"). + + Returns + ------- + Condition + 条件判断函数. + """ + + def _check() -> bool: + if Constants.IS_WINDOWS: + # Windows: 使用 tasklist 命令 + result = subprocess.run( + ["tasklist", "/nh", "/fi", f"imagename eq {app_name}"], + capture_output=True, + text=True, + check=False, + ) + return app_name.lower() in result.stdout.lower() + else: + # Linux/macOS: 使用 pgrep 命令 + result = subprocess.run( + ["pgrep", "-x", app_name], + capture_output=True, + check=False, + ) + return result.returncode == 0 + + _check.__name__ = f"IS_RUNNING({app_name!r})" + return _check + @staticmethod def HAS_INSTALLED(app_name: str) -> Condition: """检查指定应用是否已安装. diff --git a/src/pyflowx/executors.py b/src/pyflowx/executors.py index 517202b..13b1f3f 100644 --- a/src/pyflowx/executors.py +++ b/src/pyflowx/executors.py @@ -101,6 +101,10 @@ def _check_upstream_skipped( if report is None: return False, None + # 若任务允许上游跳过，则不检查上游状态 + if spec.allow_upstream_skip: + return False, None + for dep in spec.depends_on: if dep in report.results and report.results[dep].status == TaskStatus.SKIPPED: return True, f"上游任务 '{dep}' 被跳过" @@ -155,6 +159,9 @@ def _run_sync_with_retry( result.status = TaskStatus.SKIPPED result.finished_at = datetime.now() result.reason = skip_reason + _emit(on_event, result) + if spec.verbose: + print(f"[skip] 任务 '{spec.name}' 跳过: {skip_reason}", flush=True) logger.info("task %r skipped (上游任务被跳过)", spec.name) return result @@ -164,6 +171,9 @@ def _run_sync_with_retry( result.status = TaskStatus.SKIPPED result.finished_at = datetime.now() result.reason = skip_reason + _emit(on_event, result) + if spec.verbose: + print(f"[skip] 任务 '{spec.name}' 跳过: {skip_reason}", flush=True) logger.info("task %r skipped (条件不满足)", spec.name) return result @@ -232,6 +242,9 @@ async def _run_async_with_retry( result.status = TaskStatus.SKIPPED result.finished_at = datetime.now() result.reason = skip_reason + _emit(on_event, result) + if spec.verbose: + print(f"[skip] 任务 '{spec.name}' 跳过: {skip_reason}", flush=True) logger.info("task %r skipped (上游任务被跳过)", spec.name) return result @@ -241,6 +254,9 @@ async def _run_async_with_retry( result.status = TaskStatus.SKIPPED result.finished_at = datetime.now() result.reason = skip_reason + _emit(on_event, result) + if spec.verbose: + print(f"[skip] 任务 '{spec.name}' 跳过: {skip_reason}", flush=True) logger.info("task %r skipped (条件不满足)", spec.name) return result diff --git a/src/pyflowx/task.py b/src/pyflowx/task.py index 0926b84..3f5d676 100644 --- a/src/pyflowx/task.py +++ b/src/pyflowx/task.py @@ -123,6 +123,10 @@ class TaskSpec(Generic[T]): （标记为 SKIPPED）而非失败。适用于构建工具场景，避免因未安装 某些工具（如 maturin、tox）而导致整个图执行失败。 对于 ``str`` (shell) 和 ``Callable`` 类型的 ``cmd``，此参数无效。 + allow_upstream_skip: + 若为 ``True``，当上游任务因条件不满足被跳过时，本任务仍会执行 + （而非跟随跳过）。适用于清理类任务：即使某些删除操作因目标不存在 + 而跳过，后续操作（如重启服务）仍应执行。默认为 ``False``。 """ name: str @@ -137,7 +141,8 @@ class TaskSpec(Generic[T]): conditions: Tuple[Condition, ...] = () cwd: Optional[Path] = None verbose: bool = False - skip_if_missing: bool = True + skip_if_missing: bool = False + allow_upstream_skip: bool = False def __post_init__(self) -> None: if not self.name: diff --git a/src/pyflowx/tasks/system.py b/src/pyflowx/tasks/system.py index e7a4b5b..fc866e3 100644 --- a/src/pyflowx/tasks/system.py +++ b/src/pyflowx/tasks/system.py @@ -27,30 +27,40 @@ def reset_icon_cache() -> list[px.TaskSpec]: print("reset_icon_cache: 仅在 Windows 上支持") return [] + local_app_data = os.environ.get("LOCALAPPDATA", "") + icon_cache_db = Path(local_app_data) / "IconCache.db" + explorer_cache_dir = Path(local_app_data) / "Microsoft" / "Windows" / "Explorer" + return [ px.TaskSpec( "kill_explorer", - fn=lambda: subprocess.run(["taskkill", "/f", "/im", "explorer.exe"], check=False), + cmd=["taskkill", "/f", "/im", "explorer.exe"], + conditions=(BuiltinConditions.IS_RUNNING("explorer.exe"),), verbose=True, ), px.TaskSpec( "delete_icon_cache", - fn=lambda: subprocess.run(["del", "/a", "/q", r"%localappdata%\IconCache.db"], check=False), - conditions=(BuiltinConditions.DIR_EXISTS(Path(r"%localappdata%\IconCache.db")),), + cmd=["cmd", "/c", "del", "/a", "/q", str(icon_cache_db)], + conditions=(BuiltinConditions.DIR_EXISTS(icon_cache_db),), + depends_on=("kill_explorer",), verbose=True, ), px.TaskSpec( "delete_icon_cache_all", - fn=lambda: subprocess.run( - ["del", "/a", "/q", r"%localappdata%\Microsoft\Windows\Explorer\iconcache*"], check=False - ), - conditions=(BuiltinConditions.DIR_EXISTS(Path(r"%localappdata%\Microsoft\Windows\Explorer")),), + cmd=["cmd", "/c", "del", "/a", "/q", str(explorer_cache_dir / "iconcache*")], + conditions=(BuiltinConditions.DIR_EXISTS(explorer_cache_dir),), + depends_on=("kill_explorer",), verbose=True, ), px.TaskSpec( "restart_explorer", - fn=lambda: subprocess.run(["explorer.exe"], check=False), - conditions=(BuiltinConditions.HAS_INSTALLED("explorer.exe"),), + cmd=["cmd", "/c", "start", "explorer.exe"], + conditions=( + BuiltinConditions.HAS_INSTALLED("explorer.exe"), + BuiltinConditions.NOT(BuiltinConditions.IS_RUNNING("explorer.exe")), + ), + depends_on=("delete_icon_cache", "delete_icon_cache_all"), + allow_upstream_skip=True, verbose=True, ), ] diff --git a/tests/cli/test_pymake.py b/tests/cli/test_pymake.py index 7cd8e58..9e7b2d5 100644 --- a/tests/cli/test_pymake.py +++ b/tests/cli/test_pymake.py @@ -77,25 +77,25 @@ class TestTaskSpecDefinitions: """uv_build spec should be properly defined.""" assert pymake.uv_build.name == "uv_build" assert pymake.uv_build.cmd == ["uv", "build"] - assert pymake.uv_build.skip_if_missing is True + assert pymake.uv_build.skip_if_missing is False def test_maturin_build_spec(self) -> None: """maturin_build spec should be properly defined.""" assert pymake.maturin_build.name == "maturin_build" assert isinstance(pymake.maturin_build.cmd, list) - assert pymake.maturin_build.skip_if_missing is True + assert pymake.maturin_build.skip_if_missing is False def test_uv_sync_spec(self) -> None: """uv_sync spec should be properly defined.""" assert pymake.uv_sync.name == "uv_sync" assert pymake.uv_sync.cmd == ["uv", "sync"] - assert pymake.uv_sync.skip_if_missing is True + assert pymake.uv_sync.skip_if_missing is False def test_git_clean_spec(self) -> None: """git_clean spec should be properly defined.""" assert pymake.git_clean.name == "git_clean" assert pymake.git_clean.cmd == ["gitt", "c"] - assert pymake.git_clean.skip_if_missing is True + assert pymake.git_clean.skip_if_missing is False def test_test_spec(self) -> None: """test spec should be properly defined.""" @@ -135,13 +135,13 @@ class TestTaskSpecDefinitions: assert pymake.doc.name == "doc" assert isinstance(pymake.doc.cmd, list) assert "sphinx-build" in pymake.doc.cmd - assert pymake.doc.skip_if_missing is True + assert pymake.doc.skip_if_missing is False def test_hatch_publish_spec(self) -> None: """hatch_publish spec should be properly defined.""" assert pymake.hatch_publish.name == "publish_python" assert pymake.hatch_publish.cmd == ["hatch", "publish"] - assert pymake.hatch_publish.skip_if_missing is True + assert pymake.hatch_publish.skip_if_missing is False def test_twine_publish_spec(self) -> None: """twine_publish spec should be properly defined.""" @@ -149,13 +149,13 @@ class TestTaskSpecDefinitions: assert isinstance(pymake.twine_publish.cmd, list) assert "twine" in pymake.twine_publish.cmd assert "upload" in pymake.twine_publish.cmd - assert pymake.twine_publish.skip_if_missing is True + assert pymake.twine_publish.skip_if_missing is False def test_tox_spec(self) -> None: """tox spec should be properly defined.""" assert pymake.tox.name == "tox" assert pymake.tox.cmd == ["tox", "-p", "auto"] - assert pymake.tox.skip_if_missing is True + assert pymake.tox.skip_if_missing is False # ---------------------------------------------------------------------- #