"""GameGirl Color -- Pillow-based gameboard compositor.
Layers character sprites, enemies, and items onto scene
backgrounds at specified coordinates with z-ordering.
# 🎮💀 CORRUPTED CANVAS ENGINE
"""
from __future__ import annotations
import asyncio
import jsonutil as json
import logging
from io import BytesIO
from typing import Any, TYPE_CHECKING
import httpx
from tools._safe_http import safe_http_request, safe_httpx_client
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "compose_gameboard"
TOOL_DESCRIPTION = (
"Compose a gameboard scene by layering game assets (character "
"sprites, enemies, items) onto a background image at specified "
"coordinates. Uses saved game assets by name. The composed "
"image is sent to the current channel."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"background": {
"type": "string",
"description": (
"Name of a saved background asset, or a URL to " "a background image."
),
},
"sprites": {
"type": "array",
"description": ("List of sprites to layer onto the background."),
"items": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Asset name or image URL.",
},
"x": {
"type": "integer",
"description": "X position (pixels from left).",
},
"y": {
"type": "integer",
"description": "Y position (pixels from top).",
},
"scale": {
"type": "number",
"description": "Scale factor (1.0 = original). Default: 1.0.",
},
"z_order": {
"type": "integer",
"description": "Layer order (higher = on top). Default: 0.",
},
},
"required": ["name", "x", "y"],
},
},
"width": {
"type": "integer",
"description": "Output canvas width. Default: 1024.",
},
"height": {
"type": "integer",
"description": "Output canvas height. Default: 576.",
},
},
"required": ["background", "sprites"],
}
async def _download_image(url: str) -> bytes | None:
"""Download image bytes from a URL with SSRF protection.
Fetches the URL through ``tools._safe_http`` (so the target host is validated
and every redirect is re-checked against the block list, capped at five hops),
returning the body only on an HTTP 200. A blocked URL or any transport error is
logged at warning level and yields ``None``, letting the caller treat a bad
asset as simply absent.
Called by :func:`run` to fetch both the background and each sprite (via the URLs
that :func:`_resolve_asset_url` produces). Its side effect is the outbound HTTP
request made by the safe ``httpx`` client.
Args:
url (str): The image URL to fetch (whitespace-trimmed).
Returns:
bytes | None: The image bytes on success, or ``None`` when blocked, on a
non-200 status, or on error.
"""
url = (url or "").strip()
try:
async with safe_httpx_client(timeout=httpx.Timeout(30.0)) as client:
resp = await safe_http_request(client, "GET", url, max_redirects=5)
if resp.status_code == 200:
return resp.content
except ValueError as exc:
logger.warning("Blocked gameboard image URL: %s", exc)
except Exception as exc:
logger.warning("Failed to download image %s: %s", url[:80], exc)
return None
async def _resolve_asset_url(
name: str,
game_id: str | None,
redis: Any = None,
) -> str:
"""Resolve a saved game-asset name to a downloadable URL.
Lets callers refer to backgrounds and sprites by friendly asset name instead of
a raw URL: an explicit ``http(s)`` value is returned unchanged, otherwise (when
a game session and Redis client are available) it looks the name up for the
given game via ``game_assets.get_asset_by_name`` and returns that asset's URL.
If no game context or matching asset is found it returns the input as-is, since
it may already be a protocol-less URL.
Called by :func:`run` once for the background and once per sprite before
downloading. Its only side effect is the Redis-backed asset lookup, and lookup
errors are swallowed so an unknown name degrades to a passthrough.
Args:
name (str): An asset name or a URL.
game_id (str | None): Active game id used to scope the asset lookup; lookup
is skipped when ``None``.
redis (Any): Redis client used by the asset lookup; lookup is skipped when
falsy.
Returns:
str: A resolved URL, or the original ``name`` when no asset matched.
"""
if name.startswith("http://") or name.startswith("https://"):
return name
if game_id and redis:
try:
from game_assets import get_asset_by_name
asset = await get_asset_by_name(game_id, name, redis=redis)
if asset:
return asset.url
except Exception:
pass
return name # Return as-is, might be a URL without protocol
def _compose_gameboard_sync(
bg_data: bytes,
sprite_items: list[tuple[bytes, int, int, float]],
width: int,
height: int,
) -> tuple[bytes | None, str]:
"""Composite sprites over a background into a single WEBP image.
The CPU-bound core of the tool: it opens the background with Pillow, resizes it
to the requested canvas, and pastes each already-downloaded sprite at its
coordinates (scaling it first when requested and using the sprite's own alpha as
the paste mask), then encodes the result to WEBP bytes. A missing Pillow install
or a fatal error returns an error message; a single bad sprite is logged and
skipped rather than aborting the whole board.
Called by :func:`run` via :func:`asyncio.to_thread` so the blocking image work
runs off the event loop. It does no I/O of its own -- it operates purely on the
in-memory bytes it is handed.
Args:
bg_data (bytes): Background image bytes.
sprite_items (list[tuple[bytes, int, int, float]]): Per sprite the image
bytes plus ``x``, ``y``, and ``scale``, pre-sorted by z-order.
width (int): Output canvas width in pixels.
height (int): Output canvas height in pixels.
Returns:
tuple[bytes | None, str]: The encoded WEBP bytes with an empty error string
on success, or ``(None, message)`` describing the failure.
"""
try:
from PIL import Image
except ImportError:
return None, "Pillow not installed. Run: pip install Pillow"
try:
bg_img = Image.open(BytesIO(bg_data)).convert("RGBA")
bg_img = bg_img.resize((width, height), Image.LANCZOS)
canvas = bg_img.copy()
for sprite_data, x, y, scale in sprite_items:
try:
sprite_img = Image.open(BytesIO(sprite_data)).convert("RGBA")
if scale != 1.0:
new_w = int(sprite_img.width * scale)
new_h = int(sprite_img.height * scale)
sprite_img = sprite_img.resize(
(max(1, new_w), max(1, new_h)),
Image.LANCZOS,
)
canvas.paste(sprite_img, (x, y), sprite_img)
except Exception as exc:
logger.warning("Failed to composite sprite bytes: %s", exc)
output = BytesIO()
canvas.save(output, format="WEBP", quality=85)
output.seek(0)
return output.read(), ""
except Exception as exc:
return None, str(exc)
[docs]
async def run(
background: str,
sprites: list[dict[str, Any]] | None = None,
width: int = 1024,
height: int = 576,
ctx: ToolContext | None = None,
) -> str:
"""Compose a gameboard by layering sprites over a background and post it.
Entry point for the ``compose_gameboard`` tool. It looks up the active game
session for the current channel (via ``game_session.get_session``) to scope
asset resolution, resolves and downloads the background and each sprite
(:func:`_resolve_asset_url` then :func:`_download_image`), sorts the sprites by
``z_order`` so higher layers land on top, runs the Pillow compositing off-thread
with :func:`_compose_gameboard_sync`, and then uploads the resulting WEBP to the
channel through ``ctx.adapter.send_file``, also appending it to ``ctx.sent_files``
so the inference layer tracks the attachment. Missing assets are skipped rather
than failing the whole board.
Dispatched by ``tool_loader.py`` as the ``compose_gameboard`` handler (located
via ``getattr(module, "run")``); not called directly elsewhere. Side effects
include the asset HTTP downloads, the Redis-backed asset/session lookups, and
the outbound file send to the channel.
Args:
background (str): Background asset name or image URL.
sprites (list[dict[str, Any]] | None): Sprite placements, each with ``name``
plus ``x``/``y`` and optional ``scale`` and ``z_order``.
width (int): Output canvas width in pixels.
height (int): Output canvas height in pixels.
ctx (ToolContext | None): Tool execution context supplying ``channel_id``,
``redis``, ``adapter``, and ``sent_files``.
Returns:
str: A JSON object with ``success`` and board metadata (and ``file_url``
when the upload returned one), or an ``{"error": ...}`` object on a missing
context, a failed background download, or a compositing/send failure.
"""
if ctx is None:
return json.dumps({"error": "No tool context available."})
# Get game session for asset resolution # 🌀
game_id = None
redis = getattr(ctx, "redis", None)
try:
from game_session import get_session
session = get_session(str(ctx.channel_id))
if session and session.active:
game_id = session.game_id
except ImportError:
pass
# Resolve and download background # 🔥
bg_url = await _resolve_asset_url(background, game_id, redis)
bg_data = await _download_image(bg_url)
if bg_data is None:
return json.dumps(
{
"error": f"Failed to download background: {background}",
}
)
sprite_items: list[tuple[bytes, int, int, float]] = []
if sprites:
sprites_sorted = sorted(
sprites,
key=lambda s: s.get("z_order", 0),
)
for sprite_def in sprites_sorted:
name = sprite_def.get("name", "")
x = int(sprite_def.get("x", 0))
y = int(sprite_def.get("y", 0))
scale = float(sprite_def.get("scale", 1.0))
sprite_url = await _resolve_asset_url(name, game_id, redis)
sprite_data = await _download_image(sprite_url)
if sprite_data is None:
logger.warning("Skipping sprite '%s': download failed", name)
continue
sprite_items.append((sprite_data, x, y, scale))
composed_bytes, compose_err = await asyncio.to_thread(
_compose_gameboard_sync,
bg_data,
sprite_items,
width,
height,
)
if compose_err:
return json.dumps({"error": f"Failed to compose gameboard: {compose_err}"})
if not composed_bytes:
return json.dumps({"error": "Failed to compose gameboard (empty output)."})
channel_id = str(ctx.channel_id)
try:
file_url = await ctx.adapter.send_file(
channel_id,
composed_bytes,
"gameboard.webp",
"image/webp",
)
ctx.sent_files.append(
{
"data": composed_bytes,
"filename": "gameboard.webp",
"mimetype": "image/webp",
"file_url": file_url or "",
}
)
except Exception as exc:
return json.dumps({"error": f"Failed to send composed image: {exc}"})
result_info: dict = {
"success": True,
"width": width,
"height": height,
"filename": "gameboard.webp",
"sprites_count": len(sprites) if sprites else 0,
}
if file_url:
result_info["file_url"] = file_url
return json.dumps(result_info)