app.py

import gradio as gr
import subprocess
import asyncio
import threading
from queue import Queue
import os
import shlex
import signal
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

BOT_TOKEN = ""

log_queue = Queue()
MAX_LOGS = 20000
terminal_logs = []
bot_logs = []
bot_app = None
bot_running = False
bot_thread = None
current_dir = os.getcwd()

# Track current running subprocess for Ctrl+C
current_proc = None

# --- Logging ---
def add_terminal_log(entry):
    terminal_logs.append(entry)
    if len(terminal_logs) > MAX_LOGS:
        terminal_logs.pop(0)

def add_bot_log(entry):
    bot_logs.append(entry)
    if len(bot_logs) > MAX_LOGS:
        bot_logs.pop(0)

# --- Command Execution ---
def execute_command(cmd):
    global current_dir, current_proc
    output = []
    add_terminal_log(f"$ {cmd}")

    if cmd.strip() == "ctrl+c":
        if current_proc:
            current_proc.send_signal(signal.SIGINT)
            msg = "[Info] Sent Ctrl+C to running process."
            add_terminal_log(msg)
            return msg
        else:
            msg = "[Info] No running process to interrupt."
            add_terminal_log(msg)
            return msg

    if cmd.strip().startswith("cd"):
        parts = shlex.split(cmd)
        if len(parts) > 1:
            new_path = os.path.join(current_dir, parts[1])
            if os.path.isdir(new_path):
                current_dir = os.path.abspath(new_path)
                msg = f"Changed directory to {current_dir}"
                output.append(msg)
                add_terminal_log(msg)
            else:
                msg = f"Directory not found: {new_path}"
                output.append(msg)
                add_terminal_log(msg)
        else:
            msg = "Usage: cd <path>"
            output.append(msg)
            add_terminal_log(msg)
    else:
        current_proc = subprocess.Popen(cmd, cwd=current_dir, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
        for line in iter(current_proc.stdout.readline, ''):
            line = line.strip()
            if line:
                output.append(line)
                add_terminal_log(line)
        current_proc.stdout.close()
        current_proc.wait()
        current_proc = None
    return "\n".join(output)

# --- Telegram Bot Handlers ---
async def bash_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    cmd = ' '.join(context.args)
    if not cmd:
        await update.message.reply_text("Usage: /bash <command>")
        return

    prefix = f"[Bot] $ {cmd}"
    add_bot_log(prefix)

    try:
        result = execute_command(cmd)
    except Exception as e:
        result = f"Error: {e}"

    add_bot_log(result)
    await update.message.reply_text(f"$ {cmd}\n{result}")

# --- Bot Start/Stop Logic ---
def start_bot():
    global bot_app, bot_running, bot_thread, BOT_TOKEN
    if bot_running or not BOT_TOKEN:
        add_bot_log("[Bot] Already running or token missing.")
        return

    async def run_bot():
        global bot_app
        bot_app = ApplicationBuilder().token(BOT_TOKEN).build()
        bot_app.add_handler(CommandHandler("bash", bash_command))
        add_bot_log("[Bot] Starting bot...")
        await bot_app.run_polling()

    def runner():
        asyncio.run(run_bot())

    bot_thread = threading.Thread(target=runner, daemon=True)
    bot_thread.start()
    bot_running = True
    add_bot_log("[Bot] Bot started.")

def stop_bot():
    global bot_app, bot_running
    if not bot_running:
        add_bot_log("[Bot] Bot not running.")
        return

    async def shutdown():
        await bot_app.shutdown()
        await bot_app.stop()
        add_bot_log("[Bot] Bot stopped.")

    asyncio.run(shutdown())
    bot_running = False

# --- UI Update Functions ---
def update_terminal_logs():
    return "\n".join(terminal_logs[-100:])

def update_bot_logs():
    return "\n".join(bot_logs[-100:])

def live_terminal(cmd):
    result = [f"$ {cmd}"]
    try:
        result.append(execute_command(cmd))
    except Exception as e:
        result.append(f"[Error] {e}")
    result.append(f"$ {current_dir} >")
    return "\n".join(result)

# --- Gradio UI ---
with gr.Blocks() as demo:
    with gr.Tab("💻 Terminal"):
        gr.Markdown("## 🖥️ Interactive Terminal")

        with gr.Row():
            terminal_output = gr.Textbox(label="📟 Terminal Output", lines=25, interactive=False)
        with gr.Row():
            cmd_input = gr.Textbox(placeholder="Enter shell command (use 'ctrl+c' to cancel)", label="Command Input")
            run_btn = gr.Button("▶️ Run Command")

        run_btn.click(fn=live_terminal, inputs=cmd_input, outputs=terminal_output)
        demo.load(fn=update_terminal_logs, outputs=terminal_output)

    with gr.Tab("🤖 Telegram Bot"):
        gr.Markdown("## 🤖 Telegram Bot Controls")

        with gr.Row():
            bot_output = gr.Textbox(label="🤖 Telegram Bot Logs", lines=25, interactive=False)
        with gr.Row():
            token_box = gr.Textbox(label="Bot Token", placeholder="Enter Telegram Bot Token")
        with gr.Row():
            start_btn = gr.Button("🚀 Start Telegram Bot")
            stop_btn = gr.Button("🛑 Stop Telegram Bot")

        def set_token_and_start(token):
            global BOT_TOKEN
            BOT_TOKEN = token
            start_bot()
            return update_bot_logs()

        start_btn.click(fn=set_token_and_start, inputs=token_box, outputs=bot_output)

        def handle_stop():
            stop_bot()
            return update_bot_logs()

        stop_btn.click(fn=handle_stop, outputs=bot_output)
        demo.load(fn=update_bot_logs, outputs=bot_output)

# --- Launch the App ---
demo.launch()