""" Game Design Matrix Seeder for FreeExile. Pulls canonical data from all game catalogs and populates the SQLite central matrix. Strictly decoupled from engine execution. """ from __future__ import annotations import json import sqlite3 from typing import Any, Dict, List, Optional from server.world.storyline_catalog import get_canonical_acts from server.world.zone_engine import ZoneEngine from server.world.npc_catalog import create_canonical_npcs from server.world.quest_engine import create_default_quest_engine from server.world.monster_archetype_catalog import CORE_GENUS_SPECS, SUB_SPECIES_CATALOG from server.world.weapon_catalog import get_base_weapon_catalog from server.world.martial_catalog import create_standard_active_skills from server.world.item_affix_catalog import CANONICAL_AFFIX_DEFINITIONS, STANDARD_ILVLS from server.world.level_progression_curve import calculate_piecewise_exp_curve def seed_canonical_data(conn: Optional[sqlite3.Connection] = None, force: bool = False) -> Dict[str, Any]: """Nạp dữ liệu hạt giống chuẩn có kiểm tra trạng thái bảng nếu force=False.""" if conn is None: from server.world.game_design_matrix_service import GameDesignMatrixService return GameDesignMatrixService().seed_canonical_data(force=force) cur = conn.cursor() if not force: try: s_cnt = (cur.execute("SELECT COUNT(*) FROM story_acts").fetchone() or [0])[0] b_cnt = (cur.execute("SELECT COUNT(*) FROM progression_benchmarks").fetchone() or [0])[0] p_cnt = (cur.execute("SELECT COUNT(*) FROM party_scaling_matrix").fetchone() or [0])[0] if s_cnt > 0 and b_cnt >= 100 and p_cnt >= 6: return {"status": "already_seeded"} except sqlite3.OperationalError: pass res: Dict[str, Any] = seed_all_canonical_data(conn) conn.commit() return res def seed_all_canonical_data(conn: sqlite3.Connection) -> Dict[str, int]: """Nạp toàn bộ dữ liệu hạt giống từ các Catalog chuẩn vào Database.""" stats: Dict[str, int] = {} cur = conn.cursor() # Clear existing tables in reverse dependency order cur.execute("UPDATE quests SET prerequisite_quest_id = NULL") for tbl in ( "portal_quarantine_rules", "teleport_configs", "loot_allocation_modes", "party_scaling_matrix", "cross_relationships", "progression_benchmarks", "affix_tiers", "skills", "weapon_bases", "quests", "monster_archetypes", "npcs", "zones", "story_acts" ): cur.execute(f"DELETE FROM {tbl}") # 1. Story Acts acts = get_canonical_acts() for act_enum, act_def in acts.items(): lvl_min, lvl_max = _parse_level_range(act_def.recommended_level_range) cur.execute(""" INSERT OR REPLACE INTO story_acts (act_id, act_order, title, subtitle, level_min, level_max, primary_zone_id, climax_boss_id, climax_boss_name, description) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( act_def.act.value, _act_order(act_def.act.value), act_def.title, act_def.subtitle, lvl_min, lvl_max, act_def.primary_zone_id, act_def.climax_boss_id, act_def.climax_boss_name, act_def.description )) stats["story_acts"] = len(acts) # 2. Zones zone_engine = ZoneEngine() for zone_id, zone_def in zone_engine.zones.items(): act_id = _resolve_zone_act(zone_id) cur.execute(""" INSERT OR REPLACE INTO zones (zone_id, act_id, name, zone_type, environment, min_level, max_level, bounds_width, bounds_height, default_spawn_x, default_spawn_y) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( zone_def.zone_id, act_id, zone_def.name, zone_def.zone_type.name, zone_def.environment.name, zone_def.min_level, zone_def.min_level + 10, zone_def.bounds_width, zone_def.bounds_height, zone_def.default_spawn_x, zone_def.default_spawn_y )) stats["zones"] = len(zone_engine.zones) # 3. NPCs npcs = create_canonical_npcs() for npc in npcs: cur.execute(""" INSERT OR REPLACE INTO npcs (npc_id, name, title, role, zone_id, x, y, dialogue_tree_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, (npc.npc_id, npc.name, npc.title, npc.role.name, npc.zone_id, npc.x, npc.y, npc.dialogue_tree_id)) stats["npcs"] = len(npcs) # 4. Monster Archetypes monster_count = 0 for genus, (base_name, base_hp, base_dmg, base_scale, native_elem) in CORE_GENUS_SPECS.items(): sub_list = SUB_SPECIES_CATALOG.get(genus, []) for sub_code, sub_name in sub_list: arch_id = f"mob_{genus.name.lower()}_{sub_code.lower()}" cur.execute(""" INSERT OR REPLACE INTO monster_archetypes (archetype_id, genus, sub_species_code, name, base_hp, base_dmg, base_scale, native_element, min_level) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, (arch_id, genus.name, sub_code, f"{base_name} ({sub_name})", base_hp, base_dmg, base_scale, native_elem.name, 1)) monster_count += 1 stats["monster_archetypes"] = monster_count # 5. Quests quest_engine = create_default_quest_engine() for q_id, q_def in quest_engine._quest_registry.items(): # First objective target monster if any target_mob = None target_cnt = 1 if q_def.objectives: target_mob = q_def.objectives[0].target_id target_cnt = q_def.objectives[0].target_count prereq = q_def.prerequisite_quest_ids[0] if q_def.prerequisite_quest_ids else None reward_items = [ {"item_id": r.item_id, "name": r.name, "count": r.quantity} for r in q_def.reward.spirit_stones ] # Resolve zone for quest q_zone = _resolve_quest_zone(q_id, q_def.act_id) cur.execute(""" INSERT OR REPLACE INTO quests (quest_id, act_id, title, quest_type, required_level, zone_id, prerequisite_quest_id, target_monster_id, target_count, reward_exp, reward_meridian_points, reward_title, reward_items_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( q_def.quest_id, q_def.act_id, q_def.title, q_def.quest_type.name, q_def.required_level, q_zone, prereq, target_mob, target_cnt, q_def.reward.exp, q_def.reward.meridian_points, q_def.reward.title, json.dumps(reward_items, ensure_ascii=False) )) stats["quests"] = len(quest_engine._quest_registry) # 6. Weapon Bases weapons = get_base_weapon_catalog() for wid, w in weapons.items(): cur.execute(""" INSERT OR REPLACE INTO weapon_bases (weapon_id, name_vi, name_en, weapon_class, weapon_category, grip_style, sect, clan, element, required_level, base_dmg_min, base_dmg_max, attack_speed, crit_chance) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( w.weapon_id, w.name_vi, w.name_en, w.weapon_class.name, w.category.name, w.grip_style.name, w.sect_affinity.name, w.clan_affinity.name, w.element.name, w.required_level, w.min_damage, w.max_damage, w.attacks_per_second, w.base_crit_chance )) stats["weapon_bases"] = len(weapons) # 7. Skills skills = create_standard_active_skills() for sid, s in skills.items(): tags_str = json.dumps([t.name for t in s.tags], ensure_ascii=False) cur.execute(""" INSERT OR REPLACE INTO skills (skill_id, skill_name, element, sect, weapon_requirement, base_damage, energy_cost, cooldown_ms, combo_role, tags_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( s.skill_id, s.skill_name, s.element.name, s.sect.name, s.weapon_requirement.name, s.base_damage, s.energy_cost, s.cooldown_ms, s.combo_role.name, tags_str )) stats["skills"] = len(skills) # 8. Affix Tiers affix_count = 0 for affix in CANONICAL_AFFIX_DEFINITIONS: slots_str = json.dumps([slot.name for slot in affix.allowed_slots]) cur.execute(""" INSERT OR REPLACE INTO affix_tiers (mod_id, name, affix_type, tier, required_ilvl, min_value, max_value, weight, allowed_slots_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( affix.mod_id, affix.mod_name, affix.affix_type.name, affix.tier, affix.required_ilvl, affix.min_val, affix.max_val, affix.weight, slots_str )) affix_count += 1 stats["affix_tiers"] = affix_count # 9. Progression Benchmarks (Level 1 to 100) — Piecewise Exponential Curve exp_benchmarks = calculate_piecewise_exp_curve() for lvl in range(1, 101): bench = exp_benchmarks[lvl] p_hp = round(100.0 + (lvl - 1) * 28.0, 1) p_dps = round(20.0 * (1.12 ** (lvl - 1)), 1) m_hp = round(80.0 * (1.10 ** (lvl - 1)), 1) m_dps = round(12.0 * (1.09 ** (lvl - 1)), 1) max_tier = _resolve_max_affix_tier(lvl) cur.execute(""" INSERT OR REPLACE INTO progression_benchmarks (level, target_exp, exp_to_next_level, cumulative_exp, player_base_hp, player_benchmark_dps, monster_base_hp, monster_base_dps, max_affix_tier_allowed, death_penalty_ratio, level_gap_safe_range, level_gap_penalty_exp, monster_benchmark_exp) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( lvl, bench.target_exp, bench.exp_to_next_level, bench.cumulative_exp, p_hp, p_dps, m_hp, m_dps, max_tier, bench.death_penalty_ratio, bench.level_gap_safe_range, bench.level_gap_penalty_exp, bench.monster_benchmark_exp )) stats["progression_benchmarks"] = 100 # 10. Seed Cross-Relationships rel_count = _seed_cross_relationships(cur, acts, zone_engine, npcs, quest_engine) stats["cross_relationships"] = rel_count # 11. Party Scaling Matrix (PoE2 6-Player Scaled Modifiers) p_data = [ (1, 1.0, 0.0, 0.0, 0.0, 0.0, 15.0, "Solo: Base 1.0x HP, +0% EXP/Loot/Rarity"), (2, 1.5, 30.0, 50.0, 50.0, 30.0, 15.0, "Party 2: HP+50%, EXP+30%, Qty+50%, Rarity+30%"), (3, 2.0, 60.0, 100.0, 100.0, 60.0, 15.0, "Party 3: HP+100%, EXP+60%, Qty+100%, Rarity+60%"), (4, 2.5, 90.0, 150.0, 150.0, 90.0, 15.0, "Party 4: HP+150%, EXP+90%, Qty+150%, Rarity+90%"), (5, 3.0, 120.0, 200.0, 200.0, 120.0, 15.0, "Party 5: HP+200%, EXP+120%, Qty+200%, Rarity+120%"), (6, 3.5, 150.0, 250.0, 250.0, 150.0, 15.0, "Party 6: HP+250%, EXP+150%, Qty+250%, Rarity+150%"), ] cur.executemany("INSERT OR REPLACE INTO party_scaling_matrix VALUES (?,?,?,?,?,?,?,?)", p_data) stats["party_scaling_matrix"] = len(p_data) # 12. Loot Allocation Modes l_data = [ ("PERMANENT", "Permanent Allocation", 0.0, 1, "Permanent item lock for allocated player"), ("SHORT", "Short Allocation", 5.0, 0, "5.0s temporary lock, then free for all"), ("FREE_FOR_ALL", "Free-For-All", 0.0, 0, "Instant free pickup for all party members"), ] cur.executemany("INSERT OR REPLACE INTO loot_allocation_modes VALUES (?,?,?,?,?)", l_data) stats["loot_allocation_modes"] = len(l_data) # 13. Teleport Configs t_data = [ ("PARTY_MEMBER", 3.5, 10.0, 1, 1, "Teleport to party member (cast 3.5s, cd 10s, cancel on move/damage)"), ("FRIEND", 3.5, 10.0, 1, 1, "Teleport to friend (cast 3.5s, cd 10s, cancel on move/damage)"), ("GUILDMATE", 3.5, 10.0, 1, 1, "Teleport to guildmate (cast 3.5s, cd 10s, cancel on move/damage)"), ("TOWN_PORTAL", 3.5, 10.0, 1, 1, "Teleport via Town Portal to Hideout/Sanctuary (cast 3.5s, cd 10s)"), ] cur.executemany("INSERT OR REPLACE INTO teleport_configs VALUES (?,?,?,?,?,?)", t_data) stats["teleport_configs"] = len(t_data) # 14. Portal Quarantine Rules q_data = [ ("RULE_OPEN_WORLD", "WILDERNESS", 1, 1, 6, "Wilderness: 6 portals, boss & secret lockout enabled"), ("RULE_DUNGEON", "DUNGEON", 1, 1, 6, "Dungeon: 6 portals, boss & secret lockout enabled"), ("RULE_SECRET_CHAMBER", "SECRET_CHAMBER", 1, 1, 0, "Secret chamber: 0 portals, lockout enabled"), ("RULE_SAFE_HAVEN", "SAFE_HAVEN", 0, 0, 6, "Safe haven: 6 portals, no lockout"), ] cur.executemany("INSERT OR REPLACE INTO portal_quarantine_rules VALUES (?,?,?,?,?,?)", q_data) stats["portal_quarantine_rules"] = len(q_data) return stats def _act_order(act_id: str) -> int: order_map = {"ACT_I_HUNGER": 1, "ACT_II_MIASMA": 2, "ACT_III_CRUCIBLE": 3, "ACT_IV_TITAN_HUNT": 4, "ACT_V_SOVEREIGN": 5} return order_map.get(act_id, 99) def _parse_level_range(range_str: str) -> tuple[int, int]: clean = range_str.replace("Cấp", "").replace("Level", "").replace("+", "").strip() if "-" in clean: parts = clean.split("-") try: return int(parts[0].strip()), int(parts[1].strip()) except ValueError: pass return 1, 10 def _resolve_zone_act(zone_id: str) -> str: zone_act_map = { "zone_boundless_sanctuary": "ACT_I_HUNGER", "zone_player_hideout": "ACT_I_HUNGER", "zone_tang_kiem_nhai": "ACT_I_HUNGER", "zone_ancient_sword_barrow": "ACT_II_MIASMA", "zone_boundless_sandstorm": "ACT_III_CRUCIBLE", } return zone_act_map.get(zone_id, "ACT_I_HUNGER") def _resolve_quest_zone(quest_id: str, act_id: Optional[str]) -> str: if "tang_kiem" in quest_id or act_id == "ACT_I_HUNGER": return "zone_tang_kiem_nhai" if "ancient" in quest_id or act_id == "ACT_II_MIASMA": return "zone_ancient_sword_barrow" return "zone_boundless_sandstorm" if act_id == "ACT_III_CRUCIBLE" else "zone_boundless_sanctuary" def _resolve_max_affix_tier(lvl: int) -> int: for idx, ilvl in enumerate(STANDARD_ILVLS): if lvl >= ilvl: return idx + 1 return 15 def _seed_cross_relationships(cur: sqlite3.Cursor, acts: Any, zone_engine: Any, npcs: Any, quest_engine: Any) -> int: relations = [ (f"rel_act_zone_{a.act.value}", "ACT", a.act.value, "ACT_CONTAINS_ZONE", "ZONE", a.primary_zone_id) for a in acts.values() ] + [ (f"rel_act_boss_{a.act.value}", "ACT", a.act.value, "BOSS_CLIMAX_FOR_ACT", "BOSS", a.climax_boss_id) for a in acts.values() ] + [ (f"rel_zone_npc_{n.npc_id}", "ZONE", n.zone_id, "ZONE_HOUSES_NPC", "NPC", n.npc_id) for n in npcs ] + [ (f"rel_quest_prereq_{qid}_{pre}", "QUEST", qid, "QUEST_REQUIRES_PREREQUISITE", "QUEST", pre) for qid, q in quest_engine._quest_registry.items() for pre in (q.prerequisite_quest_ids or []) ] cur.executemany( "INSERT OR REPLACE INTO cross_relationships (relation_id, source_type, source_id, relation_type, target_type, target_id) VALUES (?, ?, ?, ?, ?, ?)", relations ) return len(relations)