""" Authoritative PoE2-Style Hideout Town Portal Engine with Anti-Exploit Quarantine. Manages dynamic 2-way town portal links bridging wilderness coordinates to player hideouts and hubs, 6-portal map device charge tracking, authoritative boss/secret chamber exploit quarantine, and death penalty portal handling. """ from __future__ import annotations import time from typing import Any, Dict, Optional, Tuple, Union from server.world.town_portal_types import ( PortalQuarantineStatus, TownPortalLink, TownPortalResult, ) class TownPortalEngine: """Authoritative PoE2 2-Way Town Portal Engine with Anti-Exploit Quarantine.""" def __init__(self) -> None: self.portals: Dict[str, TownPortalLink] = {} self.map_charges: Dict[str, int] = {} self.boss_active: Dict[str, bool] = {} self.secret_active: Dict[str, bool] = {} self._portal_counter: int = 0 def set_encounter_state( self, instance_id: str, boss_active: bool = False, secret_active: bool = False, ) -> None: """Register or update active boss combat and secret chamber states for an instance.""" self.boss_active[instance_id] = boss_active self.secret_active[instance_id] = secret_active def check_quarantine_lock(self, zone_id: str, instance_id: str) -> Tuple[bool, str]: """Perform authoritative pre-flight quarantine check against active encounters.""" if self.boss_active.get(instance_id, False): return True, "QUARANTINE_BOSS_ENGAGED: Cấm mở cổng/trốn thoát khi Boss đang giao tranh!" if self.secret_active.get(instance_id, False) or zone_id.startswith("zone_secret_"): return True, "QUARANTINE_SECRET_CHAMBER: Mật thất đếm ngược đang kích hoạt, không thể mở cổng!" return False, "UNLOCKED" def request_town_portal( self, player_id: str, zone_id: str, instance_id_or_wx: Union[str, float] = "inst_1", wx: float = 0.0, wy: float = 0.0, hub_zone: str = "zone_player_hideout", **kwargs: Any, ) -> TownPortalResult: """Request creation of a 2-way town portal linking wilderness coordinates to hub.""" # Parameter normalization for flexible signatures if isinstance(instance_id_or_wx, (int, float)): actual_wy = wx actual_wx = float(instance_id_or_wx) actual_instance_id = "inst_1" else: actual_instance_id = str(instance_id_or_wx) actual_wx = float(wx) actual_wy = float(wy) # Anti-exploit quarantine verification is_locked, err = self.check_quarantine_lock(zone_id, actual_instance_id) if is_locked: return TownPortalResult(False, err, None) # Personal portal exclusivity: close player's previous active portals for existing_portal in self.portals.values(): if existing_portal.owner_id == player_id and existing_portal.is_active: object.__setattr__(existing_portal, "is_active", False) self._portal_counter += 1 portal_id = f"tp_{player_id}_{int(time.time() * 1000)}_{self._portal_counter}" portal = TownPortalLink( portal_id=portal_id, owner_id=player_id, wilderness_zone_id=zone_id, wilderness_x=actual_wx, wilderness_y=actual_wy, hub_zone_id=hub_zone, hub_x=0.0, hub_y=0.0, created_at=time.time(), is_active=True, instance_id=actual_instance_id, ) self.portals[portal_id] = portal return TownPortalResult(True, "SUCCESS", portal_id) def register_astral_map_device(self, instance_id: str, initial_charges: int = 6) -> None: """Initialize 6 portal charges for an Endgame Astral Map Device instance.""" self.map_charges[instance_id] = max(0, initial_charges) def enter_map_portal(self, instance_id: str) -> Tuple[bool, str, int]: """Traverse into an Endgame Map instance, atomically consuming 1 portal charge.""" charges = self.map_charges.get(instance_id, 0) if charges <= 0: return False, "MAP_SEALED: Hết lượt cổng (0/6). Bản đồ đã phong tỏa vĩnh viễn!", 0 new_charges = charges - 1 self.map_charges[instance_id] = new_charges return True, "ENTER_SUCCESS", new_charges def on_player_death( self, player_id: str, zone_id: str, instance_id: str, is_map_instance: bool = False, ) -> Tuple[bool, str]: """Apply PoE2 death penalty: wilderness closes portal; map instance consumes 1 charge.""" if is_map_instance: charges = self.map_charges.get(instance_id, 0) if charges > 0: self.map_charges[instance_id] = charges - 1 remaining = self.map_charges.get(instance_id, 0) return True, f"MAP_DEATH_CONSUMED_PORTAL: Còn lại {remaining} lượt" # Wilderness death closes only the deceased player's active portal closed = 0 for p in self.portals.values(): if p.owner_id == player_id and p.is_active: object.__setattr__(p, "is_active", False) closed += 1 return True, f"WILDERNESS_DEATH_CLOSED_{closed}_PORTALS" def can_traverse_portal( self, player_id: str, portal_id: str, party_service: Optional[Any] = None, ) -> Tuple[bool, str]: """Verify whether player is the owner or an authorized party member to enter portal.""" portal = self.portals.get(portal_id) if not portal: return False, "PORTAL_NOT_FOUND: Cổng dịch chuyển không tồn tại" if not portal.is_active: return False, "PORTAL_INACTIVE: Cổng dịch chuyển đã đóng" if portal.owner_id == player_id: return True, "AUTHORIZED_OWNER" if party_service: party_map = getattr(party_service, "player_party_map", {}) owner_party_id = party_map.get(portal.owner_id) player_party_id = party_map.get(player_id) if owner_party_id and owner_party_id == player_party_id: return True, "AUTHORIZED_PARTY_MEMBER" return False, "PORTAL_UNAUTHORIZED: Chỉ chủ nhân hoặc đồng đội mới có thể qua cổng" def traverse_town_portal( self, player_id: str, portal_id: str, party_service: Optional[Any] = None, is_return: bool = False, ) -> Tuple[bool, str, Optional[Tuple[str, float, float]]]: """Traverse a 2-way town portal and return destination (zone_id, x, y).""" ok, reason = self.can_traverse_portal(player_id, portal_id, party_service) if not ok: return False, reason, None portal = self.portals[portal_id] if not is_return: # Wilderness -> Hub return True, "TRAVERSED_TO_HUB", (portal.hub_zone_id, portal.hub_x, portal.hub_y) # Hub -> Wilderness return True, "TRAVERSED_TO_WILDERNESS", (portal.wilderness_zone_id, portal.wilderness_x, portal.wilderness_y)