111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
aviso_voz.py — Genera y envia una nota de voz corta por Telegram al terminar
|
|
una respuesta de OpenCode, en vez del pitido/tono de notificacion.
|
|
|
|
Usa edge-tts (gratis/ilimitado, Microsoft) con la voz preferida del usuario
|
|
es-MX-DaliaNeural. Telegram exige las notas de voz en OGG/Opus, asi que el
|
|
mp3 generado por edge-tts se convierte con ffmpeg la primera vez y se cachea.
|
|
|
|
Configuracion (variables del .env del bot):
|
|
AVISO_VOZ=1 -> habilita el aviso de voz al terminar
|
|
AVISO_VOZ_FRASE=Listo. -> frase a sintetizar (opcional)
|
|
AVISO_VOZ_MP3=<ruta> -> usar un mp3 propio (voz ElevenLabs)
|
|
en vez de sintetizar (opcional)
|
|
El OGG cacheado vive en <dir>/sounds/aviso_fin.ogg (o junto al mp3 indicado).
|
|
|
|
Uso:
|
|
python3 aviso_voz.py <chat_id> -> envia la nota de voz al chat
|
|
"""
|
|
import asyncio
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
import requests
|
|
|
|
BASE = os.path.dirname(os.path.abspath(__file__))
|
|
SND = os.path.join(BASE, "sounds")
|
|
ENV = {}
|
|
try:
|
|
for linea in open(os.path.join(BASE, ".env"), encoding="utf-8"):
|
|
linea = linea.strip()
|
|
if linea and not linea.startswith("#") and "=" in linea:
|
|
k, v = linea.split("=", 1)
|
|
ENV[k.strip()] = v.strip()
|
|
except Exception:
|
|
pass
|
|
|
|
TOKEN = ENV.get("TELEGRAM_BOT_TOKEN", "")
|
|
API = f"https://api.telegram.org/bot{TOKEN}"
|
|
FRASE = ENV.get("AVISO_VOZ_FRASE", "Listo.")
|
|
MP3_SOURCE = ENV.get("AVISO_VOZ_MP3", "")
|
|
|
|
|
|
def _voz_dalia(texto, salida_mp3):
|
|
"""Sintetiza `texto` con edge-tts voz es-MX-DaliaNeural a un mp3."""
|
|
import edge_tts
|
|
asyncio.run(edge_tts.Communicate(
|
|
texto, "es-MX-DaliaNeural", rate="-8%").save(salida_mp3))
|
|
|
|
|
|
def asegurar_ogg():
|
|
"""Devuelve la ruta del OGG/Opus de aviso, generandolo/cacheandolo."""
|
|
try:
|
|
os.makedirs(SND, exist_ok=True)
|
|
except Exception:
|
|
pass
|
|
# Si hay mp3 propio de voz, el cache es su mismo nombre con .ogg
|
|
if MP3_SOURCE and os.path.isfile(MP3_SOURCE):
|
|
ogg = os.path.splitext(MP3_SOURCE)[0] + ".ogg"
|
|
fuente = MP3_SOURCE
|
|
else:
|
|
ogg = os.path.join(SND, "aviso_fin.ogg")
|
|
fuente = os.path.join(SND, "aviso_fin.mp3")
|
|
if not os.path.exists(fuente):
|
|
try:
|
|
_voz_dalia(FRASE, fuente)
|
|
except Exception as e:
|
|
print(f"[aviso_voz] edge-tts error: {e}")
|
|
return None
|
|
if not os.path.exists(ogg):
|
|
try:
|
|
r = subprocess.run(
|
|
["ffmpeg", "-y", "-loglevel", "error", "-i", fuente,
|
|
"-c:a", "libopus", "-b:a", "48k", ogg],
|
|
capture_output=True, timeout=60)
|
|
if r.returncode != 0:
|
|
print(f"[aviso_voz] ffmpeg error: {r.stderr.decode(errors='replace')[:300]}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"[aviso_voz] ffmpeg no disponible: {e}")
|
|
return None
|
|
return ogg
|
|
|
|
|
|
def enviar_voz(chat_id):
|
|
ogg = asegurar_ogg()
|
|
if not ogg:
|
|
return False
|
|
try:
|
|
with open(ogg, "rb") as f:
|
|
r = requests.post(f"{API}/sendVoice",
|
|
data={"chat_id": chat_id},
|
|
files={"voice": ("aviso.ogg", f,
|
|
"audio/ogg")},
|
|
timeout=40)
|
|
return r.json().get("ok", False)
|
|
except Exception as e:
|
|
print(f"[aviso_voz] sendVoice error: {e}")
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
chat = sys.argv[1] if len(sys.argv) > 1 else ""
|
|
if not chat or not TOKEN:
|
|
print("Uso: python3 aviso_voz.py <chat_id> (con TELEGRAM_BOT_TOKEN en .env)")
|
|
sys.exit(1)
|
|
ok = enviar_voz(chat)
|
|
print(f"[aviso_voz] {'OK' if ok else 'FALLO'} -> chat {chat}")
|
|
sys.exit(0 if ok else 2)
|