#!/usr/bin/env python3
"""
PetBot Web Server
Provides web interface for bot data including help, player stats, and pet collections
"""
import os
import sys
import asyncio
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
from threading import Thread
import time
# Add the project directory to the path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from src.database import Database
from src.rate_limiter import RateLimiter, CommandCategory
class PetBotRequestHandler(BaseHTTPRequestHandler):
"""HTTP request handler for PetBot web server"""
@property
def database(self):
"""Get database instance from server"""
return self.server.database
@property
def bot(self):
"""Get bot instance from server"""
return getattr(self.server, 'bot', None)
@property
def rate_limiter(self):
"""Get rate limiter from bot instance"""
bot = self.bot
return getattr(bot, 'rate_limiter', None) if bot else None
def get_client_ip(self):
"""Get client IP address for rate limiting"""
# Check for X-Forwarded-For header (in case of proxy)
forwarded_for = self.headers.get('X-Forwarded-For')
if forwarded_for:
return forwarded_for.split(',')[0].strip()
# Check for X-Real-IP header
real_ip = self.headers.get('X-Real-IP')
if real_ip:
return real_ip.strip()
# Fallback to client address
return self.client_address[0]
def check_rate_limit(self):
"""Check rate limit for web requests"""
if not self.rate_limiter:
return True, None
client_ip = self.get_client_ip()
# Use IP address as user identifier for web requests
user_identifier = f"web:{client_ip}"
# Run async rate limit check
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
allowed, message = loop.run_until_complete(
self.rate_limiter.check_rate_limit(user_identifier, CommandCategory.WEB)
)
return allowed, message
finally:
loop.close()
def send_rate_limit_error(self, message):
"""Send rate limit error response"""
self.send_response(429)
self.send_header('Content-type', 'text/html; charset=utf-8')
self.send_header('Retry-After', '60')
self.end_headers()
content = f"""
"""
html = self.get_page_template("PetBot Game Hub", content, "")
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(html.encode())
def serve_help(self):
"""Serve the help page using unified template"""
content = """
๐ PetBot Commands
Complete guide to Pokemon-style pet collecting in IRC
๐ Getting Started
!start
Begin your pet collecting journey! Creates your trainer account and gives you your first starter pet.
Example: !start
!help
Get a link to this comprehensive command reference page.
Example: !help
!stats
View your basic trainer information including level, experience, and money.
Example: !stats
๐ Exploration & Travel
!explore
Search your current location for wild pets or items. You might find pets to battle/catch or discover useful items!
Example: !explore
!travel <location>
Move to a different location. Each area has unique pets and gyms. Some locations require achievements to unlock.
Example: !travel whispering woods
!where / !location
See which location you're currently in and get information about the area.
Example: !where
๐บ๏ธ Available Locations
Starter Town - Peaceful starting area (Fire/Water/Grass pets)
Whispering Woods - Ancient forest (Grass pets + new species: Vinewrap, Bloomtail)
Electric Canyon - Charged valley (Electric/Rock pets)
Thunderstorm - 2.0x Electric spawns (30-60 minutes)
Blizzard - 1.7x Ice/Water spawns (1-2 hours)
Earthquake - 1.8x Rock spawns (30-90 minutes)
Calm - Normal spawns (1.5-3 hours)
โ๏ธ Battle System
!catch / !capture
Attempt to catch a wild pet that appeared during exploration. Success depends on the pet's level and rarity.
Example: !catch
!battle
Start a turn-based battle with a wild pet. Defeat it to gain experience and money for your active pet.
Example: !battle
!attack <move>
Use a specific move during battle. Each move has different power, type, and effects.
Example: !attack flamethrower
!moves
View all available moves for your active pet, including their types and power levels.
Example: !moves
!flee
Attempt to escape from the current battle. Not always successful!
Example: !flee
๐๏ธ Gym Battles NEW!
!gym
List all gyms in your current location with your progress. Shows victories and next difficulty level.
Example: !gym
!gym list
Show all gyms across all locations with your badge collection progress.
Example: !gym list
!gym challenge "<name>"
Challenge a gym leader! You must be in the same location as the gym. Difficulty increases with each victory.
Example: !gym challenge "Forest Guardian"
!gym info "<name>"
Get detailed information about a gym including leader, theme, team, and badge details.
Example: !gym info "Storm Master"
๐ก Gym Strategy: Each gym specializes in a specific type. Bring pets with type advantages! The more you beat a gym, the harder it gets, but the better the rewards!
Access your team builder web interface for drag-and-drop team management with PIN verification.
Example: !team
!pets
View your complete pet collection with detailed stats and information via web interface.
Example: !pets
!activate <pet>
Add a pet to your active battle team. You can have multiple active pets for different situations.
Example: !activate flamey
!deactivate <pet>
Remove a pet from your active team and put it in storage.
Example: !deactivate aqua
!nickname <pet> <new_name>
Give a custom nickname to one of your pets. Nicknames must be 20 characters or less.
Example: !nickname flamey FireStorm
๐ Inventory System NEW!
!inventory / !inv / !items
View all items in your inventory organized by category. Shows quantities and item descriptions.
Example: !inventory
!use <item name>
Use a consumable item from your inventory. Items can heal pets, boost stats, or provide other benefits.
Example: !use Small Potion
๐ฏ Item Categories & Rarities
โ Common (15%) - Small Potions, basic healing items
โ Uncommon (8-12%) - Large Potions, battle boosters, special berries
โ Rare (3-6%) - Super Potions, speed elixirs, location treasures
โ Epic (2-3%) - Evolution stones, rare crystals, ancient artifacts
โฆ Legendary (1%) - Lucky charms, ancient fossils, ultimate items
๐ก Item Discovery: Find items while exploring! Each location has unique treasures including rare Coin Pouches with 1-3 coins. Items stack in your inventory and can be used anytime.
๐ Achievements & Progress
!achievements
View your achievement progress and see which new locations you've unlocked.
Example: !achievements
๐ฏ Location Unlock Requirements
Pet Collector (5 pets) โ Unlocks Whispering Woods
Spark Collector (2 Electric species) โ Unlocks Electric Canyon
Rock Hound (3 Rock species) โ Unlocks Crystal Caves
Locations Guide - All areas with spawn information and current weather
Gym Badges - Display your earned badges and battle progress
Inventory Management - Visual item display with command instructions
"""
# Add command-specific CSS to the unified styles
additional_css = """
.section {
background: var(--bg-secondary);
border-radius: 15px;
margin-bottom: 30px;
box-shadow: 0 4px 20px rgba(0,0,0,0.3);
border: 1px solid var(--border-color);
overflow: hidden;
}
.section-header {
background: var(--gradient-primary);
color: white;
padding: 20px 25px;
font-size: 1.3em;
font-weight: 700;
}
.section-content {
padding: 25px;
}
.command-grid {
display: grid;
gap: 20px;
}
.command {
border: 1px solid var(--border-color);
border-radius: 12px;
overflow: hidden;
transition: all 0.3s ease;
background: var(--bg-tertiary);
}
.command:hover {
transform: translateY(-3px);
box-shadow: 0 8px 25px rgba(102, 255, 102, 0.15);
border-color: var(--text-accent);
}
.command-name {
background: var(--bg-primary);
padding: 15px 20px;
font-family: 'Fira Code', 'Courier New', monospace;
font-weight: bold;
color: var(--text-accent);
border-bottom: 1px solid var(--border-color);
font-size: 1.2em;
text-shadow: 0 0 10px rgba(102, 255, 102, 0.3);
}
.command-desc {
padding: 20px;
line-height: 1.7;
color: var(--text-primary);
}
.command-example {
background: var(--bg-primary);
padding: 12px 20px;
font-family: 'Fira Code', 'Courier New', monospace;
color: var(--text-secondary);
border-top: 1px solid var(--border-color);
font-size: 0.95em;
}
.info-box {
background: var(--bg-tertiary);
padding: 20px;
border-radius: 12px;
margin: 20px 0;
border: 1px solid var(--border-color);
}
.info-box h4 {
margin: 0 0 15px 0;
color: var(--text-accent);
font-size: 1.1em;
font-weight: 600;
}
.info-box ul {
margin: 0;
padding-left: 25px;
}
.info-box li {
margin: 8px 0;
color: var(--text-primary);
}
.info-box strong {
color: var(--text-accent);
}
.footer {
text-align: center;
margin-top: 50px;
padding: 30px;
background: var(--bg-secondary);
border-radius: 15px;
color: var(--text-secondary);
box-shadow: 0 4px 20px rgba(0,0,0,0.3);
border: 1px solid var(--border-color);
}
.tip {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 20px;
border-radius: 12px;
margin: 20px 0;
font-weight: 500;
text-shadow: 0 2px 4px rgba(0,0,0,0.3);
}
.gym-list {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(300px, 1fr));
gap: 15px;
margin: 15px 0;
}
.gym-card {
background: var(--bg-primary);
padding: 15px;
border-radius: 8px;
border: 1px solid var(--border-color);
}
.gym-card strong {
color: var(--text-accent);
}
"""
# Get the unified template with additional CSS
html_content = self.get_page_template("Command Help", content, "help")
# Insert additional CSS before closing tag
html_content = html_content.replace("", additional_css + "")
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(html_content.encode())
def serve_players(self):
"""Serve the players page with real data"""
# Get database instance from the server class
database = self.server.database if hasattr(self.server, 'database') else None
if not database:
self.serve_error_page("Players", "Database not available")
return
# Fetch players data
try:
import asyncio
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
players_data = loop.run_until_complete(self.fetch_players_data(database))
loop.close()
self.serve_players_data(players_data)
except Exception as e:
print(f"Error fetching players data: {e}")
self.serve_error_page("Players", f"Error loading players: {str(e)}")
async def fetch_players_data(self, database):
"""Fetch all players data from database"""
try:
import aiosqlite
async with aiosqlite.connect(database.db_path) as db:
# Get all players with basic stats
cursor = await db.execute("""
SELECT p.nickname, p.level, p.experience, p.money, p.created_at,
l.name as location_name,
(SELECT COUNT(*) FROM pets WHERE player_id = p.id) as pet_count,
(SELECT COUNT(*) FROM pets WHERE player_id = p.id AND is_active = 1) as active_pets,
(SELECT COUNT(*) FROM player_achievements WHERE player_id = p.id) as achievement_count
FROM players p
LEFT JOIN locations l ON p.current_location_id = l.id
ORDER BY p.level DESC, p.experience DESC
""")
rows = await cursor.fetchall()
# Convert SQLite rows to dictionaries properly
players = []
for row in rows:
player_dict = {
'nickname': row[0],
'level': row[1],
'experience': row[2],
'money': row[3],
'created_at': row[4],
'location_name': row[5],
'pet_count': row[6],
'active_pets': row[7],
'achievement_count': row[8]
}
players.append(player_dict)
return players
except Exception as e:
print(f"Database error fetching players: {e}")
return []
def serve_players_data(self, players_data):
"""Serve players page with real data"""
# Calculate statistics
total_players = len(players_data)
total_pets = sum(p['pet_count'] for p in players_data) if players_data else 0
total_achievements = sum(p['achievement_count'] for p in players_data) if players_data else 0
highest_level = max((p['level'] for p in players_data), default=0) if players_data else 0
# Build statistics cards
stats_content = f"""
๐ Total Players
{total_players}
๐พ Total Pets
{total_pets}
๐ Achievements
{total_achievements}
โญ Highest Level
{highest_level}
"""
# Build players table HTML
if players_data:
players_html = ""
for i, player in enumerate(players_data, 1):
rank_emoji = {"1": "๐ฅ", "2": "๐ฅ", "3": "๐ฅ"}.get(str(i), f"{i}.")
players_html += f"""
'
# Data rows
content += ''
for i, player in enumerate(data):
row_data = row_formatter(player, i)
rank_class = f"rank-{i+1}" if i < 3 else ""
content += f'
'
for cell in row_data:
content += f'
{cell}
'
content += '
'
content += ''
content += '
'
content += '
'
return content
def serve_locations(self):
"""Serve the locations page with real data"""
# Get database instance from the server class
database = self.server.database if hasattr(self.server, 'database') else None
if not database:
self.serve_error_page("Locations", "Database not available")
return
# Fetch locations data
try:
import asyncio
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
locations_data = loop.run_until_complete(self.fetch_locations_data(database))
loop.close()
self.serve_locations_data(locations_data)
except Exception as e:
print(f"Error fetching locations data: {e}")
self.serve_error_page("Locations", f"Error loading locations: {str(e)}")
async def fetch_locations_data(self, database):
"""Fetch all locations and their spawn data from database"""
try:
import aiosqlite
async with aiosqlite.connect(database.db_path) as db:
# Get all locations
cursor = await db.execute("""
SELECT l.*,
GROUP_CONCAT(DISTINCT ps.name || ' (' || ps.type1 ||
CASE WHEN ps.type2 IS NOT NULL THEN '/' || ps.type2 ELSE '' END || ')') as spawns
FROM locations l
LEFT JOIN location_spawns ls ON l.id = ls.location_id
LEFT JOIN pet_species ps ON ls.species_id = ps.id
GROUP BY l.id
ORDER BY l.id
""")
rows = await cursor.fetchall()
# Convert SQLite rows to dictionaries properly
locations = []
for row in rows:
location_dict = {
'id': row[0],
'name': row[1],
'description': row[2],
'level_min': row[3],
'level_max': row[4],
'spawns': row[5] if len(row) > 5 else None
}
locations.append(location_dict)
return locations
except Exception as e:
print(f"Database error fetching locations: {e}")
return []
def serve_locations_data(self, locations_data):
"""Serve locations page with real data using unified template"""
# Build locations HTML
locations_html = ""
if locations_data:
for location in locations_data:
spawns = location.get('spawns', 'No pets found')
if not spawns or spawns == 'None':
spawns = "No pets spawn here yet"
# Split spawns into a readable list and remove duplicates
if spawns != "No pets spawn here yet":
spawn_list = list(set([spawn.strip() for spawn in spawns.split(',') if spawn.strip()]))
spawn_list.sort() # Sort alphabetically for consistency
else:
spawn_list = []
spawn_badges = ""
visible_spawns = spawn_list[:6] # Show first 6
hidden_spawns = spawn_list[6:] # Hide the rest
# Add visible spawn badges
for spawn in visible_spawns:
spawn_badges += f'{spawn}'
# Add hidden spawn badges (initially hidden)
if hidden_spawns:
location_id = location['id']
for spawn in hidden_spawns:
spawn_badges += f'{spawn}'
# Add functional "show more" button
spawn_badges += f'+{len(hidden_spawns)} more'
if not spawn_badges:
spawn_badges = 'No pets spawn here yet'
locations_html += f"""
"""
# Add no-pets-specific CSS
additional_css = """
.main-container {
text-align: center;
max-width: 800px;
margin: 0 auto;
}
.no-pets-message {
background: var(--bg-secondary);
padding: 40px;
border-radius: 15px;
border: 2px solid var(--warning-color);
margin-top: 30px;
box-shadow: 0 4px 20px rgba(0,0,0,0.3);
}
.no-pets-message h2 {
color: var(--warning-color);
margin-top: 0;
}
"""
html_content = self.get_page_template(f"Team Builder - {nickname}", content, "players")
html_content = html_content.replace("", additional_css + "")
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(html_content.encode())
def serve_teambuilder_interface(self, nickname, pets):
"""Serve the full interactive team builder interface"""
active_pets = [pet for pet in pets if pet['is_active']]
inactive_pets = [pet for pet in pets if not pet['is_active']]
# Debug logging
print(f"Team Builder Debug for {nickname}:")
print(f"Total pets: {len(pets)}")
active_names = [f"{pet['nickname'] or pet['species_name']} (ID:{pet['id']}, is_active:{pet['is_active']})" for pet in active_pets]
inactive_names = [f"{pet['nickname'] or pet['species_name']} (ID:{pet['id']}, is_active:{pet['is_active']})" for pet in inactive_pets]
print(f"Active pets: {len(active_pets)} - {active_names}")
print(f"Inactive pets: {len(inactive_pets)} - {inactive_names}")
# Generate detailed pet cards with debugging
def make_pet_card(pet, is_active):
name = pet['nickname'] or pet['species_name']
status = "Active" if is_active else "Storage"
status_class = "active" if is_active else "storage"
type_str = pet['type1']
if pet['type2']:
type_str += f"/{pet['type2']}"
# Get emoji for the pet species
emoji = pet.get('emoji', '๐พ') # Default to paw emoji if none specified
# Debug logging
print(f"Making pet card for {name} (ID: {pet['id']}): is_active={pet['is_active']}, passed_is_active={is_active}, status_class={status_class}, team_order={pet.get('team_order', 'None')}")
# Calculate HP percentage for health bar
hp_percent = (pet['hp'] / pet['max_hp']) * 100 if pet['max_hp'] > 0 else 0
hp_color = "#4CAF50" if hp_percent > 60 else "#FF9800" if hp_percent > 25 else "#f44336"
return f"""
{emoji} {name}
{status}
Level {pet['level']} {pet['species_name']}
{type_str}
HP: {pet['hp']}/{pet['max_hp']}
ATK{pet['attack']}
DEF{pet['defense']}
SPD{pet['speed']}
EXP{pet['experience']}
{'๐' if pet['happiness'] > 70 else '๐' if pet['happiness'] > 40 else '๐'}Happiness: {pet['happiness']}/100
"""
# Create 6 numbered slots and place pets in their positions
team_slots = [''] * 6 # Initialize 6 empty slots
# Place active pets in their team_order positions
for pet in active_pets:
team_order = pet.get('team_order')
if team_order and 1 <= team_order <= 6:
team_slots[team_order - 1] = make_pet_card(pet, True)
storage_cards = ''.join(make_pet_card(pet, False) for pet in inactive_pets)
html = f"""
Team Builder - {nickname}โ Back to {nickname}'s Profile
๐พ Team Builder
Drag pets between Active and Storage to build your perfect team
Changes are saved securely with PIN verification via IRC
๐ PIN Verification Required
A 6-digit PIN has been sent to you via IRC private message.
Enter the PIN below to confirm your team changes:
"""
# Generate storage pets HTML first
storage_pets_html = ""
for pet in inactive_pets:
storage_pets_html += make_pet_card(pet, False)
# Generate active pets HTML for team slots
active_pets_html = ""
for pet in active_pets:
if pet.get('team_order'):
active_pets_html += make_pet_card(pet, True)
# Create content using string concatenation instead of f-strings to avoid CSS brace issues
team_builder_content = """
๐พ Team Builder
Drag pets between Active Team and Storage. Double-click as backup.
โ๏ธ Active Team (1-6 pets)
Slot 1 (Leader)
Drop pet here
Slot 2
Drop pet here
Slot 3
Drop pet here
Slot 4
Drop pet here
Slot 5
Drop pet here
Slot 6
Drop pet here
๐ฆ Storage
""" + storage_pets_html + active_pets_html + """
๐พ Team Configurations
Save up to 3 different team setups for quick switching between strategies
Editing current team (not saved to any configuration)
Changes are saved securely with PIN verification via IRC
๐ PIN Verification Required
A 6-digit PIN has been sent to you via IRC private message.
Enter the PIN below to confirm your team changes:
๐ก How to use:
โข Drag pets to team slots
โข Double-click to move pets
โข Empty slots show placeholders
"""
# Get the unified template
html_content = self.get_page_template(f"Team Builder - {nickname}", team_builder_content, "")
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(html_content.encode())
def handle_team_save(self, nickname):
"""Handle team save request and generate PIN"""
try:
# Get POST data
content_length = int(self.headers.get('Content-Length', 0))
if content_length == 0:
self.send_json_response({"success": False, "error": "No data provided"}, 400)
return
post_data = self.rfile.read(content_length).decode('utf-8')
# Parse JSON data
import json
try:
team_data = json.loads(post_data)
except json.JSONDecodeError:
self.send_json_response({"success": False, "error": "Invalid JSON data"}, 400)
return
# Run async operations
import asyncio
result = asyncio.run(self._handle_team_save_async(nickname, team_data))
if result["success"]:
self.send_json_response(result, 200)
else:
self.send_json_response(result, 400)
except Exception as e:
print(f"Error in handle_team_save: {e}")
self.send_json_response({"success": False, "error": "Internal server error"}, 500)
async def _handle_team_save_async(self, nickname, team_data):
"""Async handler for team save"""
try:
# Get player
player = await self.database.get_player(nickname)
if not player:
return {"success": False, "error": "Player not found"}
# Validate team composition
validation = await self.database.validate_team_composition(player["id"], team_data)
if not validation["valid"]:
return {"success": False, "error": validation["error"]}
# Create pending team change with PIN
import json
result = await self.database.create_pending_team_change(
player["id"],
json.dumps(team_data)
)
if result["success"]:
# Send PIN via IRC
self.send_pin_via_irc(nickname, result["pin_code"])
return {
"success": True,
"message": "PIN sent to your IRC private messages",
"expires_in_minutes": 10
}
else:
return result
except Exception as e:
print(f"Error in _handle_team_save_async: {e}")
return {"success": False, "error": str(e)}
def handle_team_verify(self, nickname):
"""Handle PIN verification and apply team changes"""
try:
# Get POST data
content_length = int(self.headers.get('Content-Length', 0))
if content_length == 0:
self.send_json_response({"success": False, "error": "No PIN provided"}, 400)
return
post_data = self.rfile.read(content_length).decode('utf-8')
# Parse JSON data
import json
try:
data = json.loads(post_data)
pin_code = data.get("pin", "").strip()
except (json.JSONDecodeError, AttributeError):
self.send_json_response({"success": False, "error": "Invalid data format"}, 400)
return
if not pin_code:
self.send_json_response({"success": False, "error": "PIN code is required"}, 400)
return
# Run async operations
import asyncio
result = asyncio.run(self._handle_team_verify_async(nickname, pin_code))
if result["success"]:
self.send_json_response(result, 200)
else:
self.send_json_response(result, 400)
except Exception as e:
print(f"Error in handle_team_verify: {e}")
self.send_json_response({"success": False, "error": "Internal server error"}, 500)
async def _handle_team_verify_async(self, nickname, pin_code):
"""Async handler for PIN verification"""
try:
# Get player
player = await self.database.get_player(nickname)
if not player:
return {"success": False, "error": "Player not found"}
# Apply team changes with PIN verification
result = await self.database.apply_team_change(player["id"], pin_code)
if result["success"]:
return {
"success": True,
"message": f"Team changes applied successfully! {result['changes_applied']} pets updated.",
"changes_applied": result["changes_applied"]
}
else:
return result
except Exception as e:
print(f"Error in _handle_team_verify_async: {e}")
return {"success": False, "error": str(e)}
def send_pin_via_irc(self, nickname, pin_code):
"""Send PIN to player via IRC private message"""
print(f"๐ PIN for {nickname}: {pin_code}")
# Try to send via IRC bot if available
if self.bot and hasattr(self.bot, 'send_message_sync'):
try:
# Send PIN via private message using sync wrapper
self.bot.send_message_sync(nickname, f"๐ Team Builder PIN: {pin_code}")
self.bot.send_message_sync(nickname, f"๐ก Enter this PIN on the web page to confirm your team changes. PIN expires in 10 minutes.")
print(f"โ PIN sent to {nickname} via IRC")
except Exception as e:
print(f"โ Failed to send PIN via IRC: {e}")
else:
print(f"โ No IRC bot available to send PIN to {nickname}")
print(f"๐ก Manual PIN for {nickname}: {pin_code}")
def handle_team_config_save(self, nickname, slot):
"""Handle saving team configuration to a slot"""
try:
# Get POST data
content_length = int(self.headers.get('Content-Length', 0))
if content_length == 0:
self.send_json_response({"success": False, "error": "No data provided"}, 400)
return
post_data = self.rfile.read(content_length).decode('utf-8')
# Parse JSON data
import json
try:
data = json.loads(post_data)
config_name = data.get("name", f"Team Config {slot}")
team_data = data.get("team", [])
except json.JSONDecodeError:
self.send_json_response({"success": False, "error": "Invalid JSON data"}, 400)
return
# Validate slot number
try:
slot_num = int(slot)
if slot_num < 1 or slot_num > 3:
self.send_json_response({"success": False, "error": "Slot must be 1, 2, or 3"}, 400)
return
except ValueError:
self.send_json_response({"success": False, "error": "Invalid slot number"}, 400)
return
# Run async operations
import asyncio
result = asyncio.run(self._handle_team_config_save_async(nickname, slot_num, config_name, team_data))
if result["success"]:
self.send_json_response(result, 200)
else:
self.send_json_response(result, 400)
except Exception as e:
print(f"Error in handle_team_config_save: {e}")
self.send_json_response({"success": False, "error": "Internal server error"}, 500)
async def _handle_team_config_save_async(self, nickname, slot_num, config_name, team_data):
"""Async handler for team configuration save"""
try:
# Get player
player = await self.database.get_player(nickname)
if not player:
return {"success": False, "error": "Player not found"}
# Save configuration
import json
success = await self.database.save_team_configuration(
player["id"], slot_num, config_name, json.dumps(team_data)
)
if success:
return {"success": True, "message": f"Team configuration '{config_name}' saved to slot {slot_num}"}
else:
return {"success": False, "error": "Failed to save team configuration"}
except Exception as e:
print(f"Error in _handle_team_config_save_async: {e}")
return {"success": False, "error": str(e)}
def handle_team_config_load(self, nickname, slot):
"""Handle loading team configuration from a slot"""
try:
# Validate slot number
try:
slot_num = int(slot)
if slot_num < 1 or slot_num > 3:
self.send_json_response({"success": False, "error": "Slot must be 1, 2, or 3"}, 400)
return
except ValueError:
self.send_json_response({"success": False, "error": "Invalid slot number"}, 400)
return
# Run async operations
import asyncio
result = asyncio.run(self._handle_team_config_load_async(nickname, slot_num))
if result["success"]:
self.send_json_response(result, 200)
else:
self.send_json_response(result, 404 if "not found" in result.get("error", "") else 400)
except Exception as e:
print(f"Error in handle_team_config_load: {e}")
self.send_json_response({"success": False, "error": "Internal server error"}, 500)
async def _handle_team_config_load_async(self, nickname, slot_num):
"""Async handler for team configuration load"""
try:
# Get player
player = await self.database.get_player(nickname)
if not player:
return {"success": False, "error": "Player not found"}
# Load configuration
config = await self.database.load_team_configuration(player["id"], slot_num)
if config:
import json
team_data = json.loads(config["team_data"])
return {
"success": True,
"config_name": config["config_name"],
"team_data": team_data,
"updated_at": config["updated_at"]
}
else:
return {"success": False, "error": f"No team configuration found in slot {slot_num}"}
except Exception as e:
print(f"Error in _handle_team_config_load_async: {e}")
return {"success": False, "error": str(e)}
def handle_team_config_rename(self, nickname, slot):
"""Handle renaming team configuration in a slot"""
try:
# Validate slot number
try:
slot_num = int(slot)
if slot_num < 1 or slot_num > 3:
self.send_json_response({"success": False, "error": "Slot must be 1, 2, or 3"}, 400)
return
except ValueError:
self.send_json_response({"success": False, "error": "Invalid slot number"}, 400)
return
# Get the new name from request body
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
import json
try:
data = json.loads(post_data.decode('utf-8'))
new_name = data.get('new_name', '').strip()
if not new_name:
self.send_json_response({"success": False, "error": "Configuration name cannot be empty"}, 400)
return
if len(new_name) > 50:
self.send_json_response({"success": False, "error": "Configuration name too long (max 50 characters)"}, 400)
return
except json.JSONDecodeError:
self.send_json_response({"success": False, "error": "Invalid JSON data"}, 400)
return
# Run async operations
import asyncio
result = asyncio.run(self._handle_team_config_rename_async(nickname, slot_num, new_name))
if result["success"]:
self.send_json_response(result, 200)
else:
self.send_json_response(result, 404 if "not found" in result.get("error", "") else 400)
except Exception as e:
print(f"Error in handle_team_config_rename: {e}")
self.send_json_response({"success": False, "error": "Internal server error"}, 500)
async def _handle_team_config_rename_async(self, nickname, slot_num, new_name):
"""Async handler for team configuration rename"""
try:
# Get player
player = await self.database.get_player(nickname)
if not player:
return {"success": False, "error": "Player not found"}
# Check if configuration exists in the slot
existing_config = await self.database.load_team_configuration(player["id"], slot_num)
if not existing_config:
return {"success": False, "error": f"No team configuration found in slot {slot_num}"}
# Rename the configuration
success = await self.database.rename_team_configuration(player["id"], slot_num, new_name)
if success:
return {
"success": True,
"message": f"Configuration renamed to '{new_name}'",
"new_name": new_name
}
else:
return {"success": False, "error": "Failed to rename configuration"}
except Exception as e:
print(f"Error in _handle_team_config_rename_async: {e}")
return {"success": False, "error": str(e)}
class PetBotWebServer:
"""Standalone web server for PetBot"""
def __init__(self, database=None, port=8080, bot=None):
self.database = database or Database()
self.port = port
self.bot = bot
self.server = None
def run(self):
"""Start the web server"""
self.server = HTTPServer(('0.0.0.0', self.port), PetBotRequestHandler)
self.server.database = self.database
self.server.bot = self.bot
print(f'๐ Starting PetBot web server on http://0.0.0.0:{self.port}')
print(f'๐ก Accessible from WSL at: http://172.27.217.61:{self.port}')
print(f'๐ก Accessible from Windows at: http://localhost:{self.port}')
print('')
print('๐ Public access at: http://petz.rdx4.com/')
print('')
self.server.serve_forever()
def start_in_thread(self):
"""Start the web server in a background thread"""
import threading
self.thread = threading.Thread(target=self.run, daemon=True)
self.thread.start()
def stop(self):
"""Stop the web server"""
if self.server:
self.server.shutdown()
self.server.server_close()
def run_standalone():
"""Run the web server in standalone mode"""
import sys
port = 8080
if len(sys.argv) > 1:
try:
port = int(sys.argv[1])
except ValueError:
print('Usage: python webserver.py [port]')
sys.exit(1)
server = PetBotWebServer(port)
print('๐ PetBot Web Server')
print('=' * 50)
print(f'Port: {port}')
print('')
print('๐ Local URLs:')
print(f' http://localhost:{port}/ - Game Hub (local)')
print(f' http://localhost:{port}/help - Command Help (local)')
print(f' http://localhost:{port}/players - Player List (local)')
print(f' http://localhost:{port}/leaderboard - Leaderboard (local)')
print(f' http://localhost:{port}/locations - Locations (local)')
print('')
print('๐ Public URLs:')
print(' http://petz.rdx4.com/ - Game Hub')
print(' http://petz.rdx4.com/help - Command Help')
print(' http://petz.rdx4.com/players - Player List')
print(' http://petz.rdx4.com/leaderboard - Leaderboard')
print(' http://petz.rdx4.com/locations - Locations')
print('')
print('Press Ctrl+C to stop')
try:
server.run()
except KeyboardInterrupt:
print('\nโ Web server stopped')
if __name__ == '__main__':
run_standalone()