""" Authoritative NPC, Dialogue Tree & Map-Quest Interconnection Engine for FreeExile. Theme: CỔ VÕ HOANG DÃ & HẮC ÁM LƯU ĐÀY (Grimdark Savage Primal Exile ARPG - 100% PoE2 Spirit). Features: 1. Spatial NPC placement, zone indexing, and Euclidean proximity gating (Zero-Trust Anti-Cheat). 2. Grimdark Feral NPC archetypes: Outcast Scavengers, Bone Blacksmiths, Blood Shamans, Feral Trackers, Dune Savages. 3. Dynamic Dialogue Trees with brutal survival dialogues, branching choices, level/quest gating, and terminal states. 4. Reactive dialogue adaptation based on real-time QuestEngine states (Available, In-Progress, Completed, Claimed). 5. Bidirectional QuestEngine synchronization (Talk NPC, Deliver to NPC, Accept, Claim). 6. Hidden Quest serendipity triggers (Cơ Duyên Đàm Đạo) through dialogue depth or blood/relic offerings. 7. Environmental & ZoneEngine linkage (Sandstorm, Miasma, and dynamic Secret Chamber gates). """ from __future__ import annotations import math import time from typing import Dict, List, Optional, Set, Tuple, Any from server.world.quest_engine import ( QuestEngine, QuestType, QuestStatus, ObjectiveType, HiddenTriggerType, ) from server.world.npc_types import ( NpcRole, NpcServiceType, NpcDialogueChoice, NpcDialogueNode, NpcDialogueTree, NpcDefinition, PlayerDialogueSession, ) from server.world.npc_catalog import populate_canonical_npcs class NpcEngine: """ Authoritative Feral NPC & Dialogue Engine. Enforces strict zero-trust distance limits, manages dialogue tree traversal, and synchronizes seamlessly with ZoneEngine and QuestEngine. """ def __init__(self) -> None: self._npc_registry: Dict[str, NpcDefinition] = {} self._dialogue_trees: Dict[str, NpcDialogueTree] = {} # zone_id -> list of npc_ids self._zone_npc_index: Dict[str, List[str]] = {} # player_id -> active PlayerDialogueSession self._active_sessions: Dict[str, PlayerDialogueSession] = {} def register_npc(self, npc: NpcDefinition) -> None: self._npc_registry[npc.npc_id] = npc if npc.zone_id not in self._zone_npc_index: self._zone_npc_index[npc.zone_id] = [] if npc.npc_id not in self._zone_npc_index[npc.zone_id]: self._zone_npc_index[npc.zone_id].append(npc.npc_id) def register_dialogue_tree(self, tree: NpcDialogueTree) -> None: self._dialogue_trees[tree.tree_id] = tree def get_npc(self, npc_id: str) -> Optional[NpcDefinition]: return self._npc_registry.get(npc_id) def get_npcs_in_zone(self, zone_id: str) -> List[NpcDefinition]: npc_ids = self._zone_npc_index.get(zone_id, []) return [self._npc_registry[nid] for nid in npc_ids if nid in self._npc_registry] def get_nearby_npcs( self, zone_id: str, x: float, y: float, radius: float = 20.0 ) -> List[NpcDefinition]: """Spatial query returning NPCs within euclidean radius in the specified zone.""" candidates = self.get_npcs_in_zone(zone_id) nearby: List[NpcDefinition] = [] radius_sq = radius * radius for npc in candidates: dx = npc.x - x dy = npc.y - y if dx * dx + dy * dy <= radius_sq: nearby.append(npc) return nearby def validate_interaction_distance( self, npc: NpcDefinition, player_x: float, player_y: float, tolerance: float = 0.5 ) -> bool: """Zero-Trust server-authoritative Euclidean distance check.""" dx = npc.x - player_x dy = npc.y - player_y dist = math.sqrt(dx * dx + dy * dy) return dist <= (npc.interaction_radius + tolerance) def get_dialogue_tree(self, tree_id: str) -> Optional[NpcDialogueTree]: return self._dialogue_trees.get(tree_id) def get_active_session(self, player_id: str) -> Optional[PlayerDialogueSession]: return self._active_sessions.get(player_id) def close_dialogue_session(self, player_id: str) -> None: if player_id in self._active_sessions: del self._active_sessions[player_id] def interact_with_npc( self, player_id: str, npc_id: str, zone_id: str, player_x: float, player_y: float, player_level: int = 1, quest_engine: Optional[QuestEngine] = None, environment: Optional[str] = None, ) -> Tuple[bool, str, Optional[NpcDialogueNode], List[str]]: """ Authoritative interaction entry point. Validates map zone, euclidean distance, level gating, and sets up dialogue session. """ npc = self.get_npc(npc_id) if not npc: return False, f"NPC '{npc_id}' không tồn tại trong Hoang Vực!", None, [] # Validate Zone membership if npc.zone_id != zone_id: return False, f"NPC '{npc.name}' không thuộc bản đồ (zone) '{zone_id}'!", None, [] # Anti-cheat distance verification if not self.validate_interaction_distance(npc, player_x, player_y): return False, f"Cự ly quá xa! Không thể tiếp cận '{npc.name}' để đối thoại.", None, [] # Level restriction check if player_level < npc.required_level: return False, f"Tu vi/căn cốt chưa đủ (Yêu cầu Cấp {npc.required_level}) để diện kiến '{npc.name}'!", None, [] # Environment condition check if npc.required_environment and npc.required_environment != environment: return False, f"Môi trường hoang dã chưa đủ dị tượng ({npc.required_environment}) để gặp '{npc.name}'!", None, [] tree = self.get_dialogue_tree(npc.dialogue_tree_id) if not tree: return False, f"Cây hội thoại '{npc.dialogue_tree_id}' chưa được đăng ký!", None, [] # Determine active root node dynamically based on player's Quest progress selected_root_id = tree.default_root_node_id if quest_engine and tree.quest_conditional_roots: for q_id, status_mapping in tree.quest_conditional_roots.items(): active_q = quest_engine.get_player_quest(player_id, q_id) if active_q and active_q.status.name in status_mapping: selected_root_id = status_mapping[active_q.status.name] break root_node = tree.nodes.get(selected_root_id) if not root_node: root_node = tree.nodes.get(tree.default_root_node_id) if not root_node: return False, "Không tìm thấy nút thoại hợp lệ!", None, [] now_ms = int(time.time() * 1000) self._active_sessions[player_id] = PlayerDialogueSession( player_id=player_id, npc_id=npc_id, current_node_id=root_node.node_id, started_at_ms=now_ms, last_interact_ms=now_ms, ) # Record TALK_NPC progress in QuestEngine updated_quests: List[str] = [] if quest_engine: updated_quests = quest_engine.record_talk_npc(player_id, npc_id) return True, f"Bắt đầu đàm đạo với '{npc.name}'", root_node, updated_quests def start_dialogue( self, player_id: str, npc_id: str, player_x: float, player_y: float, player_level: int = 1, quest_engine: Optional[QuestEngine] = None, environment: Optional[str] = None, zone_id: Optional[str] = None, ) -> Tuple[bool, str, Optional[NpcDialogueNode], List[str]]: """Convenience alias for interact_with_npc.""" npc = self.get_npc(npc_id) actual_zone = zone_id if zone_id is not None else (npc.zone_id if npc else "") return self.interact_with_npc( player_id=player_id, npc_id=npc_id, zone_id=actual_zone, player_x=player_x, player_y=player_y, player_level=player_level, quest_engine=quest_engine, environment=environment, ) def select_dialogue_choice( self, player_id: str, choice_id: str, npc_id: Optional[str] = None, current_node_id: Optional[str] = None, quest_engine: Optional[QuestEngine] = None, player_level: int = 1, ) -> Tuple[bool, str, Optional[NpcDialogueNode], Optional[NpcServiceType], Optional[str]]: """ Executes a dialogue choice, updates session node, and triggers quest/services. Returns: (success, message, next_node, triggered_service, triggered_param) """ session = self._active_sessions.get(player_id) if not session: if npc_id and current_node_id: now_ms = int(time.time() * 1000) session = PlayerDialogueSession( player_id=player_id, npc_id=npc_id, current_node_id=current_node_id, started_at_ms=now_ms, last_interact_ms=now_ms, ) self._active_sessions[player_id] = session else: return False, "Chưa mở phiên đối thoại với NPC nào!", None, None, None target_npc_id = npc_id or session.npc_id target_node_id = current_node_id or session.current_node_id npc = self.get_npc(target_npc_id) if not npc: return False, "NPC không hợp lệ!", None, None, None tree = self.get_dialogue_tree(npc.dialogue_tree_id) if not tree: return False, "Cây đối thoại không tồn tại!", None, None, None current_node = tree.nodes.get(target_node_id) if not current_node: return False, "Nút thoại hiện tại không hợp lệ!", None, None, None matched_choice: Optional[NpcDialogueChoice] = None for ch in current_node.choices: if ch.choice_id == choice_id: matched_choice = ch break if not matched_choice: return False, f"Lựa chọn '{choice_id}' không có trong nút thoại hiện tại!", None, None, None if player_level < matched_choice.required_level: return False, f"Cấp độ chưa đủ (Yêu cầu Cấp {matched_choice.required_level}) để chọn phương án này!", None, None, None if quest_engine and matched_choice.required_quest_id and matched_choice.required_quest_status: q = quest_engine.get_active_quest(player_id, matched_choice.required_quest_id) if not q or q.status != matched_choice.required_quest_status: return False, "Trạng thái nhiệm vụ không thỏa mãn điều kiện đối thoại!", None, None, None triggered_service = matched_choice.trigger_service triggered_payload = matched_choice.trigger_param # Quest serendipity & Quest dispatch integration if quest_engine: quest_engine.trigger_hidden_quest( player_id=player_id, trigger_type=HiddenTriggerType.NPC_DIALOGUE_CHOICE, npc_id=npc.npc_id, choice_id=choice_id, zone=npc.zone_id, ) if triggered_service == NpcServiceType.QUEST_DISPATCH and triggered_payload: quest_engine.accept_quest(player_id, triggered_payload) if matched_choice.is_terminal: self.close_dialogue_session(player_id) return True, "Kết thúc đối thoại.", None, triggered_service, triggered_payload next_node: Optional[NpcDialogueNode] = None if matched_choice.target_node_id: next_node = tree.nodes.get(matched_choice.target_node_id) if next_node: session.current_node_id = next_node.node_id session.last_interact_ms = int(time.time() * 1000) return True, "Chọn đáp án thành công!", next_node, triggered_service, triggered_payload def deliver_item_to_npc( self, player_id: str, npc_id: str, item_id: str, count: int = 1, quest_engine: Optional[QuestEngine] = None, ) -> Tuple[bool, str, List[str]]: """ Delivers an item to an NPC, updating DELIVER_TO_NPC quest objectives and checking potential NPC_GIFT_OFFERING hidden triggers. """ npc = self.get_npc(npc_id) if not npc: return False, f"NPC '{npc_id}' không tồn tại!", [] updated_quests: List[str] = [] if quest_engine: updated_quests = quest_engine.record_deliver_npc(player_id, npc_id, count) # Check hidden quest gift offering quest_engine.trigger_hidden_quest( player_id=player_id, trigger_type=HiddenTriggerType.NPC_GIFT_OFFERING, npc_id=npc_id, item_id=item_id, zone=npc.zone_id, ) return True, f"Ném {count}x '{item_id}' cho '{npc.name}' thành công!", updated_quests def create_canonical_npc_engine() -> NpcEngine: """ Factory creating an NpcEngine populated with FreeExile's Grimdark Feral NPCs across Outcast Camp, Bone-Cliff Chasm, Ancient Barrow, Dune Desert, and Ruins. """ engine = NpcEngine() populate_canonical_npcs(engine) return engine