ptarrant/Code_of_Conquest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement Phase 5 Quest System (100% complete)

Browse Source 
Add YAML-driven quest system with context-aware offering:

Core Implementation:
- Quest data models (Quest, QuestObjective, QuestReward, QuestTriggers)
- QuestService for YAML loading and caching
- QuestEligibilityService with level, location, and probability filtering
- LoreService stub (MockLoreService) ready for Phase 6 Weaviate integration

Quest Content:
- 5 example quests across difficulty tiers (2 easy, 2 medium, 1 hard)
- Quest-centric design: quests define their NPC givers
- Location-based probability weights for natural quest offering

AI Integration:
- Quest offering section in npc_dialogue.j2 template
- Response parser extracts [QUEST_OFFER:quest_id] markers
- AI naturally weaves quest offers into NPC conversations

API Endpoints:
- POST /api/v1/quests/accept - Accept quest offer
- POST /api/v1/quests/decline - Decline quest offer
- POST /api/v1/quests/progress - Update objective progress
- POST /api/v1/quests/complete - Complete quest, claim rewards
- POST /api/v1/quests/abandon - Abandon active quest
- GET /api/v1/characters/{id}/quests - List character quests
- GET /api/v1/quests/{quest_id} - Get quest details

Frontend:
- Quest tracker sidebar with HTMX integration
- Quest offer modal for accept/decline flow
- Quest detail modal for viewing progress
- Combat service integration for kill objective tracking

Testing:
- Unit tests for Quest models and serialization
- Integration tests for full quest lifecycle
- Comprehensive test coverage for eligibility service

Documentation:
- Reorganized docs into /docs/phases/ structure
- Added Phase 5-12 planning documents
- Updated ROADMAP.md with new structure

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Phillip Tarrant
2025-11-29 15:42:55 -06:00
parent e7e329e6ed
commit df26abd207
42 changed files with 8421 additions and 2227 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
api/app/services/combat_service.py
View File

@@ -34,6 +34,8 @@ from app.services.combat_repository import (
get_combat_repository,
CombatRepository
)
from app.services.quest_service import get_quest_service
from app.models.quest import ObjectiveType
from app.utils.logging import get_logger
logger = get_logger(__file__)
@@ -1290,6 +1292,9 @@ class CombatService:
items=len(rewards.items),
level_ups=rewards.level_ups)
# Update quest progress for kill objectives
self._update_quest_kill_progress(encounter, session, user_id)
return rewards
def _build_loot_context(self, encounter: CombatEncounter) -> LootContext:
@@ -1339,6 +1344,114 @@ class CombatService:
# Helper Methods
# =========================================================================
def _update_quest_kill_progress(
self,
encounter: CombatEncounter,
session,
user_id: str
) -> None:
"""
Update quest progress for kill objectives based on defeated enemies.
Scans all defeated enemies in the encounter, identifies their types,
and updates any active quests that have kill objectives matching
those enemy types.
Args:
encounter: Completed combat encounter
session: Game session
user_id: User ID for character updates
"""
# Collect killed enemy types and counts
killed_enemy_types: Dict[str, int] = {}
for combatant in encounter.combatants:
if not combatant.is_player and combatant.is_dead():
enemy_id = combatant.combatant_id.split("_")[0]
enemy = self.enemy_loader.load_enemy(enemy_id)
if enemy:
# Use enemy_id as the type identifier (matches target_enemy_type in quests)
killed_enemy_types[enemy_id] = killed_enemy_types.get(enemy_id, 0) + 1
if not killed_enemy_types:
return # No enemies killed
# Get character
if session.is_solo():
char_id = session.solo_character_id
else:
# For multiplayer, we'd need to update all players
# For now, just handle solo
return
try:
character = self.character_service.get_character(char_id, user_id)
if not character:
return
# Get active quests
active_quests = getattr(character, 'active_quests', [])
if not active_quests:
return
quest_service = get_quest_service()
quest_states = getattr(character, 'quest_states', {})
updated = False
for quest_id in active_quests:
quest = quest_service.load_quest(quest_id)
if not quest:
continue
quest_state = quest_states.get(quest_id, {})
objectives_progress = quest_state.get('objectives_progress', {})
for objective in quest.objectives:
# Check if this is a kill objective with a matching enemy type
if objective.objective_type != ObjectiveType.KILL:
continue
target_enemy = objective.target_enemy_type
if not target_enemy or target_enemy not in killed_enemy_types:
continue
# Update progress for this objective
kill_count = killed_enemy_types[target_enemy]
current_progress = objectives_progress.get(objective.objective_id, 0)
new_progress = min(
current_progress + kill_count,
objective.required_progress
)
if new_progress > current_progress:
objectives_progress[objective.objective_id] = new_progress
updated = True
logger.info(
"Quest kill progress updated",
character_id=char_id,
quest_id=quest_id,
objective_id=objective.objective_id,
enemy_type=target_enemy,
kills=kill_count,
progress=f"{new_progress}/{objective.required_progress}"
)
# Update quest state
if quest_id in quest_states:
quest_states[quest_id]['objectives_progress'] = objectives_progress
# Save character if any quests were updated
if updated:
character.quest_states = quest_states
self.character_service.update_character(character, user_id)
except Exception as e:
logger.error(
"Failed to update quest kill progress",
char_id=char_id,
error=str(e)
)
def _create_combatant_from_character(
self,
character: Character

374
api/app/services/lore_service.py Normal file
View File

@@ -0,0 +1,374 @@
"""
LoreService for retrieving contextual lore for NPC conversations.
This module provides an interface for lore retrieval that will be
implemented with Weaviate in Phase 6. For now, it provides a mock
implementation that returns embedded lore from quest definitions.
The service follows a three-tier knowledge hierarchy:
1. World Lore - Global history, mythology, kingdoms
2. Regional Lore - Local history, landmarks, rumors
3. NPC Persona - Individual NPC knowledge (already in NPC YAML)
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Dict, List, Any, Optional, Protocol
import structlog
from app.models.quest import Quest, QuestLoreContext
from app.models.npc import NPC
logger = structlog.get_logger(__name__)
@dataclass
class LoreEntry:
"""
A single piece of lore information.
Attributes:
content: The lore text content
title: Optional title for the lore entry
knowledge_type: Type of knowledge (common, academic, secret)
source: Where this lore came from (world, regional, quest)
metadata: Additional metadata for filtering
"""
content: str
title: str = ""
knowledge_type: str = "common"
source: str = "quest"
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
"""Serialize to dictionary."""
return {
"content": self.content,
"title": self.title,
"knowledge_type": self.knowledge_type,
"source": self.source,
"metadata": self.metadata,
}
@dataclass
class LoreContext:
"""
Aggregated lore context for AI prompts.
Contains lore from multiple sources, filtered for the current
NPC and conversation context.
Attributes:
world_lore: Global world knowledge entries
regional_lore: Local/regional knowledge entries
quest_lore: Quest-specific lore entries
"""
world_lore: List[LoreEntry] = field(default_factory=list)
regional_lore: List[LoreEntry] = field(default_factory=list)
quest_lore: List[LoreEntry] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
"""Serialize to dictionary for AI prompt injection."""
return {
"world": [entry.to_dict() for entry in self.world_lore],
"regional": [entry.to_dict() for entry in self.regional_lore],
"quest": [entry.to_dict() for entry in self.quest_lore],
}
def has_content(self) -> bool:
"""Check if any lore content is available."""
return bool(self.world_lore or self.regional_lore or self.quest_lore)
class LoreServiceInterface(ABC):
"""
Abstract interface for lore services.
This interface allows swapping between MockLoreService (current)
and WeaviateLoreService (Phase 6) without changing calling code.
"""
@abstractmethod
def get_lore_context(
self,
npc: NPC,
quest: Optional[Quest] = None,
topic: str = "",
region_id: str = "",
) -> LoreContext:
"""
Get lore context for an NPC conversation.
Args:
npc: The NPC in the conversation
quest: Optional quest being discussed
topic: Topic of conversation for semantic search
region_id: Region to filter lore by
Returns:
LoreContext with relevant lore entries
"""
pass
@abstractmethod
def filter_lore_for_npc(
self,
lore_entries: List[LoreEntry],
npc: NPC,
) -> List[LoreEntry]:
"""
Filter lore entries based on what an NPC would know.
Args:
lore_entries: Raw lore entries
npc: The NPC to filter for
Returns:
Filtered list of lore entries
"""
pass
class MockLoreService(LoreServiceInterface):
"""
Mock implementation of LoreService using embedded quest lore.
This implementation returns lore directly from quest YAML files
until Weaviate is implemented in Phase 6. It provides a working
lore system without external dependencies.
"""
# NPC roles that have access to academic knowledge
ACADEMIC_ROLES = ["scholar", "wizard", "sage", "librarian", "priest", "mage"]
# NPC roles that might know secrets
SECRET_KEEPER_ROLES = ["mayor", "noble", "spy", "elder", "priest"]
def __init__(self):
"""Initialize the mock lore service."""
logger.info("MockLoreService initialized")
def get_lore_context(
self,
npc: NPC,
quest: Optional[Quest] = None,
topic: str = "",
region_id: str = "",
) -> LoreContext:
"""
Get lore context for an NPC conversation.
For the mock implementation, this primarily returns quest
embedded lore and some static regional/world hints.
Args:
npc: The NPC in the conversation
quest: Optional quest being discussed
topic: Topic of conversation (unused in mock)
region_id: Region to filter by
Returns:
LoreContext with available lore
"""
context = LoreContext()
# Add quest-specific lore if a quest is provided
if quest and quest.lore_context:
quest_entries = self._extract_quest_lore(quest)
context.quest_lore = self.filter_lore_for_npc(quest_entries, npc)
# Add some mock regional lore based on NPC location
regional_entries = self._get_mock_regional_lore(npc.location_id, region_id)
context.regional_lore = self.filter_lore_for_npc(regional_entries, npc)
# Add world lore if NPC is knowledgeable
if npc.role in self.ACADEMIC_ROLES:
world_entries = self._get_mock_world_lore()
context.world_lore = self.filter_lore_for_npc(world_entries, npc)
logger.debug(
"Lore context built",
npc_id=npc.npc_id,
quest_lore_count=len(context.quest_lore),
regional_lore_count=len(context.regional_lore),
world_lore_count=len(context.world_lore),
)
return context
def filter_lore_for_npc(
self,
lore_entries: List[LoreEntry],
npc: NPC,
) -> List[LoreEntry]:
"""
Filter lore entries based on what an NPC would know.
Args:
lore_entries: Raw lore entries
npc: The NPC to filter for
Returns:
Filtered list of lore entries
"""
filtered = []
for entry in lore_entries:
knowledge_type = entry.knowledge_type
# Academic knowledge requires appropriate role
if knowledge_type == "academic":
if npc.role not in self.ACADEMIC_ROLES:
continue
# Secret knowledge requires special role or tag
if knowledge_type == "secret":
if npc.role not in self.SECRET_KEEPER_ROLES:
if "secret_keeper" not in npc.tags:
continue
filtered.append(entry)
return filtered
def _extract_quest_lore(self, quest: Quest) -> List[LoreEntry]:
"""
Extract lore entries from a quest's embedded lore context.
Args:
quest: The quest to extract lore from
Returns:
List of LoreEntry objects
"""
entries = []
lore = quest.lore_context
# Add backstory as a lore entry
if lore.backstory:
entries.append(LoreEntry(
content=lore.backstory,
title=f"Background: {quest.name}",
knowledge_type="common",
source="quest",
metadata={"quest_id": quest.quest_id},
))
# Add world connections
for connection in lore.world_connections:
entries.append(LoreEntry(
content=connection,
title="World Connection",
knowledge_type="common",
source="quest",
metadata={"quest_id": quest.quest_id},
))
# Add regional hints
for hint in lore.regional_hints:
entries.append(LoreEntry(
content=hint,
title="Local Knowledge",
knowledge_type="common",
source="quest",
metadata={"quest_id": quest.quest_id},
))
return entries
def _get_mock_regional_lore(
self,
location_id: str,
region_id: str,
) -> List[LoreEntry]:
"""
Get mock regional lore for a location.
This provides basic regional context until Weaviate is implemented.
Args:
location_id: Specific location
region_id: Region identifier
Returns:
List of LoreEntry objects
"""
entries = []
# Crossville regional lore
if "crossville" in location_id.lower() or region_id == "crossville":
entries.extend([
LoreEntry(
content="Crossville was founded two hundred years ago as a trading post.",
title="Crossville History",
knowledge_type="common",
source="regional",
),
LoreEntry(
content="The Old Mines were sealed fifty years ago after a tragic accident.",
title="The Old Mines",
knowledge_type="common",
source="regional",
),
LoreEntry(
content="The Thornwood family has led the village for three generations.",
title="Village Leadership",
knowledge_type="common",
source="regional",
),
])
return entries
def _get_mock_world_lore(self) -> List[LoreEntry]:
"""
Get mock world-level lore.
This provides basic world context until Weaviate is implemented.
Returns:
List of LoreEntry objects
"""
return [
LoreEntry(
content="The Five Kingdoms united against the Shadow Empire two hundred years ago in a great war.",
title="The Great War",
knowledge_type="academic",
source="world",
),
LoreEntry(
content="Magic flows from the ley lines that crisscross the land, concentrated at ancient sites.",
title="The Nature of Magic",
knowledge_type="academic",
source="world",
),
LoreEntry(
content="The ancient ruins predate human settlement and are believed to be from the First Age.",
title="First Age Ruins",
knowledge_type="academic",
source="world",
),
]
# Global singleton instance
_service_instance: Optional[LoreServiceInterface] = None
def get_lore_service() -> LoreServiceInterface:
"""
Get the global LoreService instance.
Currently returns MockLoreService. In Phase 6, this will
be updated to return WeaviateLoreService.
Returns:
Singleton LoreService instance
"""
global _service_instance
if _service_instance is None:
_service_instance = MockLoreService()
return _service_instance

408
api/app/services/quest_eligibility_service.py Normal file
View File

@@ -0,0 +1,408 @@
"""
QuestEligibilityService for determining quest offering eligibility.
This service handles the core logic for determining which quests an NPC
can offer to a specific character, including:
- Finding quests where the NPC is a quest giver
- Filtering by character level and prerequisites
- Checking relationship and flag conditions
- Applying probability rolls based on location
"""
import random
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
import structlog
from app.models.quest import Quest, QuestLoreContext
from app.models.character import Character
from app.services.quest_service import get_quest_service
logger = structlog.get_logger(__name__)
@dataclass
class QuestOfferContext:
"""
Context for offering a specific quest during NPC conversation.
Contains all the information needed to inject quest offering
context into the AI prompt for natural dialogue generation.
Attributes:
quest: The quest being offered
offer_dialogue: NPC-specific dialogue for offering this quest
npc_quest_knowledge: Facts the NPC knows about this quest
lore_context: Embedded lore for AI context
narrative_hooks: Hints for natural conversation weaving
"""
quest: Quest
offer_dialogue: str
npc_quest_knowledge: List[str]
lore_context: QuestLoreContext
narrative_hooks: List[str]
def to_dict(self) -> Dict[str, Any]:
"""Serialize to dictionary for AI prompt injection."""
return {
"quest_id": self.quest.quest_id,
"quest_name": self.quest.name,
"quest_description": self.quest.description,
"offer_dialogue": self.offer_dialogue,
"npc_quest_knowledge": self.npc_quest_knowledge,
"lore_backstory": self.lore_context.backstory,
"narrative_hooks": self.narrative_hooks,
"rewards": {
"gold": self.quest.rewards.gold,
"experience": self.quest.rewards.experience,
},
}
@dataclass
class QuestEligibilityResult:
"""
Result of checking quest eligibility for an NPC conversation.
Attributes:
eligible_quests: Quests that can be offered
should_offer_quest: Whether to actually offer a quest (after probability roll)
selected_quest_context: Context for the quest to offer (if should_offer)
blocking_reasons: Why certain quests were filtered out
"""
eligible_quests: List[Quest] = field(default_factory=list)
should_offer_quest: bool = False
selected_quest_context: Optional[QuestOfferContext] = None
blocking_reasons: Dict[str, str] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
"""Serialize to dictionary."""
return {
"eligible_quest_count": len(self.eligible_quests),
"should_offer_quest": self.should_offer_quest,
"selected_quest": (
self.selected_quest_context.to_dict()
if self.selected_quest_context
else None
),
}
class QuestEligibilityService:
"""
Service for determining quest offering eligibility.
This is the core service that implements the quest-centric design:
given an NPC and a character, determine which quests can be offered
and whether an offer should be made based on probability.
"""
# Default probability weights by location type
DEFAULT_PROBABILITY_WEIGHTS = {
"tavern": 0.35,
"town": 0.25,
"shop": 0.20,
"wilderness": 0.05,
"dungeon": 0.10,
"road": 0.10,
}
# Maximum active quests a character can have
MAX_ACTIVE_QUESTS = 2
def __init__(self):
"""Initialize the eligibility service."""
self.quest_service = get_quest_service()
logger.info("QuestEligibilityService initialized")
def check_eligibility(
self,
npc_id: str,
character: Character,
location_type: str = "town",
location_id: str = "",
force_probability: Optional[float] = None,
) -> QuestEligibilityResult:
"""
Check which quests an NPC can offer to a character.
This is the main entry point for quest eligibility checking.
It performs the full pipeline:
1. Find quests where NPC is quest_giver
2. Filter by character eligibility
3. Apply probability roll
4. Build offer context if successful
Args:
npc_id: The NPC's identifier
character: The player character
location_type: Type of current location (for probability)
location_id: Specific location ID
force_probability: Override probability roll (for testing)
Returns:
QuestEligibilityResult with eligible quests and offer context
"""
result = QuestEligibilityResult()
# Check if character already has max quests
if len(character.active_quests) >= self.MAX_ACTIVE_QUESTS:
logger.debug(
"Character at max active quests",
character_id=character.character_id,
active_quests=len(character.active_quests),
)
result.blocking_reasons["max_quests"] = "Character already has maximum active quests"
return result
# Get quests this NPC can offer
npc_quests = self.quest_service.get_quests_for_npc(npc_id)
if not npc_quests:
logger.debug("NPC has no quests to offer", npc_id=npc_id)
return result
# Filter quests by eligibility
for quest in npc_quests:
eligible, reason = self._check_quest_eligibility(
quest, character, npc_id, location_id
)
if eligible:
result.eligible_quests.append(quest)
else:
result.blocking_reasons[quest.quest_id] = reason
if not result.eligible_quests:
logger.debug(
"No eligible quests after filtering",
npc_id=npc_id,
character_id=character.character_id,
)
return result
# Apply probability roll
probability = self._get_offer_probability(
result.eligible_quests, location_type, force_probability
)
roll = random.random()
if roll > probability:
logger.debug(
"Probability roll failed",
probability=probability,
roll=roll,
location_type=location_type,
)
return result
# Select a quest to offer
selected_quest = self._select_quest_to_offer(result.eligible_quests)
if not selected_quest:
return result
# Build offer context
result.should_offer_quest = True
result.selected_quest_context = self._build_offer_context(
selected_quest, npc_id
)
logger.info(
"Quest offer prepared",
quest_id=selected_quest.quest_id,
npc_id=npc_id,
character_id=character.character_id,
)
return result
def _check_quest_eligibility(
self,
quest: Quest,
character: Character,
npc_id: str,
location_id: str,
) -> tuple[bool, str]:
"""
Check if a specific quest is eligible for a character.
Args:
quest: The quest to check
character: The player character
npc_id: The NPC offering the quest
location_id: Current location ID
Returns:
Tuple of (is_eligible, reason_if_not)
"""
triggers = quest.offering_triggers
# Check character level
if character.level < triggers.min_character_level:
return False, f"Character level too low (need {triggers.min_character_level})"
if character.level > triggers.max_character_level:
return False, f"Character level too high (max {triggers.max_character_level})"
# Check prerequisite quests
# Note: We need to check completed_quests on character
# For now, check active_quests doesn't contain prereqs (they should be completed)
completed_quests = getattr(character, 'completed_quests', [])
for prereq_id in triggers.required_quests_completed:
if prereq_id not in completed_quests:
return False, f"Prerequisite quest not completed: {prereq_id}"
# Check quest is not already active
if quest.quest_id in character.active_quests:
return False, "Quest already active"
# Check quest is not already completed
if quest.quest_id in completed_quests:
return False, "Quest already completed"
# Check NPC-specific conditions
npc_conditions = quest.get_offer_conditions(npc_id)
npc_interaction = character.npc_interactions.get(npc_id, {})
# Check relationship level
relationship = npc_interaction.get("relationship_level", 50)
if relationship < npc_conditions.min_relationship:
return False, f"Relationship too low (need {npc_conditions.min_relationship})"
# Check required flags
custom_flags = npc_interaction.get("custom_flags", {})
for required_flag in npc_conditions.required_flags:
if not custom_flags.get(required_flag):
return False, f"Required flag not set: {required_flag}"
# Check forbidden flags
for forbidden_flag in npc_conditions.forbidden_flags:
if custom_flags.get(forbidden_flag):
return False, f"Forbidden flag is set: {forbidden_flag}"
return True, ""
def _get_offer_probability(
self,
eligible_quests: List[Quest],
location_type: str,
force_probability: Optional[float] = None,
) -> float:
"""
Calculate the probability of offering a quest.
Uses the highest probability weight among eligible quests
for the given location type.
Args:
eligible_quests: Quests that passed eligibility check
location_type: Type of current location
force_probability: Override value (for testing)
Returns:
Probability between 0.0 and 1.0
"""
if force_probability is not None:
return force_probability
# Find the highest probability weight for this location
max_probability = 0.0
for quest in eligible_quests:
quest_prob = quest.offering_triggers.get_probability(location_type)
if quest_prob > max_probability:
max_probability = quest_prob
# Fall back to default if no quest defines a probability
if max_probability == 0.0:
max_probability = self.DEFAULT_PROBABILITY_WEIGHTS.get(location_type, 0.10)
return max_probability
def _select_quest_to_offer(self, eligible_quests: List[Quest]) -> Optional[Quest]:
"""
Select which quest to offer from eligible quests.
Currently uses simple random selection, but could be enhanced
to use AI selection or priority weights.
Args:
eligible_quests: Quests that can be offered
Returns:
Selected quest or None
"""
if not eligible_quests:
return None
# Simple random selection for now
# TODO: Could use AI selection or difficulty-based priority
return random.choice(eligible_quests)
def _build_offer_context(
self,
quest: Quest,
npc_id: str,
) -> QuestOfferContext:
"""
Build the context for offering a quest.
Args:
quest: The quest being offered
npc_id: The NPC offering it
Returns:
QuestOfferContext with all necessary data for AI prompt
"""
return QuestOfferContext(
quest=quest,
offer_dialogue=quest.get_offer_dialogue(npc_id),
npc_quest_knowledge=quest.get_npc_knowledge(npc_id),
lore_context=quest.lore_context,
narrative_hooks=quest.dialogue_templates.narrative_hooks,
)
def get_available_quests_for_display(
self,
npc_id: str,
character: Character,
) -> List[Dict[str, Any]]:
"""
Get a list of available quests for UI display.
This is a simpler check that just returns eligible quests
without probability rolls, for showing in a quest list UI.
Args:
npc_id: The NPC's identifier
character: The player character
Returns:
List of quest display data
"""
result = self.check_eligibility(
npc_id=npc_id,
character=character,
force_probability=1.0, # Always include eligible quests
)
return [
quest.to_offer_dict()
for quest in result.eligible_quests
]
# Global singleton instance
_service_instance: Optional[QuestEligibilityService] = None
def get_quest_eligibility_service() -> QuestEligibilityService:
"""
Get the global QuestEligibilityService instance.
Returns:
Singleton QuestEligibilityService instance
"""
global _service_instance
if _service_instance is None:
_service_instance = QuestEligibilityService()
return _service_instance

423
api/app/services/quest_service.py Normal file
View File

@@ -0,0 +1,423 @@
"""
QuestService for loading quest definitions from YAML files.
This service reads quest configuration files and converts them into Quest
dataclass instances, providing caching for performance. Quests are organized
by difficulty subdirectories (easy, medium, hard, epic).
Follows the same pattern as NPCLoader for consistency.
"""
import yaml
from pathlib import Path
from typing import Dict, List, Optional
import structlog
from app.models.quest import (
Quest,
QuestObjective,
QuestReward,
QuestOfferingTriggers,
QuestOfferConditions,
NPCOfferDialogue,
QuestLoreContext,
QuestDialogueTemplates,
QuestDifficulty,
ObjectiveType,
)
logger = structlog.get_logger(__name__)
class QuestService:
"""
Loads quest definitions from YAML configuration files.
Quests are organized in difficulty subdirectories:
/app/data/quests/
easy/
cellar_rats.yaml
missing_item.yaml
medium/
bandit_threat.yaml
hard/
dungeon_depths.yaml
epic/
ancient_evil.yaml
This allows game designers to define quests without touching code.
"""
def __init__(self, data_dir: Optional[str] = None):
"""
Initialize the quest service.
Args:
data_dir: Path to directory containing quest YAML files.
Defaults to /app/data/quests/
"""
if data_dir is None:
# Default to app/data/quests relative to this file
current_file = Path(__file__)
app_dir = current_file.parent.parent # Go up to /app
data_dir = str(app_dir / "data" / "quests")
self.data_dir = Path(data_dir)
self._quest_cache: Dict[str, Quest] = {}
self._npc_quest_cache: Dict[str, List[str]] = {} # npc_id -> [quest_ids]
self._difficulty_cache: Dict[str, List[str]] = {} # difficulty -> [quest_ids]
logger.info("QuestService initialized", data_dir=str(self.data_dir))
def load_quest(self, quest_id: str) -> Optional[Quest]:
"""
Load a single quest by ID.
Searches all difficulty subdirectories for the quest file.
Args:
quest_id: Unique quest identifier (e.g., "quest_cellar_rats")
Returns:
Quest instance or None if not found
"""
# Check cache first
if quest_id in self._quest_cache:
logger.debug("Quest loaded from cache", quest_id=quest_id)
return self._quest_cache[quest_id]
# Search in difficulty subdirectories
if not self.data_dir.exists():
logger.error("Quest data directory does not exist", data_dir=str(self.data_dir))
return None
for difficulty_dir in self.data_dir.iterdir():
if not difficulty_dir.is_dir():
continue
# Try both with and without quest_ prefix
for filename in [f"{quest_id}.yaml", f"{quest_id.replace('quest_', '')}.yaml"]:
file_path = difficulty_dir / filename
if file_path.exists():
return self._load_quest_file(file_path)
logger.warning("Quest not found", quest_id=quest_id)
return None
def _load_quest_file(self, file_path: Path) -> Optional[Quest]:
"""
Load a quest from a specific file.
Args:
file_path: Path to the YAML file
Returns:
Quest instance or None if loading fails
"""
try:
with open(file_path, 'r') as f:
data = yaml.safe_load(f)
quest = self._parse_quest_data(data)
self._quest_cache[quest.quest_id] = quest
# Update NPC-to-quest cache
for npc_id in quest.quest_giver_npc_ids:
if npc_id not in self._npc_quest_cache:
self._npc_quest_cache[npc_id] = []
if quest.quest_id not in self._npc_quest_cache[npc_id]:
self._npc_quest_cache[npc_id].append(quest.quest_id)
# Update difficulty cache
difficulty = quest.difficulty.value
if difficulty not in self._difficulty_cache:
self._difficulty_cache[difficulty] = []
if quest.quest_id not in self._difficulty_cache[difficulty]:
self._difficulty_cache[difficulty].append(quest.quest_id)
logger.info("Quest loaded successfully", quest_id=quest.quest_id)
return quest
except Exception as e:
logger.error("Failed to load quest", file=str(file_path), error=str(e))
return None
def _parse_quest_data(self, data: Dict) -> Quest:
"""
Parse YAML data into a Quest dataclass.
Args:
data: Dictionary loaded from YAML file
Returns:
Quest instance
Raises:
ValueError: If data is invalid or missing required fields
"""
# Validate required fields
required_fields = ["quest_id", "name"]
for field in required_fields:
if field not in data:
raise ValueError(f"Missing required field: {field}")
# Parse objectives
objectives = []
for obj_data in data.get("objectives", []):
objectives.append(QuestObjective(
objective_id=obj_data["objective_id"],
description=obj_data["description"],
objective_type=ObjectiveType(obj_data.get("objective_type", "kill")),
required_progress=obj_data.get("required_progress", 1),
current_progress=0,
target_enemy_type=obj_data.get("target_enemy_type"),
target_item_id=obj_data.get("target_item_id"),
target_location_id=obj_data.get("target_location_id"),
target_npc_id=obj_data.get("target_npc_id"),
))
# Parse rewards
rewards_data = data.get("rewards", {})
rewards = QuestReward(
gold=rewards_data.get("gold", 0),
experience=rewards_data.get("experience", 0),
items=rewards_data.get("items", []),
relationship_bonuses=rewards_data.get("relationship_bonuses", {}),
unlocks_quests=rewards_data.get("unlocks_quests", []),
reveals_locations=rewards_data.get("reveals_locations", []),
)
# Parse offering triggers
triggers_data = data.get("offering_triggers", {})
offering_triggers = QuestOfferingTriggers(
location_types=triggers_data.get("location_types", []),
specific_locations=triggers_data.get("specific_locations", []),
min_character_level=triggers_data.get("min_character_level", 1),
max_character_level=triggers_data.get("max_character_level", 100),
required_quests_completed=triggers_data.get("required_quests_completed", []),
probability_weights=triggers_data.get("probability_weights", {}),
)
# Parse NPC offer dialogues
npc_offer_dialogues = {}
for npc_id, dialogue_data in data.get("npc_offer_dialogues", {}).items():
conditions_data = dialogue_data.get("conditions", {})
conditions = QuestOfferConditions(
min_relationship=conditions_data.get("min_relationship", 0),
required_flags=conditions_data.get("required_flags", []),
forbidden_flags=conditions_data.get("forbidden_flags", []),
)
npc_offer_dialogues[npc_id] = NPCOfferDialogue(
dialogue=dialogue_data.get("dialogue", ""),
conditions=conditions,
)
# Parse lore context
lore_data = data.get("lore_context", {})
lore_context = QuestLoreContext(
backstory=lore_data.get("backstory", ""),
world_connections=lore_data.get("world_connections", []),
regional_hints=lore_data.get("regional_hints", []),
)
# Parse dialogue templates
templates_data = data.get("dialogue_templates", {})
dialogue_templates = QuestDialogueTemplates(
narrative_hooks=templates_data.get("narrative_hooks", []),
ambient_hints=templates_data.get("ambient_hints", []),
)
return Quest(
quest_id=data["quest_id"],
name=data["name"],
description=data.get("description", ""),
difficulty=QuestDifficulty(data.get("difficulty", "easy")),
quest_giver_npc_ids=data.get("quest_giver_npc_ids", []),
quest_giver_name=data.get("quest_giver_name", ""),
location_id=data.get("location_id", ""),
region_id=data.get("region_id", ""),
objectives=objectives,
rewards=rewards,
offering_triggers=offering_triggers,
npc_offer_dialogues=npc_offer_dialogues,
npc_quest_knowledge=data.get("npc_quest_knowledge", {}),
lore_context=lore_context,
dialogue_templates=dialogue_templates,
completion_dialogue=data.get("completion_dialogue", {}),
tags=data.get("tags", []),
)
def load_all_quests(self) -> List[Quest]:
"""
Load all quests from all difficulty directories.
Returns:
List of Quest instances
"""
quests = []
if not self.data_dir.exists():
logger.error("Quest data directory does not exist", data_dir=str(self.data_dir))
return quests
for difficulty_dir in self.data_dir.iterdir():
if not difficulty_dir.is_dir():
continue
for file_path in difficulty_dir.glob("*.yaml"):
quest = self._load_quest_file(file_path)
if quest:
quests.append(quest)
logger.info("All quests loaded", count=len(quests))
return quests
def get_quests_for_npc(self, npc_id: str) -> List[Quest]:
"""
Get all quests that a specific NPC can offer.
This is the primary method for the quest-centric design:
quests define which NPCs can offer them, and this method
finds all quests for a given NPC.
Args:
npc_id: NPC identifier
Returns:
List of Quest instances that this NPC can offer
"""
# Ensure all quests are loaded
if not self._quest_cache:
self.load_all_quests()
quest_ids = self._npc_quest_cache.get(npc_id, [])
return [
self._quest_cache[quest_id]
for quest_id in quest_ids
if quest_id in self._quest_cache
]
def get_quests_by_difficulty(self, difficulty: str) -> List[Quest]:
"""
Get all quests of a specific difficulty.
Args:
difficulty: Difficulty level ("easy", "medium", "hard", "epic")
Returns:
List of Quest instances with this difficulty
"""
# Ensure all quests are loaded
if not self._quest_cache:
self.load_all_quests()
quest_ids = self._difficulty_cache.get(difficulty, [])
return [
self._quest_cache[quest_id]
for quest_id in quest_ids
if quest_id in self._quest_cache
]
def get_quests_for_location(self, location_id: str, location_type: str = "") -> List[Quest]:
"""
Get quests that can be offered at a specific location.
Args:
location_id: Location identifier
location_type: Type of location (e.g., "tavern", "town")
Returns:
List of Quest instances that can be offered here
"""
# Ensure all quests are loaded
if not self._quest_cache:
self.load_all_quests()
matching_quests = []
for quest in self._quest_cache.values():
triggers = quest.offering_triggers
# Check specific locations
if location_id in triggers.specific_locations:
matching_quests.append(quest)
continue
# Check location types
if location_type and location_type in triggers.location_types:
matching_quests.append(quest)
continue
return matching_quests
def get_all_quest_ids(self) -> List[str]:
"""
Get a list of all available quest IDs.
Returns:
List of quest IDs
"""
# Ensure all quests are loaded
if not self._quest_cache:
self.load_all_quests()
return list(self._quest_cache.keys())
def reload_quest(self, quest_id: str) -> Optional[Quest]:
"""
Force reload a quest from disk, bypassing cache.
Useful for development/testing when quest definitions change.
Args:
quest_id: Unique quest identifier
Returns:
Quest instance or None if not found
"""
# Remove from caches if present
if quest_id in self._quest_cache:
old_quest = self._quest_cache[quest_id]
# Remove from NPC cache
for npc_id in old_quest.quest_giver_npc_ids:
if npc_id in self._npc_quest_cache:
self._npc_quest_cache[npc_id] = [
qid for qid in self._npc_quest_cache[npc_id]
if qid != quest_id
]
# Remove from difficulty cache
difficulty = old_quest.difficulty.value
if difficulty in self._difficulty_cache:
self._difficulty_cache[difficulty] = [
qid for qid in self._difficulty_cache[difficulty]
if qid != quest_id
]
del self._quest_cache[quest_id]
return self.load_quest(quest_id)
def clear_cache(self) -> None:
"""Clear all cached data. Useful for testing."""
self._quest_cache.clear()
self._npc_quest_cache.clear()
self._difficulty_cache.clear()
logger.info("Quest cache cleared")
# Global singleton instance
_service_instance: Optional[QuestService] = None
def get_quest_service() -> QuestService:
"""
Get the global QuestService instance.
Returns:
Singleton QuestService instance
"""
global _service_instance
if _service_instance is None:
_service_instance = QuestService()
return _service_instance