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.
102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Comprueba la configuración de Telegram del Pet Monitor.
|
|
|
|
Hace todo el trabajo pesado de la puesta en marcha:
|
|
1. Valida el token contra la API (getMe) y te dice de qué bot se trata.
|
|
2. Deduce el chat_id a partir del último mensaje que haya recibido el bot.
|
|
3. Manda un mensaje de prueba y una foto de prueba.
|
|
4. Te dice exactamente qué líneas añadir al .env.
|
|
|
|
Uso:
|
|
uv run scripts/telegram_check.py
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from pet_common import load_env, mask_secrets # noqa: E402
|
|
from pet_telegram import API, TelegramNotifier, descubrir_chat_id # noqa: E402
|
|
import requests # noqa: E402
|
|
|
|
|
|
def main():
|
|
env = load_env()
|
|
token = env.get("TELEGRAM_BOT_TOKEN")
|
|
if not token:
|
|
print("No hay TELEGRAM_BOT_TOKEN en el .env.\n")
|
|
print("Para crearlo:")
|
|
print(" 1. Abre Telegram y habla con @BotFather")
|
|
print(" 2. Envía /newbot, ponle nombre y usuario al bot")
|
|
print(" 3. Copia el token que te devuelve y añádelo a ~/pet_monitor/.env:")
|
|
print(" TELEGRAM_BOT_TOKEN=123456:ABC...")
|
|
print(" 4. Escríbele cualquier mensaje a tu bot (para que él te vea)")
|
|
print(" 5. Vuelve a lanzar este script")
|
|
return 1
|
|
|
|
# 1. ¿El token es válido y de quién es?
|
|
try:
|
|
respuesta = requests.get(API % (token, "getMe"), timeout=(5, 15))
|
|
respuesta.raise_for_status()
|
|
datos = respuesta.json()
|
|
if not datos.get("ok"):
|
|
raise RuntimeError(datos.get("description", "respuesta sin ok"))
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"[ERROR] El token no funciona: {mask_secrets(exc, [token])}")
|
|
return 1
|
|
|
|
bot = datos["result"]
|
|
print(f"[OK] Token válido. Bot: @{bot.get('username')} ({bot.get('first_name')})")
|
|
|
|
# 2. chat_id: del .env o deducido del último mensaje recibido
|
|
chat_id = env.get("TELEGRAM_CHAT_ID")
|
|
if chat_id:
|
|
print(f"[OK] TELEGRAM_CHAT_ID tomado del .env: {chat_id}")
|
|
else:
|
|
chat_id, detalle = descubrir_chat_id(token)
|
|
if chat_id is None:
|
|
print(f"[ERROR] No se pudo deducir el chat_id: {detalle}")
|
|
print(" Escríbele un mensaje a tu bot desde Telegram y vuelve a intentarlo.")
|
|
return 1
|
|
print(f"[OK] chat_id deducido: {chat_id} ({detalle})")
|
|
|
|
# 3. Mensaje y foto de prueba
|
|
notifier = TelegramNotifier(token, chat_id).start()
|
|
notifier.send_message(
|
|
"🐾 Pet Monitor: prueba de conexión correcta.\n"
|
|
f"Enviado el {time.strftime('%Y-%m-%d a las %H:%M:%S')} desde la Raspberry Pi.\n"
|
|
"Si ves esta foto, los avisos ya funcionan."
|
|
)
|
|
|
|
imagen = np.full((360, 640, 3), 40, dtype=np.uint8)
|
|
cv2.rectangle(imagen, (40, 40), (600, 320), (0, 200, 0), 3)
|
|
cv2.putText(imagen, "PRUEBA DE CAPTURA", (70, 160), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255, 255, 255), 2)
|
|
cv2.putText(imagen, time.strftime("%Y-%m-%d %H:%M:%S"), (70, 220), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (200, 200, 200), 2)
|
|
ok, buffer = cv2.imencode(".jpg", imagen)
|
|
if ok:
|
|
notifier.send_photo(
|
|
buffer.tobytes(),
|
|
"Captura de prueba (un aviso real llevará la foto de la cámara con las cajas).",
|
|
)
|
|
|
|
notifier.stop(timeout=30)
|
|
print(f"[OK] Enviados: {notifier.sent} | fallos: {notifier.errors}")
|
|
if notifier.last_error:
|
|
print(f"[WARN] Último error: {notifier.last_error}")
|
|
|
|
# 4. Qué falta por apuntar
|
|
if not env.get("TELEGRAM_CHAT_ID"):
|
|
print("\nAñade esta línea a ~/pet_monitor/.env para no deducirlo cada vez:")
|
|
print(f" TELEGRAM_CHAT_ID={chat_id}")
|
|
print("\nComprueba el móvil: deberías tener un mensaje y una foto del bot.")
|
|
return 0 if notifier.errors else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|