Điều khiển máy tính từ xa bằng Telegram

import os import platform import socket import time import logging import psutil import pyautogui from datetime import datetime from PIL import ImageGrab import subprocess import sys from telegram import Update from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters # Cấu hình logging logging.basicConfig( format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO ) logger = logging.getLogger(__name__) # Thay thế bằng token của bot của bạn từ BotFather TOKEN = "YOUR_TOKEN_HERE" # Thay thế bằng chat_id của bạn (để giới hạn quyền truy cập) ALLOWED_CHAT_ID = YOUR_CHAT_ID # Thay bằng chat_id của bạn # Đường dẫn để lưu ảnh chụp màn hình SCREENSHOT_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "screenshot.png") # Kiểm tra người dùng được phép def restricted(func): async def wrapped(update: Update, context: ContextTypes.DEFAULT_TYPE, *args, **kwargs): user_id = update.effective_chat.id if user_id != ALLOWED_CHAT_ID: logger.warning(f"Unauthorized access denied for {user_id}") await update.message.reply_text("Bạn không có quyền sử dụng bot này.") return return await func(update, context, *args, **kwargs) return wrapped # Hàm thông báo khi bot khởi động async def notify_startup(context: ContextTypes.DEFAULT_TYPE) -> None: """Thông báo khi bot khởi động.""" try: system_info = get_system_info() await context.bot.send_message(chat_id=ALLOWED_CHAT_ID, text=f"🤖 Bot đã khởi động!\n\n{system_info}") except Exception as e: logger.error(f"Lỗi khi gửi thông báo khởi động: {e}") # Lấy thông tin hệ thống def get_system_info(): hostname = socket.gethostname() ip_address = socket.gethostbyname(hostname) os_name = platform.system() + " " + platform.release() cpu_usage = psutil.cpu_percent() memory = psutil.virtual_memory() memory_usage = f"{memory.percent}% ({memory.used // (1024**2)} MB / {memory.total // (1024**2)} MB)" return (f"🖥️ Máy tính: {hostname}\n" f"🌐 IP: {ip_address}\n" f"💻 Hệ điều hành: {os_name}\n" f"⚙️ CPU: {cpu_usage}%\n" f"🧠 RAM: {memory_usage}") # Xử lý lệnh /start @restricted async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Gửi tin nhắn khi lệnh /start được đưa ra.""" await update.message.reply_text( "👋 Chào mừng đến với Bot điều khiển PC!\n\n" "Sử dụng /help để xem danh sách lệnh." ) # Xử lý lệnh /help @restricted async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Gửi tin nhắn khi lệnh /help được đưa ra.""" help_text = ( "📋 *Danh sách lệnh:*\n\n" "/status - Xem trạng thái hệ thống\n" "/screenshot - Chụp màn hình\n" "/shutdown - Tắt máy tính\n" "/restart - Khởi động lại máy tính\n" "/cmd <lệnh> - Thực thi lệnh CMD\n" "/type <văn bản> - Gõ văn bản\n" "/press <phím> - Nhấn phím (vd: enter, space, ctrl+alt+delete)\n" "/click <x> <y> - Click chuột tại tọa độ\n" "/rightclick <x> <y> - Click chuột phải tại tọa độ\n" "/doubleclick <x> <y> - Double click tại tọa độ\n" "/moveto <x> <y> - Di chuyển chuột đến tọa độ\n" "/open <tên ứng dụng> - Mở ứng dụng\n" "/kill <tên tiến trình> - Đóng ứng dụng\n" "/processes - Liệt kê các tiến trình đang chạy\n" ) await update.message.reply_text(help_text, parse_mode='Markdown') # Xử lý lệnh /status @restricted async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Gửi thông tin trạng thái hệ thống.""" system_info = get_system_info() current_time = datetime.now().strftime("%H:%M:%S %d/%m/%Y") await update.message.reply_text(f"⏰ Thời gian: {current_time}\n\n{system_info}") # Xử lý lệnh /screenshot @restricted async def screenshot_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Chụp và gửi ảnh màn hình.""" await update.message.reply_text("📸 Đang chụp màn hình...") try: # Chụp màn hình screenshot = ImageGrab.grab() screenshot.save(SCREENSHOT_PATH) # Gửi ảnh await update.message.reply_photo(photo=open(SCREENSHOT_PATH, 'rb'), caption="🖼️ Ảnh chụp màn hình") # Xóa file sau khi gửi os.remove(SCREENSHOT_PATH) except Exception as e: await update.message.reply_text(f"❌ Lỗi khi chụp màn hình: {e}") # Xử lý lệnh /shutdown @restricted async def shutdown_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Tắt máy tính.""" await update.message.reply_text("⚠️ Đang chuẩn bị tắt máy tính sau 10 giây...") try: if platform.system() == "Windows": os.system("shutdown /s /t 10 /c \"Tắt máy từ Telegram Bot\"") elif platform.system() == "Linux": os.system("shutdown -h 1") elif platform.system() == "Darwin": # macOS os.system("sudo shutdown -h +1") else: await update.message.reply_text("❌ Không hỗ trợ hệ điều hành này.") return await update.message.reply_text("✅ Lệnh tắt máy đã được thực thi.") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi tắt máy: {e}") # Xử lý lệnh /restart @restricted async def restart_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Khởi động lại máy tính.""" await update.message.reply_text("⚠️ Đang chuẩn bị khởi động lại máy tính sau 10 giây...") try: if platform.system() == "Windows": os.system("shutdown /r /t 10 /c \"Khởi động lại từ Telegram Bot\"") elif platform.system() == "Linux": os.system("shutdown -r 1") elif platform.system() == "Darwin": # macOS os.system("sudo shutdown -r +1") else: await update.message.reply_text("❌ Không hỗ trợ hệ điều hành này.") return await update.message.reply_text("✅ Lệnh khởi động lại đã được thực thi.") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi khởi động lại: {e}") # Xử lý lệnh /cmd @restricted async def cmd_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Thực thi lệnh CMD.""" if not context.args: await update.message.reply_text("❌ Vui lòng nhập lệnh. Ví dụ: /cmd ipconfig") return command = ' '.join(context.args) await update.message.reply_text(f"⚙️ Đang thực thi lệnh: `{command}`", parse_mode='Markdown') try: # Thực thi lệnh và lấy kết quả result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30) # Xử lý kết quả output = result.stdout if result.stdout else result.stderr if not output: output = "Lệnh đã được thực thi nhưng không có kết quả trả về." # Giới hạn kết quả nếu quá dài if len(output) > 4000: output = output[:4000] + "...\n(Kết quả đã bị cắt ngắn vì quá dài)" await update.message.reply_text(f"📄 Kết quả:\n```\n{output}\n```", parse_mode='Markdown') except subprocess.TimeoutExpired: await update.message.reply_text("❌ Lệnh thực thi quá thời gian (30 giây).") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi thực thi lệnh: {e}") # Xử lý lệnh /type @restricted async def type_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Gõ văn bản.""" if not context.args: await update.message.reply_text("❌ Vui lòng nhập văn bản cần gõ. Ví dụ: /type Hello World") return text = ' '.join(context.args) await update.message.reply_text(f"⌨️ Đang gõ văn bản: \"{text}\"") try: pyautogui.write(text) await update.message.reply_text("✅ Đã gõ văn bản thành công.") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi gõ văn bản: {e}") # Xử lý lệnh /press @restricted async def press_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Nhấn phím.""" if not context.args: await update.message.reply_text("❌ Vui lòng nhập phím cần nhấn. Ví dụ: /press enter") return key = ' '.join(context.args) await update.message.reply_text(f"⌨️ Đang nhấn phím: {key}") try: # Xử lý tổ hợp phím if '+' in key: keys = key.split('+') pyautogui.hotkey(*keys) else: pyautogui.press(key) await update.message.reply_text("✅ Đã nhấn phím thành công.") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi nhấn phím: {e}") # Xử lý lệnh /click @restricted async def click_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Click chuột tại tọa độ.""" if len(context.args) != 2: await update.message.reply_text("❌ Vui lòng nhập tọa độ x và y. Ví dụ: /click 500 500") return try: x = int(context.args[0]) y = int(context.args[1]) await update.message.reply_text(f"🖱️ Đang click tại tọa độ ({x}, {y})") pyautogui.click(x, y) await update.message.reply_text("✅ Đã click thành công.") except ValueError: await update.message.reply_text("❌ Tọa độ phải là số.") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi click chuột: {e}") # Xử lý lệnh /rightclick @restricted async def rightclick_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Click chuột phải tại tọa độ.""" if len(context.args) != 2: await update.message.reply_text("❌ Vui lòng nhập tọa độ x và y. Ví dụ: /rightclick 500 500") return try: x = int(context.args[0]) y = int(context.args[1]) await update.message.reply_text(f"🖱️ Đang click chuột phải tại tọa độ ({x}, {y})") pyautogui.rightClick(x, y) await update.message.reply_text("✅ Đã click chuột phải thành công.") except ValueError: await update.message.reply_text("❌ Tọa độ phải là số.") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi click chuột phải: {e}") # Xử lý lệnh /doubleclick @restricted async def doubleclick_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Double click tại tọa độ.""" if len(context.args) != 2: await update.message.reply_text("❌ Vui lòng nhập tọa độ x và y. Ví dụ: /doubleclick 500 500") return try: x = int(context.args[0]) y = int(context.args[1]) await update.message.reply_text(f"🖱️ Đang double click tại tọa độ ({x}, {y})") pyautogui.doubleClick(x, y) await update.message.reply_text("✅ Đã double click thành công.") except ValueError: await update.message.reply_text("❌ Tọa độ phải là số.") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi double click: {e}") # Xử lý lệnh /moveto @restricted async def moveto_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Di chuyển chuột đến tọa độ.""" if len(context.args) != 2: await update.message.reply_text("❌ Vui lòng nhập tọa độ x và y. Ví dụ: /moveto 500 500") return try: x = int(context.args[0]) y = int(context.args[1]) await update.message.reply_text(f"🖱️ Đang di chuyển chuột đến tọa độ ({x}, {y})") pyautogui.moveTo(x, y) await update.message.reply_text("✅ Đã di chuyển chuột thành công.") except ValueError: await update.message.reply_text("❌ Tọa độ phải là số.") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi di chuyển chuột: {e}") # Xử lý tin nhắn không phải lệnh async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Phản hồi tin nhắn không phải lệnh.""" if update.effective_chat.id != ALLOWED_CHAT_ID: return await update.message.reply_text( "❓ Tôi không hiểu lệnh này. Sử dụng /help để xem danh sách lệnh." ) # Xử lý lệnh /open @restricted async def open_app_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Mở một ứng dụng.""" if not context.args: await update.message.reply_text("❌ Vui lòng nhập tên ứng dụng. Ví dụ: /open notepad") return app_name = ' '.join(context.args) await update.message.reply_text(f"🚀 Đang mở ứng dụng: {app_name}") try: if platform.system() == "Windows": # Mở ứng dụng trên Windows subprocess.Popen(app_name, shell=True) elif platform.system() == "Linux": # Mở ứng dụng trên Linux subprocess.Popen([app_name], shell=True) elif platform.system() == "Darwin": # macOS # Mở ứng dụng trên macOS subprocess.Popen(["open", "-a", app_name]) else: await update.message.reply_text("❌ Không hỗ trợ hệ điều hành này.") return await update.message.reply_text("✅ Đã mở ứng dụng thành công.") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi mở ứng dụng: {e}") # Xử lý lệnh /kill @restricted async def kill_app_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Đóng một ứng dụng.""" if not context.args: await update.message.reply_text("❌ Vui lòng nhập tên tiến trình. Ví dụ: /kill notepad.exe") return process_name = ' '.join(context.args) await update.message.reply_text(f"🛑 Đang đóng ứng dụng: {process_name}") try: if platform.system() == "Windows": # Đóng ứng dụng trên Windows os.system(f"taskkill /f /im {process_name}") elif platform.system() == "Linux" or platform.system() == "Darwin": # Linux hoặc macOS # Đóng ứng dụng trên Linux/macOS os.system(f"pkill -f {process_name}") else: await update.message.reply_text("❌ Không hỗ trợ hệ điều hành này.") return await update.message.reply_text("✅ Đã đóng ứng dụng thành công.") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi đóng ứng dụng: {e}") # Xử lý lệnh /processes @restricted async def list_processes_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Liệt kê các tiến trình đang chạy.""" await update.message.reply_text("📋 Đang lấy danh sách tiến trình...") try: processes = [] for proc in psutil.process_iter(['pid', 'name']): try: processes.append(f"{proc.info['pid']}: {proc.info['name']}") except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): pass # Sắp xếp theo tên processes.sort(key=lambda x: x.split(': ')[1].lower()) # Giới hạn số lượng tiến trình hiển thị if len(processes) > 50: processes = processes[:50] processes.append("... (còn nhiều tiến trình khác)") process_list = "\n".join(processes) await update.message.reply_text(f"🔄 Các tiến trình đang chạy:\n\n{process_list}") except Exception as e: await update.message.reply_text(f"❌ Lỗi khi liệt kê tiến trình: {e}") def main() -> None: """Khởi động bot.""" # Tạo ứng dụng và truyền token của bot application = Application.builder().token(TOKEN).build() # Đăng ký các bộ xử lý lệnh application.add_handler(CommandHandler("start", start_command)) application.add_handler(CommandHandler("help", help_command)) application.add_handler(CommandHandler("status", status_command)) application.add_handler(CommandHandler("screenshot", screenshot_command)) application.add_handler(CommandHandler("shutdown", shutdown_command)) application.add_handler(CommandHandler("restart", restart_command)) application.add_handler(CommandHandler("cmd", cmd_command)) application.add_handler(CommandHandler("type", type_command)) application.add_handler(CommandHandler("press", press_command)) application.add_handler(CommandHandler("click", click_command)) application.add_handler(CommandHandler("rightclick", rightclick_command)) application.add_handler(CommandHandler("doubleclick", doubleclick_command)) application.add_handler(CommandHandler("moveto", moveto_command)) application.add_handler(CommandHandler("open", open_app_command)) application.add_handler(CommandHandler("kill", kill_app_command)) application.add_handler(CommandHandler("processes", list_processes_command)) # Đăng ký bộ xử lý tin nhắn không phải lệnh application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo)) # Thông báo khi bot khởi động (sử dụng phương pháp an toàn) try: # Bắt đầu polling application.run_polling(allowed_updates=Update.ALL_TYPES) # Gửi thông báo khởi động ngay sau khi bot bắt đầu if application.job_queue: application.job_queue.run_once(notify_startup, 1) else: # Nếu job_queue không khả dụng, gọi trực tiếp asyncio.run(notify_startup(ContextTypes.DEFAULT_TYPE(application))) except Exception as e: logger.error(f"Lỗi khi khởi động bot: {e}") if __name__ == "__main__": main()