02-ago: [proyecto] vigilancia - bot Telegram de la Pi (estado/temp/tv/alarma) + tv_api modo simple + deploy en Raspberry Pi minipc

vigilancia/pi_bot.py

@@ -0,0 +1,174 @@
+#!/usr/bin/env python3
+import os
+import subprocess
+import threading
+import time
+import urllib.parse
+
+import requests
+
+BASE = os.path.dirname(os.path.abspath(__file__))
+LOG = os.path.join(BASE, "bot.log")
+TV_API = "http://127.0.0.1:8600/tv/mensaje"
+
+
+def log(linea):
+    try:
+        with open(LOG, "a") as f:
+            f.write(f"{time.strftime('%H:%M:%S')} {linea}\n")
+    except Exception:
+        pass
+
+
+def load_env():
+    env = {}
+    try:
+        for line in open(os.path.join(BASE, ".env")):
+            line = line.strip()
+            if line and not line.startswith("#") and "=" in line:
+                k, v = line.split("=", 1)
+                env[k.strip()] = v.strip()
+    except Exception:
+        pass
+    return env
+
+
+ENV = load_env()
+TOKEN = ENV.get("TELEGRAM_TOKEN", "")
+CHAT_ID = ENV.get("TELEGRAM_CHAT_ID", "")
+ALERTA_TEMP = float(ENV.get("ALERTA_TEMP", "80"))
+
+
+def sh(cmd):
+    try:
+        return subprocess.run(cmd, shell=True, capture_output=True,
+                              text=True, timeout=10).stdout.strip()
+    except Exception:
+        return "n/a"
+
+
+def estado():
+    temp = sh("vcgencmd measure_temp")
+    upt = sh("uptime -p")
+    load = open("/proc/loadavg").read().split()[:3]
+    free = sh("free -h | awk 'NR==2{print $2\" total, \"$7\" disponible\"}'")
+    disk = sh("df -h / | awk 'NR==2{print $3\" usados de \"$2\" (\"$5\")\"}'")
+    ips = sh("hostname -I")
+    temp_c = None
+    try:
+        temp_c = float(temp.replace("'C", "").replace("temp=", ""))
+    except Exception:
+        pass
+    msg = (
+        "📊 Estado de <b>minipc</b>:\n"
+        f"🌡️ Temperatura: <b>{temp}</b>\n"
+        f"🔌 Uptime: {upt}\n"
+        f"⚙️ Carga: {', '.join(load)}\n"
+        f"🧠 RAM: {free}\n"
+        f"💾 Disco /: {disk}\n"
+        f"🌐 IPs: {ips}"
+    )
+    return msg, temp_c
+
+
+def enviar(texto, chat_id):
+    try:
+        requests.post(
+            f"https://api.telegram.org/bot{TOKEN}/sendMessage",
+            data={"chat_id": chat_id, "text": texto, "parse_mode": "HTML"},
+            timeout=10,
+        )
+    except Exception as e:
+        print("Error enviando:", e)
+
+
+def mandar_tv(mensaje, simple):
+    try:
+        url = f"{TV_API}?texto={urllib.parse.quote(mensaje)}"
+        if simple:
+            url += "&simple=1"
+        r = requests.get(url, timeout=8)
+        return r.status_code == 200
+    except Exception as e:
+        log(f"tv error: {e}")
+        return False
+
+
+def procesar(texto, chat_id):
+    t = texto.lower().strip()
+    log(f"RECIBIDO chat={chat_id} texto={texto!r}")
+    if t in ("estado", "/estado", "status", "/status"):
+        msg, _ = estado()
+        enviar(msg, chat_id)
+    elif "temperatura" in t or "temp" in t:
+        temp = sh("vcgencmd measure_temp")
+        enviar(f"🌡️ Temperatura de la Pi: <b>{temp}</b>", chat_id)
+    elif t.startswith("tv ") or t.startswith("/tv "):
+        msj = texto.split(" ", 1)[1].strip()
+        if not msj:
+            enviar("Uso: <b>tv &lt;mensaje&gt;</b> (mensaje rapido a la tele)", chat_id)
+        else:
+            ok = mandar_tv(msj, simple=True)
+            enviar("📺 Mensaje enviado a la tele ✅" if ok
+                   else "❌ No se pudo enviar a la tele", chat_id)
+    elif t.startswith("alarma ") or t.startswith("/alarma "):
+        msj = texto.split(" ", 1)[1].strip()
+        if not msj:
+            enviar("Uso: <b>alarma &lt;mensaje&gt;</b> (secuencia completa: sirena x3)", chat_id)
+        else:
+            ok = mandar_tv(msj, simple=False)
+            enviar("🚨 Alarma lanzada en la tele ✅" if ok
+                   else "❌ No se pudo lanzar la alarma", chat_id)
+    elif t in ("ayuda", "/ayuda", "help", "/help"):
+        enviar("Comandos:\n"
+               "<b>estado</b> — estado de la Pi\n"
+               "<b>temperatura</b> — temperatura\n"
+               "<b>tv &lt;mensaje&gt;</b> — mensaje rapido a la tele\n"
+               "<b>alarma &lt;mensaje&gt;</b> — secuencia completa (sirena x3)",
+               chat_id)
+
+
+def monitor_temperatura():
+    while True:
+        time.sleep(300)
+        msg, temp_c = estado()
+        if temp_c is not None and temp_c > ALERTA_TEMP:
+            enviar(f"🚨 ALERTA: la Pi está a <b>{temp_c:.1f}°C</b> "
+                   f"(límite {ALERTA_TEMP}°C)\n{msg}", CHAT_ID)
+
+
+def main():
+    if not TOKEN:
+        print("Falta TELEGRAM_TOKEN en .env")
+        return
+    threading.Thread(target=monitor_temperatura, daemon=True).start()
+    offset = None
+    log("Bot iniciado")
+    print("Bot iniciado. Esperando mensajes...")
+    while True:
+        try:
+            params = {"timeout": 25}
+            if offset:
+                params["offset"] = offset
+            r = requests.get(
+                f"https://api.telegram.org/bot{TOKEN}/getUpdates",
+                params=params, timeout=35,
+            ).json()
+            for upd in r.get("result", []):
+                offset = upd["update_id"] + 1
+                log(f"UPDATE {upd.get('update_id')}")
+                msg = upd.get("message") or upd.get("channel_post")
+                if not msg:
+                    continue
+                texto = msg.get("text")
+                chat_id = msg["chat"]["id"]
+                if texto:
+                    procesar(texto, chat_id)
+        except Exception as e:
+            log(f"ERROR {e}")
+            print("Error:", e)
+            time.sleep(3)
+
+
+if __name__ == "__main__":
+    main()

vigilancia/tv_api.py

@@ -9,7 +9,7 @@ import sys
 sys.path.insert(0, sys_dir)
 
 import config
-from voz import anuncio_en_teles
+from voz import anuncio_en_teles, anuncio_simple
 
 
 class Handler(BaseHTTPRequestHandler):
@@ -19,9 +19,11 @@ class Handler(BaseHTTPRequestHandler):
         if not texto:
             self._resp(400, {"error": "falta parametro texto"})
             return
-        threading.Thread(target=anuncio_en_teles, args=(texto,), daemon=True).start()
+        simple = (q.get("simple") or ["0"])[0] == "1"
-        print(f"[TV-API] Mensaje lanzado: {texto[:80]}")
+        fn = anuncio_simple if simple else anuncio_en_teles
-        self._resp(200, {"ok": True, "mensaje": texto})
+        threading.Thread(target=fn, args=(texto,), daemon=True).start()
+        print(f"[TV-API] Mensaje lanzado ({'simple' if simple else 'alarma'}): {texto[:80]}")
+        self._resp(200, {"ok": True, "mensaje": texto, "modo": "simple" if simple else "alarma"})
 
     def _resp(self, code, obj):
         body = json.dumps(obj).encode("utf-8")

vigilancia/voz.py

@@ -320,3 +320,38 @@ def anuncio_en_teles(texto):
 
 def reproducir_en_teles(texto):
     anuncio_en_teles(texto)
+
+
+def anuncio_simple(texto):
+    """Mensaje rapido: enciende, reproduce el mensaje una vez y apaga."""
+    if not config.GOOGLE_TV_ACTIVO:
+        return
+    url_mensaje, tipo = _url_audio(texto)
+    if url_mensaje is None:
+        return
+    try:
+        import pychromecast
+
+        chromecasts, browser = pychromecast.get_chromecasts(timeout=8)
+        try:
+            if config.GOOGLE_TV_NOMBRES:
+                objetivos = [c for c in chromecasts
+                             if c.cast_info.friendly_name in config.GOOGLE_TV_NOMBRES]
+            else:
+                objetivos = chromecasts
+            for c in objetivos:
+                try:
+                    c.wait(timeout=10)
+                    c.set_volume(config.GOOGLE_TV_VOLUMEN / 100.0)
+                    _reproducir_y_esperar(
+                        c, url_mensaje, tipo, texto[:80],
+                        f"Mensaje simple en {c.cast_info.friendly_name}")
+                    if config.GOOGLE_TV_APAGAR_VIA_ADB:
+                        from tv_adb import apagar_tv
+                        apagar_tv()
+                except Exception as e:
+                    print(f"[Voz] Error en {c.cast_info.friendly_name}: {e}")
+        finally:
+            pychromecast.discovery.stop_discovery(browser)
+    except Exception as e:
+        print(f"[Voz] Error de casting: {e}")