""" Trading System Plugin for MMORPG Adds player-to-player trading and marketplace functionality """ from src.interfaces.plugin_interfaces import IEconomyPlugin, PluginMetadata, PluginType from typing import Dict, List, Any, Optional import time import uuid class TradingSystemPlugin(IEconomyPlugin): """Plugin that adds comprehensive trading and marketplace features.""" def __init__(self): self._metadata = PluginMetadata( id="trading_system", name="Trading System", version="1.1.0", author="MMORPG Dev Team", description="Adds player trading, marketplace, and auction features", plugin_type=PluginType.SERVICE, dependencies=["enhanced_chat"], # Depends on chat for trade messages config={ "enable_direct_trading": True, "enable_marketplace": True, "trade_tax_rate": 0.05, "max_trade_distance": 100 } ) self._enabled = False self._active_trades = {} # Trade sessions self._marketplace_listings = {} # Marketplace items self._trade_history = [] # Sample items for trading self._tradeable_items = { "health_potion": {"name": "Health Potion", "base_value": 10, "stackable": True}, "mana_potion": {"name": "Mana Potion", "base_value": 15, "stackable": True}, "iron_sword": {"name": "Iron Sword", "base_value": 50, "stackable": False}, "wooden_shield": {"name": "Wooden Shield", "base_value": 30, "stackable": False}, "magic_scroll": {"name": "Magic Scroll", "base_value": 25, "stackable": True}, "gold_coin": {"name": "Gold Coin", "base_value": 1, "stackable": True} } @property def metadata(self) -> PluginMetadata: return self._metadata def initialize(self, context: Dict[str, Any]) -> bool: """Initialize the trading system plugin.""" try: self._config = self._metadata.config self._enabled = True print(f"🏪 Trading System Plugin initialized with {len(self._tradeable_items)} tradeable items") return True except Exception as e: print(f"🏪 Failed to initialize Trading System Plugin: {e}") return False def cleanup(self) -> None: """Clean up trading system resources.""" # Cancel all active trades for trade_id in list(self._active_trades.keys()): self.cancel_trade(trade_id) self._enabled = False print("💰 Trading System Plugin cleaned up") def is_enabled(self) -> bool: return self._enabled def initiate_trade(self, player1_id: str, player2_id: str, player1_pos: tuple, player2_pos: tuple) -> Dict[str, Any]: """Initiate a trade between two players.""" if not self._enabled or not self._config.get("enable_direct_trading", True): return {"success": False, "error": "Trading is disabled"} # Check distance max_distance = self._config.get("max_trade_distance", 100) distance = ((player1_pos[0] - player2_pos[0]) ** 2 + (player1_pos[1] - player2_pos[1]) ** 2) ** 0.5 if distance > max_distance: return {"success": False, "error": "Players are too far apart to trade"} # Create trade session trade_id = str(uuid.uuid4()) trade_session = { "id": trade_id, "player1": player1_id, "player2": player2_id, "player1_items": {}, "player2_items": {}, "player1_gold": 0, "player2_gold": 0, "player1_accepted": False, "player2_accepted": False, "status": "active", "created_at": time.time() } self._active_trades[trade_id] = trade_session return { "success": True, "trade_id": trade_id, "message": f"Trade initiated between players {player1_id} and {player2_id}" } def add_item_to_trade(self, trade_id: str, player_id: str, item_id: str, quantity: int = 1) -> Dict[str, Any]: """Add an item to a trade.""" if trade_id not in self._active_trades: return {"success": False, "error": "Trade not found"} trade = self._active_trades[trade_id] if trade["status"] != "active": return {"success": False, "error": "Trade is not active"} # Determine which player if player_id == trade["player1"]: items_dict = trade["player1_items"] elif player_id == trade["player2"]: items_dict = trade["player2_items"] else: return {"success": False, "error": "Player not part of this trade"} # Add item if item_id not in items_dict: items_dict[item_id] = 0 items_dict[item_id] += quantity # Reset acceptance status trade["player1_accepted"] = False trade["player2_accepted"] = False return { "success": True, "message": f"Added {quantity}x {self._tradeable_items.get(item_id, {}).get('name', item_id)} to trade" } def add_gold_to_trade(self, trade_id: str, player_id: str, amount: int) -> Dict[str, Any]: """Add gold to a trade.""" if trade_id not in self._active_trades: return {"success": False, "error": "Trade not found"} trade = self._active_trades[trade_id] if trade["status"] != "active": return {"success": False, "error": "Trade is not active"} # Determine which player and update gold if player_id == trade["player1"]: trade["player1_gold"] += amount elif player_id == trade["player2"]: trade["player2_gold"] += amount else: return {"success": False, "error": "Player not part of this trade"} # Reset acceptance status trade["player1_accepted"] = False trade["player2_accepted"] = False return {"success": True, "message": f"Added {amount} gold to trade"} def accept_trade(self, trade_id: str, player_id: str) -> Dict[str, Any]: """Accept a trade.""" if trade_id not in self._active_trades: return {"success": False, "error": "Trade not found"} trade = self._active_trades[trade_id] if trade["status"] != "active": return {"success": False, "error": "Trade is not active"} # Mark player as accepted if player_id == trade["player1"]: trade["player1_accepted"] = True elif player_id == trade["player2"]: trade["player2_accepted"] = True else: return {"success": False, "error": "Player not part of this trade"} # Check if both players accepted if trade["player1_accepted"] and trade["player2_accepted"]: return self._complete_trade(trade_id) return {"success": True, "message": "Trade accepted, waiting for other player"} def cancel_trade(self, trade_id: str) -> Dict[str, Any]: """Cancel a trade.""" if trade_id not in self._active_trades: return {"success": False, "error": "Trade not found"} trade = self._active_trades[trade_id] trade["status"] = "cancelled" del self._active_trades[trade_id] return {"success": True, "message": "Trade cancelled"} def get_trade_status(self, trade_id: str) -> Dict[str, Any]: """Get the status of a trade.""" if trade_id not in self._active_trades: return {"success": False, "error": "Trade not found"} trade = self._active_trades[trade_id] return { "success": True, "trade": { "id": trade["id"], "status": trade["status"], "player1": trade["player1"], "player2": trade["player2"], "player1_items": trade["player1_items"], "player2_items": trade["player2_items"], "player1_gold": trade["player1_gold"], "player2_gold": trade["player2_gold"], "player1_accepted": trade["player1_accepted"], "player2_accepted": trade["player2_accepted"] } } def create_marketplace_listing(self, player_id: str, item_id: str, quantity: int, price: int) -> Dict[str, Any]: """Create a marketplace listing.""" if not self._enabled or not self._config.get("enable_marketplace", True): return {"success": False, "error": "Marketplace is disabled"} listing_id = str(uuid.uuid4()) listing = { "id": listing_id, "seller_id": player_id, "item_id": item_id, "quantity": quantity, "price": price, "created_at": time.time(), "status": "active" } self._marketplace_listings[listing_id] = listing item_name = self._tradeable_items.get(item_id, {}).get("name", item_id) return { "success": True, "listing_id": listing_id, "message": f"Listed {quantity}x {item_name} for {price} gold" } def get_marketplace_listings(self, item_filter: Optional[str] = None) -> List[Dict[str, Any]]: """Get marketplace listings.""" if not self._enabled or not self._config.get("enable_marketplace", True): return [] listings = [] for listing in self._marketplace_listings.values(): if listing["status"] != "active": continue if item_filter and listing["item_id"] != item_filter: continue item_info = self._tradeable_items.get(listing["item_id"], {}) listing_info = { "id": listing["id"], "seller_id": listing["seller_id"], "item_id": listing["item_id"], "item_name": item_info.get("name", listing["item_id"]), "quantity": listing["quantity"], "price": listing["price"], "price_per_unit": listing["price"] / listing["quantity"], "created_at": listing["created_at"] } listings.append(listing_info) # Sort by price per unit listings.sort(key=lambda x: x["price_per_unit"]) return listings def purchase_from_marketplace(self, buyer_id: str, listing_id: str) -> Dict[str, Any]: """Purchase an item from the marketplace.""" if listing_id not in self._marketplace_listings: return {"success": False, "error": "Listing not found"} listing = self._marketplace_listings[listing_id] if listing["status"] != "active": return {"success": False, "error": "Listing is no longer active"} if listing["seller_id"] == buyer_id: return {"success": False, "error": "Cannot buy your own listing"} # Calculate tax tax_rate = self._config.get("trade_tax_rate", 0.05) tax_amount = int(listing["price"] * tax_rate) seller_receives = listing["price"] - tax_amount # Mark listing as sold listing["status"] = "sold" listing["buyer_id"] = buyer_id listing["sold_at"] = time.time() # Add to trade history trade_record = { "type": "marketplace", "seller_id": listing["seller_id"], "buyer_id": buyer_id, "item_id": listing["item_id"], "quantity": listing["quantity"], "price": listing["price"], "tax_amount": tax_amount, "timestamp": time.time() } self._trade_history.append(trade_record) item_name = self._tradeable_items.get(listing["item_id"], {}).get("name", listing["item_id"]) return { "success": True, "message": f"Purchased {listing['quantity']}x {item_name} for {listing['price']} gold", "seller_receives": seller_receives, "tax_paid": tax_amount } def get_tradeable_items(self) -> Dict[str, Dict[str, Any]]: """Get list of tradeable items.""" return self._tradeable_items.copy() def _complete_trade(self, trade_id: str) -> Dict[str, Any]: """Complete a trade between two players.""" trade = self._active_trades[trade_id] # Mark trade as completed trade["status"] = "completed" trade["completed_at"] = time.time() # Add to trade history trade_record = { "type": "direct_trade", "player1": trade["player1"], "player2": trade["player2"], "player1_items": trade["player1_items"], "player2_items": trade["player2_items"], "player1_gold": trade["player1_gold"], "player2_gold": trade["player2_gold"], "timestamp": time.time() } self._trade_history.append(trade_record) # Remove from active trades del self._active_trades[trade_id] return { "success": True, "message": "Trade completed successfully!", "trade_summary": trade_record } def get_status(self) -> Dict[str, Any]: """Get trading system status.""" return { "enabled": self._enabled, "active_trades": len(self._active_trades), "marketplace_listings": len(self._marketplace_listings), "total_items": len(self._tradeable_items), "trade_history_count": len(self._trade_history) } def shutdown(self) -> bool: """Shutdown the trading system.""" try: self.cleanup() return True except Exception as e: print(f"🏪 Error shutting down Trading System Plugin: {e}") return False # Plugin entry point def create_plugin(): """Factory function to create the plugin instance.""" return TradingSystemPlugin()