| """Relationship affinity / stage / memory (R6).""" |
| from __future__ import annotations |
|
|
| from typing import Dict, List, Optional |
|
|
| STAGE_ORDER = ["STRANGER", "ACQUAINTANCE", "FAMILIAR", "TRUSTED", "BOND"] |
|
|
| class RelationshipStore: |
| def __init__(self, data: Optional[Dict[str, dict]] = None, rules: Optional[dict] = None): |
| self.data: Dict[str, dict] = {} |
| self.rules = rules or {} |
| for npc_id, raw in (data or {}).items(): |
| self.data[npc_id] = { |
| "affinity": int(raw.get("affinity", 0)), |
| "visit_count": int(raw.get("visit_count", 0)), |
| "stage": raw.get("stage", "STRANGER"), |
| "memories": list(raw.get("memories") or []), |
| "deep_chats_done": list(raw.get("deep_chats_done") or []), |
| } |
| self.recompute_stage(npc_id, day_index=0) |
|
|
| @classmethod |
| def from_state(cls, state: dict, rules: Optional[dict] = None) -> "RelationshipStore": |
| return cls(state.get("relationships") or {}, rules=rules) |
|
|
| def ensure(self, npc_id: str) -> dict: |
| if npc_id not in self.data: |
| self.data[npc_id] = { |
| "affinity": 0, |
| "visit_count": 0, |
| "stage": "STRANGER", |
| "memories": [], |
| "deep_chats_done": [], |
| } |
| return self.data[npc_id] |
|
|
| def apply_visit(self, npc_id: str, day_index: int) -> None: |
| r = self.ensure(npc_id) |
| |
| r["visit_count"] += 1 |
| r["affinity"] = min(100, r["affinity"] + int(self._delta("visit", 1))) |
| self.recompute_stage(npc_id, day_index) |
|
|
| def apply_purchase(self, npc_id: str, item_match_score: float, day_index: int) -> None: |
| r = self.ensure(npc_id) |
| if item_match_score >= 0.5: |
| d = int(self._delta("buy_good", 3)) |
| else: |
| d = int(self._delta("buy_weak", 1)) |
| r["affinity"] = min(100, r["affinity"] + d) |
| self.recompute_stage(npc_id, day_index) |
|
|
| def can_deep_chat(self, npc_id: str) -> bool: |
| """Deep chat unlocks at FAMILIAR (affinity ≥ 35) or higher.""" |
| r = self.ensure(npc_id) |
| stage = r.get("stage") or "STRANGER" |
| return STAGE_ORDER.index(stage) >= STAGE_ORDER.index("FAMILIAR") |
|
|
| def apply_deep_chat(self, npc_id: str, chat_id: str, kind: bool, day_index: int) -> bool: |
| r = self.ensure(npc_id) |
| if not self.can_deep_chat(npc_id): |
| return False |
| if chat_id in r["deep_chats_done"]: |
| return False |
| r["deep_chats_done"].append(chat_id) |
| d = int(self._delta("deep_chat_kind" if kind else "deep_chat_neutral", 5 if kind else 1)) |
| r["affinity"] = min(100, r["affinity"] + d) |
| self.recompute_stage(npc_id, day_index) |
| return True |
|
|
| def add_memory(self, npc_id: str, memory_id: str, text: str) -> None: |
| r = self.ensure(npc_id) |
| if any(m.get("id") == memory_id for m in r["memories"]): |
| return |
| r["memories"].append({"id": memory_id, "text": text}) |
|
|
| def _delta(self, key: str, default: int) -> int: |
| return int((self.rules.get("affinity_delta") or {}).get(key, default)) |
|
|
| def recompute_stage(self, npc_id: str, day_index: int = 0) -> str: |
| r = self.ensure(npc_id) |
| aff = r["affinity"] |
| visits = r["visit_count"] |
| chats = len(r["deep_chats_done"]) |
| has_arc = any(m.get("id", "").startswith("arc_") for m in r["memories"]) |
|
|
| best = "STRANGER" |
| |
| candidates = [ |
| ("STRANGER", aff >= 0, True), |
| ("ACQUAINTANCE", aff >= 15, visits >= 2), |
| ("FAMILIAR", aff >= 35, True), |
| ("TRUSTED", aff >= 60, chats >= 1), |
| ("BOND", aff >= 85, day_index >= 5 and has_arc), |
| ] |
| for name, aff_ok, hard in candidates: |
| if aff_ok and hard: |
| best = name |
| |
| if STAGE_ORDER.index(best) < STAGE_ORDER.index(r.get("stage", "STRANGER")): |
| best = r["stage"] |
| r["stage"] = best |
| return best |
|
|
| def sync_to_state(self, state: dict) -> None: |
| state["relationships"] = {k: dict(v, memories=list(v["memories"]), deep_chats_done=list(v["deep_chats_done"])) for k, v in self.data.items()} |
|
|
