322 lines
11 KiB
Python
322 lines
11 KiB
Python
"""
|
|
Bot de Telegram para controlar el PC remotamente.
|
|
|
|
Comandos:
|
|
- /apagar - Apaga PC completo (luces + regleta)
|
|
- /encender - Enciende PC (regleta + WoL)
|
|
- /estado - Verifica si el PC está encendido
|
|
- /menu - Muestra menú con botones
|
|
|
|
Uso:
|
|
python telegram_bot_pc.py
|
|
"""
|
|
import os
|
|
import sys
|
|
import asyncio
|
|
import requests
|
|
import subprocess
|
|
from dotenv import load_dotenv
|
|
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
|
|
from telegram.ext import Application, CommandHandler, CallbackQueryHandler, ContextTypes
|
|
|
|
# Cargar .env
|
|
dotenv_path = r"C:\Users\juanm\Documents\GitHub\biblioteca_negocio_prolongo\.env"
|
|
load_dotenv(dotenv_path)
|
|
|
|
# Configuración
|
|
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_PC_TOKEN")
|
|
TELEGRAM_USER_ID = os.getenv("TELEGRAM_USER_ID") # ID de Juanjo
|
|
HA_WEBHOOK_APAGAR = "http://185.187.169.109:8123/api/webhook/apagar-pc-juanjo"
|
|
HA_WEBHOOK_ENCENDER = "http://185.187.169.109:8123/api/webhook/encender-pc-juanjo"
|
|
PC_IP = "192.168.1.239" # IP del PC sobremesa
|
|
|
|
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Comando /start"""
|
|
await menu(update, context)
|
|
|
|
async def menu(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Muestra el menú principal"""
|
|
keyboard = [
|
|
[
|
|
InlineKeyboardButton("🔴 Apagar PC", callback_data='apagar'),
|
|
InlineKeyboardButton("🟢 Encender PC", callback_data='encender')
|
|
],
|
|
[
|
|
InlineKeyboardButton("📊 Estado PC", callback_data='estado')
|
|
]
|
|
]
|
|
reply_markup = InlineKeyboardMarkup(keyboard)
|
|
|
|
mensaje = (
|
|
"🖥️ <b>Control Remoto PC</b>\n\n"
|
|
"Elige una opción:"
|
|
)
|
|
|
|
if update.callback_query:
|
|
await update.callback_query.message.edit_text(
|
|
mensaje,
|
|
parse_mode='HTML',
|
|
reply_markup=reply_markup
|
|
)
|
|
else:
|
|
await update.message.reply_text(
|
|
mensaje,
|
|
parse_mode='HTML',
|
|
reply_markup=reply_markup
|
|
)
|
|
|
|
async def apagar_pc(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Apaga el PC con confirmación"""
|
|
query = update.callback_query if update.callback_query else None
|
|
|
|
# CONFIRMAR antes de apagar
|
|
keyboard = [
|
|
[
|
|
InlineKeyboardButton("✅ SÍ, apagar", callback_data='apagar_confirmar'),
|
|
InlineKeyboardButton("❌ Cancelar", callback_data='cancelar')
|
|
]
|
|
]
|
|
reply_markup = InlineKeyboardMarkup(keyboard)
|
|
|
|
mensaje = (
|
|
"⚠️ <b>CONFIRMAR APAGADO</b>\n\n"
|
|
"¿Seguro que quieres apagar el PC?\n\n"
|
|
"Se ejecutará:\n"
|
|
"🔹 Cerrar ventanas\n"
|
|
"🔹 Apagar luces (barras + cuartillo)\n"
|
|
"🔹 Apagar pantalla\n"
|
|
"🔹 Apagar regleta\n"
|
|
"🔹 Shutdown completo"
|
|
)
|
|
|
|
if query:
|
|
await query.answer()
|
|
await query.message.edit_text(mensaje, parse_mode='HTML', reply_markup=reply_markup)
|
|
else:
|
|
await update.message.reply_text(mensaje, parse_mode='HTML', reply_markup=reply_markup)
|
|
|
|
async def apagar_pc_confirmar(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Ejecuta el apagado tras confirmación"""
|
|
query = update.callback_query
|
|
await query.answer("🔴 Apagando PC...")
|
|
|
|
msg = await query.message.edit_text("🔴 Apagando PC...")
|
|
|
|
try:
|
|
response = requests.post(HA_WEBHOOK_APAGAR, timeout=10)
|
|
if response.status_code == 200:
|
|
mensaje = (
|
|
"✅ <b>Apagado iniciado</b>\n\n"
|
|
"🔹 Luces: apagando\n"
|
|
"🔹 Pantalla: apagando\n"
|
|
"🔹 PC: shutdown en 10s\n\n"
|
|
"⏳ Verificando estado en 30s..."
|
|
)
|
|
await msg.edit_text(mensaje, parse_mode='HTML')
|
|
|
|
# WATCHDOG: comprobar estado tras 30 segundos
|
|
await asyncio.sleep(30)
|
|
|
|
# Verificar si realmente se apagó
|
|
try:
|
|
result = subprocess.run(
|
|
["powershell", "-Command", f"Test-Connection -ComputerName {PC_IP} -Count 1 -Quiet"],
|
|
capture_output=True,
|
|
timeout=5,
|
|
text=True
|
|
)
|
|
online = "True" in result.stdout
|
|
|
|
if online:
|
|
mensaje_final = (
|
|
"⚠️ <b>APAGADO FALLIDO</b>\n\n"
|
|
f"El PC sigue respondiendo en {PC_IP}\n"
|
|
"Puede que el apagado se cancelara."
|
|
)
|
|
else:
|
|
mensaje_final = (
|
|
"✅ <b>PC APAGADO CONFIRMADO</b>\n\n"
|
|
f"🔹 IP {PC_IP}: sin respuesta\n"
|
|
"🔹 Estado: Offline"
|
|
)
|
|
except:
|
|
mensaje_final = (
|
|
"✅ <b>Apagado completado</b>\n\n"
|
|
"No se pudo verificar estado (timeout)"
|
|
)
|
|
|
|
await msg.edit_text(mensaje_final, parse_mode='HTML')
|
|
else:
|
|
mensaje = f"❌ Error al apagar: HTTP {response.status_code}"
|
|
await msg.edit_text(mensaje, parse_mode='HTML')
|
|
except Exception as e:
|
|
mensaje = f"❌ Error: {e}"
|
|
await msg.edit_text(mensaje, parse_mode='HTML')
|
|
|
|
async def encender_pc(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Enciende el PC y verifica que arrancó"""
|
|
query = update.callback_query if update.callback_query else None
|
|
|
|
if query:
|
|
await query.answer("🟢 Encendiendo PC...")
|
|
msg = await query.message.reply_text("🟢 Encendiendo PC...")
|
|
else:
|
|
msg = await update.message.reply_text("🟢 Encendiendo PC...")
|
|
|
|
try:
|
|
response = requests.post(HA_WEBHOOK_ENCENDER, timeout=10)
|
|
if response.status_code == 200:
|
|
mensaje = (
|
|
"✅ <b>Proceso de encendido iniciado</b>\n\n"
|
|
"🔹 Regleta: encendiendo\n"
|
|
"🔹 Esperando 60 segundos...\n"
|
|
"🔹 Wake-on-LAN: enviando\n\n"
|
|
"⏳ Verificando en 90 segundos..."
|
|
)
|
|
await msg.edit_text(mensaje, parse_mode='HTML')
|
|
|
|
# WATCHDOG: esperar 90s y verificar si arrancó
|
|
await asyncio.sleep(90)
|
|
|
|
# Verificar si el PC arrancó
|
|
intentos = 0
|
|
max_intentos = 3
|
|
online = False
|
|
|
|
while intentos < max_intentos and not online:
|
|
try:
|
|
result = subprocess.run(
|
|
["powershell", "-Command", f"Test-Connection -ComputerName {PC_IP} -Count 2 -Quiet"],
|
|
capture_output=True,
|
|
timeout=10,
|
|
text=True
|
|
)
|
|
online = "True" in result.stdout
|
|
if not online:
|
|
intentos += 1
|
|
if intentos < max_intentos:
|
|
await asyncio.sleep(15)
|
|
except:
|
|
intentos += 1
|
|
if intentos < max_intentos:
|
|
await asyncio.sleep(15)
|
|
|
|
if online:
|
|
mensaje_final = (
|
|
"✅ <b>PC ENCENDIDO CORRECTAMENTE</b>\n\n"
|
|
f"🔹 IP {PC_IP}: responde\n"
|
|
"🔹 Estado: Online\n"
|
|
"🔹 Tiempo: ~{} segundos".format(90 + intentos * 15)
|
|
)
|
|
else:
|
|
mensaje_final = (
|
|
"❌ <b>ENCENDIDO FALLIDO</b>\n\n"
|
|
f"El PC no responde en {PC_IP}\n"
|
|
"Verifica:\n"
|
|
"🔹 Regleta encendida\n"
|
|
"🔹 PC físicamente apagado\n"
|
|
"🔹 Wake-on-LAN habilitado en BIOS"
|
|
)
|
|
|
|
await msg.edit_text(mensaje_final, parse_mode='HTML')
|
|
else:
|
|
mensaje = f"❌ Error al encender: HTTP {response.status_code}"
|
|
await msg.edit_text(mensaje, parse_mode='HTML')
|
|
except Exception as e:
|
|
mensaje = f"❌ Error: {e}"
|
|
await msg.edit_text(mensaje, parse_mode='HTML')
|
|
|
|
async def estado_pc(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Verifica si el PC está encendido"""
|
|
query = update.callback_query if update.callback_query else None
|
|
|
|
if query:
|
|
await query.answer("📊 Verificando...")
|
|
msg = await query.message.reply_text("📊 Verificando estado...")
|
|
else:
|
|
msg = await update.message.reply_text("📊 Verificando estado...")
|
|
|
|
# Ping al PC usando PowerShell (más confiable en Windows)
|
|
try:
|
|
result = subprocess.run(
|
|
["powershell", "-Command", f"Test-Connection -ComputerName {PC_IP} -Count 1 -Quiet"],
|
|
capture_output=True,
|
|
timeout=5,
|
|
text=True
|
|
)
|
|
|
|
online = "True" in result.stdout
|
|
|
|
if online:
|
|
mensaje = (
|
|
"✅ <b>PC ENCENDIDO</b>\n\n"
|
|
f"🔹 IP: {PC_IP}\n"
|
|
"🔹 Responde a ping\n"
|
|
"🔹 Estado: Online"
|
|
)
|
|
else:
|
|
mensaje = (
|
|
"🔴 <b>PC APAGADO</b>\n\n"
|
|
f"🔹 IP: {PC_IP}\n"
|
|
"🔹 No responde a ping\n"
|
|
"🔹 Estado: Offline"
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
mensaje = (
|
|
"⚠️ <b>TIMEOUT</b>\n\n"
|
|
f"🔹 IP: {PC_IP}\n"
|
|
"🔹 No responde (>5s)\n"
|
|
"🔹 Estado: Probablemente apagado"
|
|
)
|
|
except Exception as e:
|
|
mensaje = f"❌ Error al verificar estado: {e}"
|
|
|
|
await msg.edit_text(mensaje, parse_mode='HTML')
|
|
|
|
async def cancelar(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Cancela una operación y vuelve al menú"""
|
|
query = update.callback_query
|
|
await query.answer("Cancelado")
|
|
await menu(update, context)
|
|
|
|
async def button_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Maneja los botones del menú"""
|
|
query = update.callback_query
|
|
|
|
if query.data == 'apagar':
|
|
await apagar_pc(update, context)
|
|
elif query.data == 'apagar_confirmar':
|
|
await apagar_pc_confirmar(update, context)
|
|
elif query.data == 'encender':
|
|
await encender_pc(update, context)
|
|
elif query.data == 'estado':
|
|
await estado_pc(update, context)
|
|
elif query.data == 'cancelar':
|
|
await cancelar(update, context)
|
|
|
|
def main():
|
|
"""Arranca el bot"""
|
|
if not TELEGRAM_BOT_TOKEN:
|
|
print("ERROR: Falta TELEGRAM_BOT_PC_TOKEN en .env")
|
|
return
|
|
|
|
print("Iniciando bot de Telegram para control de PC...")
|
|
|
|
app = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
|
|
|
|
# Comandos
|
|
app.add_handler(CommandHandler("start", start))
|
|
app.add_handler(CommandHandler("menu", menu))
|
|
app.add_handler(CommandHandler("apagar", apagar_pc))
|
|
app.add_handler(CommandHandler("encender", encender_pc))
|
|
app.add_handler(CommandHandler("estado", estado_pc))
|
|
|
|
# Botones
|
|
app.add_handler(CallbackQueryHandler(button_handler))
|
|
|
|
print("Bot iniciado. Presiona Ctrl+C para detener.")
|
|
app.run_polling()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|