""" Storyline and Beggar's Chronicle Engine for FreeExile. Manages narrative progression, journal unlocks, cutscene triggers, and evolution from a destitute mendicant to the sovereign of the deadlands. """ from __future__ import annotations import time from typing import Dict, List, Optional, Any, Tuple from server.world.storyline_types import ( StoryAct, BeggarAscentStage, JournalEntry, StoryActDefinition, StoryCutscene, PlayerStoryState, ) from server.world.storyline_catalog import ( get_canonical_acts, get_canonical_journals, get_canonical_cutscenes, ) __all__ = [ "StorylineEngine", "create_default_storyline_engine", ] class StorylineEngine: """Server-Authoritative engine for the Beggar's Chronicle narrative system.""" def __init__(self) -> None: self._acts: Dict[StoryAct, StoryActDefinition] = {} self._journals: Dict[str, JournalEntry] = {} self._cutscenes: Dict[str, StoryCutscene] = {} self._quest_to_journal: Dict[str, List[str]] = {} self._quest_to_cutscene: Dict[str, List[str]] = {} self._player_states: Dict[str, PlayerStoryState] = {} def register_act(self, act_def: StoryActDefinition) -> None: self._acts[act_def.act] = act_def def register_journal(self, entry: JournalEntry) -> None: self._journals[entry.journal_id] = entry if entry.unlock_quest_id not in self._quest_to_journal: self._quest_to_journal[entry.unlock_quest_id] = [] self._quest_to_journal[entry.unlock_quest_id].append(entry.journal_id) def register_cutscene(self, cutscene: StoryCutscene) -> None: self._cutscenes[cutscene.cutscene_id] = cutscene if cutscene.trigger_quest_id not in self._quest_to_cutscene: self._quest_to_cutscene[cutscene.trigger_quest_id] = [] self._quest_to_cutscene[cutscene.trigger_quest_id].append(cutscene.cutscene_id) def get_act(self, act: StoryAct) -> Optional[StoryActDefinition]: return self._acts.get(act) def get_journal(self, journal_id: str) -> Optional[JournalEntry]: return self._journals.get(journal_id) def initialize_player_story(self, player_id: str) -> PlayerStoryState: if player_id in self._player_states: return self._player_states[player_id] now = time.time() initial_state = PlayerStoryState( player_id=player_id, current_act=StoryAct.ACT_I_HUNGER, current_stage=BeggarAscentStage.MENDICANT, completed_acts=[], unlocked_journal_ids=[], read_journal_ids=[], completed_quest_ids=[], reputation_title="Kẻ Ăn Mày Rách Rưới", started_at=now, updated_at=now, ) self._player_states[player_id] = initial_state return initial_state def get_player_story_state(self, player_id: str) -> PlayerStoryState: return self.initialize_player_story(player_id) def on_quest_completed(self, player_id: str, quest_id: str) -> Dict[str, Any]: """Invoked when a quest or milestone is completed or claimed.""" state = self.initialize_player_story(player_id) if quest_id not in state.completed_quest_ids: state.completed_quest_ids.append(quest_id) new_journals: List[JournalEntry] = [] for j_id in self._quest_to_journal.get(quest_id, []): if j_id not in state.unlocked_journal_ids: state.unlocked_journal_ids.append(j_id) entry = self.get_journal(j_id) if entry: new_journals.append(entry) triggered_cutscenes: List[StoryCutscene] = [] for c_id in self._quest_to_cutscene.get(quest_id, []): cutscene = self._cutscenes.get(c_id) if cutscene: triggered_cutscenes.append(cutscene) stage_changed, new_stage = self._evaluate_stage_progression(state) act_changed, new_act = self._evaluate_act_progression(state) state.updated_at = time.time() return { "player_id": player_id, "quest_id": quest_id, "unlocked_journals": [j.journal_id for j in new_journals], "triggered_cutscenes": [c.cutscene_id for c in triggered_cutscenes], "stage_changed": stage_changed, "current_stage": new_stage.value, "act_changed": act_changed, "current_act": new_act.value, "reputation_title": state.reputation_title, } def _evaluate_stage_progression(self, state: PlayerStoryState) -> Tuple[bool, BeggarAscentStage]: """Calculates stage advancement based on completed quests.""" completed = set(state.completed_quest_ids) prev_stage = state.current_stage target_stage = BeggarAscentStage.MENDICANT title = "Kẻ Ăn Mày Rách Rưới" if "ms_tier_5_van_gioi_chi_ton" in completed or "quest_epic_04_van_gioi_than_ma_tranh_ba" in completed: target_stage = BeggarAscentStage.SOVEREIGN_OUTCAST title = "Bá Chủ Tối Cao Cõi Lưu Đày" elif "quest_epic_02_nghich_menh_dai_huyet" in completed or "quest_hidden_04_u_hon_oan_kiem" in completed: target_stage = BeggarAscentStage.PRIMAL_HUNTER title = "Thợ Săn Cự Thú Cổ Đại" elif "quest_epic_01_phong_ma_co_tran" in completed or "quest_hidden_02_tan_bi_than_tuong" in completed: target_stage = BeggarAscentStage.BONE_WARRIOR title = "Cuồng Cốt Chiến Giả" elif "quest_norm_04_diet_u_hon_kiem_no" in completed or "quest_hidden_01_huyen_anh_to_su" in completed: target_stage = BeggarAscentStage.SCAVENGER title = "Kẻ Mót Xương Bùn Độc" state.current_stage = target_stage state.reputation_title = title return (target_stage != prev_stage, target_stage) def _evaluate_act_progression(self, state: PlayerStoryState) -> Tuple[bool, StoryAct]: """Calculates current narrative act based on journal entries and milestones.""" unlocked = set(state.unlocked_journal_ids) prev_act = state.current_act target_act = StoryAct.ACT_I_HUNGER if any(j in unlocked for j in ["journal_13_sau_canh_cong_hu_khong", "journal_14_chua_te_quy_vuong", "journal_15_vinh_quang_dinh_phong"]): target_act = StoryAct.ACT_V_SOVEREIGN elif any(j in unlocked for j in ["journal_10_dau_truong_xe_thit", "journal_11_ha_guc_cuong_vien", "journal_12_khac_an_nghich_menh"]): target_act = StoryAct.ACT_IV_TITAN_HUNT elif any(j in unlocked for j in ["journal_07_con_khat_sa_mac", "journal_08_bo_cap_khong_lo", "journal_09_nhung_ke_cung_kho"]): target_act = StoryAct.ACT_III_CRUCIBLE elif any(j in unlocked for j in ["journal_04_bun_ngap_ngang_nguc", "journal_05_gap_lai_cai_nguc", "journal_06_doan_truy_co_thi"]): target_act = StoryAct.ACT_II_MIASMA state.current_act = target_act act_sequence = ( StoryAct.ACT_I_HUNGER, StoryAct.ACT_II_MIASMA, StoryAct.ACT_III_CRUCIBLE, StoryAct.ACT_IV_TITAN_HUNT, StoryAct.ACT_V_SOVEREIGN, ) target_idx = act_sequence.index(target_act) for idx, act in enumerate(act_sequence[:-1]): if target_idx > idx and act not in state.completed_acts: state.completed_acts.append(act) return (target_act != prev_act, target_act) def read_journal(self, player_id: str, journal_id: str) -> Optional[JournalEntry]: """Marks a journal entry as read and returns its data.""" state = self.initialize_player_story(player_id) if journal_id not in state.unlocked_journal_ids: return None if journal_id not in state.read_journal_ids: state.read_journal_ids.append(journal_id) state.updated_at = time.time() return self.get_journal(journal_id) def get_unlocked_journals(self, player_id: str) -> List[JournalEntry]: state = self.initialize_player_story(player_id) results: List[JournalEntry] = [] for j_id in state.unlocked_journal_ids: entry = self.get_journal(j_id) if entry: results.append(entry) results.sort(key=lambda x: x.chapter_index) return results def get_chronicle_summary(self, player_id: str) -> Dict[str, Any]: """Provides a comprehensive overview of the player's Beggar's Journey.""" state = self.initialize_player_story(player_id) current_act_def = self.get_act(state.current_act) return { "player_id": player_id, "current_act": state.current_act.value, "current_act_title": current_act_def.title if current_act_def else "", "current_act_subtitle": current_act_def.subtitle if current_act_def else "", "current_stage": state.current_stage.value, "reputation_title": state.reputation_title, "total_journals_unlocked": len(state.unlocked_journal_ids), "total_journals_available": len(self._journals), "total_acts_completed": len(state.completed_acts), "completed_quests_count": len(state.completed_quest_ids), "unlocked_journal_ids": list(state.unlocked_journal_ids), "read_journal_ids": list(state.read_journal_ids), } def create_default_storyline_engine() -> StorylineEngine: """Factory creating a StorylineEngine populated with canonical Beggar's Chronicle data.""" engine = StorylineEngine() for act_def in get_canonical_acts().values(): engine.register_act(act_def) for journal_def in get_canonical_journals().values(): engine.register_journal(journal_def) for cutscene_def in get_canonical_cutscenes().values(): engine.register_cutscene(cutscene_def) return engine