From 2a1f2f7175d969c3e95e7120bf03dda78a6f9789 Mon Sep 17 00:00:00 2001 From: gooker_young Date: Sat, 27 Jun 2026 18:29:40 +0800 Subject: [PATCH] =?UTF-8?q?refactor(envdev,=20conditions):=20=E9=87=8D?= =?UTF-8?q?=E6=9E=84=E7=8E=AF=E5=A2=83=E9=85=8D=E7=BD=AE=E8=84=9A=E6=9C=AC?= =?UTF-8?q?=EF=BC=8C=E6=96=B0=E5=A2=9E=E5=B9=B3=E5=8F=B0=E5=92=8C=E6=96=87?= =?UTF-8?q?=E4=BB=B6=E6=9D=A1=E4=BB=B6=E6=A3=80=E6=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 移除废弃的envqt命令入口 2. 新增IS_WINDOWS、IS_LINUX等平台检测条件 3. 新增FILE_CONTENT_EXISTS文件内容检查条件 4. 使用内置条件替代硬编码的平台判断 5. 为任务添加条件控制，仅在符合场景时执行 --- pyproject.toml | 1 - src/pyflowx/cli/envdev.py | 69 +++++++++++++++++++++++++-------------- src/pyflowx/conditions.py | 31 ++++++++++++++++++ 3 files changed, 76 insertions(+), 25 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index e9fdfe3..db4a2af 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,7 +30,6 @@ clr = "pyflowx.cli.clearscreen:main" emlman = "pyflowx.cli.emlmanager:main" envdev = "pyflowx.cli.envdev:main" envpy = "pyflowx.cli.envpy:main" -envqt = "pyflowx.cli.envqt:main" envrs = "pyflowx.cli.envrs:main" filedate = "pyflowx.cli.filedate:main" filelvl = "pyflowx.cli.filelevel:main" diff --git a/src/pyflowx/cli/envdev.py b/src/pyflowx/cli/envdev.py index e8ed0a0..09f2e6e 100644 --- a/src/pyflowx/cli/envdev.py +++ b/src/pyflowx/cli/envdev.py @@ -5,7 +5,8 @@ from pathlib import Path from typing import Literal, get_args import pyflowx as px -from pyflowx.tasks.system import Constants, setenv_group, write_file +from pyflowx.conditions import BuiltinConditions +from pyflowx.tasks.system import setenv_group, write_file # ============================================================================ # Mirror 配置 @@ -27,7 +28,7 @@ PIP_TRUSTED_HOSTS: dict[PyMirrorType, str] = { "tsinghua": "pypi.tuna.tsinghua.edu.cn", "aliyun": "mirrors.aliyun.com", } -PIP_CONFIG_PATH = Path.home() / ".pip" / "pip.conf" if not Constants.IS_WINDOWS else Path.home() / "pip" / "pip.ini" +PIP_CONFIG_PATH = Path.home() / ".pip" / "pip.conf" if BuiltinConditions.IS_LINUX() else Path.home() / "pip" / "pip.ini" UV_INDEX_URLS: dict[PyMirrorType, str] = { "tsinghua": "https://pypi.tuna.tsinghua.edu.cn/simple", @@ -146,52 +147,72 @@ def main() -> None: args = parser.parse_args() python_mirror = args.python_mirror - python_envs: dict[str, str] = { - "PIP_INDEX_URL": PIP_INDEX_URLS[python_mirror], - "PIP_TRUSTED_HOSTS": PIP_TRUSTED_HOSTS[python_mirror], - "UV_INDEX_URL": UV_INDEX_URLS[python_mirror], - "UV_PYTHON_INSTALL_MIRROR": UV_PYTHON_INSTALL_MIRROR, - "UV_HTTP_TIMEOUT": "600", - "UV_LINK_MODE": "copy", - } - conda_mirror_urls = CONDA_MIRROR_URLS[args.conda_mirror] # 确保配置文件目录存在 PIP_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) CONDA_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) - # 使用更安全的分步执行方式，便于调试和捕获错误 + # 使用 conditions 自动控制任务执行 graph = px.Graph.from_specs([ - # 下载镜像 - px.TaskSpec("download_mirror", cmd=DOWNLOAD_MIRROR_SCRIPT, verbose=True), - # 安装镜像 - px.TaskSpec("install_mirror", cmd=INSTALL_MIRROR_SCRIPT, verbose=True, depends_on=("download_mirror",)), - # 安装 PyQt 相关依赖 + # 系统镜像配置（仅 Linux 且未配置） + px.TaskSpec( + "download_mirror", + cmd=DOWNLOAD_MIRROR_SCRIPT, + conditions=( + BuiltinConditions.IS_LINUX(), + BuiltinConditions.NOT( + BuiltinConditions.OR( + BuiltinConditions.FILE_CONTENT_EXISTS("/etc/apt/sources.list", "tsinghua"), + BuiltinConditions.FILE_CONTENT_EXISTS("/etc/apt/sources.list", "aliyun"), + BuiltinConditions.FILE_CONTENT_EXISTS("/etc/apt/sources.list", "ustc"), + ) + ), + ), + verbose=True, + ), + px.TaskSpec( + "install_mirror", + cmd=INSTALL_MIRROR_SCRIPT, + depends_on=("download_mirror",), + verbose=True, + ), + # 安装 Qt 依赖（仅 Linux） px.TaskSpec( "install_qt_libs", cmd=["sudo", "apt", "install", "-y", *QT_LIBS], - verbose=True, + conditions=(BuiltinConditions.IS_LINUX(),), depends_on=("install_mirror",), + allow_upstream_skip=True, + verbose=True, ), - # 安装中文字体 + # 安装中文字体（仅 Linux） px.TaskSpec( "install_fonts", cmd=["sudo", "apt", "install", "-y", *CHINESE_FONTS], - verbose=True, + conditions=(BuiltinConditions.IS_LINUX(),), depends_on=("install_mirror",), + allow_upstream_skip=True, + verbose=True, ), # 设置 Python 环境变量 - *setenv_group(python_envs), - # 写入 Python 配置 + *setenv_group({ + "PIP_INDEX_URL": PIP_INDEX_URLS[python_mirror], + "PIP_TRUSTED_HOSTS": PIP_TRUSTED_HOSTS[python_mirror], + "UV_INDEX_URL": UV_INDEX_URLS[python_mirror], + "UV_PYTHON_INSTALL_MIRROR": UV_PYTHON_INSTALL_MIRROR, + "UV_HTTP_TIMEOUT": "600", + "UV_LINK_MODE": "copy", + }), + # 写入 Python 配置（仅当未配置） write_file( str(PIP_CONFIG_PATH), f"[global]\nindex-url = {PIP_INDEX_URLS[python_mirror]}\ntrusted-host = {PIP_TRUSTED_HOSTS[python_mirror]}", ), - # 写入 Conda 配置 + # 写入 Conda 配置（仅当未配置） write_file( str(CONDA_CONFIG_PATH), - "show_channel_urls: true\nchannels:" + "\n".join(conda_mirror_urls) + "\n - defaults", + "show_channel_urls: true\nchannels:\n - " + "\n - ".join(conda_mirror_urls) + "\n - defaults", ), ]) px.run(graph, strategy="thread", verbose=True) diff --git a/src/pyflowx/conditions.py b/src/pyflowx/conditions.py index 0c3f976..590e462 100644 --- a/src/pyflowx/conditions.py +++ b/src/pyflowx/conditions.py @@ -61,6 +61,22 @@ class BuiltinConditions: # ------------------------------------------------------------------ # # 静态条件 # ------------------------------------------------------------------ # + def IS_WINDOWS() -> Condition: + """检查是否为 Windows 平台.""" + return _static(lambda: Constants.IS_WINDOWS, "IS_WINDOWS") + + def IS_LINUX() -> Condition: + """检查是否为 Linux 平台.""" + return _static(lambda: Constants.IS_LINUX, "IS_LINUX") + + def IS_MACOS() -> Condition: + """检查是否为 macOS 平台.""" + return _static(lambda: Constants.IS_MACOS, "IS_MACOS") + + def IS_POSIX() -> Condition: + """检查是否为 POSIX 平台.""" + return _static(lambda: Constants.IS_POSIX, "IS_POSIX") + @staticmethod def PYTHON_VERSION(major: int, minor: int | None = None) -> Condition: """检查 Python 版本是否匹配.""" @@ -118,6 +134,21 @@ class BuiltinConditions: f"ENV_VAR_EQUALS({var_name!r},{value!r})", ) + @staticmethod + def FILE_CONTENT_EXISTS(path: Path | str, content: str) -> Condition: + """检查文件是否包含指定内容.""" + + def _check() -> bool: + p = Path(path) + if not p.exists(): + return False + try: + return content in p.read_text(encoding="utf-8") + except Exception: + return False + + return _static(_check, f"FILE_CONTENT_EXISTS({path!r},{content!r})") + # ------------------------------------------------------------------ # # 上下文条件：基于上游依赖结果 # ------------------------------------------------------------------ #