nexus/tools/image_processor.py

"""
Image Processor Tool - Resize/crop/rotate/convert/watermark/info ảnh.
===========================================
Dùng Pillow (PIL) lazy import. Hỗ trợ JPEG/PNG/GIF/WEBP/BMP/TIFF.

Author: Hieu Louis (2026)
"""
from __future__ import annotations

import os
from typing import Any, Dict, List, Optional

from .base import Tool, ToolResult, ToolContext, ToolCategory, ToolSafety


OPERATIONS = {"resize", "crop", "rotate", "convert", "watermark", "info", "thumbnail"}
SUPPORTED_FORMATS = {"JPEG", "PNG", "GIF", "WEBP", "BMP", "TIFF"}


class ImageProcessorTool(Tool):
    """Xử lý ảnh: resize, crop, rotate, convert, watermark, info."""

    category = ToolCategory.MEDIA
    safety = ToolSafety.MODERATE
    requires_confirmation = True

    @property
    def name(self) -> str:
        return "image_processor"

    @property
    def description(self) -> str:
        return "Image ops (resize/crop/rotate/convert/watermark/info) bằng Pillow."

    @property
    def parameters(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "input_path": {"type": "string"},
                "output_path": {"type": "string", "description": "Output file (bỏ qua cho 'info')"},
                "operation": {
                    "type": "string",
                    "enum": sorted(OPERATIONS),
                    "default": "info",
                },
                "params": {
                    "type": "object",
                    "description": "Operation-specific params (width/height/crop_box/angle/format/...)",
                },
            },
            "required": ["input_path", "operation"],
        }

    def validate_args(self, args: Dict[str, Any]) -> Optional[str]:
        if not args.get("input_path"):
            return "Missing required arg: input_path"
        op = args.get("operation", "info")
        if op not in OPERATIONS:
            return f"Invalid operation='{op}'. Supported: {sorted(OPERATIONS)}"
        if op != "info" and not args.get("output_path"):
            return f"Missing required arg: output_path (cho operation='{op}')"
        return None

    # ---- Helpers --------------------------------------------------------

    def _watermark(self, img: Any, text: str) -> Any:
        """Vẽ watermark text lên ảnh bằng ImageDraw (stdlib của PIL)."""
        from PIL import ImageDraw, ImageFont  # type: ignore
        draw = ImageDraw.Draw(img)
        try:
            font = ImageFont.truetype("DejaVuSans.ttf", max(12, img.width // 30))
        except Exception:
            font = ImageFont.load_default()
        # Tính bounding box text / compute text bbox
        try:
            bbox = draw.textbbox((0, 0), text, font=font)
            tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1]
        except AttributeError:
            tw, th = font.getsize(text)
        x = img.width - tw - 10
        y = img.height - th - 10
        # Shadow + text / shadow + main text
        draw.text((x + 2, y + 2), text, fill=(0, 0, 0, 128), font=font)
        draw.text((x, y), text, fill=(255, 255, 255, 200), font=font)
        return img

    # ---- Execute --------------------------------------------------------

    def execute(self, args: Dict[str, Any], context: ToolContext) -> ToolResult:
        input_path = args["input_path"]
        output_path = args.get("output_path")
        op = args.get("operation", "info")
        params: Dict[str, Any] = args.get("params", {}) or {}

        if not os.path.exists(input_path):
            return ToolResult(success=False, error=f"Input không tồn tại: {input_path}", return_code=1)

        if context.dry_run:
            return ToolResult(
                success=True,
                output=f"[dry-run] Sẽ thực hiện '{op}' trên {input_path}",
                metadata={"operation": op, "input_path": input_path, "dry_run": True},
            )

        try:
            from PIL import Image  # type: ignore
        except ImportError:
            return ToolResult(
                success=False,
                error="Pillow chưa cài. Cài đặt: pip install Pillow",
                return_code=127,
            )

        try:
            img = Image.open(input_path)
            # Force load để detect lỗi sớm / force load to catch errors early
            img.load()
            fmt = img.format or os.path.splitext(input_path)[1][1:].upper()

            if op == "info":
                meta: Dict[str, Any] = {
                    "path": input_path,
                    "format": fmt,
                    "size": list(img.size),
                    "width": img.width,
                    "height": img.height,
                    "mode": img.mode,
                    "is_animated": getattr(img, "is_animated", False),
                    "n_frames": getattr(img, "n_frames", 1),
                    "file_size_bytes": os.path.getsize(input_path),
                }
                # EXIF / read EXIF if available
                try:
                    exif = img._getexif()  # type: ignore[attr-defined]
                    if exif:
                        meta["exif"] = {k: str(v) for k, v in exif.items()}
                except Exception:
                    pass
                return ToolResult(
                    success=True,
                    output=f"{fmt} {img.width}x{img.height} mode={img.mode} ({os.path.getsize(input_path)} bytes)",
                    metadata=meta,
                )

            if op == "resize":
                w = int(params.get("width", img.width // 2))
                h = int(params.get("height", img.height // 2))
                resample_name = params.get("resample", "LANCZOS")
                resample = getattr(Image, resample_name, Image.LANCZOS)
                result = img.resize((w, h), resample)
            elif op == "thumbnail":
                w = int(params.get("width", 256))
                h = int(params.get("height", 256))
                result = img.copy()
                result.thumbnail((w, h), Image.LANCZOS)
            elif op == "crop":
                box = params.get("crop_box") or (
                    int(params.get("left", 0)),
                    int(params.get("top", 0)),
                    int(params.get("right", img.width)),
                    int(params.get("bottom", img.height)),
                )
                result = img.crop(box)
            elif op == "rotate":
                angle = float(params.get("angle", 90))
                expand = bool(params.get("expand", True))
                result = img.rotate(angle, expand=expand)
            elif op == "convert":
                target_fmt = str(params.get("format", "PNG")).upper()
                if target_fmt not in SUPPORTED_FORMATS:
                    return ToolResult(success=False, error=f"Unsupported format: {target_fmt}", return_code=1)
                # Convert mode nếu cần / convert mode for format
                if target_fmt == "JPEG" and img.mode in ("RGBA", "P", "LA"):
                    result = img.convert("RGB")
                else:
                    result = img.copy()
                fmt = target_fmt
            elif op == "watermark":
                text = str(params.get("text", "© Nexus Coder"))
                result = img.copy()
                if result.mode != "RGBA":
                    result = result.convert("RGBA")
                result = self._watermark(result, text)
            else:
                return ToolResult(success=False, error=f"Unknown operation: {op}", return_code=1)

            # Tự suy ra format từ extension / infer format from output extension
            save_fmt = fmt
            ext = os.path.splitext(output_path)[1][1:].upper() if output_path else None
            if ext and ext in SUPPORTED_FORMATS:
                save_fmt = ext
            # Đảm bảo thư mục cha / ensure parent dir
            os.makedirs(os.path.dirname(os.path.abspath(output_path)), exist_ok=True)
            save_kwargs: Dict[str, Any] = {}
            if save_fmt == "JPEG":
                save_kwargs["quality"] = int(params.get("quality", 85))
            result.save(output_path, format=save_fmt, **save_kwargs)
            return ToolResult(
                success=True,
                output=f"{op} → {output_path} ({result.width}x{result.height}, {save_fmt})",
                artifacts=[output_path],
                metadata={
                    "operation": op,
                    "input_path": input_path,
                    "output_path": output_path,
                    "input_size": list(img.size),
                    "output_size": list(result.size),
                    "format": save_fmt,
                },
            )
        except Exception as e:
            return ToolResult(success=False, error=f"Image op failed: {e}", return_code=1)