jjminguez/biblioteca_conocimiento_laboratorio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

02-ago: [proyecto] vigilancia - notificacion personalizada por Telegram a destinatarios + aviso Discord (DM+TTS) a Carolina/Ines segun persona detectada

Browse Source
This commit is contained in:
minguezsanzjuanjose
2026-08-02 13:53:38 +02:00
parent fa44dcc943
commit 8215f21773
5 changed files with 126 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
vigilancia/.env.example
View File

@@ -30,3 +30,11 @@ SNAPSHOT_DIR=snapshots
# Zonas autorizadas (nombres de camara separados por coma).
# Si una camara esta aqui, sus presencias se marcan como "zona autorizada".
ZONAS_AUTORIZADAS=
# Destinatarios del mensaje personalizado cuando se identifica a una persona
# conocida (separados por coma). Cada uno debe haber hablado con el bot antes.
TELEGRAM_RECIPIENTES=
# Discord: bot + usuarios a los que se envia el aviso personalizado (TTS).
DISCORD_TOKEN=
DISCORD_IDS=

7
vigilancia/config.py
View File

@@ -35,3 +35,10 @@ SNAPSHOT_DIR = os.getenv("SNAPSHOT_DIR", "snapshots")
ZONAS_AUTORIZADAS = {
n.strip() for n in os.getenv("ZONAS_AUTORIZADAS", "").split(",") if n.strip()
}
RECIPIENTES = [
x.strip() for x in os.getenv("TELEGRAM_RECIPIENTES", "").split(",") if x.strip()
]
DISCORD_TOKEN = os.getenv("DISCORD_TOKEN", "").strip()
DISCORD_IDS = [
x.strip() for x in os.getenv("DISCORD_IDS", "").split(",") if x.strip()
]

55
vigilancia/discord_notif.py Normal file
View File

@@ -0,0 +1,55 @@
import config
import requests
from datetime import datetime
URL = "https://discord.com/api/v10"
_dm_cache = {}
def _headers():
return {"Authorization": f"Bot {config.DISCORD_TOKEN}"}
def _dm_channel(user_id):
if user_id in _dm_cache:
return _dm_cache[user_id]
r = requests.post(
f"{URL}/users/@me/channels",
headers=_headers(),
json={"recipient_id": user_id},
timeout=10,
)
r.raise_for_status()
ch = r.json()["id"]
_dm_cache[user_id] = ch
return ch
def enviar_discord(texto, user_id, tts=True):
if not config.DISCORD_TOKEN or not user_id:
return
try:
ch = _dm_channel(user_id)
r = requests.post(
f"{URL}/channels/{ch}/messages",
headers=_headers(),
json={"content": texto, "tts": tts},
timeout=10,
)
if r.status_code not in (200, 201):
print(f"[Discord] Error {r.status_code}: {r.text[:200]}")
except Exception as e:
print(f"[Discord] Excepción: {e}")
def notificar_discord_personales(resultados, cam_nombre):
conocidas = {r["nombre"] for r in resultados if "nombre" in r}
if not conocidas or not config.DISCORD_IDS:
return
ts = datetime.now().strftime("%H:%M:%S")
for uid in config.DISCORD_IDS:
for nombre in sorted(conocidas):
texto = (f"🚨 <b>{nombre}</b>, ¿qué estás haciendo? "
f"(vista en {cam_nombre} a las {ts})")
enviar_discord(texto, uid)

33
vigilancia/main.py
View File

@@ -7,12 +7,35 @@ import cv2
import config
from camara import CamaraRTSP
from detector_movimiento import DetectorMovimiento
from notificador import disparar_webhook_alexa, enviar_telegram
from notificador import (disparar_webhook_alexa, enviar_telegram,
enviar_texto_telegram)
USAR_YOLO = True
identificador = None
def notificar_personales(resultados, cam_nombre):
"""Mensaje personalizado por Telegram a cada destinatario segun la persona."""
conocidas = {r["nombre"] for r in resultados if "nombre" in r}
if not conocidas or not config.RECIPIENTES:
return
nombres = ", ".join(sorted(conocidas))
ts = datetime.now().strftime("%H:%M:%S")
for chat in config.RECIPIENTES:
for nombre in sorted(conocidas):
texto = (f"📢 <b>{nombre}</b>, ¿qué estás haciendo? "
f"(vista en {cam_nombre} a las {ts})")
enviar_texto_telegram(texto, chat_id=chat)
def notificar_discord(resultados, cam_nombre):
try:
from discord_notif import notificar_discord_personales
notificar_discord_personales(resultados, cam_nombre)
except Exception as e:
print(f"[Discord] Error: {e}")
def cargar_identificador():
global identificador
if identificador is None:
@@ -129,9 +152,9 @@ def main():
)
texto = f"🚨 Presencia en {cam.nombre} - {ts}"
resultados = []
if ident is not None:
try:
resultados = []
for _ in range(3):
f2 = cam.leer_frame()
if f2 is None:
@@ -155,6 +178,12 @@ def main():
if cv2_ok:
enviar_telegram(ruta, texto)
if resultados:
try:
notificar_personales(resultados, cam.nombre)
notificar_discord(resultados, cam.nombre)
except Exception as e:
print(f"[{cam.nombre}] Error notificando personales: {e}")
disparar_webhook_alexa(texto)
ultima_alerta[i] = ahora
print(texto)

30
vigilancia/notificador.py
View File

@@ -6,9 +6,10 @@ import requests
import config
def enviar_telegram(ruta_imagen: str, texto: str):
if not config.TELEGRAM_TOKEN or not config.TELEGRAM_CHAT_ID:
print("[Telegram] Falta TELEGRAM_TOKEN o TELEGRAM_CHAT_ID")
def enviar_telegram(ruta_imagen: str, texto: str, chat_id=None):
chat_id = chat_id or config.TELEGRAM_CHAT_ID
if not config.TELEGRAM_TOKEN or not chat_id:
print("[Telegram] Falta TELEGRAM_TOKEN o chat_id")
return
url = f"https://api.telegram.org/bot{config.TELEGRAM_TOKEN}/sendPhoto"
@@ -16,12 +17,31 @@ def enviar_telegram(ruta_imagen: str, texto: str):
with open(ruta_imagen, "rb") as f:
r = requests.post(
url,
data={"chat_id": config.TELEGRAM_CHAT_ID, "caption": texto},
data={"chat_id": chat_id, "caption": texto},
files={"photo": f},
timeout=10,
)
if not r.ok:
print(f"[Telegram] Error {r.status_code}: {r.text}")
print(f"[Telegram] Error {r.status_code}: {r.text[:200]}")
except Exception as e:
print(f"[Telegram] Excepción al enviar: {e}")
def enviar_texto_telegram(texto: str, chat_id=None):
chat_id = chat_id or config.TELEGRAM_CHAT_ID
if not config.TELEGRAM_TOKEN or not chat_id:
print("[Telegram] Falta TELEGRAM_TOKEN o chat_id")
return
url = f"https://api.telegram.org/bot{config.TELEGRAM_TOKEN}/sendMessage"
try:
r = requests.post(
url,
data={"chat_id": chat_id, "text": texto, "parse_mode": "HTML"},
timeout=10,
)
if not r.ok:
print(f"[Telegram] Error {r.status_code}: {r.text[:200]}")
except Exception as e:
print(f"[Telegram] Excepción al enviar: {e}")