160 lines
4.8 KiB
Python
160 lines
4.8 KiB
Python
import asyncio
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
import config
|
|
|
|
MEDIA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "voz_media")
|
|
_server = None
|
|
|
|
|
|
def _python_firewall_ok():
|
|
"""Python del sistema (con regla de firewall Allow) para servir el mp3."""
|
|
if hasattr(sys, "base_prefix") and sys.base_prefix != sys.prefix:
|
|
ruta = os.path.join(sys.base_prefix, "python.exe")
|
|
if os.path.exists(ruta):
|
|
return ruta
|
|
return sys.executable
|
|
|
|
|
|
def _ip_lan():
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
try:
|
|
s.connect(("8.8.8.8", 80))
|
|
return s.getsockname()[0]
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def sintetizar(texto, nombre="anuncio.mp3"):
|
|
os.makedirs(MEDIA_DIR, exist_ok=True)
|
|
ruta = os.path.join(MEDIA_DIR, nombre)
|
|
try:
|
|
import edge_tts
|
|
|
|
async def _gen():
|
|
comunicador = edge_tts.Communicate(
|
|
texto,
|
|
config.VOZ_EDGE,
|
|
rate=config.VOZ_RATE,
|
|
pitch=config.VOZ_PITCH,
|
|
)
|
|
await comunicador.save(ruta)
|
|
|
|
asyncio.run(_gen())
|
|
return ruta
|
|
except Exception as e:
|
|
print(f"[Voz] Error sintetizando: {e}")
|
|
return None
|
|
|
|
|
|
def _arrancar_servidor():
|
|
global _server
|
|
if _server is not None and _server.poll() is None:
|
|
return
|
|
try:
|
|
py = _python_firewall_ok()
|
|
_server = subprocess.Popen(
|
|
[py, "-m", "http.server", str(config.GOOGLE_TV_PUERTO),
|
|
"--directory", MEDIA_DIR],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
)
|
|
time.sleep(1)
|
|
except Exception as e:
|
|
print(f"[Voz] No se pudo arrancar el servidor HTTP: {e}")
|
|
|
|
|
|
def _descubrir():
|
|
import pychromecast
|
|
|
|
chromecasts, browser = pychromecast.get_chromecasts(timeout=8)
|
|
try:
|
|
if config.GOOGLE_TV_NOMBRES:
|
|
return [c for c in chromecasts
|
|
if c.cast_info.friendly_name in config.GOOGLE_TV_NOMBRES]
|
|
return chromecasts
|
|
finally:
|
|
pychromecast.discovery.stop_discovery(browser)
|
|
|
|
|
|
def _url_audio(texto, nombre="anuncio.mp3"):
|
|
ruta = sintetizar(texto, nombre=nombre)
|
|
if ruta is None:
|
|
return None
|
|
_arrancar_servidor()
|
|
if _server is None:
|
|
return None
|
|
return f"http://{_ip_lan()}:{config.GOOGLE_TV_PUERTO}/{os.path.basename(ruta)}"
|
|
|
|
|
|
def _esperar_fin(cast, timeout=90):
|
|
mc = cast.media_controller
|
|
t0 = time.time()
|
|
while time.time() - t0 < timeout:
|
|
try:
|
|
st = mc.status
|
|
if st.player_state == "IDLE":
|
|
return True
|
|
except Exception:
|
|
pass
|
|
time.sleep(1)
|
|
return False
|
|
|
|
|
|
def anuncio_en_teles(texto):
|
|
"""Enciende la tele con el volumen configurado, reproduce el mensaje y la apaga."""
|
|
if not config.GOOGLE_TV_ACTIVO:
|
|
return
|
|
url = _url_audio(texto)
|
|
if url 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
|
|
|
|
if not objetivos:
|
|
print("[Voz] No hay teles Google detectadas")
|
|
return
|
|
|
|
for c in objetivos:
|
|
try:
|
|
c.wait(timeout=10)
|
|
c.set_volume(config.GOOGLE_TV_VOLUMEN / 100.0)
|
|
c.media_controller.play_media(url, "audio/mpeg", title=texto[:80])
|
|
c.media_controller.block_until_active(timeout=10)
|
|
print(f"[Voz] Anuncio en {c.cast_info.friendly_name} "
|
|
f"(vol {config.GOOGLE_TV_VOLUMEN})")
|
|
if _esperar_fin(c):
|
|
print(f"[Voz] Anuncio terminado en {c.cast_info.friendly_name}")
|
|
|
|
if config.ALEXA_APAGAR_TV_MONKEY:
|
|
print("[Voz] Apagando la tele via Alexa...")
|
|
from notificador import disparar_monkey_alexa
|
|
disparar_monkey_alexa(config.ALEXA_APAGAR_TV_MONKEY)
|
|
if config.GOOGLE_TV_APAGAR:
|
|
c.quit_app()
|
|
print(f"[Voz] {c.cast_info.friendly_name} apagada (standby)")
|
|
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}")
|
|
|
|
|
|
def reproducir_en_teles(texto):
|
|
anuncio_en_teles(texto)
|