"""Flavor Memory — Per-channel gustatory history in Redis.
Tracks which flavors a user has referenced, how often, and when.
Provides familiarity bonuses: more mentions → stronger OXT/MOR on re-encounter.
"""
from __future__ import annotations
import json
import logging
import time
from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
# Redis key pattern
_KEY_PREFIX = "ncm:flavor_history"
_MAX_HISTORY = 20
def _make_key(channel_id: str) -> str:
"""Internal helper: make key.
Args:
channel_id (str): Discord/Matrix channel identifier.
Returns:
str: Result string.
"""
return f"{_KEY_PREFIX}:{channel_id}"
[docs]
async def record_flavor(
redis_client: Any,
channel_id: str,
flavor_name: str,
) -> None:
"""Log a flavor reference for a channel.
Each entry: {"flavor": name, "ts": timestamp, "count": N}
Maintains a running history of the last _MAX_HISTORY unique flavors.
"""
if redis_client is None:
return
key = _make_key(channel_id)
fname = flavor_name.upper().replace(" ", "_")
try:
raw = await redis_client.get(key)
history: List[Dict[str, Any]] = json.loads(raw) if raw else []
except Exception:
history = []
# Find existing entry for this flavor
found = False
for entry in history:
if entry.get("flavor") == fname:
entry["count"] = entry.get("count", 1) + 1
entry["last_ts"] = time.time()
found = True
break
if not found:
history.append({
"flavor": fname,
"count": 1,
"first_ts": time.time(),
"last_ts": time.time(),
})
# Trim to max history (keep most recent by last_ts)
history.sort(key=lambda e: e.get("last_ts", 0), reverse=True)
history = history[:_MAX_HISTORY]
try:
await redis_client.set(
key,
json.dumps(history),
ex=60 * 60 * 24 * 90, # 90 day TTL
)
except Exception as e:
logger.error("Failed to write flavor history: %s", e)
[docs]
async def get_flavor_affinity(
redis_client: Any,
channel_id: str,
flavor_name: str,
) -> float:
"""Get familiarity bonus for a specific flavor in this channel.
Returns a multiplier: 1.0 for unknown, up to ~1.5 for very familiar.
The bonus follows a log curve: rapid early gain, diminishing returns.
Used as a multiplier on OXT and MOR deltas during Phase 4 NCM computation.
"""
if redis_client is None:
return 1.0
key = _make_key(channel_id)
fname = flavor_name.upper().replace(" ", "_")
try:
raw = await redis_client.get(key)
history: List[Dict[str, Any]] = json.loads(raw) if raw else []
except Exception:
return 1.0
for entry in history:
if entry.get("flavor") == fname:
count = entry.get("count", 1)
# Log curve: 1.0 at count=1, ~1.3 at count=5, ~1.5 at count=20
import math
return 1.0 + 0.15 * math.log(count + 1)
return 1.0
[docs]
async def get_history(
redis_client: Any,
channel_id: str,
) -> List[Dict[str, Any]]:
"""Get the full flavor history for a channel."""
if redis_client is None:
return []
key = _make_key(channel_id)
try:
raw = await redis_client.get(key)
return json.loads(raw) if raw else []
except Exception:
return []
[docs]
async def get_top_flavors(
redis_client: Any,
channel_id: str,
n: int = 5,
) -> List[str]:
"""Get the N most frequently referenced flavors for a channel."""
history = await get_history(redis_client, channel_id)
history.sort(key=lambda e: e.get("count", 0), reverse=True)
return [e["flavor"] for e in history[:n]]