Add comprehensive NPC events system with community collaboration
- Implement NPC events module with full IRC command support: - \!events: View all active community events - \!event <id>: Get detailed event information and leaderboard - \!contribute <id>: Participate in community events - \!eventhelp: Comprehensive event system documentation - Add NPC events backend system with automatic spawning: - Configurable event types (resource gathering, pet rescue, exploration) - Difficulty levels (easy, medium, hard) with scaled rewards - Community collaboration mechanics with shared progress - Automatic event spawning and expiration management - Database integration for event tracking and player contributions - Expandable system supporting future event types and mechanics - Admin \!startevent command for manual event creation - Comprehensive error handling and user feedback 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
megaproxy
parent
530134bd36
commit
cd2ad10aec
2 changed files with 529 additions and 0 deletions
293
src/npc_events.py
Normal file
293
src/npc_events.py
Normal file
|
|
@ -0,0 +1,293 @@
|
|||
"""
|
||||
NPC Events System
|
||||
Manages random collaborative events that all players can participate in
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import random
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Optional
|
||||
from src.database import Database
|
||||
|
||||
class NPCEventsManager:
|
||||
def __init__(self, database: Database):
|
||||
self.database = database
|
||||
self.active_events = {}
|
||||
self.event_templates = {
|
||||
1: [ # Difficulty 1 - Easy
|
||||
{
|
||||
"event_type": "resource_gathering",
|
||||
"title": "Village Supply Run",
|
||||
"description": "The village needs supplies! Help gather resources by exploring and finding items.",
|
||||
"target_contributions": 25,
|
||||
"reward_experience": 50,
|
||||
"reward_money": 100,
|
||||
"completion_message": "🎉 The village has enough supplies! Everyone who helped gets rewarded!",
|
||||
"duration_hours": 4
|
||||
},
|
||||
{
|
||||
"event_type": "pet_rescue",
|
||||
"title": "Lost Pet Search",
|
||||
"description": "A pet has gone missing! Help search different locations to find clues.",
|
||||
"target_contributions": 20,
|
||||
"reward_experience": 40,
|
||||
"reward_money": 80,
|
||||
"completion_message": "🐾 The lost pet has been found safe! Thanks to everyone who helped search!",
|
||||
"duration_hours": 3
|
||||
},
|
||||
{
|
||||
"event_type": "community_project",
|
||||
"title": "Park Cleanup",
|
||||
"description": "The local park needs cleaning! Help by contributing your time and effort.",
|
||||
"target_contributions": 30,
|
||||
"reward_experience": 35,
|
||||
"reward_money": 75,
|
||||
"completion_message": "🌳 The park is clean and beautiful again! Great teamwork everyone!",
|
||||
"duration_hours": 5
|
||||
}
|
||||
],
|
||||
2: [ # Difficulty 2 - Medium
|
||||
{
|
||||
"event_type": "emergency_response",
|
||||
"title": "Storm Recovery",
|
||||
"description": "A storm has damaged the town! Help with recovery efforts by contributing resources and time.",
|
||||
"target_contributions": 50,
|
||||
"reward_experience": 100,
|
||||
"reward_money": 200,
|
||||
"completion_message": "⛈️ The town has recovered from the storm! Everyone's hard work paid off!",
|
||||
"duration_hours": 6
|
||||
},
|
||||
{
|
||||
"event_type": "festival_preparation",
|
||||
"title": "Annual Festival Setup",
|
||||
"description": "The annual pet festival is coming! Help set up decorations and prepare activities.",
|
||||
"target_contributions": 40,
|
||||
"reward_experience": 80,
|
||||
"reward_money": 150,
|
||||
"completion_message": "🎪 The festival is ready! Thanks to everyone who helped prepare!",
|
||||
"duration_hours": 8
|
||||
},
|
||||
{
|
||||
"event_type": "research_expedition",
|
||||
"title": "Scientific Discovery",
|
||||
"description": "Researchers need help documenting rare pets! Contribute by exploring and reporting findings.",
|
||||
"target_contributions": 35,
|
||||
"reward_experience": 90,
|
||||
"reward_money": 180,
|
||||
"completion_message": "🔬 The research is complete! Your discoveries will help future generations!",
|
||||
"duration_hours": 7
|
||||
}
|
||||
],
|
||||
3: [ # Difficulty 3 - Hard
|
||||
{
|
||||
"event_type": "crisis_response",
|
||||
"title": "Regional Emergency",
|
||||
"description": "A regional crisis requires immediate community response! All trainers needed!",
|
||||
"target_contributions": 75,
|
||||
"reward_experience": 150,
|
||||
"reward_money": 300,
|
||||
"completion_message": "🚨 Crisis averted! The entire region is safe thanks to your heroic efforts!",
|
||||
"duration_hours": 12
|
||||
},
|
||||
{
|
||||
"event_type": "legendary_encounter",
|
||||
"title": "Legendary Pet Sighting",
|
||||
"description": "A legendary pet has been spotted! Help researchers track and document this rare encounter.",
|
||||
"target_contributions": 60,
|
||||
"reward_experience": 200,
|
||||
"reward_money": 400,
|
||||
"completion_message": "✨ The legendary pet has been successfully documented! History has been made!",
|
||||
"duration_hours": 10
|
||||
},
|
||||
{
|
||||
"event_type": "ancient_mystery",
|
||||
"title": "Ancient Ruins Discovery",
|
||||
"description": "Ancient ruins have been discovered! Help archaeologists uncover the secrets within.",
|
||||
"target_contributions": 80,
|
||||
"reward_experience": 180,
|
||||
"reward_money": 350,
|
||||
"completion_message": "🏛️ The ancient secrets have been revealed! Your efforts uncovered lost knowledge!",
|
||||
"duration_hours": 14
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
async def start_background_task(self):
|
||||
"""Start the background task that manages NPC events"""
|
||||
while True:
|
||||
try:
|
||||
# Check for expired events
|
||||
await self.expire_events()
|
||||
|
||||
# Distribute rewards for completed events
|
||||
await self.distribute_completed_rewards()
|
||||
|
||||
# Maybe spawn a new event
|
||||
await self.maybe_spawn_event()
|
||||
|
||||
# Wait 30 minutes before next check
|
||||
await asyncio.sleep(30 * 60)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error in NPC events background task: {e}")
|
||||
await asyncio.sleep(5 * 60) # Wait 5 minutes on error
|
||||
|
||||
async def expire_events(self):
|
||||
"""Mark expired events as expired"""
|
||||
try:
|
||||
expired_count = await self.database.expire_npc_events()
|
||||
if expired_count > 0:
|
||||
print(f"🕐 {expired_count} NPC events expired")
|
||||
except Exception as e:
|
||||
print(f"Error expiring NPC events: {e}")
|
||||
|
||||
async def distribute_completed_rewards(self):
|
||||
"""Distribute rewards for completed events"""
|
||||
try:
|
||||
# Get completed events that haven't distributed rewards yet
|
||||
completed_events = await self.database.get_active_npc_events()
|
||||
|
||||
for event in completed_events:
|
||||
if event['status'] == 'completed':
|
||||
result = await self.database.distribute_event_rewards(event['id'])
|
||||
if result['success']:
|
||||
print(f"🎁 Distributed rewards for event '{event['title']}' to {result['participants_rewarded']} players")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error distributing event rewards: {e}")
|
||||
|
||||
async def maybe_spawn_event(self):
|
||||
"""Maybe spawn a new event based on conditions"""
|
||||
try:
|
||||
# Check if we have any active events
|
||||
active_events = await self.database.get_active_npc_events()
|
||||
|
||||
# Don't spawn if we already have 2 or more active events
|
||||
if len(active_events) >= 2:
|
||||
return
|
||||
|
||||
# 20% chance to spawn a new event each check (every 30 minutes)
|
||||
if random.random() < 0.2:
|
||||
await self.spawn_random_event()
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error in maybe_spawn_event: {e}")
|
||||
|
||||
async def spawn_random_event(self):
|
||||
"""Spawn a random event based on difficulty"""
|
||||
try:
|
||||
# Choose difficulty (weighted towards easier events)
|
||||
difficulty_weights = {1: 0.6, 2: 0.3, 3: 0.1}
|
||||
difficulty = random.choices(list(difficulty_weights.keys()),
|
||||
weights=list(difficulty_weights.values()))[0]
|
||||
|
||||
# Choose random event template
|
||||
templates = self.event_templates[difficulty]
|
||||
template = random.choice(templates)
|
||||
|
||||
# Create event data
|
||||
event_data = {
|
||||
'event_type': template['event_type'],
|
||||
'title': template['title'],
|
||||
'description': template['description'],
|
||||
'difficulty': difficulty,
|
||||
'target_contributions': template['target_contributions'],
|
||||
'reward_experience': template['reward_experience'],
|
||||
'reward_money': template['reward_money'],
|
||||
'completion_message': template['completion_message'],
|
||||
'expires_at': (datetime.now() + timedelta(hours=template['duration_hours'])).isoformat()
|
||||
}
|
||||
|
||||
# Create the event
|
||||
event_id = await self.database.create_npc_event(event_data)
|
||||
|
||||
print(f"🎯 New NPC event spawned: '{template['title']}' (ID: {event_id}, Difficulty: {difficulty})")
|
||||
|
||||
return event_id
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error spawning random event: {e}")
|
||||
return None
|
||||
|
||||
async def get_active_events(self) -> List[Dict]:
|
||||
"""Get all active events"""
|
||||
return await self.database.get_active_npc_events()
|
||||
|
||||
async def contribute_to_event(self, event_id: int, player_id: int, contribution: int = 1) -> Dict:
|
||||
"""Add a player's contribution to an event"""
|
||||
return await self.database.contribute_to_npc_event(event_id, player_id, contribution)
|
||||
|
||||
async def get_event_details(self, event_id: int) -> Optional[Dict]:
|
||||
"""Get detailed information about an event"""
|
||||
event = await self.database.get_npc_event_by_id(event_id)
|
||||
if not event:
|
||||
return None
|
||||
|
||||
# Add leaderboard
|
||||
leaderboard = await self.database.get_event_leaderboard(event_id)
|
||||
event['leaderboard'] = leaderboard
|
||||
|
||||
return event
|
||||
|
||||
async def get_player_contributions(self, player_id: int, event_id: int) -> int:
|
||||
"""Get player's contributions to a specific event"""
|
||||
return await self.database.get_player_event_contributions(player_id, event_id)
|
||||
|
||||
async def force_spawn_event(self, difficulty: int = 1) -> Optional[int]:
|
||||
"""Force spawn an event (admin command)"""
|
||||
if difficulty not in self.event_templates:
|
||||
return None
|
||||
|
||||
templates = self.event_templates[difficulty]
|
||||
template = random.choice(templates)
|
||||
|
||||
event_data = {
|
||||
'event_type': template['event_type'],
|
||||
'title': template['title'],
|
||||
'description': template['description'],
|
||||
'difficulty': difficulty,
|
||||
'target_contributions': template['target_contributions'],
|
||||
'reward_experience': template['reward_experience'],
|
||||
'reward_money': template['reward_money'],
|
||||
'completion_message': template['completion_message'],
|
||||
'expires_at': (datetime.now() + timedelta(hours=template['duration_hours'])).isoformat()
|
||||
}
|
||||
|
||||
return await self.database.create_npc_event(event_data)
|
||||
|
||||
async def force_spawn_specific_event(self, event_type: str, difficulty: int = 1) -> Optional[int]:
|
||||
"""Force spawn a specific event type (admin command)"""
|
||||
if difficulty not in self.event_templates:
|
||||
return None
|
||||
|
||||
# Find template matching the event type
|
||||
templates = self.event_templates[difficulty]
|
||||
template = None
|
||||
for t in templates:
|
||||
if t['event_type'] == event_type:
|
||||
template = t
|
||||
break
|
||||
|
||||
if not template:
|
||||
return None
|
||||
|
||||
event_data = {
|
||||
'event_type': template['event_type'],
|
||||
'title': template['title'],
|
||||
'description': template['description'],
|
||||
'difficulty': difficulty,
|
||||
'target_contributions': template['target_contributions'],
|
||||
'reward_experience': template['reward_experience'],
|
||||
'reward_money': template['reward_money'],
|
||||
'completion_message': template['completion_message'],
|
||||
'expires_at': (datetime.now() + timedelta(hours=template['duration_hours'])).isoformat()
|
||||
}
|
||||
|
||||
return await self.database.create_npc_event(event_data)
|
||||
|
||||
def get_progress_bar(self, current: int, target: int, width: int = 20) -> str:
|
||||
"""Generate a progress bar for event progress"""
|
||||
filled = int((current / target) * width)
|
||||
bar = "█" * filled + "░" * (width - filled)
|
||||
percentage = min(100, int((current / target) * 100))
|
||||
return f"[{bar}] {percentage}% ({current}/{target})"
|
||||
Loading…
Add table
Add a link
Reference in a new issue