abraham
762699f41e
- pet_telegram.py: cliente de Telegram con cola y hilo de envio propio, para que la red nunca bloquee la vision ni la escucha de audio. Reintentos con retroceso exponencial, respeto del retry_after ante limites de frecuencia y descarte de los avisos mas viejos si la cola se llena. - pet_alerts.py: punto unico de reparto de avisos. Toma el fotograma mas reciente, dibuja la caja de la mascota detectada y la marca de tiempo, guarda la captura en capturas/ y la manda a Telegram. Anti-repeticion de 10 s en consola y de ALERT_COOLDOWN (60 s) para capturas y Telegram. - scripts/telegram_check.py: valida el token, deduce el chat_id del ultimo mensaje recibido por el bot y manda mensaje y foto de prueba. - pet_speed.py: integra el reparto de avisos y anade capturas y estado de Telegram a las estadisticas. - Readme.md: seccion de Telegram y estado de la Fase B.
92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
"""Utilidades compartidas por los módulos del Pet Monitor."""
|
|
|
|
import os
|
|
import urllib.parse
|
|
|
|
ENV_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
|
|
|
|
# Claves que se pueden sobreescribir desde el entorno real (systemd, pruebas
|
|
# puntuales). Las variables de entorno tienen prioridad sobre el fichero .env.
|
|
ENV_KEYS = (
|
|
"CAMERA_IP",
|
|
"CAMERA_USER",
|
|
"CAMERA_PW",
|
|
"CAMERA_STREAM",
|
|
"CAMERA_RTSP_URL",
|
|
"PET_AUDIO",
|
|
"BARK_THRESHOLD",
|
|
"BARK_COOLDOWN",
|
|
"TELEGRAM_BOT_TOKEN",
|
|
"TELEGRAM_CHAT_ID",
|
|
"ALERT_COOLDOWN",
|
|
"SNAPSHOTS",
|
|
"SNAPSHOT_DIR",
|
|
)
|
|
|
|
|
|
def load_env(path=ENV_PATH):
|
|
"""Lee un .env sencillo (CLAVE=valor) sin dependencias externas."""
|
|
env = {}
|
|
if os.path.exists(path):
|
|
with open(path) as fh:
|
|
for line in fh:
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
env[key.strip()] = value.strip().strip('"').strip("'")
|
|
for key in ENV_KEYS:
|
|
if os.environ.get(key):
|
|
env[key] = os.environ[key]
|
|
return env
|
|
|
|
|
|
def build_rtsp_url(env, stream):
|
|
"""Construye la URL RTSP de la cámara, escapando usuario y contraseña."""
|
|
if env.get("CAMERA_RTSP_URL"):
|
|
return env["CAMERA_RTSP_URL"]
|
|
ip = env.get("CAMERA_IP")
|
|
user = env.get("CAMERA_USER")
|
|
password = env.get("CAMERA_PW")
|
|
missing = [k for k, v in (("CAMERA_IP", ip), ("CAMERA_USER", user), ("CAMERA_PW", password)) if not v]
|
|
if missing:
|
|
raise SystemExit(f"[ERROR] Faltan variables en {ENV_PATH}: {', '.join(missing)}")
|
|
return "rtsp://%s:%s@%s:554/%s" % (
|
|
urllib.parse.quote(user, safe=""),
|
|
urllib.parse.quote(password, safe=""),
|
|
ip,
|
|
stream,
|
|
)
|
|
|
|
|
|
def mask_url(url):
|
|
"""Oculta las credenciales de la URL para poder mostrarla por consola."""
|
|
parts = urllib.parse.urlsplit(url)
|
|
if not parts.hostname:
|
|
return "***"
|
|
port = f":{parts.port}" if parts.port else ""
|
|
return f"{parts.scheme}://***@{parts.hostname}{port}{parts.path}"
|
|
|
|
|
|
def secrets_of(url):
|
|
"""Devuelve los trozos de una URL que nunca deben aparecer en un log."""
|
|
parts = urllib.parse.urlsplit(url)
|
|
valores = [parts.password, parts.username]
|
|
if parts.password:
|
|
valores.append(urllib.parse.quote(parts.password, safe=""))
|
|
return [v for v in valores if v]
|
|
|
|
|
|
def mask_secrets(text, secrets):
|
|
"""
|
|
Tapa las credenciales dentro de un texto.
|
|
|
|
PyAV incluye la URL completa (con contraseña) en sus excepciones, así que
|
|
todo lo que se imprima desde el hilo de audio tiene que pasar por aquí: si
|
|
no, un fallo de red dejaría la contraseña escrita en journalctl.
|
|
"""
|
|
text = str(text)
|
|
for secret in secrets:
|
|
if secret:
|
|
text = text.replace(secret, "***")
|
|
return text
|