143 lines
5.6 KiB
Python
143 lines
5.6 KiB
Python
"""Persistent Kirara state that decays only during an active live session."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
|
||
EMOTION_DELTAS = {
|
||
"开心": 1,
|
||
"很开心": 3,
|
||
"平静": 0,
|
||
"生气": -1,
|
||
"很生气": -3,
|
||
}
|
||
EMOTION_PREFIX = re.compile(r"^\s*\[(很开心|很生气|开心|平静|生气)\]\s*")
|
||
GIFT_RECOVERY = {
|
||
"饭团": (5, 0),
|
||
"薯条": (5, 0),
|
||
"人气票": (0, 5),
|
||
"粉丝团灯牌": (0, 5),
|
||
"灯牌": (0, 5), # 直播事件中常用的简写
|
||
}
|
||
|
||
|
||
def clamp(value: int | float) -> int:
|
||
return max(0, min(100, int(value)))
|
||
|
||
|
||
def gift_recovery(gift_name: str, quantity: int = 1) -> tuple[int, int]:
|
||
"""Return nominal stamina/cleanliness points for a gift batch."""
|
||
stamina, cleanliness = GIFT_RECOVERY.get(str(gift_name).strip(), (0, 0))
|
||
count = max(0, int(quantity))
|
||
return stamina * count, cleanliness * count
|
||
|
||
|
||
def split_emotion(reply: str) -> tuple[str, str | None]:
|
||
"""Return the spoken reply and its recognized leading emotion tag."""
|
||
match = EMOTION_PREFIX.match(reply)
|
||
if not match:
|
||
return reply, None
|
||
return reply[match.end():].strip(), match.group(1)
|
||
|
||
|
||
class PetState:
|
||
def __init__(self, path: str | Path):
|
||
self.path = Path(path)
|
||
self.values = {"stamina": 100, "cleanliness": 100, "mood": 100}
|
||
# Short-lived events let open frontends play each AI mood reaction once.
|
||
self.recent_emotions: list[dict] = []
|
||
self.live_seconds = 0.0
|
||
# A restart must never count the offline gap as live time.
|
||
self.live_active = False
|
||
self.updated_at = time.time()
|
||
try:
|
||
data = json.loads(self.path.read_text(encoding="utf-8"))
|
||
self.values = {key: clamp(data.get(key, 100)) for key in self.values}
|
||
self.live_seconds = max(0.0, min(3599.999, float(data.get("live_seconds", 0))))
|
||
self.updated_at = min(self.updated_at, float(data.get("updated_at", self.updated_at)))
|
||
except (OSError, ValueError, TypeError):
|
||
pass
|
||
self._saved_at = self.updated_at
|
||
|
||
def _save(self) -> None:
|
||
"""持久化。保存失败只降级为日志，绝不影响礼物/直播主流程。"""
|
||
try:
|
||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||
payload = json.dumps({**self.values, "live_seconds": self.live_seconds,
|
||
"updated_at": self.updated_at}, ensure_ascii=False)
|
||
try:
|
||
pending = self.path.with_suffix(self.path.suffix + ".tmp")
|
||
pending.write_text(payload, encoding="utf-8")
|
||
pending.replace(self.path)
|
||
except OSError:
|
||
# 原子替换被拒（杀软锁定/权限）时退回直接覆写
|
||
self.path.write_text(payload, encoding="utf-8")
|
||
Path(str(self.path) + ".tmp").unlink(missing_ok=True)
|
||
self._saved_at = self.updated_at
|
||
except Exception:
|
||
logging.getLogger("pet_state").warning(
|
||
"[桌宠] 状态保存失败（不影响礼物处理）: %s", self.path, exc_info=True)
|
||
|
||
def _record_emotion(self, delta: int, now: float) -> None:
|
||
if not delta:
|
||
return
|
||
self.recent_emotions.append({
|
||
"id": uuid.uuid4().hex,
|
||
"ts": now,
|
||
"delta": delta,
|
||
})
|
||
del self.recent_emotions[:-32]
|
||
|
||
def sync(self, now: float | None = None) -> dict[str, int]:
|
||
now = time.time() if now is None else now
|
||
if now > self.updated_at:
|
||
added = now - self.updated_at if self.live_active else 0.0
|
||
self.live_seconds += added
|
||
hours = int(self.live_seconds // 3600)
|
||
self.live_seconds %= 3600
|
||
for _ in range(hours):
|
||
self.values["stamina"] = clamp(self.values["stamina"] - 10)
|
||
self.values["cleanliness"] = clamp(self.values["cleanliness"] - 10)
|
||
if self.values["stamina"] < 40 and self.values["cleanliness"] < 40:
|
||
self.values["mood"] = clamp(self.values["mood"] - 5)
|
||
elif self.values["stamina"] > 80 and self.values["cleanliness"] > 80:
|
||
self.values["mood"] = clamp(self.values["mood"] + 3)
|
||
self._record_emotion(3, now)
|
||
self.updated_at = now
|
||
if hours or (added and now - self._saved_at >= 60):
|
||
self._save()
|
||
return dict(self.values)
|
||
|
||
def set_live_active(self, active: bool, now: float | None = None) -> dict[str, int]:
|
||
"""Settle the previous session state before switching at this instant."""
|
||
now = time.time() if now is None else now
|
||
self.sync(now)
|
||
if self.live_active != active:
|
||
self.live_active = active
|
||
self._save()
|
||
return dict(self.values)
|
||
|
||
def apply_emotion(self, emotion: str) -> dict[str, int]:
|
||
self.sync()
|
||
if emotion in EMOTION_DELTAS:
|
||
delta = EMOTION_DELTAS[emotion]
|
||
self.values["mood"] = clamp(self.values["mood"] + delta)
|
||
self._record_emotion(delta, time.time())
|
||
self._save()
|
||
return dict(self.values)
|
||
|
||
def restore_gift(self, stamina: int = 0, cleanliness: int = 0) -> dict[str, int]:
|
||
"""Apply gift recovery while keeping stored values within 0–100."""
|
||
self.sync()
|
||
if stamina or cleanliness:
|
||
self.values["stamina"] = clamp(self.values["stamina"] + stamina)
|
||
self.values["cleanliness"] = clamp(self.values["cleanliness"] + cleanliness)
|
||
self._save()
|
||
return dict(self.values)
|