"""Per-user character persistence for S.N.E.S. game sessions.
Characters are stored per-user (not per-game) and automatically
loaded into every game session. The last created character is the
default. Character images are persisted to the filesystem at
/home/star/large_files/assets/ for permanence.
# ðŸŽðŸ’€ POSSESSED AVATAR REGISTRY
"""
from __future__ import annotations
import jsonutil as json
import logging
import os
import time
import uuid
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
# Redis key patterns # \ud83d\udd77\ufe0f
_CHARACTERS_KEY = "game:characters:{user_id}" # Hash: char_id -> JSON
_ACTIVE_CHAR_KEY = "game:active_char:{user_id}" # String: char_id
# Persistent file storage -- web-accessible for img2img # 💀🔥
_ASSET_DIR = Path("/home/star/large_files/assets/game")
_PUBLIC_URL_BASE = "https://sg.neko.li/assets/egregores/game"
def _safe_filename(name: str) -> str:
"""Slugify a character name into a filesystem- and URL-safe stem.
Replaces every non-alphanumeric/underscore character with an underscore,
collapses runs of underscores, trims leading/trailing underscores, lower-cases
the result, and caps it at 50 characters; an empty result falls back to
``"unnamed"``. This keeps generated image filenames (and the public URLs derived
from them) clean and predictable. Pure string transformation with no side
effects.
Called by :func:`create_character` to build the stored image filename.
Args:
name: The raw, user-supplied character name.
Returns:
str: A lower-cased, underscore-safe stem (1-50 chars), or ``"unnamed"``.
"""
import re as _re
safe = _re.sub(r"[^a-zA-Z0-9_]", "_", name.strip()).strip("_")
safe = _re.sub(r"_+", "_", safe)
return safe.lower()[:50] or "unnamed"
[docs]
async def create_character(
user_id: str,
name: str,
description: str,
redis: Any,
image_data: bytes | None = None,
image_url: str | None = None,
) -> dict[str, Any]:
"""Create a new character for a user.
Images are saved to the web-accessible game assets directory so
img2img can reference them via public URL.
"""
char_id = uuid.uuid4().hex[:10]
image_path = ""
public_image_url = ""
# Save image to web-accessible game assets # 💀🔥
if image_data:
try:
_ASSET_DIR.mkdir(parents=True, exist_ok=True)
safe_name = _safe_filename(name)
ext = "png"
filename = f"{safe_name}_{char_id}.{ext}"
image_path = str(_ASSET_DIR / filename)
with open(image_path, "wb") as f:
f.write(image_data)
public_image_url = f"{_PUBLIC_URL_BASE}/{filename}"
logger.info(
"Saved character image: %s (%d bytes) -> %s",
image_path,
len(image_data),
public_image_url,
)
except Exception as exc:
logger.error("Failed to save character image: %s", exc)
image_path = ""
public_image_url = ""
character = {
"char_id": char_id,
"name": name,
"description": description,
"image_url": public_image_url or image_url or "",
"image_path": image_path,
"public_url": public_image_url,
"created_at": time.time(),
"last_used": time.time(),
}
key = _CHARACTERS_KEY.format(user_id=user_id)
try:
await redis.hset(key, char_id, json.dumps(character))
await set_active_character(user_id, char_id, redis)
logger.info(
"Created character '%s' (%s) for user %s -> %s",
name,
char_id,
user_id,
public_image_url or "(no image)",
)
except Exception as exc:
logger.error("Failed to save character: %s", exc)
return {"success": False, "error": str(exc)}
return {
"success": True,
"char_id": char_id,
"name": name,
"description": description,
"public_url": public_image_url,
}
[docs]
async def list_characters(
user_id: str,
redis: Any,
) -> list[dict[str, Any]]:
"""List a user's saved characters, most-recently-used first.
Reads the per-user Redis hash ``game:characters:<user_id>`` (field = ``char_id``,
value = JSON), decodes every entry (handling both ``str`` and ``bytes`` from the
client), stamps each dict with its ``char_id``, and returns them sorted by the
``last_used`` timestamp descending so the active/default character sorts first.
Read-only; any decode or connection error is logged and reported as an empty
list.
Called by :func:`get_active_character` as a fallback (most-recent character) and
by ``game_ui/components.py`` to render the character roster.
Args:
user_id: Owner whose character hash to read.
redis: Async Redis client.
Returns:
list[dict[str, Any]]: Character dicts newest-used first, or an empty list
when none exist or on error.
""" # \U0001f3ad
key = _CHARACTERS_KEY.format(user_id=user_id)
try:
raw = await redis.hgetall(key)
if not raw:
return []
characters = []
for cid, data in raw.items():
cid_str = cid if isinstance(cid, str) else cid.decode()
data_str = data if isinstance(data, str) else data.decode()
char = json.loads(data_str)
char["char_id"] = cid_str
characters.append(char)
return sorted(
characters,
key=lambda c: c.get("last_used", 0),
reverse=True,
)
except Exception as exc:
logger.error("Failed to list characters: %s", exc)
return []
[docs]
async def get_character(
user_id: str,
char_id: str,
redis: Any,
) -> dict[str, Any] | None:
"""Fetch one of a user's characters by its identifier.
Reads a single field from the per-user Redis hash
``game:characters:<user_id>`` keyed by ``char_id`` and JSON-decodes it (handling
both ``str`` and ``bytes`` values). Returns ``None`` if the field is missing.
Read-only; any decode/connection error is logged and surfaced as ``None``.
Called by :func:`get_active_character` to resolve the stored active id; no
external callers were found via grep.
Args:
user_id: Owner of the character hash.
char_id: Identifier of the character to fetch.
redis: Async Redis client.
Returns:
dict[str, Any] | None: The decoded character dict, or ``None`` if absent or
on error.
""" # \U0001f464
key = _CHARACTERS_KEY.format(user_id=user_id)
try:
raw = await redis.hget(key, char_id)
if raw is None:
return None
data_str = raw if isinstance(raw, str) else raw.decode()
return json.loads(data_str)
except Exception as exc:
logger.error("Failed to get character: %s", exc)
return None
[docs]
async def get_active_character(
user_id: str,
redis: Any,
) -> dict[str, Any] | None:
"""Resolve a user's active character, falling back to most-recent.
Reads the active-pointer string ``game:active_char:<user_id>`` from Redis; if it
is set, resolves the character via :func:`get_character`. If no active pointer
exists, it falls back to the most-recently-used character returned by
:func:`list_characters` so a user who never explicitly chose one still gets a
sensible default. Read-only; any error is logged and surfaced as ``None``.
This is the most widely used accessor in the module: it is invoked across the
message pipeline and platforms -- e.g. ``message_processor/context_injections.py``,
``message_processor/processor.py``, ``message_processor/generate_and_send.py``,
``platforms/discord.py``, ``background_agents/game_art_agent.py``,
``tools/game_controls.py``, and ``openrouter_client/executor.py`` -- to inject the
player's character into prompts and image generation.
Args:
user_id: Owner whose active character to resolve.
redis: Async Redis client.
Returns:
dict[str, Any] | None: The active (or most-recent) character dict, or
``None`` if the user has none or on error.
""" # \U0001f300
active_key = _ACTIVE_CHAR_KEY.format(user_id=user_id)
try:
char_id_raw = await redis.get(active_key)
if char_id_raw is None:
# Fall back to most recently created # \ud83d\udc80
chars = await list_characters(user_id, redis)
return chars[0] if chars else None
char_id = char_id_raw if isinstance(char_id_raw, str) else char_id_raw.decode()
return await get_character(user_id, char_id, redis)
except Exception as exc:
logger.error("Failed to get active character: %s", exc)
return None
[docs]
async def set_active_character(
user_id: str,
char_id: str,
redis: Any,
) -> None:
"""Mark a character active for a user and refresh its last-used time.
Writes the ``char_id`` into the active-pointer string
``game:active_char:<user_id>``, then reads that character back from the per-user
hash ``game:characters:<user_id>``, updates its ``last_used`` timestamp, and
persists the modified JSON. Keeping ``last_used`` current also influences the
most-recent fallback in :func:`get_active_character` and the ordering in
:func:`list_characters`. Errors are logged and swallowed (best-effort write).
Called by :func:`create_character` (to auto-activate a freshly created
character) and by the UI/platform layers ``game_ui/modals/lobby.py`` and
``platforms/discord.py`` when a user selects a character.
Args:
user_id: Owner of the character.
char_id: Identifier of the character to make active.
redis: Async Redis client.
Returns:
None.
""" # \U0001f525
active_key = _ACTIVE_CHAR_KEY.format(user_id=user_id)
try:
await redis.set(active_key, char_id)
# Update last_used timestamp # \ud83d\udc80
key = _CHARACTERS_KEY.format(user_id=user_id)
raw = await redis.hget(key, char_id)
if raw:
data_str = raw if isinstance(raw, str) else raw.decode()
char = json.loads(data_str)
char["last_used"] = time.time()
await redis.hset(key, char_id, json.dumps(char))
except Exception as exc:
logger.error("Failed to set active character: %s", exc)
[docs]
async def delete_character(
user_id: str,
char_id: str,
redis: Any,
) -> bool:
"""Delete a user's character, its image file, and any active pointer.
Reads the character from the per-user hash ``game:characters:<user_id>`` to find
its persisted ``image_path`` and best-effort removes that file from disk, then
deletes the character field from the hash with ``HDEL``. If the deleted character
was the one
recorded in the active-pointer string ``game:active_char:<user_id>``, that
pointer is cleared so no dangling active id remains. Touches both Redis and the
filesystem; errors are logged and treated as a failed delete.
No external callers were found via grep; this is the public teardown counterpart
to :func:`create_character`.
Args:
user_id: Owner of the character.
char_id: Identifier of the character to delete.
redis: Async Redis client.
Returns:
bool: ``True`` if a character field was removed, ``False`` if it was absent
or on error.
""" # \U0001f480
key = _CHARACTERS_KEY.format(user_id=user_id)
try:
# Delete image file if exists
raw = await redis.hget(key, char_id)
if raw:
data_str = raw if isinstance(raw, str) else raw.decode()
char = json.loads(data_str)
image_path = char.get("image_path", "")
if image_path and os.path.exists(image_path):
try:
os.remove(image_path)
except Exception:
pass
count = await redis.hdel(key, char_id)
# If deleted char was active, clear active # \ud83c\udf00
active_key = _ACTIVE_CHAR_KEY.format(user_id=user_id)
active_raw = await redis.get(active_key)
if active_raw:
active_id = (
active_raw if isinstance(active_raw, str) else active_raw.decode()
)
if active_id == char_id:
await redis.delete(active_key)
return count > 0
except Exception as exc:
logger.error("Failed to delete character: %s", exc)
return False
[docs]
async def get_character_image_data(
char: dict[str, Any],
) -> bytes | None:
"""Read a character's persisted image bytes from disk, if present.
Pulls the ``image_path`` recorded on the character dict (written by
:func:`create_character` when an image was supplied) and, if that path exists,
reads and returns the raw file bytes -- e.g. for re-use as an img2img reference.
Returns ``None`` when the character has no stored image or the file is missing,
and logs and swallows any read error. Touches the filesystem only (no Redis).
No callers were found via grep; intended for consumers needing the original
image bytes rather than the public URL stored on the character.
Args:
char: A character dict, typically from :func:`get_active_character` or
:func:`get_character`, expected to carry an ``image_path`` key.
Returns:
bytes | None: The image file contents, or ``None`` if absent/unreadable.
""" # \U0001f4be
image_path = char.get("image_path", "")
if not image_path or not os.path.exists(image_path):
return None
try:
with open(image_path, "rb") as f:
return f.read()
except Exception as exc:
logger.error("Failed to read character image: %s", exc)
return None