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 - Try to catch a pet in a location", "!team - View your active pets", "!wild - See what pets are in an area", "!battle - 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())