megaproxy/Petbot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Initial commit: Complete PetBot IRC Game

Browse source 
🎮 Features implemented:
- Pokemon-style pet collection and battles
- Multi-location exploration system
- Dynamic weather with background updates
- Achievement system with location unlocks
- Web dashboard for player stats
- Modular command system
- Async database with SQLite
- PM flood prevention
- Persistent player data

🌤️ Weather System:
- 6 weather types with spawn modifiers
- 30min-3hour dynamic durations
- Background task for automatic updates
- Location-specific weather patterns

🐛 Recent Bug Fixes:
- Database persistence on restart
- Player page SQLite row conversion
- Achievement count calculations
- Travel requirement messages
- Battle move color coding
- Locations page display

🔧 Generated with Claude Code
🤖 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
megaproxy 2025-07-13 23:57:39 +01:00
commit 47f160a295
31 changed files with 6235 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

191
src/bot.py Normal file
View file

@ -0,0 +1,191 @@
import asyncio
import irc.client_aio
import irc.connection
import threading
from typing import Dict, List, Optional
import json
import random
from datetime import datetime
from .database import Database
from .game_engine import GameEngine
class PetBot:
def __init__(self, config_path: str = "config/settings.json"):
self.config = self.load_config(config_path)
self.database = Database()
self.game_engine = GameEngine(self.database)
self.client = None
self.connection = None
def load_config(self, config_path: str) -> Dict:
try:
with open(config_path, 'r') as f:
return json.load(f)
except FileNotFoundError:
return {
"server": "irc.libera.chat",
"port": 6667,
"nickname": "PetBot",
"channel": "#petz",
"command_prefix": "!"
}
async def start(self):
await self.database.init_database()
await self.game_engine.load_game_data()
self.client = irc.client_aio.AioReactor()
try:
self.connection = await self.client.server().connect(
self.config["server"],
self.config["port"],
self.config["nickname"]
)
except irc.client.ServerConnectionError:
print(f"Could not connect to {self.config['server']}:{self.config['port']}")
return
self.connection.add_global_handler("welcome", self.on_connect)
self.connection.add_global_handler("pubmsg", self.on_message)
self.connection.add_global_handler("privmsg", self.on_private_message)
await self.client.process_forever(timeout=0.2)
def on_connect(self, connection, event):
connection.join(self.config["channel"])
print(f"Connected to {self.config['channel']}")
async def on_message(self, connection, event):
message = event.arguments[0]
nickname = event.source.nick
channel = event.target
if message.startswith(self.config["command_prefix"]):
await self.handle_command(connection, channel, nickname, message)
async def on_private_message(self, connection, event):
message = event.arguments[0]
nickname = event.source.nick
if message.startswith(self.config["command_prefix"]):
await self.handle_command(connection, nickname, nickname, message)
async def handle_command(self, connection, target, nickname, message):
command_parts = message[1:].split()
if not command_parts:
return
command = command_parts[0].lower()
args = command_parts[1:]
try:
if command == "help":
await self.cmd_help(connection, target, nickname)
elif command == "start":
await self.cmd_start(connection, target, nickname)
elif command == "catch":
await self.cmd_catch(connection, target, nickname, args)
elif command == "team":
await self.cmd_team(connection, target, nickname)
elif command == "wild":
await self.cmd_wild(connection, target, nickname, args)
elif command == "battle":
await self.cmd_battle(connection, target, nickname, args)
elif command == "stats":
await self.cmd_stats(connection, target, nickname, args)
else:
await self.send_message(connection, target, f"{nickname}: Unknown command. Use !help for available commands.")
except Exception as e:
await self.send_message(connection, target, f"{nickname}: Error processing command: {str(e)}")
async def send_message(self, connection, target, message):
connection.privmsg(target, message)
await asyncio.sleep(0.5)
async def cmd_help(self, connection, target, nickname):
help_text = [
"Available commands:",
"!start - Begin your pet journey",
"!catch <location> - Try to catch a pet in a location",
"!team - View your active pets",
"!wild <location> - See what pets are in an area",
"!battle <player> - Challenge another player",
"!stats [pet_name] - View pet or player stats"
]
for line in help_text:
await self.send_message(connection, target, line)
async def cmd_start(self, connection, target, nickname):
player = await self.database.get_player(nickname)
if player:
await self.send_message(connection, target, f"{nickname}: You already have an account! Use !team to see your pets.")
return
player_id = await self.database.create_player(nickname)
starter_pet = await self.game_engine.give_starter_pet(player_id)
await self.send_message(connection, target,
f"{nickname}: Welcome to the world of pets! You received a {starter_pet['species_name']}!")
async def cmd_catch(self, connection, target, nickname, args):
if not args:
await self.send_message(connection, target, f"{nickname}: Specify a location to catch pets in!")
return
location_name = " ".join(args)
player = await self.database.get_player(nickname)
if not player:
await self.send_message(connection, target, f"{nickname}: Use !start to begin your journey first!")
return
result = await self.game_engine.attempt_catch(player["id"], location_name)
await self.send_message(connection, target, f"{nickname}: {result}")
async def cmd_team(self, connection, target, nickname):
player = await self.database.get_player(nickname)
if not player:
await self.send_message(connection, target, f"{nickname}: Use !start to begin your journey first!")
return
pets = await self.database.get_player_pets(player["id"], active_only=True)
if not pets:
await self.send_message(connection, target, f"{nickname}: You don't have any active pets! Use !catch to find some.")
return
team_info = []
for pet in pets:
name = pet["nickname"] or pet["species_name"]
team_info.append(f"{name} (Lv.{pet['level']}) - {pet['hp']}/{pet['max_hp']} HP")
await self.send_message(connection, target, f"{nickname}'s team: " + " | ".join(team_info))
async def cmd_wild(self, connection, target, nickname, args):
if not args:
await self.send_message(connection, target, f"{nickname}: Specify a location to explore!")
return
location_name = " ".join(args)
wild_pets = await self.game_engine.get_location_spawns(location_name)
if wild_pets:
pet_list = ", ".join([pet["name"] for pet in wild_pets])
await self.send_message(connection, target, f"Wild pets in {location_name}: {pet_list}")
else:
await self.send_message(connection, target, f"{nickname}: No location found called '{location_name}'")
async def cmd_battle(self, connection, target, nickname, args):
await self.send_message(connection, target, f"{nickname}: Battle system coming soon!")
async def cmd_stats(self, connection, target, nickname, args):
player = await self.database.get_player(nickname)
if not player:
await self.send_message(connection, target, f"{nickname}: Use !start to begin your journey first!")
return
await self.send_message(connection, target,
f"{nickname}: Level {player['level']} | {player['experience']} XP | ${player['money']}")
if __name__ == "__main__":
bot = PetBot()
asyncio.run(bot.start())