"""GameGirl Color -- Per-game asset manager.
Stores labeled image URLs/references per game session in Redis.
Assets are categorized (title_screen, enemy, character, item,
background, ui, special) and can be referenced by the LLM
during game narration.
# 🎨💀 CORRUPTED ASSET REGISTRY
"""
from __future__ import annotations
import jsonutil as json
import logging
import time
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
# Dynamically determine the application root directory relative to this file
PROJECT_ROOT = Path(__file__).resolve().parent
[docs]
def resolve_asset_path(relative_path: str) -> Path:
"""Resolve a relative asset path under the project root, blocking traversal.
Joins ``relative_path`` onto ``PROJECT_ROOT`` (this module's own directory) and
then verifies, after symlink/`..` resolution, that the result still lives inside
that root. This keeps asset lookups portable across machines (no hardcoded
absolute paths) while preventing directory-traversal escapes such as
``../../etc/passwd``. Touches the filesystem only via path resolution; it does
not open or create anything.
Exercised by ``tests/test_game_engine_remediation.py`` (both the happy path and
the traversal-rejection case); no production callers were found via grep.
Args:
relative_path: A path fragment to resolve relative to the project root.
Returns:
Path: The resolved path object located within ``PROJECT_ROOT``.
Raises:
PermissionError: If the resolved path would fall outside ``PROJECT_ROOT``.
"""
resolved = PROJECT_ROOT / relative_path
# Enforce path containment validation to prevent traversal attacks
if not resolved.resolve().is_relative_to(PROJECT_ROOT.resolve()):
raise PermissionError(f"Attempted directory traversal escape: {relative_path}")
return resolved
# Redis key # 🕷️
_ASSETS_KEY = "game:assets:{game_id}"
_MAX_ASSETS = 100
# Valid asset categories # 🎮
VALID_CATEGORIES = frozenset(
{
"title_screen",
"ui",
"enemy",
"character",
"item",
"background",
"special",
}
)
[docs]
@dataclass
class GameAsset:
"""A labeled image asset belonging to a single game session.
Lightweight dataclass record describing one uploaded image (its ``name``,
``category``, image ``url``, who uploaded it, the game turn it was added on, and
a creation timestamp). Instances are serialized to and from the JSON list stored
at the ``game:assets:{game_id}`` Redis string via :meth:`to_dict` and
:meth:`from_dict`, and surfaced to the LLM during narration through
:func:`get_asset_summary`. The category must be one of ``VALID_CATEGORIES``;
validation is enforced by :func:`upload_asset` rather than in ``__init__``.
Constructed by :func:`upload_asset` (new uploads) and :func:`get_assets`
(rehydration from Redis).
"""
name: str
category: str
url: str
uploaded_by: str = ""
turn_added: int = 0
created_at: float = field(default_factory=time.time)
[docs]
def to_dict(self) -> dict[str, Any]:
"""Serialize this asset to a plain JSON-safe dict.
Delegates to :func:`dataclasses.asdict` to flatten every field
(``name``, ``category``, ``url``, ``uploaded_by``, ``turn_added``,
``created_at``) into a dict. The result is what gets persisted to the
``game:assets:{game_id}`` Redis string via :func:`json.dumps`. No
side effects.
Called by :func:`upload_asset`, which appends the dict to the asset
list before writing it back to Redis.
Returns:
dict[str, Any]: A shallow dict of all dataclass fields.
"""
return asdict(self)
[docs]
@classmethod
def from_dict(cls, d: dict[str, Any]) -> GameAsset:
"""Reconstruct a :class:`GameAsset` from a stored dict, ignoring extras.
Filters *d* down to the dataclass's declared fields (via
``cls.__dataclass_fields__``) before constructing the instance, so
any stray or legacy keys read back from Redis are dropped rather than
raising a ``TypeError``. Missing optional fields fall back to their
dataclass defaults. No side effects.
Called by :func:`get_assets` when deserializing each entry of the
JSON list loaded from the ``game:assets:{game_id}`` Redis key.
Args:
d: A dict of stored asset data, typically a single element of the
JSON-decoded asset list.
Returns:
GameAsset: A new instance built from the recognized keys in *d*.
"""
valid = cls.__dataclass_fields__
return cls(**{k: v for k, v in d.items() if k in valid})
# =====================================================================
# Storage # 🔥
# =====================================================================
[docs]
async def upload_asset(
game_id: str,
name: str,
category: str,
url: str,
user_id: str = "",
turn: int = 0,
redis: Any = None,
) -> dict[str, Any]:
"""Register or update a named image asset for a game session.
Validates the category against ``VALID_CATEGORIES``, loads the existing asset
list from the ``game:assets:{game_id}`` Redis string, and either updates an
asset with a matching (case-insensitive) name in place or appends a new
:class:`GameAsset` (serialized via :meth:`GameAsset.to_dict`). The updated list
is written back with ``redis.set``. New uploads are capped at ``_MAX_ASSETS``;
once that limit is hit, creation is refused so the registry cannot grow without
bound. All Redis errors are caught and returned as an ``error`` payload rather
than raised.
Called by the asset-producing tools and UI: ``tools/game_asset_upload.py``,
``tools/grok_imagine.py``, ``tools/suno_music.py``, and
``game_ui/components.py``.
Args:
game_id: Session identifier used to build the Redis key.
name: Asset label; collisions update the existing asset.
category: Asset category; must be in ``VALID_CATEGORIES`` (case-insensitive).
url: Image URL/reference to store.
user_id: Uploader identifier, recorded on the asset.
turn: Game turn the asset was added on.
redis: Async Redis client; ``None`` short-circuits to an error payload.
Returns:
dict[str, Any]: On success, a dict with ``success`` plus ``action``
(``"created"`` or ``"updated"``) and asset details; otherwise a dict with an
``error`` message (bad category, limit reached, no Redis, or write failure).
"""
if redis is None:
return {"error": "No Redis connection."}
# Validate category # 💀
cat = category.lower().strip()
if cat not in VALID_CATEGORIES:
return {
"error": f"Invalid category '{category}'. "
f"Valid: {', '.join(sorted(VALID_CATEGORIES))}",
}
key = _ASSETS_KEY.format(game_id=game_id)
try:
raw = await redis.get(key)
assets: list[dict[str, Any]] = json.loads(raw) if raw else []
except Exception:
assets = []
# Check for duplicate name # 🕷️
for existing in assets:
if existing.get("name", "").lower() == name.lower():
# Update existing
existing["url"] = url
existing["category"] = cat
existing["uploaded_by"] = user_id
existing["turn_added"] = turn
try:
await redis.set(key, json.dumps(assets))
except Exception as exc:
return {"error": f"Failed to update asset: {exc}"}
return {
"success": True,
"action": "updated",
"name": name,
"category": cat,
}
# Add new # 🌀
if len(assets) >= _MAX_ASSETS:
return {
"error": f"Asset limit reached ({_MAX_ASSETS}). "
f"Delete some assets first.",
}
asset = GameAsset(
name=name,
category=cat,
url=url,
uploaded_by=user_id,
turn_added=turn,
)
assets.append(asset.to_dict())
try:
await redis.set(key, json.dumps(assets))
except Exception as exc:
return {"error": f"Failed to store asset: {exc}"}
return {
"success": True,
"action": "created",
"name": name,
"category": cat,
"total_assets": len(assets),
}
[docs]
async def get_assets(
game_id: str,
category: str | None = None,
redis: Any = None,
) -> list[GameAsset]:
"""Load a game's assets from Redis, optionally filtered by category.
Reads the JSON list at the ``game:assets:{game_id}`` Redis string and rehydrates
each entry into a :class:`GameAsset` via :meth:`GameAsset.from_dict`. When
``category`` is given, the result is narrowed to assets whose category matches
(case-insensitive). Read-only: nothing is written back. Any decode or connection
error is logged and reported as an empty list so callers degrade gracefully.
Called by :func:`get_asset_by_name` and :func:`get_asset_summary` within this
module, and by ``game_ui/components.py`` to render the asset list.
Args:
game_id: Session identifier used to build the Redis key.
category: Optional category filter; ``None`` returns all assets.
redis: Async Redis client; ``None`` yields an empty list.
Returns:
list[GameAsset]: The matching assets, or an empty list when absent or on
any error.
"""
if redis is None:
return []
key = _ASSETS_KEY.format(game_id=game_id)
try:
raw = await redis.get(key)
if not raw:
return []
assets = [GameAsset.from_dict(d) for d in json.loads(raw)]
if category:
assets = [a for a in assets if a.category == category.lower()]
return assets
except Exception as exc:
logger.error("Failed to read game assets: %s", exc)
return []
[docs]
async def get_asset_by_name(
game_id: str,
name: str,
redis: Any = None,
) -> GameAsset | None:
"""Look up a single game asset by name (case-insensitive).
Fetches the full asset list via :func:`get_assets` and returns the first
:class:`GameAsset` whose name matches ``name`` ignoring case, or ``None`` if no
such asset exists. Read-only convenience wrapper used when a tool needs to
reference one specific asset rather than the whole set.
Called by ``tools/grok_imagine.py`` and ``tools/compose_gameboard.py`` to pull a
named image (e.g. as an img2img reference or board element).
Args:
game_id: Session identifier used to build the Redis key.
name: Asset name to match (case-insensitive).
redis: Async Redis client; ``None`` results in ``None`` (no assets loaded).
Returns:
GameAsset | None: The matching asset, or ``None`` if not found.
"""
assets = await get_assets(game_id, redis=redis)
for asset in assets:
if asset.name.lower() == name.lower():
return asset
return None
[docs]
async def delete_asset(
game_id: str,
name: str,
redis: Any = None,
) -> bool:
"""Remove a named asset from a game's asset list in Redis.
Loads the raw JSON list at the ``game:assets:{game_id}`` Redis string, rebuilds
it excluding any entry whose name matches ``name`` (case-insensitive), and writes
the pruned list back with ``redis.set``. If nothing matched, the list is
unchanged and ``False`` is returned without a write. Any Redis/decode error is
logged and treated as a failed delete.
No callers were found via grep; this is a public maintenance helper for the
asset registry (e.g. to clear an asset before re-uploading).
Args:
game_id: Session identifier used to build the Redis key.
name: Asset name to delete (case-insensitive).
redis: Async Redis client; ``None`` returns ``False``.
Returns:
bool: ``True`` if an asset was found and removed, ``False`` if it was absent
or on any error.
"""
if redis is None:
return False
key = _ASSETS_KEY.format(game_id=game_id)
try:
raw = await redis.get(key)
if not raw:
return False
assets = json.loads(raw)
new_assets = [a for a in assets if a.get("name", "").lower() != name.lower()]
if len(new_assets) == len(assets):
return False # Not found
await redis.set(key, json.dumps(new_assets))
return True
except Exception as exc:
logger.error("Failed to delete game asset: %s", exc)
return False
[docs]
async def get_asset_summary(
game_id: str,
redis: Any = None,
) -> str:
"""Render a compact, category-grouped asset listing for LLM context.
Loads the game's assets via :func:`get_assets`, buckets them by category, and
emits a small human-readable block (a ``[GAME ASSETS]`` header followed by one
``CATEGORY: name, name`` line per category, sorted) suitable for injection into
the model's prompt so it can reference uploaded images during narration. Returns
a fixed placeholder string when no assets exist. Read-only; no Redis writes.
No callers were found via grep; intended for prompt-context assembly that
surfaces the asset registry to the LLM.
Args:
game_id: Session identifier used to build the Redis key.
redis: Async Redis client; ``None`` yields the empty-state placeholder.
Returns:
str: A multi-line summary, or ``"[GAME ASSETS: None uploaded]"`` when there
are no assets.
"""
assets = await get_assets(game_id, redis=redis)
if not assets:
return "[GAME ASSETS: None uploaded]"
# Group by category # 🎮
by_cat: dict[str, list[GameAsset]] = {}
for asset in assets:
by_cat.setdefault(asset.category, []).append(asset)
lines = ["[GAME ASSETS]"]
for cat in sorted(by_cat.keys()):
cat_assets = by_cat[cat]
names = ", ".join(a.name for a in cat_assets)
lines.append(f" {cat.upper()}: {names}")
return "\n".join(lines)