diff --git a/.trae/documents/p10-image-processing-plan.md b/.trae/documents/p10-image-processing-plan.md new file mode 100644 index 0000000..e2d0a48 --- /dev/null +++ b/.trae/documents/p10-image-processing-plan.md @@ -0,0 +1,174 @@ +# P10 图片处理开发计划 + +## Context + +PyFlowX 现有 `ops/files/` 工具集已覆盖 PDF（pdftool）、截图（screenshot）、文件日期（filedate）等场景，但缺少图片处理能力。`pyproject.toml` 的 `office` extra 已声明 `pillow>=10.4.0`，却无对应工具消费它。 + +本计划新增 `imagetool` 工具，提供基础操作（resize/crop/rotate/flip/convert/watermark/compress）、元数据与信息（info/exif/histogram/colors）、批量 DAG 编排（`image_pipeline` 便捷构造器）三类能力。用户已确认范围：基础操作 + 元数据 + 批量 DAG（不含 OCR），工具形态为单一 `imagetool` + office extra。 + +## 实现步骤 + +### P10.1 基础操作子命令 + +**新建 `src/pyflowx/ops/files/imagetool.py`**，参照 [pdftool.py](file:///home/zhou/pyflowx/src/pyflowx/ops/files/pdftool.py) 模式： + +- 顶部 `try: from PIL import Image; HAS_PIL = True; except ImportError: HAS_PIL = False` +- `_require_pil() -> bool` 检查并打印 `pip install pyflowx[office]` 提示 +- 7 个 `@px.tool("imagetool", subcommand=..., help=...)` 子命令： + - `resize` (`r`) — `image_resize(input: Path, output: Path, width: int, height: int | None = None, keep_ratio: bool = True)` + - `crop` (`c`) — `image_crop(input: Path, output: Path, left: int, top: int, right: int, bottom: int)` + - `rotate` (`ro`) — `image_rotate(input: Path, output: Path, degrees: float, expand: bool = False)` + - `flip` (`fl`) — `image_flip(input: Path, output: Path, direction: str = "horizontal")` + - `convert` (`cv`) — `image_convert(input: Path, output: Path, format: str | None = None, quality: int = 85)` + - `watermark` (`wm`) — `image_watermark(input: Path, output: Path, text: str, position: str = "bottom-right", opacity: float = 0.5, font_size: int = 32)` + - `compress` (`cp`) — `image_compress(input: Path, output: Path, quality: int = 85)` + +实现要点： +- 用 `Image.open(input)` 打开，操作后 `img.save(output, format=..., quality=...)` 保存 +- `resize` + `keep_ratio=True` 时用 `Image.thumbnail((width, height))` 保比缩放；`keep_ratio=False` 用 `Image.resize((width, height))` +- `watermark` 用 `ImageDraw.Draw(img).text(...)` + `ImageFont.truetype()`（字体缺失回退 `load_default()`） +- `convert` 根据 `output.suffix` 推断格式；JPEG 系列需先 `img.convert("RGB")` 去 alpha 通道 +- `__all__` 列出所有 `image_*` 函数名 + +### P10.2 元数据与信息子命令 + +继续在 `imagetool.py` 追加 4 个子命令： + +- `info` (`i`) — `image_info(input: Path, json: bool = False)`：打印尺寸/格式/模式/色彩空间/EXIF 摘要。`json=True` 输出 JSON 格式 +- `exif` (`e`) — `image_exif(input: Path, output: Path | None = None, show: bool = True, set: list[str] | None = None, clear: bool = False)`：读写 EXIF。`--set KEY=VAL` 修改，`--clear` 清空，`output` 另存 +- `histogram` (`hi`) — `image_histogram(input: Path, channel: str = "rgb")`：打印 RGB/Luminance 直方图统计（每通道 8 桶，纯文本表格输出） +- `colors` (`co`) — `image_colors(input: Path, count: int = 5)`：主色调提取，用 `Image.quantize(colors=count).getpalette()` 提取并打印十六进制色值 + +实现要点： +- `info` 用 `img.size` / `img.format` / `img.mode` / `img.info.get("exif")` +- `exif` 用 `img.getexif()` / `img.save(exif=bytes(exif))` +- `histogram` 用 `img.histogram()` 分桶统计 +- `colors` 用 `img.quantize(colors=count).getpalette()` 取前 N 色 + +### P10.3 批量 DAG 编排 + +**新建 `src/pyflowx/imaging.py`**（类似 [compose.py](file:///home/zhou/pyflowx/src/pyflowx/compose.py) 的独立模块），提供链式 DAG 构造器： + +```python +def image_pipeline( + source: str | Path, + steps: list[tuple[str, dict[str, Any]]], + *, + output_dir: str | Path | None = None, + naming: str = "{stem}_{step}{ext}", +) -> Graph: + """图片处理流水线 DAG 构造器。 + + 每个 step 是 (操作名, 参数字典)，操作名对应 imagetool 子命令 + （resize/crop/rotate/flip/convert/watermark/compress）。 + 前一步输出作为后一步输入，形成链式 DAG。 + """ +``` + +示例： +```python +graph = px.image_pipeline( + "input.jpg", + steps=[ + ("resize", {"width": 800}), + ("watermark", {"text": "© 2026"}), + ("convert", {"format": "webp", "quality": 85}), + ], +) +report = px.run(graph, strategy="sequential") +``` + +实现要点： +- 每个 step 生成一个 `TaskSpec`，`fn` 调用对应的 `image_*` 函数 +- 任务名按 `naming` 模板生成（默认 `{stem}_{step}{ext}`，如 `input_resize.jpg`） +- 任务间用 `depends_on` 链式依赖，上游输出路径 = 下游输入路径 +- `output_dir` 默认为源文件所在目录 +- 在 `__init__.py` 导出 `image_pipeline` + +### P10.4 验证与文档 + +**测试 `tests/cli/test_imagetool.py`**： +- 用 `pytest.importorskip("PIL")` 跳过未安装 Pillow 的环境 +- 用 `tmp_path` 创建真实测试图片（`Image.new("RGB", (100, 100), "red").save(tmp_path / "test.png")`） +- 每个子命令至少 1 个测试（基础操作验证输出文件存在 + 尺寸/格式正确） +- 元数据测试验证输出内容（info 的 JSON 结构、exif 读写往返、histogram 桶数、colors 色值数） +- `image_pipeline` 测试验证 DAG 拓扑（3 步链式，每步输出存在且最终输出为 webp） +- 测试 mock 优先级遵循 `python-standards.md`：优先用真实 Pillow 操作（非 subprocess），仅字体加载等不可控部分用 `monkeypatch` + +**新增 `tests/imaging/test_image_pipeline.py`**（或在 `tests/test_imaging.py`）： +- 测试 `image_pipeline` 生成的 Graph 拓扑正确（任务数、依赖链） +- 测试实际执行后输出文件链完整 + +**注册到 CLI**： +- 编辑 [src/pyflowx/cli/pf.py](file:///home/zhou/pyflowx/src/pyflowx/cli/pf.py) 的 `_TOOL_MODULES`：添加 `"imagetool": "pyflowx.ops.files.imagetool"` +- 编辑 `_TOOL_ALIASES`：添加 `"imagetool": "imagetool"`、`"image": "imagetool"`、`"img": "imagetool"` +- 编辑 [src/pyflowx/__init__.py](file:///home/zhou/pyflowx/src/pyflowx/__init__.py) 导出 `image_pipeline` + +**文档同步**： +- `.trae/docs/iter-22-image-processing.md` 迭代记录 +- `.trae/skills/pyflowx-development/SKILL.md` 同步行为变更（新增"十五、图片处理工具"小节，或在十二章 CLI 工具下追加） +- `project_memory.md` 追加 P10 工程约定 + +## 关键文件 + +| 操作 | 文件 | +|------|------| +| 新建 | `src/pyflowx/ops/files/imagetool.py`（11 子命令） | +| 新建 | `src/pyflowx/imaging.py`（`image_pipeline` 构造器） | +| 新建 | `tests/cli/test_imagetool.py`（子命令测试） | +| 新建 | `tests/test_imaging.py`（DAG 构造器测试） | +| 修改 | `src/pyflowx/cli/pf.py`（注册 imagetool + 别名） | +| 修改 | `src/pyflowx/__init__.py`（导出 `image_pipeline`） | +| 新建 | `.trae/docs/iter-22-image-processing.md` | +| 修改 | `.trae/skills/pyflowx-development/SKILL.md`（行为变更同步） | + +## 复用的现有模式 + +- **`@px.tool` 装饰器 + ToolSpec**：[tools.py L116-L180](file:///home/zhou/pyflowx/src/pyflowx/tools.py#L116-L180)，每个子命令一个装饰器调用 +- **可选依赖 try/except + `_require_*`**：[pdftool.py L31-L59](file:///home/zhou/pyflowx/src/pyflowx/ops/files/pdftool.py#L31-L59) +- **`_TOOL_MODULES` 注册**：[pf.py L112-L137](file:///home/zhou/pyflowx/src/pyflowx/cli/pf.py#L112-L137) +- **`pytest.importorskip` 跳过缺依赖**：[test_pdftool.py L22](file:///home/zhou/pyflowx/tests/cli/test_pdftool.py#L22) +- **`image_pipeline` 委托模式**：参照 [compose.py](file:///home/zhou/pyflowx/src/pyflowx/compose.py) 的 `compose()` 函数式构造器 +- **TaskSpec 链式依赖**：用 `Graph.from_specs` + `depends_on` 构建链式 DAG（P9.2 `pipeline()` 同模式） + +## 验证方法 + +```bash +# 1. 单元测试 +uv run pytest tests/cli/test_imagetool.py tests/test_imaging.py -v + +# 2. 全套门禁 +uv run ruff check . +uv run ruff format --check . +uv run pyrefly check . +uv run pytest --cov=pyflowx --cov-branch + +# 3. CLI 实测（需 pip install pyflowx[office]） +pf imagetool info test.png +pf imagetool resize test.png out.png --width 50 +pf imagetool watermark test.png wm.png --text "TEST" +pf imagetool convert test.png out.webp --quality 80 + +# 4. DAG 编排实测 +python -c " +import pyflowx as px +g = px.image_pipeline('test.png', steps=[ + ('resize', {'width': 50}), + ('watermark', {'text': 'X'}), + ('convert', {'format': 'webp', 'quality': 80}), +]) +r = px.run(g, strategy='sequential') +print(r.success) +" + +# 5. 覆盖率门槛 ≥ 95%（branch） +``` + +## 验收标准 + +- ruff/pyrefly 0 错误，pytest 全绿，覆盖率 ≥ 95%（branch） +- `pf imagetool --help` 列出全部 11 个子命令 +- `pf imagetool --help` 显示参数说明 +- `image_pipeline()` 生成的 DAG 可执行，链式输出完整 +- 未安装 office extra 时 `pf imagetool ` 打印友好提示而非崩溃 +- `tests/cli/test_tool_modules.py` 自动覆盖新工具注册（无需额外修改） diff --git a/src/pyflowx/__init__.py b/src/pyflowx/__init__.py index d0fa581..0a46b96 100644 --- a/src/pyflowx/__init__.py +++ b/src/pyflowx/__init__.py @@ -77,6 +77,7 @@ from .errors import ( from .executors import Strategy, run, run_iter from .graph import Graph, GraphDefaults from .history import RunHistory +from .imaging import image_pipeline from .monitoring import MetricsCollector, health_check, start_metrics_server from .notification import ( ALL_LEVELS, @@ -180,6 +181,7 @@ __all__ = [ "describe_injection", "diagnose", "health_check", + "image_pipeline", "list_subcommands", "list_tools", "load_yaml", diff --git a/src/pyflowx/cli/pf.py b/src/pyflowx/cli/pf.py index 7335655..4c4d5a8 100644 --- a/src/pyflowx/cli/pf.py +++ b/src/pyflowx/cli/pf.py @@ -63,6 +63,9 @@ class PfApp: "gitt": "gittool", "gittool": "gittool", "gt": "gittool", + "image": "imagetool", + "imagetool": "imagetool", + "img": "imagetool", "ls": "lscalc", "lscalc": "lscalc", "msdown": "msdownload", @@ -120,6 +123,7 @@ class PfApp: "folderback": "pyflowx.ops.files.folderback", "folderzip": "pyflowx.ops.files.folderzip", "gittool": "pyflowx.ops.dev.gittool", + "imagetool": "pyflowx.ops.files.imagetool", "lscalc": "pyflowx.ops.dev.lscalc", "msdownload": "pyflowx.ops.infra.msdownload", "packtool": "pyflowx.ops.dev.packtool", diff --git a/src/pyflowx/imaging.py b/src/pyflowx/imaging.py new file mode 100644 index 0000000..2fcbcab --- /dev/null +++ b/src/pyflowx/imaging.py @@ -0,0 +1,130 @@ +"""图片处理流水线 DAG 构造器. + +将 :mod:`pyflowx.ops.files.imagetool` 的操作封装为链式 :class:`Graph`, +便于在 DAG 中组合多步图片处理 (如 resize → watermark → convert). + +设计参照 :mod:`pyflowx.compose` 的函数式构造器模式. +""" + +from __future__ import annotations + +__all__ = ["image_pipeline"] + +from pathlib import Path +from typing import Any + +from .graph import Graph +from .task import TaskSpec + + +def image_pipeline( + source: str | Path, + steps: list[tuple[str, dict[str, Any]]], + *, + output_dir: str | Path | None = None, + naming: str = "{stem}_{step}{ext}", +) -> Graph: + """图片处理流水线 DAG 构造器. + + 每个 step 是 ``(操作名, 参数字典)``, 操作名对应 imagetool 子命令 + (resize/crop/rotate/flip/convert/watermark/compress). + 前一步输出作为后一步输入, 形成链式 DAG. + + Parameters + ---------- + source : str | Path + 源图片路径 + steps : list[tuple[str, dict[str, Any]]] + 操作步骤列表, 如 ``[("resize", {"width": 800}), ("watermark", {"text": "X"})]`` + output_dir : str | Path | None + 输出目录 (None 时用源文件所在目录) + naming : str + 输出文件名模板, 可用占位符 ``{stem}`` (累积 stem) / ``{step}`` (操作名) / + ``{ext}`` (当前扩展名, convert 步骤可能改变). 默认 ``"{stem}_{step}{ext}"`` + + Returns + ------- + Graph + 链式 DAG, 每步为一个 :class:`TaskSpec`, 通过 ``depends_on`` 链接 + + Raises + ------ + ValueError + 操作名不在支持列表中时 + + Example + ------- + >>> graph = px.image_pipeline( + ... "input.jpg", + ... steps=[ + ... ("resize", {"width": 800}), + ... ("watermark", {"text": "© 2026"}), + ... ("convert", {"format": "webp", "quality": 85}), + ... ], + ... ) + >>> report = px.run(graph, strategy="sequential") + """ + from .ops.files import imagetool + + operations = { + "resize": imagetool.image_resize, + "crop": imagetool.image_crop, + "rotate": imagetool.image_rotate, + "flip": imagetool.image_flip, + "convert": imagetool.image_convert, + "watermark": imagetool.image_watermark, + "compress": imagetool.image_compress, + } + + source_path = Path(source) + out_dir = Path(output_dir) if output_dir else source_path.parent + + specs: list[TaskSpec[Any]] = [] + prev_name: str | None = None + prev_output: Path | None = None + current_stem = source_path.stem + current_ext = source_path.suffix + + for i, (op_name, params) in enumerate(steps): + if op_name not in operations: + raise ValueError(f"未知图片操作: {op_name!r}, 支持: {sorted(operations)}") + + if op_name == "convert" and "format" in params: + current_ext = f".{params['format'].lower()}" + + output_name = naming.format(stem=current_stem, step=op_name, ext=current_ext) + output_path = out_dir / output_name + input_path = source_path if prev_output is None else prev_output + + task_name = f"step_{i:02d}_{op_name}" + base_fn = operations[op_name] + fn = _make_step_fn(base_fn, input_path, output_path, params) + fn.__name__ = task_name + + depends_on = (prev_name,) if prev_name else () + spec = TaskSpec(task_name, fn=fn, depends_on=depends_on) + specs.append(spec) + + prev_name = task_name + prev_output = output_path + current_stem = output_path.stem + + return Graph.from_specs(specs) + + +def _make_step_fn( + base_fn: Any, + input_path: Path, + output_path: Path, + params: dict[str, Any], +) -> Any: + """创建步骤执行闭包, 绑定输入/输出路径与参数. + + 闭包捕获的是 :func:`_make_step_fn` 的参数 (每次调用独立作用域), + 不会受外层循环变量重绑定影响. + """ + + def _run() -> None: + base_fn(input_path, output_path, **params) + + return _run diff --git a/src/pyflowx/ops/files/imagetool.py b/src/pyflowx/ops/files/imagetool.py new file mode 100644 index 0000000..a0b51f7 --- /dev/null +++ b/src/pyflowx/ops/files/imagetool.py @@ -0,0 +1,520 @@ +"""imagetool - 图片处理工具集. + +提供 resize/crop/rotate/flip/convert/watermark/compress/info/exif/ +histogram/colors 子命令, 基于 Pillow 实现. + +依赖 ``pyflowx[office]`` extra 中的 ``pillow``. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pyflowx as px + +__all__ = [ + "image_colors", + "image_compress", + "image_convert", + "image_crop", + "image_exif", + "image_flip", + "image_histogram", + "image_info", + "image_resize", + "image_rotate", + "image_watermark", +] + +try: + from PIL import Image, ImageDraw, ImageFont + + HAS_PIL = True +except ImportError: + HAS_PIL = False + + +def _require_pil() -> bool: + """Pillow 未安装时打印提示, 返回是否可用.""" + if not HAS_PIL: + print("未安装 Pillow 库, 请安装: pip install pyflowx[office]") + return False + return True + + +def _save_image(img: Any, output: Path, fmt: str | None = None, quality: int = 85) -> None: + """保存图片, 自动处理 JPEG 不支持 alpha 通道的情况.""" + output.parent.mkdir(parents=True, exist_ok=True) + save_fmt = fmt or output.suffix.lstrip(".").upper() + if save_fmt == "JPG": + save_fmt = "JPEG" + if save_fmt == "JPEG": + img = img.convert("RGB") + if save_fmt in ("JPEG", "WEBP"): + img.save(output, format=save_fmt, quality=quality) + else: + img.save(output, format=save_fmt) + + +# ---------------------------------------------------------------------- # +# 基础操作 +# ---------------------------------------------------------------------- # +@px.tool("imagetool", subcommand="r", help="调整尺寸") +def image_resize( + input_path: Path, + output_path: Path, + width: int, + height: int | None = None, + keep_ratio: bool = True, +) -> None: + """调整图片尺寸. + + Parameters + ---------- + input_path : Path + 输入图片 + output_path : Path + 输出图片 + width : int + 目标宽度 + height : int | None + 目标高度 (keep_ratio=True 时仅作上限, 默认 None 表示按宽度等比) + keep_ratio : bool + 是否保持宽高比 (默认 True) + """ + if not _require_pil(): + return + + img = Image.open(input_path) + if keep_ratio: + target_height = height if height is not None else width + img.thumbnail((width, target_height)) + else: + if height is None: + height = width + img = img.resize((width, height)) + _save_image(img, output_path) + print(f"调整尺寸完成: {output_path} ({img.size[0]}x{img.size[1]})") + + +@px.tool("imagetool", subcommand="c", help="裁剪图片") +def image_crop( + input_path: Path, + output_path: Path, + left: int, + top: int, + right: int, + bottom: int, +) -> None: + """裁剪图片到指定矩形. + + Parameters + ---------- + input_path : Path + 输入图片 + output_path : Path + 输出图片 + left, top, right, bottom : int + 裁剪矩形坐标 (左上为原点) + """ + if not _require_pil(): + return + + img = Image.open(input_path) + cropped = img.crop((left, top, right, bottom)) + _save_image(cropped, output_path) + print(f"裁剪完成: {output_path} ({right - left}x{bottom - top})") + + +@px.tool("imagetool", subcommand="ro", help="旋转图片") +def image_rotate( + input_path: Path, + output_path: Path, + degrees: float, + expand: bool = False, +) -> None: + """旋转图片. + + Parameters + ---------- + input_path : Path + 输入图片 + output_path : Path + 输出图片 + degrees : float + 旋转角度 (正值逆时针) + expand : bool + 是否扩展画布以容纳整个旋转后的图片 (默认 False) + """ + if not _require_pil(): + return + + img = Image.open(input_path) + rotated = img.rotate(degrees, expand=expand) + _save_image(rotated, output_path) + print(f"旋转完成: {output_path} ({degrees}度)") + + +@px.tool("imagetool", subcommand="fl", help="翻转图片") +def image_flip( + input_path: Path, + output_path: Path, + direction: str = "horizontal", +) -> None: + """翻转图片. + + Parameters + ---------- + input_path : Path + 输入图片 + output_path : Path + 输出图片 + direction : str + 翻转方向: "horizontal" (水平镜像) / "vertical" (垂直镜像) + """ + if not _require_pil(): + return + + img = Image.open(input_path) + method = Image.Transpose.FLIP_LEFT_RIGHT if direction == "horizontal" else Image.Transpose.FLIP_TOP_BOTTOM + flipped = img.transpose(method) + _save_image(flipped, output_path) + print(f"翻转完成: {output_path} ({direction})") + + +@px.tool("imagetool", subcommand="cv", help="格式转换") +def image_convert( + input_path: Path, + output_path: Path, + format: str | None = None, + quality: int = 85, +) -> None: + """转换图片格式. + + Parameters + ---------- + input_path : Path + 输入图片 + output_path : Path + 输出图片 (后缀决定格式, 除非 format 显式指定) + format : str | None + 目标格式 (如 "PNG"/"JPEG"/"WEBP"), None 时按 output_path 后缀推断 + quality : int + 压缩质量 (1-100, 仅对 JPEG/WEBP 有效) + """ + if not _require_pil(): + return + + img = Image.open(input_path) + _save_image(img, output_path, fmt=format, quality=quality) + actual_fmt = format or output_path.suffix.lstrip(".").upper() + print(f"格式转换完成: {output_path} ({actual_fmt})") + + +@px.tool("imagetool", subcommand="wm", help="添加文字水印") +def image_watermark( + input_path: Path, + output_path: Path, + text: str, + position: str = "bottom-right", + opacity: float = 0.5, + font_size: int = 32, +) -> None: + """添加文字水印. + + Parameters + ---------- + input_path : Path + 输入图片 + output_path : Path + 输出图片 + text : str + 水印文字 + position : str + 水印位置: top-left/top-right/bottom-left/bottom-right/center + opacity : float + 不透明度 (0.0-1.0) + font_size : int + 字体大小 + """ + if not _require_pil(): + return + + img = Image.open(input_path).convert("RGBA") + overlay = Image.new("RGBA", img.size, (0, 0, 0, 0)) + draw = ImageDraw.Draw(overlay) + + font = _load_font(font_size) + bbox = draw.textbbox((0, 0), text, font=font) + text_w = int(bbox[2] - bbox[0]) + text_h = int(bbox[3] - bbox[1]) + margin = 10 + x, y = _resolve_position(position, img.size, text_w, text_h, margin) + + alpha = int(255 * max(0.0, min(1.0, opacity))) + draw.text((x, y), text, font=font, fill=(255, 255, 255, alpha)) + result = Image.alpha_composite(img, overlay) + _save_image(result, output_path) + print(f"水印添加完成: {output_path}") + + +def _load_font(size: int) -> Any: + """加载字体, 优先 truetype, 失败回退默认字体.""" + candidates = ("DejaVuSans.ttf", "Arial.ttf", "LiberationSans-Regular.ttf") + for name in candidates: + try: + return ImageFont.truetype(name, size) + except OSError: + continue + return ImageFont.load_default() + + +def _resolve_position( + position: str, + img_size: tuple[int, int], + text_w: int, + text_h: int, + margin: int, +) -> tuple[int, int]: + """根据位置描述符计算水印坐标.""" + w, h = img_size + pos_map = { + "top-left": (margin, margin), + "top-right": (w - text_w - margin, margin), + "bottom-left": (margin, h - text_h - margin), + "bottom-right": (w - text_w - margin, h - text_h - margin), + "center": ((w - text_w) // 2, (h - text_h) // 2), + } + return pos_map.get(position, pos_map["bottom-right"]) + + +@px.tool("imagetool", subcommand="cp", help="压缩图片") +def image_compress( + input_path: Path, + output_path: Path, + quality: int = 85, +) -> None: + """压缩图片 (重新编码以减小体积). + + Parameters + ---------- + input_path : Path + 输入图片 + output_path : Path + 输出图片 + quality : int + 压缩质量 (1-100) + """ + if not _require_pil(): + return + + img = Image.open(input_path) + fmt = input_path.suffix.lstrip(".").upper() + _save_image(img, output_path, fmt=fmt, quality=quality) + in_size = input_path.stat().st_size + out_size = output_path.stat().st_size + ratio = (1 - out_size / in_size) * 100 if in_size > 0 else 0.0 + print(f"压缩完成: {output_path} (原 {in_size}B → 新 {out_size}B, 节省 {ratio:.1f}%)") + + +# ---------------------------------------------------------------------- # +# 元数据与信息 +# ---------------------------------------------------------------------- # +@px.tool("imagetool", subcommand="i", help="查看图片信息") +def image_info(input_path: Path, json: bool = False) -> None: + """打印图片信息 (尺寸/格式/模式/EXIF 摘要). + + Parameters + ---------- + input_path : Path + 输入图片 + json : bool + 是否以 JSON 格式输出 (默认纯文本表格) + """ + if not _require_pil(): + return + + img = Image.open(input_path) + exif = img.getexif() + exif_count = len(exif) if exif else 0 + + import json as json_mod + + data = { + "path": str(input_path), + "format": img.format, + "mode": img.mode, + "width": img.size[0], + "height": img.size[1], + "exif_tags": exif_count, + } + if json: + print(json_mod.dumps(data, ensure_ascii=False, indent=2)) + else: + print(f"文件: {data['path']}") + print(f"格式: {data['format']}") + print(f"模式: {data['mode']}") + print(f"尺寸: {data['width']}x{data['height']}") + print(f"EXIF 标签数: {data['exif_tags']}") + + +@px.tool("imagetool", subcommand="e", help="读取/修改 EXIF") +def image_exif( + input_path: Path, + output_path: Path | None = None, + show: bool = True, + set: list[str] | None = None, + clear: bool = False, +) -> None: + """读取或修改 EXIF 元数据. + + Parameters + ---------- + input_path : Path + 输入图片 + output_path : Path | None + 输出路径 (None 时原地覆盖; 仅 show=True 时可省略) + show : bool + 打印全部 EXIF 标签 (默认 True) + set : list[str] | None + 设置标签, 格式 ["KEY=VALUE", ...] (KEY 为数字标签号) + clear : bool + 清空所有 EXIF 标签 (在 set 之前执行) + """ + if not _require_pil(): + return + + img = Image.open(input_path) + exif = img.getexif() + + if show: + _print_exif(exif) + + modified = _apply_exif_modifications(exif, set, clear) + if modified: + _save_exif(img, exif, output_path if output_path is not None else input_path) + + +def _print_exif(exif: Any) -> None: + """打印 EXIF 标签.""" + if exif: + for tag, value in exif.items(): + print(f" {tag}: {value}") + else: + print(" (无 EXIF 数据)") + + +def _apply_exif_modifications(exif: Any, set_items: list[str] | None, clear: bool) -> bool: + """应用 EXIF 修改 (clear + set), 返回是否有改动.""" + if clear: + for tag in list(exif.keys()): + del exif[tag] + if set_items: + for item in set_items: + _apply_single_exif_set(exif, item) + return bool(set_items or clear) + + +def _apply_single_exif_set(exif: Any, item: str) -> None: + """解析并应用单个 KEY=VALUE 设置项.""" + if "=" not in item: + print(f"跳过无效项 (缺少 =): {item}") + return + key_str, value = item.split("=", 1) + try: + tag = int(key_str) + except ValueError: + print(f"跳过无效标签号: {key_str}") + return + exif[tag] = value + + +def _save_exif(img: Any, exif: Any, output_path: Path) -> None: + """保存图片与 EXIF.""" + exif_bytes = exif.tobytes() if exif else b"" + img.save(output_path, exif=exif_bytes) + print(f"EXIF 已保存: {output_path}") + + +@px.tool("imagetool", subcommand="hi", help="颜色直方图") +def image_histogram( + input_path: Path, + channel: str = "rgb", +) -> None: + """打印颜色直方图统计 (每通道 8 桶). + + Parameters + ---------- + input_path : Path + 输入图片 + channel : str + 通道: "rgb" (R/G/B 三通道) / "luminance" (亮度单通道) + """ + if not _require_pil(): + return + + img = Image.open(input_path) + hist = img.histogram() + buckets = 8 + + if channel == "luminance": + gray = img.convert("L") + gray_hist = gray.histogram() + print("亮度直方图 (8 桶):") + _print_histogram_buckets(gray_hist, buckets, "L") + else: + print("RGB 直方图 (8 桶):") + if len(hist) == 256: + _print_histogram_buckets(hist, buckets, "L") + else: + for idx, name in enumerate(("R", "G", "B")): + start = idx * 256 + _print_histogram_buckets(hist[start : start + 256], buckets, name) + + +def _print_histogram_buckets(channel_hist: list[int], buckets: int, name: str) -> None: + """将单通道 256 桶直方图聚合为指定桶数并打印.""" + bucket_size = max(1, len(channel_hist) // buckets) + print(f" {name}:") + for i in range(buckets): + start = i * bucket_size + end = min((i + 1) * bucket_size, len(channel_hist)) + count = sum(channel_hist[start:end]) + slice_max = max(channel_hist[start:end] or [1]) + bar = "#" * min(40, count * 40 // max(1, slice_max)) + print(f" [{start:3d}-{end:3d}] {count:>8d} {bar}") + + +@px.tool("imagetool", subcommand="co", help="提取主色调") +def image_colors( + input_path: Path, + count: int = 5, +) -> None: + """提取并打印主色调. + + Parameters + ---------- + input_path : Path + 输入图片 + count : int + 提取的颜色数 (默认 5) + """ + if not _require_pil(): + return + + img = Image.open(input_path).convert("RGB") + quantized = img.quantize(colors=count) + palette = quantized.getpalette() + if palette is None: + print("无法提取调色板") + return + + actual_count = len(palette) // 3 + print(f"主色调 (前 {min(count, actual_count)} 色):") + for i in range(min(count, actual_count)): + r = palette[i * 3] + g = palette[i * 3 + 1] + b = palette[i * 3 + 2] + hex_color = f"#{r:02X}{g:02X}{b:02X}" + print(f" {i + 1}. {hex_color} rgb({r}, {g}, {b})") diff --git a/tests/cli/test_imagetool.py b/tests/cli/test_imagetool.py new file mode 100644 index 0000000..68792a3 --- /dev/null +++ b/tests/cli/test_imagetool.py @@ -0,0 +1,316 @@ +"""imagetool 子命令测试. + +用 ``pytest.importorskip("PIL")`` 跳过未安装 Pillow 的环境. +用真实 Pillow 操作 (非 subprocess), 创建测试图片验证. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from pyflowx.ops.files import imagetool + +pytest.importorskip("PIL") +from PIL import Image + + +def _make_test_image(path: Path, size: tuple[int, int] = (100, 80), color: str = "red") -> None: + """创建测试图片.""" + Image.new("RGB", size, color).save(path) + + +def _make_test_image_rgba(path: Path, size: tuple[int, int] = (100, 80)) -> None: + """创建带 alpha 通道的测试图片.""" + Image.new("RGBA", size, (255, 0, 0, 128)).save(path) + + +class TestImageResize: + """image_resize 测试.""" + + def test_resize_with_keep_ratio(self, tmp_path: Path) -> None: + """keep_ratio=True 时按宽度等比缩放.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + out = tmp_path / "out.png" + imagetool.image_resize(src, out, width=50) + assert out.exists() + with Image.open(out) as img: + assert img.size[0] == 50 + + def test_resize_without_keep_ratio(self, tmp_path: Path) -> None: + """keep_ratio=False 时强制缩放到指定尺寸.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + out = tmp_path / "out.png" + imagetool.image_resize(src, out, width=50, height=40, keep_ratio=False) + assert out.exists() + with Image.open(out) as img: + assert img.size == (50, 40) + + +class TestImageCrop: + """image_crop 测试.""" + + def test_crop(self, tmp_path: Path) -> None: + """裁剪到指定矩形.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + out = tmp_path / "out.png" + imagetool.image_crop(src, out, left=10, top=10, right=60, bottom=50) + assert out.exists() + with Image.open(out) as img: + assert img.size == (50, 40) + + +class TestImageRotate: + """image_rotate 测试.""" + + def test_rotate_90(self, tmp_path: Path) -> None: + """旋转 90 度.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + out = tmp_path / "out.png" + imagetool.image_rotate(src, out, degrees=90, expand=True) + assert out.exists() + with Image.open(out) as img: + assert img.size[0] == 80 + assert img.size[1] == 100 + + +class TestImageFlip: + """image_flip 测试.""" + + def test_flip_horizontal(self, tmp_path: Path) -> None: + """水平翻转.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + out = tmp_path / "out.png" + imagetool.image_flip(src, out, direction="horizontal") + assert out.exists() + + def test_flip_vertical(self, tmp_path: Path) -> None: + """垂直翻转.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + out = tmp_path / "out.png" + imagetool.image_flip(src, out, direction="vertical") + assert out.exists() + + +class TestImageConvert: + """image_convert 测试.""" + + def test_convert_to_jpeg(self, tmp_path: Path) -> None: + """PNG → JPEG 转换.""" + src = tmp_path / "src.png" + _make_test_image(src, (50, 50)) + out = tmp_path / "out.jpg" + imagetool.image_convert(src, out) + assert out.exists() + with Image.open(out) as img: + assert img.format == "JPEG" + + def test_convert_to_webp(self, tmp_path: Path) -> None: + """PNG → WEBP 转换.""" + src = tmp_path / "src.png" + _make_test_image(src, (50, 50)) + out = tmp_path / "out.webp" + imagetool.image_convert(src, out, quality=80) + assert out.exists() + with Image.open(out) as img: + assert img.format == "WEBP" + + def test_convert_rgba_to_jpeg(self, tmp_path: Path) -> None: + """RGBA → JPEG 自动去 alpha 通道.""" + src = tmp_path / "src.png" + _make_test_image_rgba(src, (50, 50)) + out = tmp_path / "out.jpg" + imagetool.image_convert(src, out) + assert out.exists() + with Image.open(out) as img: + assert img.format == "JPEG" + assert img.mode == "RGB" + + +class TestImageWatermark: + """image_watermark 测试.""" + + def test_watermark_text(self, tmp_path: Path) -> None: + """添加文字水印.""" + src = tmp_path / "src.png" + _make_test_image(src, (200, 150)) + out = tmp_path / "out.png" + imagetool.image_watermark(src, out, text="TEST", position="bottom-right", opacity=0.7) + assert out.exists() + with Image.open(out) as img: + assert img.size == (200, 150) + + def test_watermark_center_position(self, tmp_path: Path) -> None: + """center 位置水印.""" + src = tmp_path / "src.png" + _make_test_image(src, (200, 150)) + out = tmp_path / "out.png" + imagetool.image_watermark(src, out, text="CENTER", position="center") + assert out.exists() + + +class TestImageCompress: + """image_compress 测试.""" + + def test_compress_jpeg(self, tmp_path: Path) -> None: + """JPEG 压缩后文件存在.""" + src = tmp_path / "src.jpg" + _make_test_image(src, (100, 80)) + out = tmp_path / "out.jpg" + imagetool.image_compress(src, out, quality=50) + assert out.exists() + assert out.stat().st_size > 0 + + +class TestImageInfo: + """image_info 测试.""" + + def test_info_text(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """文本格式信息输出.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + imagetool.image_info(src, json=False) + captured = capsys.readouterr() + assert "格式: PNG" in captured.out + assert "100x80" in captured.out + + def test_info_json(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """JSON 格式信息输出.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + imagetool.image_info(src, json=True) + captured = capsys.readouterr() + import json + + data = json.loads(captured.out) + assert data["format"] == "PNG" + assert data["width"] == 100 + assert data["height"] == 80 + + +class TestImageExif: + """image_exif 测试.""" + + def test_exif_show_empty(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """无 EXIF 时打印提示.""" + src = tmp_path / "src.png" + _make_test_image(src, (50, 50)) + imagetool.image_exif(src, show=True) + captured = capsys.readouterr() + assert "无 EXIF 数据" in captured.out + + def test_exif_set_and_read(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """设置 EXIF 后能读取.""" + src = tmp_path / "src.jpg" + _make_test_image(src, (50, 50)) + out = tmp_path / "out.jpg" + imagetool.image_exif(src, output_path=out, show=False, set=["271=TestCamera"]) + assert out.exists() + + capsys.readouterr() + imagetool.image_exif(out, show=True) + captured = capsys.readouterr() + assert "271: TestCamera" in captured.out + + def test_exif_clear(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """清空 EXIF.""" + src = tmp_path / "src.jpg" + _make_test_image(src, (50, 50)) + out_set = tmp_path / "set.jpg" + imagetool.image_exif(src, output_path=out_set, show=False, set=["271=TestCamera"]) + + out_clear = tmp_path / "clear.jpg" + imagetool.image_exif(out_set, output_path=out_clear, show=False, clear=True) + + capsys.readouterr() + imagetool.image_exif(out_clear, show=True) + captured = capsys.readouterr() + assert "无 EXIF 数据" in captured.out + + def test_exif_invalid_set_item(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """无效设置项被跳过.""" + src = tmp_path / "src.jpg" + _make_test_image(src, (50, 50)) + out = tmp_path / "out.jpg" + imagetool.image_exif(src, output_path=out, show=False, set=["invalid", "abc=val"]) + captured = capsys.readouterr() + assert "跳过" in captured.out + + +class TestImageHistogram: + """image_histogram 测试.""" + + def test_histogram_rgb(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """RGB 直方图输出 3 通道.""" + src = tmp_path / "src.png" + _make_test_image(src, (50, 50), color="red") + imagetool.image_histogram(src, channel="rgb") + captured = capsys.readouterr() + assert "R:" in captured.out + assert "G:" in captured.out + assert "B:" in captured.out + + def test_histogram_luminance(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """亮度直方图输出 L 通道.""" + src = tmp_path / "src.png" + _make_test_image(src, (50, 50), color="red") + imagetool.image_histogram(src, channel="luminance") + captured = capsys.readouterr() + assert "L:" in captured.out + + +class TestImageColors: + """image_colors 测试.""" + + def test_colors_extraction(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """主色调提取输出指定数量颜色.""" + src = tmp_path / "src.png" + _make_test_image(src, (50, 50), color="red") + imagetool.image_colors(src, count=3) + captured = capsys.readouterr() + assert "主色调" in captured.out + assert "#FF0000" in captured.out or "#" in captured.out + + def test_colors_count(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """主色调数量符合请求.""" + src = tmp_path / "src.png" + _make_test_image(src, (50, 50), color="blue") + imagetool.image_colors(src, count=5) + captured = capsys.readouterr() + lines = [line for line in captured.out.splitlines() if line.strip().startswith(tuple("12345"))] + assert len(lines) <= 5 + + +class TestRequirePil: + """_require_pil 未安装时的行为.""" + + def test_require_pil_returns_false_when_missing( + self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + ) -> None: + """HAS_PIL=False 时打印提示并返回 False.""" + monkeypatch.setattr(imagetool, "HAS_PIL", False) + result = imagetool._require_pil() + assert result is False + captured = capsys.readouterr() + assert "pip install pyflowx[office]" in captured.out + + def test_require_pil_returns_true_when_installed(self, monkeypatch: pytest.MonkeyPatch) -> None: + """HAS_PIL=True 时返回 True.""" + monkeypatch.setattr(imagetool, "HAS_PIL", True) + assert imagetool._require_pil() is True + + def test_resize_no_pil_noop(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """未安装 PIL 时 resize 不创建输出文件.""" + monkeypatch.setattr(imagetool, "HAS_PIL", False) + src = tmp_path / "src.png" + _make_test_image(src, (50, 50)) + out = tmp_path / "out.png" + imagetool.image_resize(src, out, width=25) + assert not out.exists() diff --git a/tests/test_imaging.py b/tests/test_imaging.py new file mode 100644 index 0000000..9db5eee --- /dev/null +++ b/tests/test_imaging.py @@ -0,0 +1,158 @@ +"""image_pipeline DAG 构造器测试. + +验证: +* 生成的 Graph 拓扑正确 (任务数/依赖链). +* 实际执行后输出文件链完整. +* 未知操作抛 ValueError. +* convert 步骤改变扩展名. +* 自定义 output_dir 与 naming 模板. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +import pyflowx as px + +pytest.importorskip("PIL") +from PIL import Image + + +def _make_test_image(path: Path, size: tuple[int, int] = (100, 80), color: str = "red") -> None: + """创建测试图片.""" + Image.new("RGB", size, color).save(path) + + +class TestImagePipelineTopology: + """image_pipeline 生成的 Graph 拓扑测试.""" + + def test_single_step(self, tmp_path: Path) -> None: + """单步流水线生成 1 个任务无依赖.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + graph = px.image_pipeline(src, steps=[("resize", {"width": 50})]) + assert len(graph.all_specs()) == 1 + spec = next(iter(graph.all_specs().values())) + assert spec.depends_on == () + + def test_multi_step_chain(self, tmp_path: Path) -> None: + """多步流水线形成链式依赖.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + graph = px.image_pipeline( + src, + steps=[ + ("resize", {"width": 50}), + ("watermark", {"text": "X"}), + ("compress", {"quality": 80}), + ], + ) + specs = graph.all_specs() + assert len(specs) == 3 + names = list(specs.keys()) + assert specs[names[0]].depends_on == () + assert specs[names[1]].depends_on == (names[0],) + assert specs[names[2]].depends_on == (names[1],) + + def test_unknown_operation_raises(self, tmp_path: Path) -> None: + """未知操作抛 ValueError.""" + src = tmp_path / "src.png" + _make_test_image(src, (50, 50)) + with pytest.raises(ValueError, match="未知图片操作"): + px.image_pipeline(src, steps=[("unknown_op", {})]) + + +class TestImagePipelineExecution: + """image_pipeline 实际执行测试.""" + + def test_execute_resize_only(self, tmp_path: Path) -> None: + """单步 resize 执行后输出文件存在且尺寸正确.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + graph = px.image_pipeline(src, steps=[("resize", {"width": 50})]) + report = px.run(graph, strategy="sequential") + assert report.success + out = tmp_path / "src_resize.png" + assert out.exists() + with Image.open(out) as img: + assert img.size[0] == 50 + + def test_execute_chain(self, tmp_path: Path) -> None: + """链式执行: resize → watermark → convert.""" + src = tmp_path / "src.png" + _make_test_image(src, (200, 150)) + graph = px.image_pipeline( + src, + steps=[ + ("resize", {"width": 100}), + ("watermark", {"text": "TEST"}), + ("convert", {"format": "webp", "quality": 80}), + ], + ) + report = px.run(graph, strategy="sequential") + assert report.success + final_out = tmp_path / "src_resize_watermark_convert.webp" + assert final_out.exists() + with Image.open(final_out) as img: + assert img.format == "WEBP" + + def test_execute_with_output_dir(self, tmp_path: Path) -> None: + """自定义 output_dir 输出到指定目录.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + out_dir = tmp_path / "output" + graph = px.image_pipeline( + src, + steps=[("resize", {"width": 50})], + output_dir=out_dir, + ) + report = px.run(graph, strategy="sequential") + assert report.success + assert (out_dir / "src_resize.png").exists() + + def test_execute_convert_changes_extension(self, tmp_path: Path) -> None: + """convert 步骤改变输出扩展名.""" + src = tmp_path / "src.png" + _make_test_image(src, (50, 50)) + graph = px.image_pipeline( + src, + steps=[ + ("convert", {"format": "jpeg"}), + ], + ) + report = px.run(graph, strategy="sequential") + assert report.success + out = tmp_path / "src_convert.jpeg" + assert out.exists() + + def test_execute_custom_naming(self, tmp_path: Path) -> None: + """自定义 naming 模板.""" + src = tmp_path / "src.png" + _make_test_image(src, (50, 50)) + graph = px.image_pipeline( + src, + steps=[("resize", {"width": 25})], + naming="processed_{step}_{stem}{ext}", + ) + report = px.run(graph, strategy="sequential") + assert report.success + out = tmp_path / "processed_resize_src.png" + assert out.exists() + + def test_execute_flip_and_rotate(self, tmp_path: Path) -> None: + """flip + rotate 组合执行.""" + src = tmp_path / "src.png" + _make_test_image(src, (100, 80)) + graph = px.image_pipeline( + src, + steps=[ + ("flip", {"direction": "horizontal"}), + ("rotate", {"degrees": 90, "expand": True}), + ], + ) + report = px.run(graph, strategy="sequential") + assert report.success + out = tmp_path / "src_flip_rotate.png" + assert out.exists()