111 lines
4.2 KiB
Python
111 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
import requests
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import config
|
|
|
|
# Segundos tras los que se borran las fotos del panorama en Telegram (default 10 min)
|
|
DELETE_AFTER = int(os.getenv("PANORAMA_DELETE_AFTER", "600"))
|
|
|
|
if not config.CAMARAS:
|
|
print("No hay cámaras configuradas")
|
|
sys.exit(1)
|
|
|
|
def programar_borrado(message_ids):
|
|
"""Borra los mensajes (fotos del panorama) tras DELETE_AFTER segundos.
|
|
Se lanza en un proceso separado para que sobreviva al cierre de este script."""
|
|
codigo = (
|
|
"import time, requests\n"
|
|
f"time.sleep({DELETE_AFTER})\n"
|
|
f"chat={config.TELEGRAM_CHAT_ID!r}\n"
|
|
f"ids={list(message_ids)!r}\n"
|
|
f"tok={config.TELEGRAM_TOKEN!r}\n"
|
|
"for mid in ids:\n"
|
|
" try:\n"
|
|
" requests.post('https://api.telegram.org/bot'+tok+'/deleteMessage',"
|
|
" data={'chat_id': chat, 'message_id': mid}, timeout=10)\n"
|
|
" print('borrado', mid)\n"
|
|
" except Exception as e:\n"
|
|
" print('err', mid, e)\n"
|
|
)
|
|
subprocess.Popen(["python3", "-c", codigo],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
start_new_session=True,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
|
|
|
|
def capturar_snapshot():
|
|
os.makedirs(config.SNAPSHOT_DIR, exist_ok=True)
|
|
resultados = []
|
|
for cam in config.CAMARAS:
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
ruta = os.path.join(config.SNAPSHOT_DIR, f"panorama_{cam['nombre']}_{ts}.jpg")
|
|
cmd = ["ffmpeg", "-i", cam["rtsp"], "-frames:v", "1", "-q:v", "2",
|
|
"-y", "-hide_banner", "-loglevel", "error", ruta]
|
|
try:
|
|
subprocess.run(cmd, timeout=15, check=True)
|
|
print(f"[{cam['nombre']}] Snapshot: {ruta}")
|
|
resultados.append((cam["nombre"], ruta))
|
|
except Exception as e:
|
|
print(f"[{cam['nombre']}] Error: {e}")
|
|
return resultados
|
|
|
|
def enviar_album(fotos):
|
|
if not fotos:
|
|
print("No hay fotos para enviar")
|
|
return
|
|
if len(fotos) == 1:
|
|
nombre, ruta = fotos[0]
|
|
url = f"https://api.telegram.org/bot{config.TELEGRAM_TOKEN}/sendPhoto"
|
|
try:
|
|
with open(ruta, "rb") as f:
|
|
r = requests.post(url,
|
|
data={"chat_id": config.TELEGRAM_CHAT_ID,
|
|
"caption": f"📷 {nombre} — {datetime.now().strftime('%H:%M:%S')}"},
|
|
files={"photo": f}, timeout=30)
|
|
print(f"Foto enviada: HTTP {r.status_code}")
|
|
if r.status_code == 200:
|
|
programar_borrado([r.json()["result"]["message_id"]])
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
else:
|
|
url = f"https://api.telegram.org/bot{config.TELEGRAM_TOKEN}/sendMediaGroup"
|
|
media = []
|
|
files_dict = {}
|
|
for i, (nombre, ruta) in enumerate(fotos):
|
|
media.append({
|
|
"type": "photo",
|
|
"media": f"attach://{i}",
|
|
"caption": f"📷 {nombre} — {datetime.now().strftime('%H:%M:%S')}"
|
|
})
|
|
files_dict[str(i)] = open(ruta, "rb")
|
|
try:
|
|
r = requests.post(url, data={"chat_id": config.TELEGRAM_CHAT_ID,
|
|
"media": json.dumps(media)}, files=files_dict, timeout=30)
|
|
print(f"Album enviado: HTTP {r.status_code}")
|
|
if r.status_code == 200:
|
|
ids = [m["message_id"] for m in r.json()["result"]]
|
|
print(f"Borrado programado dentro de {DELETE_AFTER}s para {len(ids)} foto(s)")
|
|
programar_borrado(ids)
|
|
else:
|
|
print(f"Error: {r.text[:300]}")
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
finally:
|
|
for f in files_dict.values():
|
|
f.close()
|
|
|
|
if __name__ == "__main__":
|
|
print("Capturando snapshots...")
|
|
fotos = capturar_snapshot()
|
|
if fotos:
|
|
print(f"Enviando {len(fotos)} foto(s) a IDD...")
|
|
enviar_album(fotos)
|
|
else:
|
|
print("No se capturaron fotos")
|