admin/cyclop
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Refactor: Make reboot file paths configurable in TelegramHelper

Browse Source
This commit is contained in:
cyclop-bot
2025-06-02 16:35:52 -05:00
parent 75381f5f11
commit 53d1ec8923
1 changed files with 18 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram_helper.py
+18 -12
View File
@@ -8,17 +8,20 @@ from telegram.ext import Application, CommandHandler, MessageHandler, filters, C
from browse_command import browse_command, button_callback
class TelegramHelper:
# --- Constants for configurable paths and magic strings ---
REBOOT_CLAUDE_FILE = '.reboot_claude'
REBOOT_FILE = '.doreboot'
# --- Constants for magic strings (paths are now instance vars) ---
CLAUDE_REBOOT_TARGET = 'claude'
HTML_QUOTE_BLOCK_START = '<blockquote expandable><b>Thinking...</b>'
HTML_QUOTE_BLOCK_END = '</blockquote>'
DEFAULT_REBOOT_CLAUDE_FILE = '.reboot_claude' # Default value
DEFAULT_REBOOT_FILE = '.doreboot' # Default value
def __init__(self, bot):
def __init__(self, bot, reboot_claude_file_path: str | None = None, reboot_file_path: str | None = None): # MODIFIED
self.bot = bot
self.telegram_bot_token = os.getenv('TELEGRAM_BOT_TOKEN')
self.start_time = time.time()
# MODIFIED: Store configurable paths
self.reboot_claude_file = reboot_claude_file_path or self.DEFAULT_REBOOT_CLAUDE_FILE
self.reboot_file = reboot_file_path or self.DEFAULT_REBOOT_FILE
async def start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
await self.bot.start()
@@ -94,25 +97,28 @@ class TelegramHelper:
async def reboot(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
user_message = update.message.text.split()
if len(user_message) > 1 and user_message[1].lower() == self.CLAUDE_REBOOT_TARGET:
open(self.REBOOT_CLAUDE_FILE, 'w').close()
# MODIFIED: Use instance variable
open(self.reboot_claude_file, 'w').close()
if update:
await update.message.reply_text("Rebooting the bot...")
logging.info("Received reboot command. Exiting process...")
reboot_file_path = self.REBOOT_FILE
if not os.path.exists(reboot_file_path):
with open(reboot_file_path, 'w') as f:
# MODIFIED: Use instance variable
reboot_f_path = self.reboot_file
if not os.path.exists(reboot_f_path):
with open(reboot_f_path, 'w') as f:
f.write(str(update.effective_chat.id) if update else "")
sys.exit(0)
async def check_doreboot_file(self, application: Application):
reboot_file_path = self.REBOOT_FILE
if os.path.exists(reboot_file_path):
with open(reboot_file_path, 'r') as f:
# MODIFIED: Use instance variable
reboot_f_path = self.reboot_file
if os.path.exists(reboot_f_path):
with open(reboot_f_path, 'r') as f:
chat_id = f.read().strip()
if chat_id:
await application.bot.send_message(chat_id=chat_id, text="The application has finished initializing.")
os.remove(reboot_file_path)
os.remove(reboot_f_path)
async def browse(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
await browse_command(update, context, self.bot)