""" Read2Burn Mailbox Add-on - Self-destructing secure messaging system. """ import time import random import string from typing import Dict, List, Tuple from dataclasses import dataclass from abc import ABC, abstractmethod import gradio as gr from ..interfaces.npc_addon import NPCAddon @dataclass class Read2BurnMessage: """Data class for Read2Burn messages.""" id: str creator_id: str content: str created_at: float expires_at: float reads_left: int burned: bool = False class IRead2BurnService(ABC): """Interface for Read2Burn service operations.""" @abstractmethod def create_message(self, creator_id: str, content: str) -> str: """Create a new self-destructing message.""" pass @abstractmethod def read_message(self, reader_id: str, message_id: str) -> str: """Read and potentially burn a message.""" pass @abstractmethod def list_player_messages(self, player_id: str) -> str: """List messages created by a player.""" pass class Read2BurnService(IRead2BurnService, NPCAddon): """Service for managing Read2Burn secure messaging.""" def __init__(self): super().__init__() self.messages: Dict[str, Read2BurnMessage] = {} self.access_log: List[Dict] = [] @property def addon_id(self) -> str: """Unique identifier for this add-on""" return "read2burn_mailbox" @property def addon_name(self) -> str: """Display name for this add-on""" return "Read2Burn Secure Mailbox" def get_interface(self) -> gr.Component: """Return Gradio interface for this add-on""" # This will be implemented later for the UI return gr.Markdown("šŸ“§ **Read2Burn Secure Mailbox**\n\nUse private messages to the read2burn NPC to manage secure messages.") def generate_message_id(self) -> str: """Generate a unique message ID.""" return ''.join(random.choices(string.ascii_uppercase + string.digits, k=8)) def create_message(self, creator_id: str, content: str) -> str: """Create a new self-destructing message.""" message_id = self.generate_message_id() message = Read2BurnMessage( id=message_id, creator_id=creator_id, content=content, # In production, encrypt this created_at=time.time(), expires_at=time.time() + (24 * 3600), # 24 hours reads_left=1, burned=False ) self.messages[message_id] = message self.access_log.append({ 'action': 'create', 'message_id': message_id, 'player_id': creator_id, 'timestamp': time.time() }) return f"āœ… **Message Created Successfully!**\n\nšŸ“ **Message ID:** `{message_id}`\nšŸ”— Share this ID with the recipient\nā° Expires in 24 hours\nšŸ”„ Burns after 1 read" def read_message(self, reader_id: str, message_id: str) -> str: """Read and burn a message.""" if message_id not in self.messages: return "āŒ Message not found or already burned" message = self.messages[message_id] # Check expiry if time.time() > message.expires_at: del self.messages[message_id] return "āŒ Message expired and has been burned" # Check if already burned if message.burned or message.reads_left <= 0: del self.messages[message_id] return "āŒ Message has already been burned" # Read the message content = message.content message.reads_left -= 1 self.access_log.append({ 'action': 'read', 'message_id': message_id, 'player_id': reader_id, 'timestamp': time.time() }) # Burn the message after reading if message.reads_left <= 0: message.burned = True del self.messages[message_id] return f"šŸ”„ **Message Self-Destructed After Reading**\n\nšŸ“– **Content:** {content}\n\nšŸ’Ø This message has been permanently destroyed." return f"šŸ“– **Message Content:** {content}\n\nāš ļø Reads remaining: {message.reads_left}" def list_player_messages(self, player_id: str) -> str: """List messages created by a player.""" player_messages = [msg for msg in self.messages.values() if msg.creator_id == player_id] if not player_messages: return "šŸ“Ŗ No messages found. Create one with: `create Your message here`" result = "šŸ“‹ **Your Created Messages:**\n\n" for msg in player_messages: status = "šŸ”„ Burned" if msg.burned else f"āœ… Active ({msg.reads_left} reads left)" created_time = time.strftime("%Y-%m-%d %H:%M", time.localtime(msg.created_at)) expires_time = time.strftime("%Y-%m-%d %H:%M", time.localtime(msg.expires_at)) result += f"**ID:** `{msg.id}`\n" result += f"**Status:** {status}\n" result += f"**Created:** {created_time}\n" result += f"**Expires:** {expires_time}\n" result += f"**Preview:** {msg.content[:50]}{'...' if len(msg.content) > 50 else ''}\n\n" return result def handle_command(self, player_id: str, command: str) -> str: """Handle Read2Burn mailbox commands.""" parts = command.strip().split(' ', 1) cmd = parts[0].lower() if cmd == "create" and len(parts) > 1: return self.create_message(player_id, parts[1]) elif cmd == "read" and len(parts) > 1: return self.read_message(player_id, parts[1]) elif cmd == "list": return self.list_player_messages(player_id) elif cmd == "help": return """šŸ“š **Read2Burn Mailbox Commands:** **create** `Your secret message here` - Create new message **read** `MESSAGE_ID` - Read message (destroys it!) **list** - Show your created messages **help** - Show this help šŸ”„ **Features:** • Messages self-destruct after reading • 24-hour automatic expiration • Secure delivery system • Anonymous messaging support""" else: return "ā“ Invalid command. Try: `create `, `read `, `list`, or `help`" def get_player_message_history(self, player_id: str) -> List[List[str]]: """Get message history for display in a dataframe.""" player_messages = [msg for msg in self.messages.values() if msg.creator_id == player_id] history = [] for msg in player_messages: status = "šŸ”„ Burned" if msg.burned else "āœ… Active" created_time = time.strftime("%H:%M", time.localtime(msg.created_at)) reads_left = str(msg.reads_left) if not msg.burned else "0" history.append([msg.id, created_time, status, reads_left]) return history def cleanup_expired_messages(self): """Clean up expired messages.""" current_time = time.time() expired_ids = [ msg_id for msg_id, msg in self.messages.items() if current_time > msg.expires_at ] for msg_id in expired_ids: del self.messages[msg_id] if expired_ids: print(f"[Read2Burn] Cleaned up {len(expired_ids)} expired messages") # Global Read2Burn service instance read2burn_service = Read2BurnService()