6 Commits

Author SHA1 Message Date
zhou c3b86b603d bump version to 0.2.10
Release / build (push) Failing after 11m58s
Release / publish-pypi (push) Has been skipped
Release / release (push) Has been skipped
2026-06-27 19:41:24 +08:00
zhou 327bd6e069 feat: 优化条件不满足时的报错信息展示
1. 新增格式化reason的工具函数统一处理报错信息
2. 支持从条件函数中提取自定义的失败原因
3. 完善NOT和OR条件的失败原因传递逻辑
4. 移除任务跳过的冗余打印输出
2026-06-27 19:40:51 +08:00
zhou 22f8d2110d chore: add pysnooper dev dependency and update configs
1. add pysnooper>=1.2.3 to dev dependencies in pyproject.toml and uv.lock
2. update type hints in task.py from Iterator to Generator
3. add more PyPI mirrors and update envdev.py comments and checks
4. fix trailing whitespace in executors.py
2026-06-27 19:35:11 +08:00
zhou 2a1f2f7175 refactor(envdev, conditions): 重构环境配置脚本,新增平台和文件条件检查
1. 移除废弃的envqt命令入口
2. 新增IS_WINDOWS、IS_LINUX等平台检测条件
3. 新增FILE_CONTENT_EXISTS文件内容检查条件
4. 使用内置条件替代硬编码的平台判断
5. 为任务添加条件控制,仅在符合场景时执行
2026-06-27 18:29:40 +08:00
zhou 9d033e1c0b refactor(system): add setenv_group and write_file task helpers
1. 为setenv和which函数添加正确的返回类型注解
2. 新增setenv_group批量设置环境变量的任务组
3. 新增write_file写入文件的任务工具函数
4. 更新__all__导出所有新增的工具函数

feat(cli/envdev): rewrite envdev cli with proper config and args
1. 重构环境开发CLI脚本,使用argparse替换原有TypedDict配置
2. 新增Python和Conda镜像源选择参数
3. 自动生成并写入Python pip和Conda配置文件
4. 优化任务依赖和命名,统一使用系统工具函数
2026-06-27 17:12:53 +08:00
zhou 336f7b7292 -envqt 2026-06-27 16:45:02 +08:00
9 changed files with 305 additions and 100 deletions
+5 -3
View File
@@ -21,7 +21,7 @@ license = { text = "MIT" }
name = "pyflowx"
readme = "README.md"
requires-python = ">=3.8"
version = "0.2.9"
version = "0.2.10"
[project.scripts]
autofmt = "pyflowx.cli.autofmt:main"
@@ -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"
@@ -94,7 +93,10 @@ packages = ["src/pyflowx"]
pyflowx = { workspace = true }
[dependency-groups]
dev = ["pyflowx[dev,office,llm]"]
dev = [
"pyflowx[dev,office,llm]",
"pysnooper>=1.2.3",
]
[tool.coverage.run]
branch = true
+1 -1
View File
@@ -95,7 +95,7 @@ from .task import (
task_template,
)
__version__ = "0.3.3"
__version__ = "0.3.4"
__all__ = [
"IS_LINUX",
+193 -26
View File
@@ -1,59 +1,226 @@
from typing import TypedDict
from __future__ import annotations
import argparse
from pathlib import Path
from typing import Literal, get_args
import pyflowx as px
class EnvConfig(TypedDict):
"""环境配置项."""
name: str
value: str
description: str
PIP_INDEX_URL_CONFIG: EnvConfig = {
"name": "PIP_INDEX_URL",
"value": "https://pypi.tuna.tsinghua.edu.cn/simple",
"description": "PIP索引URL",
}
from pyflowx.conditions import BuiltinConditions
from pyflowx.tasks.system import setenv_group, write_file
# ============================================================================
# 配置
# Mirror 配置
# ============================================================================
DOWNLOAD_MIRROR_SCRIPT: str = "curl -sSL https://linuxmirrors.cn/main.sh -o /tmp/linuxmirrors.sh"
INSTALL_MIRROR_SCRIPT: str = "sudo bash /tmp/linuxmirrors.sh"
PIP_INDEX_URLS: dict[str, str] = {
# ============================================================================
# Python 配置
# ============================================================================
PyMirrorType = Literal["tsinghua", "aliyun", "huaweicloud", "ustc", "zju"]
PIP_INDEX_URLS: dict[PyMirrorType, str] = {
"tsinghua": "https://pypi.tuna.tsinghua.edu.cn/simple",
"aliyun": "https://mirrors.aliyun.com/pypi/simple/",
"huaweicloud": "https://mirrors.huaweicloud.com/repository/pypi/simple/",
"ustc": "https://pypi.mirrors.ustc.edu.cn/simple/",
"zju": "https://mirrors.zju.edu.cn/pypi/simple/",
}
PIP_TRUSTED_HOSTS: dict[str, str] = {
PIP_TRUSTED_HOSTS: dict[PyMirrorType, str] = {
"tsinghua": "pypi.tuna.tsinghua.edu.cn",
"aliyun": "mirrors.aliyun.com",
"huaweicloud": "mirrors.huaweicloud.com",
"ustc": "pypi.mirrors.ustc.edu.cn",
"zju": "mirrors.zju.edu.cn",
}
PIP_CONFIG_PATH = Path.home() / ".pip" / "pip.conf" if BuiltinConditions.IS_LINUX() else Path.home() / "pip" / "pip.ini"
UV_INDEX_URL: str = "https://mirrors.aliyun.com/pypi/simple/"
UV_INDEX_URLS = PIP_INDEX_URLS
UV_PYTHON_INSTALL_MIRROR: str = "https://registry.npmmirror.com/-/binary/python-build-standalone"
CONDA_MIRROR_URLS: dict[str, list[str]] = {
# ============================================================================
# Conda 配置
# ============================================================================
CondaMirrorType = Literal["tsinghua", "ustc", "bsfu", "aliyun"]
CONDA_MIRROR_URLS: dict[CondaMirrorType, list[str]] = {
"tsinghua": [
"https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/",
"https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/",
"https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/r/",
"https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/msys2/",
"https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/pro/",
"https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/conda-forge/",
"https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/bioconda/",
"https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/menpo/",
"https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/pytorch/",
],
"ustc": [
"https://mirrors.ustc.edu.cn/anaconda/pkgs/main/",
"https://mirrors.ustc.edu.cn/anaconda/pkgs/free/",
"https://mirrors.ustc.edu.cn/anaconda/pkgs/r/",
"https://mirrors.ustc.edu.cn/anaconda/pkgs/msys2/",
"https://mirrors.ustc.edu.cn/anaconda/pkgs/pro/",
"https://mirrors.ustc.edu.cn/anaconda/pkgs/dev/",
"https://mirrors.ustc.edu.cn/anaconda/cloud/conda-forge/",
"https://mirrors.ustc.edu.cn/anaconda/cloud/bioconda/",
"https://mirrors.ustc.edu.cn/anaconda/cloud/menpo/",
"https://mirrors.ustc.edu.cn/anaconda/cloud/pytorch/",
],
"bsfu": [
"https://mirrors.bsfu.edu.cn/anaconda/pkgs/main/",
"https://mirrors.bsfu.edu.cn/anaconda/pkgs/free/",
"https://mirrors.bsfu.edu.cn/anaconda/pkgs/r/",
"https://mirrors.bsfu.edu.cn/anaconda/pkgs/msys2/",
"https://mirrors.bsfu.edu.cn/anaconda/pkgs/pro/",
"https://mirrors.bsfu.edu.cn/anaconda/pkgs/dev/",
"https://mirrors.bsfu.edu.cn/anaconda/cloud/conda-forge/",
"https://mirrors.bsfu.edu.cn/anaconda/cloud/bioconda/",
"https://mirrors.bsfu.edu.cn/anaconda/cloud/menpo/",
"https://mirrors.bsfu.edu.cn/anaconda/cloud/pytorch/",
],
"aliyun": [
"https://mirrors.aliyun.com/anaconda/pkgs/main/",
"https://mirrors.aliyun.com/anaconda/pkgs/free/",
"https://mirrors.aliyun.com/anaconda/pkgs/r/",
"https://mirrors.aliyun.com/anaconda/pkgs/msys2/",
"https://mirrors.aliyun.com/anaconda/pkgs/pro/",
"https://mirrors.aliyun.com/anaconda/pkgs/dev/",
"https://mirrors.aliyun.com/anaconda/cloud/conda-forge/",
"https://mirrors.aliyun.com/anaconda/cloud/bioconda/",
"https://mirrors.aliyun.com/anaconda/cloud/menpo/",
"https://mirrors.aliyun.com/anaconda/cloud/pytorch/",
],
}
CONDA_CONFIG_PATH = Path.home() / ".condarc"
# ============================================================================
# Qt 配置
# ============================================================================
QT_LIBS: list[str] = [
"build-essential",
"libgl1",
"libegl1",
"libglib2.0-0",
"libfontconfig1",
"libfreetype6",
"libxkbcommon0",
"libdbus-1-3",
"libxcb-xinerama0",
"libxcb-icccm4",
"libxcb-image0",
"libxcb-keysyms1",
"libxcb-randr0",
"libxcb-render-util0",
"libxcb-shape0",
"libxcb-xfixes0",
"libxcb-cursor0",
]
CHINESE_FONTS: list[str] = [
"fonts-noto-cjk",
"fonts-wqy-microhei",
"fonts-wqy-zenhei",
"fonts-noto-color-emoji",
]
def main() -> None:
"""主函数."""
# 使用更安全的分步执行方式,便于调试和捕获错误
parser = argparse.ArgumentParser(description="环境开发工具")
parser.add_argument(
"--python-mirror",
nargs="?",
type=str,
default="tsinghua",
choices=get_args(PyMirrorType),
help="Python 镜像源",
)
parser.add_argument(
"--conda-mirror",
nargs="?",
type=str,
default="tsinghua",
choices=get_args(CondaMirrorType),
help="Conda 镜镜像源",
)
args = parser.parse_args()
python_mirror = args.python_mirror
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", cmd="curl -sSL https://linuxmirrors.cn/main.sh -o /tmp/linuxmirrors.sh", verbose=True),
px.TaskSpec("install", cmd="sudo bash /tmp/linuxmirrors.sh", verbose=True, depends_on=("download",)),
# 系统镜像配置(仅 Linux 且未配置国内镜像)
px.TaskSpec(
"download_mirror",
cmd=DOWNLOAD_MIRROR_SCRIPT,
conditions=(
BuiltinConditions.IS_LINUX(),
BuiltinConditions.NOT(
BuiltinConditions.OR(
*[
BuiltinConditions.FILE_CONTENT_EXISTS(f, m)
for f in [
"/etc/apt/sources.list",
"/etc/apt/sources.list.d/ubuntu.sources",
]
for m in get_args(PyMirrorType)
],
)
),
),
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],
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],
conditions=(BuiltinConditions.IS_LINUX(),),
depends_on=("install_mirror",),
allow_upstream_skip=True,
verbose=True,
),
# 设置 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 配置(仅当未配置)
write_file(
str(CONDA_CONFIG_PATH),
"show_channel_urls: true\nchannels:\n - " + "\n - ".join(conda_mirror_urls) + "\n - defaults",
),
])
px.run(graph, strategy="thread")
px.run(graph, strategy="thread", verbose=True)
-57
View File
@@ -1,57 +0,0 @@
"""PyQt 环境配置工具.
用于设置 PyQt 相关环境变量, 安装依赖环境.
"""
from __future__ import annotations
import pyflowx as px
from pyflowx.conditions import Constants
QT_LIBS: list[str] = [
"build-essential",
"libgl1",
"libegl1",
"libglib2.0-0",
"libfontconfig1",
"libfreetype6",
"libxkbcommon0",
"libdbus-1-3",
"libxcb-xinerama0",
"libxcb-icccm4",
"libxcb-image0",
"libxcb-keysyms1",
"libxcb-randr0",
"libxcb-render-util0",
"libxcb-shape0",
"libxcb-xfixes0",
"libxcb-cursor0",
]
CHINESE_FONTS: list[str] = [
"fonts-noto-cjk",
"fonts-wqy-microhei",
"fonts-wqy-zenhei",
"fonts-noto-color-emoji",
]
def main() -> None:
"""PyQt 环境配置工具主函数."""
graph = px.Graph.from_specs(
[
px.TaskSpec(
"envqt_install",
cmd=["sudo", "apt", "install", "-y", *QT_LIBS],
conditions=(lambda _: Constants.IS_LINUX,),
verbose=True,
),
px.TaskSpec(
"envqt_fonts",
cmd=["sudo", "apt", "install", "-y", *CHINESE_FONTS],
conditions=(lambda _: Constants.IS_LINUX,),
verbose=True,
),
],
)
px.run(graph, strategy="thread", verbose=True)
+48 -2
View File
@@ -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})")
# ------------------------------------------------------------------ #
# 上下文条件:基于上游依赖结果
# ------------------------------------------------------------------ #
@@ -180,7 +211,13 @@ class BuiltinConditions:
"""对条件取反."""
def _cond(ctx: Context) -> bool:
return not condition(ctx)
result = condition(ctx)
if result:
# inner 为 True 时 NOT 会失败,记录 inner 的具体原因
inner_reason = getattr(condition, "_reason", None)
if inner_reason is not None:
_cond._reason = inner_reason # type: ignore[attr-defined]
return not result
_cond.__name__ = f"NOT({getattr(condition, '__name__', repr(condition))})"
return _cond
@@ -201,7 +238,16 @@ class BuiltinConditions:
"""多个条件的逻辑或."""
def _cond(ctx: Context) -> bool:
return any(c(ctx) for c in conditions)
matched: list[str] = []
for c in conditions:
if c(ctx):
matched.append(
getattr(c, "_reason", None) or getattr(c, "__name__", repr(c)),
)
if matched:
_cond._reason = matched # type: ignore[attr-defined]
return True
return False
names = [getattr(c, "__name__", repr(c)) for c in conditions]
_cond.__name__ = f"OR({', '.join(names)})"
+13 -3
View File
@@ -116,6 +116,13 @@ def _check_upstream_skipped(
return False, None # pragma: no cover
def _format_reason(reason: Any) -> str:
"""将 _reason 格式化为可读字符串."""
if isinstance(reason, list):
return ", ".join(str(r) for r in reason)
return str(reason)
def _evaluate_conditions(spec: TaskSpec[Any], context: Mapping[str, Any]) -> str | None:
"""求值所有条件,返回跳过原因或 ``None``。
@@ -130,8 +137,13 @@ def _evaluate_conditions(spec: TaskSpec[Any], context: Mapping[str, Any]) -> str
name = getattr(condition, "__name__", None) or "匿名条件(执行错误)"
failed_conditions.append(name)
continue
if not ok:
failed_conditions.append(getattr(condition, "__name__", None) or "匿名条件")
reason = getattr(condition, "_reason", None)
if reason is not None:
failed_conditions.append(_format_reason(reason))
else:
failed_conditions.append(getattr(condition, "__name__", None) or "匿名条件")
if failed_conditions:
if len(failed_conditions) <= 2:
@@ -158,8 +170,6 @@ def _make_skipped_result(
reason=reason,
)
_emit(on_event, result)
if spec.verbose:
print(f"[skip] 任务 '{spec.name}' 跳过: {reason}", flush=True)
logger.info("task %r skipped (%s)", spec.name, reason)
return result
+9 -3
View File
@@ -31,8 +31,8 @@ from typing import (
Callable,
ContextManager,
Coroutine,
Generator,
Generic,
Iterator,
List,
Mapping,
Union,
@@ -327,7 +327,13 @@ class TaskSpec(Generic[T]):
failed_conditions.append(name)
continue
if not ok:
failed_conditions.append(getattr(condition, "__name__", None) or "匿名条件")
reason = getattr(condition, "_reason", None)
if reason is not None:
failed_conditions.append(
", ".join(str(r) for r in reason) if isinstance(reason, list) else str(reason),
)
else:
failed_conditions.append(getattr(condition, "__name__", None) or "匿名条件")
if failed_conditions:
return False, f"条件不满足: {', '.join(failed_conditions)}"
@@ -367,7 +373,7 @@ class TaskSpec(Generic[T]):
def _env_and_cwd(
env: Mapping[str, str] | None,
cwd: Path | None,
) -> Iterator[None]:
) -> Generator[None, None, None]:
"""临时设置环境变量与工作目录。"""
saved_env: dict[str, str] = {}
saved_cwd: str | None = None
+21 -3
View File
@@ -66,7 +66,7 @@ def reset_icon_cache() -> list[px.TaskSpec]:
]
def setenv(name: str, value: str, default: bool = False):
def setenv(name: str, value: str, default: bool = False) -> px.TaskSpec:
"""设置环境变量任务."""
def set_env():
@@ -78,7 +78,12 @@ def setenv(name: str, value: str, default: bool = False):
return px.TaskSpec(f"setenv_{name.lower()}", fn=set_env, verbose=True)
def which(cmd: str):
def setenv_group(envs: dict[str, str], default: bool = False) -> list[px.TaskSpec]:
"""设置环境变量组任务."""
return [setenv(name, value, default) for name, value in envs.items()]
def which(cmd: str) -> px.TaskSpec:
"""查找命令路径任务."""
which_cmd = "where" if Constants.IS_WINDOWS else "which"
@@ -95,4 +100,17 @@ def which(cmd: str):
return px.TaskSpec(f"which_{cmd}", fn=find_command)
__all__ = ["clr", "setenv", "which"]
def write_file(path: str, content: str, encoding: str = "utf-8") -> px.TaskSpec:
"""写入文件任务."""
def write():
try:
with open(path, "w", encoding=encoding) as f:
f.write(content)
except Exception as e:
print(f"写入文件 {path} 失败: {e}")
return px.TaskSpec(f"write_file_{path}", fn=write, verbose=True)
__all__ = ["clr", "reset_icon_cache", "setenv", "setenv_group", "which", "write_file"]
Generated
+15 -2
View File
@@ -5603,7 +5603,7 @@ pycountry = [
[[package]]
name = "pyflowx"
version = "0.2.8"
version = "0.2.9"
source = { editable = "." }
dependencies = [
{ name = "graphlib-backport", marker = "python_full_version < '3.9'" },
@@ -5658,6 +5658,7 @@ office = [
[package.dev-dependencies]
dev = [
{ name = "pyflowx", extra = ["dev", "llm", "office"] },
{ name = "pysnooper" },
]
[package.metadata]
@@ -5686,7 +5687,10 @@ requires-dist = [
provides-extras = ["dev", "llm", "office"]
[package.metadata.requires-dev]
dev = [{ name = "pyflowx", extras = ["dev", "office", "llm"], editable = "." }]
dev = [
{ name = "pyflowx", extras = ["dev", "office", "llm"], editable = "." },
{ name = "pysnooper", specifier = ">=1.2.3" },
]
[[package]]
name = "pygments"
@@ -6003,6 +6007,15 @@ wheels = [
{ url = "https://mirrors.aliyun.com/pypi/packages/42/3d/4c6bcb3d456835f51445d3662a428f56c3ea5643ec798c577030ae34298c/pyrefly-1.1.1-py3-none-win_arm64.whl", hash = "sha256:83baf0db71e172665db1fca0ced50b8f7773f5192ca57e8ac6773a772b6d2fc5" },
]
[[package]]
name = "pysnooper"
version = "1.2.3"
source = { registry = "https://mirrors.aliyun.com/pypi/simple/" }
sdist = { url = "https://mirrors.aliyun.com/pypi/packages/d2/4a/be3c144f58de6b78911c417cc4a3b3fe5eb6d13cae4c12daf3ca17a8d473/pysnooper-1.2.3.tar.gz", hash = "sha256:1fa1425444a7af45108aaed860b5ca8b62b25bba25b0b037c059ba353d8f1e74" }
wheels = [
{ url = "https://mirrors.aliyun.com/pypi/packages/69/87/df62c8a998216e6749b67d548dae0967906036c61457510ef49667927c49/PySnooper-1.2.3-py2.py3-none-any.whl", hash = "sha256:546372f0e72da89f8d1b89e758b7c05a478d65288569a1ca2cc1620e7b1b1944" },
]
[[package]]
name = "pytesseract"
version = "0.3.13"