aed6720dab
run.py: - built-in adapter creation: 9-branch if/elif -> _BUILTIN_ADAPTERS table - idle slash-command routing: 35 `if canonical == ...` branches -> _gateway_idle_command_handlers() - shared helpers: _send_command_ack (4 sites), _command_origin_for_source (2), _session_entry_for_manager (goal/heartbeat), _toggle_adapter_auto_tts_set (2), _load_env_or_agent_cfg_timeout (2), _float_env reuse (2), _resolve_session_key_or_none (3), _running_agent_ids (4), _schedule_rename_from_title_thread (2), _write_runtime_status_quiet (5), _AUTO_RESET_CONTEXT_NOTES/_auto_reset_reason_text - ruff SIM102/SIM103/SIM105/SIM108/SIM118 + F401 across gateway/ (semantics re-reviewed; sqlite Row `.keys()` and side-effecting assignments kept) - two hand-reviewed comment/docstring compaction passes (AST-identical, rationale kept) slash_commands.py: - /model: typed path and picker callback shared one 200-line commit block -> _perform_model_switch + _commit_model_switch - comment/docstring compaction (AST-identical)
84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
"""Sticker description cache for Telegram.
|
|
|
|
Stickers are described via the vision tool once and cached by file_unique_id
|
|
(``~/.hermes/sticker_cache.json``) so the same image is never re-analyzed.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from typing import Optional
|
|
|
|
from hermes_cli.config import get_hermes_home
|
|
import contextlib
|
|
|
|
|
|
CACHE_PATH = get_hermes_home() / "sticker_cache.json"
|
|
|
|
# Kept concise to save tokens.
|
|
STICKER_VISION_PROMPT = (
|
|
"Describe this sticker in 1-2 sentences. Focus on what it depicts -- "
|
|
"character, action, emotion. Be concise and objective."
|
|
)
|
|
|
|
|
|
def _load_cache() -> dict:
|
|
if CACHE_PATH.exists():
|
|
try:
|
|
return json.loads(CACHE_PATH.read_text(encoding="utf-8"))
|
|
except (json.JSONDecodeError, OSError):
|
|
return {}
|
|
return {}
|
|
|
|
|
|
def _save_cache(cache: dict) -> None:
|
|
"""Write the cache atomically (temp file + fsync + replace)."""
|
|
CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, tmp_path = tempfile.mkstemp(dir=str(CACHE_PATH.parent), suffix=".tmp")
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
json.dump(cache, f, indent=2, ensure_ascii=False)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp_path, str(CACHE_PATH))
|
|
except BaseException:
|
|
with contextlib.suppress(OSError):
|
|
os.unlink(tmp_path)
|
|
raise
|
|
|
|
|
|
def get_cached_description(file_unique_id: str) -> Optional[dict]:
|
|
"""Return ``{description, emoji, set_name, cached_at}`` or None."""
|
|
return _load_cache().get(file_unique_id)
|
|
|
|
|
|
def cache_sticker_description(file_unique_id: str, description: str, emoji: str = "", set_name: str = "") -> None:
|
|
"""Store a vision-generated description under Telegram's stable sticker id."""
|
|
cache = _load_cache()
|
|
cache[file_unique_id] = {"description": description, "emoji": emoji, "set_name": set_name, "cached_at": time.time()}
|
|
_save_cache(cache)
|
|
|
|
|
|
def build_sticker_injection(description: str, emoji: str = "", set_name: str = "") -> str:
|
|
"""Warm-style injection text, e.g.
|
|
``[The user sent a sticker 😀 from "MyPack"~ It shows: "A cat waving" (=^.w.^=)]``.
|
|
``set_name`` is only shown together with an emoji.
|
|
"""
|
|
context = ""
|
|
if set_name and emoji:
|
|
context = f" {emoji} from \"{set_name}\""
|
|
elif emoji:
|
|
context = f" {emoji}"
|
|
return f"[The user sent a sticker{context}~ It shows: \"{description}\" (=^.w.^=)]"
|
|
|
|
|
|
def build_animated_sticker_injection(emoji: str = "") -> str:
|
|
"""Injection text for animated/video stickers we can't analyze."""
|
|
if emoji:
|
|
return (
|
|
f"[The user sent an animated sticker {emoji}~ "
|
|
f"I can't see animated ones yet, but the emoji suggests: {emoji}]"
|
|
)
|
|
return "[The user sent an animated sticker~ I can't see animated ones yet]"
|