Commit inicial: gr-locutor (Locutor IA - GRP)
Locuciones de radio con voz IA local (XTTS-v2) integradas a G Radio Player: interfaz Rust/GTK4 de configuración, motor de generación y enriquecimiento de texto en Python, daemon de monitoreo que dispara locuciones sin pausar playlist-refill. Requiere radio-player >= 0.4.11.
This commit is contained in:
@@ -0,0 +1,368 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Toma la ruta de un tema, extrae metadata y datos online, genera texto enriquecido.
|
||||
|
||||
Uso:
|
||||
./enriquecer-locucion.py /ruta/al/tema.mp3
|
||||
|
||||
Salida (stdout): texto enriquecido listo para locución
|
||||
|
||||
Estructura de la locución:
|
||||
1. Opener: una línea aleatoria, alternando entre curiosidades.txt y
|
||||
frases-motivacionales.txt (nunca se repite el mismo banco dos veces seguidas).
|
||||
2. Tema: título/artista + info enriquecida (año/género/país/dato Wikipedia).
|
||||
3. Cierre opcional: una línea de identificaciones.txt (con {radio} sustituido
|
||||
por el nombre real de la emisora, leído de gradio.config) — aparece según
|
||||
el porcentaje configurado en locutor.config (50% por defecto).
|
||||
"""
|
||||
|
||||
import json, os, random, re, requests, sys
|
||||
|
||||
try:
|
||||
import mutagen
|
||||
except ImportError:
|
||||
mutagen = None
|
||||
|
||||
DATA_DIR = os.environ.get("GR_LOCUTOR_DIR", os.path.expanduser("~/.gradio/data/locutor-ia"))
|
||||
CACHE_FILE = os.path.expanduser("~/.gradio/data/tmp/locuciones_cache.json")
|
||||
GRADIO_CONFIG_FILE = os.path.expanduser("~/.gradio/data/tmp/gradio.config")
|
||||
LOCUTOR_CONFIG_FILE = os.path.join(DATA_DIR, "locutor.config")
|
||||
ALTERNANCIA_FILE = os.path.expanduser("~/.gradio/data/tmp/locutor_ultimo_banco")
|
||||
MB_USER_AGENT = "GRadio-LocutorIA/1.0"
|
||||
TIMEOUT = 8
|
||||
|
||||
|
||||
def log(msg):
|
||||
print(f"[enriquecer] {msg}", file=sys.stderr, flush=True)
|
||||
|
||||
|
||||
def station_name():
|
||||
"""Lee el nombre de la radio desde gradio.config (línea 3), igual que GRP.
|
||||
Default 'G Radio' si no existe el archivo o la línea está vacía."""
|
||||
try:
|
||||
with open(GRADIO_CONFIG_FILE) as f:
|
||||
lines = f.read().splitlines()
|
||||
if len(lines) >= 3 and lines[2].strip():
|
||||
return lines[2].strip()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
return "G Radio"
|
||||
|
||||
|
||||
def porcentaje_identificacion():
|
||||
"""Lee porcentaje_identificacion de locutor.config (clave=valor). Default 50."""
|
||||
try:
|
||||
with open(LOCUTOR_CONFIG_FILE) as f:
|
||||
for line in f:
|
||||
if line.strip().startswith("porcentaje_identificacion"):
|
||||
return int(line.split("=", 1)[1].strip())
|
||||
except (FileNotFoundError, ValueError, IndexError):
|
||||
pass
|
||||
return 50
|
||||
|
||||
|
||||
def leer_banco(nombre_archivo):
|
||||
path = os.path.join(DATA_DIR, nombre_archivo)
|
||||
try:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
return [l.strip() for l in f if l.strip()]
|
||||
except FileNotFoundError:
|
||||
return []
|
||||
|
||||
|
||||
def elegir_opener():
|
||||
"""Alterna entre curiosidades.txt y frases-motivacionales.txt: nunca
|
||||
repite el mismo banco dos veces seguidas (se persiste cuál tocó última vez)."""
|
||||
bancos = {"curiosidades": "curiosidades.txt", "frases": "frases-motivacionales.txt"}
|
||||
ultimo = ""
|
||||
try:
|
||||
with open(ALTERNANCIA_FILE) as f:
|
||||
ultimo = f.read().strip()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
opciones = [b for b in bancos if b != ultimo] or list(bancos)
|
||||
elegido = random.choice(opciones)
|
||||
lineas = leer_banco(bancos[elegido])
|
||||
if not lineas:
|
||||
# Fallback: probar el otro banco si el elegido está vacío/falta
|
||||
otro = [b for b in bancos if b != elegido]
|
||||
if otro:
|
||||
lineas = leer_banco(bancos[otro[0]])
|
||||
elegido = otro[0]
|
||||
if not lineas:
|
||||
return ""
|
||||
|
||||
try:
|
||||
with open(ALTERNANCIA_FILE, "w") as f:
|
||||
f.write(elegido)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
return random.choice(lineas)
|
||||
|
||||
|
||||
def elegir_identificacion():
|
||||
lineas = leer_banco("identificaciones.txt")
|
||||
if not lineas:
|
||||
return f"Estás en sintonía con {station_name()}."
|
||||
return random.choice(lineas).replace("{radio}", station_name())
|
||||
|
||||
|
||||
def extract_metadata(song_path):
|
||||
title = artist = album = year = genre = None
|
||||
if mutagen:
|
||||
try:
|
||||
tags = mutagen.File(song_path, easy=True)
|
||||
if tags:
|
||||
title = tags.get("title", [None])[0]
|
||||
artist = tags.get("artist", [None])[0]
|
||||
album = tags.get("album", [None])[0]
|
||||
year = tags.get("date", [None])[0]
|
||||
genre = tags.get("genre", [None])[0]
|
||||
if title and artist:
|
||||
return title.strip(), artist.strip(), album, year, genre
|
||||
except Exception as e:
|
||||
log(f"mutagen error: {e}")
|
||||
|
||||
name = os.path.splitext(os.path.basename(song_path))[0]
|
||||
name = re.sub(r"\(.*?\)", "", name)
|
||||
name = re.sub(r"www\..*", "", name)
|
||||
name = name.strip()
|
||||
artist = title = None
|
||||
if " - " in name:
|
||||
parts = name.split(" - ", 1)
|
||||
artist = parts[0].strip()
|
||||
title = parts[1].strip()
|
||||
else:
|
||||
title = name
|
||||
return title, artist, album, year, genre
|
||||
|
||||
|
||||
def load_cache():
|
||||
try:
|
||||
with open(CACHE_FILE) as f:
|
||||
return json.load(f)
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
def save_cache(cache):
|
||||
try:
|
||||
with open(CACHE_FILE, "w") as f:
|
||||
json.dump(cache, f)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def lookup_wikipedia(artist, title, cache):
|
||||
if not title:
|
||||
return None
|
||||
key = f"wiki:{artist}:{title}"
|
||||
if key in cache:
|
||||
return cache[key]
|
||||
|
||||
queries = []
|
||||
if artist:
|
||||
queries.append(f'"{artist}" "{title}" canción')
|
||||
queries.append(f'"{artist}" "{title}" música')
|
||||
queries.append(f'"{artist}" músico')
|
||||
else:
|
||||
queries.append(f'"{title}" canción')
|
||||
|
||||
WIKI = "https://es.wikipedia.org/w/api.php"
|
||||
for q in queries:
|
||||
try:
|
||||
r = requests.get(WIKI,
|
||||
params={
|
||||
"action": "query",
|
||||
"list": "search",
|
||||
"srsearch": q,
|
||||
"format": "json",
|
||||
"srlimit": 3,
|
||||
},
|
||||
timeout=TIMEOUT,
|
||||
headers={"User-Agent": "GRadio-LocutorIA/1.0 (radio-automation)"},
|
||||
)
|
||||
data = r.json()
|
||||
pages = data.get("query", {}).get("search", [])
|
||||
if not pages:
|
||||
continue
|
||||
|
||||
for page in pages:
|
||||
page_title = page["title"]
|
||||
r2 = requests.get(
|
||||
WIKI,
|
||||
params={
|
||||
"action": "query",
|
||||
"prop": "extracts",
|
||||
"exintro": True,
|
||||
"explaintext": True,
|
||||
"titles": page_title,
|
||||
"format": "json",
|
||||
"exsentences": 3,
|
||||
},
|
||||
timeout=TIMEOUT,
|
||||
headers={"User-Agent": "GRadio-LocutorIA/1.0 (radio-automation)"},
|
||||
)
|
||||
pg_data = r2.json().get("query", {}).get("pages", {})
|
||||
for pid, pg in pg_data.items():
|
||||
if pid == "-1" or "extract" not in pg:
|
||||
continue
|
||||
extract = pg["extract"].strip()
|
||||
sentences = re.split(r"(?<=[.!?])\s+", extract)
|
||||
for s in sentences:
|
||||
s = s.strip()
|
||||
if 40 <= len(s) <= 200:
|
||||
cache[key] = s
|
||||
save_cache(cache)
|
||||
return s
|
||||
except Exception as e:
|
||||
log(f"wiki error ({q[:30]}...): {e}")
|
||||
|
||||
cache[key] = None
|
||||
save_cache(cache)
|
||||
return None
|
||||
|
||||
|
||||
def lookup_musicbrainz(artist, title, cache):
|
||||
if not artist or not title:
|
||||
return None
|
||||
key = f"mb:{artist}:{title}"
|
||||
if key in cache:
|
||||
return cache[key]
|
||||
|
||||
try:
|
||||
r = requests.get(
|
||||
"https://musicbrainz.org/ws/2/recording/",
|
||||
params={
|
||||
"query": f'artist:"{artist}" AND recording:"{title}"',
|
||||
"fmt": "json",
|
||||
"limit": 1,
|
||||
},
|
||||
timeout=TIMEOUT + 2,
|
||||
headers={"User-Agent": MB_USER_AGENT},
|
||||
)
|
||||
data = r.json()
|
||||
recs = data.get("recordings", [])
|
||||
if not recs:
|
||||
cache[key] = None
|
||||
save_cache(cache)
|
||||
return None
|
||||
|
||||
rec = recs[0]
|
||||
info = {}
|
||||
if "releases" in rec and rec["releases"]:
|
||||
rel = rec["releases"][0]
|
||||
date = rel.get("date", "")
|
||||
info["year"] = date[:4] if date and len(date) >= 4 else ""
|
||||
info["country"] = rel.get("country", "")
|
||||
if "artist-credit" in rec:
|
||||
ac = rec["artist-credit"][0]
|
||||
if isinstance(ac, dict) and "artist" in ac:
|
||||
info["artist_type"] = ac["artist"].get("type", "")
|
||||
|
||||
cache[key] = info if info else None
|
||||
save_cache(cache)
|
||||
return info
|
||||
except Exception as e:
|
||||
log(f"musicbrainz error: {e}")
|
||||
cache[key] = None
|
||||
save_cache(cache)
|
||||
return None
|
||||
|
||||
|
||||
_PAISES = {
|
||||
"US": "Estados Unidos", "GB": "Reino Unido", "UK": "Reino Unido",
|
||||
"ES": "España", "MX": "México", "AR": "Argentina", "CO": "Colombia",
|
||||
"CL": "Chile", "PE": "Perú", "CU": "Cuba", "PR": "Puerto Rico",
|
||||
"DE": "Alemania", "FR": "Francia", "IT": "Italia", "BR": "Brasil",
|
||||
"CA": "Canadá", "AU": "Australia", "JP": "Japón", "MY": "Malasia",
|
||||
"SE": "Suecia", "NL": "Países Bajos", "IE": "Irlanda", "NO": "Noruega",
|
||||
"DK": "Dinamarca", "PT": "Portugal", "RU": "Rusia",
|
||||
}
|
||||
|
||||
|
||||
def generate_text(artist, title, album, year, genre, wiki_fact, mb_info):
|
||||
anyo = mb_info.get("year", "") if mb_info and isinstance(mb_info, dict) else ""
|
||||
|
||||
nombre_pais = ""
|
||||
pa_mb = mb_info.get("country", "") if mb_info and isinstance(mb_info, dict) else ""
|
||||
if pa_mb:
|
||||
nombre_pais = _PAISES.get(pa_mb, pa_mb)
|
||||
|
||||
# --- 1. Opener (alternando curiosidades.txt / frases-motivacionales.txt) ---
|
||||
opener = elegir_opener()
|
||||
|
||||
# --- 2. Tema ---
|
||||
if artist and title:
|
||||
tema = f"Escuchas {title} de {artist}"
|
||||
elif title:
|
||||
tema = f"Escuchas {title}"
|
||||
else:
|
||||
tema = ""
|
||||
|
||||
# --- 3. Info enriquecida ---
|
||||
infos = []
|
||||
if anyo and anyo not in ("?", ""):
|
||||
infos.append(f"grabada en {anyo}")
|
||||
if genre and genre not in ("?", ""):
|
||||
infos.append(f"un tema de {genre}")
|
||||
if nombre_pais:
|
||||
infos.append(f"lanzada en {nombre_pais}")
|
||||
|
||||
info_str = ", ".join(infos) if infos else ""
|
||||
|
||||
wiki_str = ""
|
||||
if wiki_fact:
|
||||
fact = wiki_fact.strip()
|
||||
if len(fact) > 160:
|
||||
fact = fact[:157] + "..."
|
||||
wiki_str = fact
|
||||
|
||||
if tema and info_str:
|
||||
cuerpo = f"{tema}, {info_str}."
|
||||
elif tema:
|
||||
cuerpo = f"{tema}."
|
||||
elif info_str:
|
||||
cuerpo = info_str
|
||||
else:
|
||||
cuerpo = ""
|
||||
|
||||
if wiki_str:
|
||||
cuerpo = f"{cuerpo} {wiki_str}" if cuerpo else wiki_str
|
||||
|
||||
# --- 4. Cierre opcional (según porcentaje_identificacion de locutor.config) ---
|
||||
cierre = ""
|
||||
if random.randint(1, 100) <= porcentaje_identificacion():
|
||||
cierre = elegir_identificacion()
|
||||
|
||||
oraciones = [o for o in (opener, cuerpo, cierre) if o]
|
||||
return " ".join(oraciones)
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
log("Uso: enriquecer-locucion.py /ruta/al/tema.mp3")
|
||||
sys.exit(1)
|
||||
|
||||
song_path = os.path.abspath(sys.argv[1])
|
||||
if not os.path.exists(song_path):
|
||||
log(f"no existe: {song_path}")
|
||||
sys.exit(1)
|
||||
|
||||
title, artist, album, year, genre = extract_metadata(song_path)
|
||||
log(f"metadata: artist={artist} title={title} album={album} year={year} genre={genre}")
|
||||
|
||||
cache = load_cache()
|
||||
wiki_fact = lookup_wikipedia(artist or title, title, cache)
|
||||
mb_info = lookup_musicbrainz(artist or title, title, cache)
|
||||
|
||||
text = generate_text(artist, title, album, year, genre, wiki_fact, mb_info)
|
||||
log(f"wiki={'sí' if wiki_fact else 'no'} mb={'sí' if mb_info else 'no'}")
|
||||
log(f"texto ({len(text)} chars): {text[:100]}...")
|
||||
|
||||
print(text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,60 @@
|
||||
#!/bin/bash
|
||||
# Genera una locución de introducción para un tema musical
|
||||
# Uso: ./generar-intro.sh "/ruta/al/tema.mp3"
|
||||
# Imprime la ruta del MP3 generado en stdout
|
||||
#
|
||||
# 1. Lee metadata ID3 del tema + enriquece con Wikipedia/MusicBrainz
|
||||
# 2. Sintetiza con el motor instalado (~/.gradio/locutor/gr-locutor.sh),
|
||||
# usando la voz de referencia configurada (genérica de GRP o la propia
|
||||
# de la estación si fue reemplazada)
|
||||
|
||||
set -e
|
||||
cd "$(dirname "$0")"
|
||||
|
||||
TEMA="$1"
|
||||
LOCUTOR_BIN="$HOME/.gradio/locutor-venv/gr-locutor.sh"
|
||||
ENRIQUECER="enriquecer-locucion.py"
|
||||
SALIDA_DIR="$HOME/.gradio/data/locutor-ia/salida"
|
||||
|
||||
if [ -z "$TEMA" ]; then
|
||||
echo "Uso: $0 \"/ruta/al/tema.mp3\""
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if [ ! -f "$TEMA" ]; then
|
||||
echo "Error: no existe el archivo $TEMA"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if [ ! -x "$LOCUTOR_BIN" ]; then
|
||||
echo "Error: no está instalado el motor de voz ($LOCUTOR_BIN)."
|
||||
echo "Ejecuta primero el instalador (instalar-locutor-amd64.sh)."
|
||||
exit 1
|
||||
fi
|
||||
|
||||
TEXTO=""
|
||||
if [ -f "$ENRIQUECER" ]; then
|
||||
TEXTO=$(python3 "$ENRIQUECER" "$TEMA" 2>/dev/null || true)
|
||||
fi
|
||||
|
||||
if [ -z "$TEXTO" ]; then
|
||||
BASENAME=$(basename "$TEMA" .mp3)
|
||||
BASENAME="${BASENAME%.*}"
|
||||
ARTISTA=""
|
||||
TITULO="$BASENAME"
|
||||
if echo "$BASENAME" | grep -q " - "; then
|
||||
ARTISTA=$(echo "$BASENAME" | sed 's/ - .*//')
|
||||
TITULO=$(echo "$BASENAME" | sed 's/.* - //')
|
||||
fi
|
||||
if [ -n "$ARTISTA" ]; then
|
||||
TEXTO="Y ahora, disfruta de ${TITULO} de ${ARTISTA}."
|
||||
else
|
||||
TEXTO="Y ahora, disfruta de ${TITULO}."
|
||||
fi
|
||||
fi
|
||||
|
||||
NOMBRE="intro_$(date +%Y%m%d_%H%M%S)"
|
||||
mkdir -p "$SALIDA_DIR"
|
||||
|
||||
MP3=$("$LOCUTOR_BIN" "$TEXTO" "$NOMBRE")
|
||||
echo "$MP3"
|
||||
@@ -0,0 +1,339 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Monitor de Locuciones — Ventana corrediza configurable.
|
||||
|
||||
Cuando una locución+song se consume (desaparece del tope de playlist4),
|
||||
genera la siguiente locución para el próximo song sin locución que le
|
||||
corresponda narrar (según `cada_cuantos_temas`). Solo genera UNA a la vez
|
||||
para no saturar el motor de TTS.
|
||||
|
||||
Lee `ventana_previas` y `cada_cuantos_temas` de locutor.config en cada
|
||||
ciclo (permite reconfigurar sin reiniciar el daemon).
|
||||
|
||||
No pausa `playlist-refill` (el daemon de GRP que rellena la cola desde la
|
||||
parrilla): ese proceso solo agrega canciones al final hasta un largo
|
||||
objetivo, sin importar el contenido existente — conviven sin conflicto,
|
||||
igual que si se agregaran temas a mano. La única carrera real posible
|
||||
(el player consume temas mientras este monitor genera una locución, que
|
||||
puede tardar decenas de segundos) se resuelve re-leyendo `playlist4` justo
|
||||
antes de escribir, no pausando nada — pausar indefinidamente dejaría de
|
||||
rellenar la cola y la estación se quedaría sin música.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
TMP_DIR = os.path.expanduser("~/.gradio/data/tmp")
|
||||
DATA_DIR = os.path.expanduser("~/.gradio/data/locutor-ia")
|
||||
ESTADO_FILE = os.path.join(TMP_DIR, "estado.json")
|
||||
PLAYLIST_FILE = os.path.join(TMP_DIR, "playlist4")
|
||||
LOCUTOR_AUTO_FILE = os.path.join(TMP_DIR, "locutor_auto")
|
||||
LOCUTOR_CONFIG_FILE = os.path.join(DATA_DIR, "locutor.config")
|
||||
CONTADOR_FILE = os.path.join(TMP_DIR, "locutor_song_counter.json")
|
||||
SCRIPT_DIR = os.path.join(DATA_DIR, "scripts")
|
||||
GENERAR_SCRIPT = os.path.join(SCRIPT_DIR, "generar-intro.sh")
|
||||
LOCK_FILE = os.path.join(TMP_DIR, "locuciones_generating")
|
||||
|
||||
POLL_INTERVAL = 3
|
||||
DEFAULT_WINDOW_SIZE = 3
|
||||
DEFAULT_CADA_CUANTOS = 3
|
||||
|
||||
|
||||
def log(msg):
|
||||
ts = time.strftime("%Y-%m-%d %H:%M:%S")
|
||||
print(f"[{ts}] {msg}", flush=True)
|
||||
|
||||
|
||||
def leer_locutor_config():
|
||||
"""Lee ventana_previas y cada_cuantos_temas de locutor.config (clave=valor)."""
|
||||
ventana = DEFAULT_WINDOW_SIZE
|
||||
cada_cuantos = DEFAULT_CADA_CUANTOS
|
||||
try:
|
||||
with open(LOCUTOR_CONFIG_FILE) as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if "=" not in line or line.startswith("#"):
|
||||
continue
|
||||
clave, valor = line.split("=", 1)
|
||||
clave, valor = clave.strip(), valor.strip()
|
||||
if clave == "ventana_previas":
|
||||
ventana = max(1, int(valor))
|
||||
elif clave == "cada_cuantos_temas":
|
||||
cada_cuantos = max(1, int(valor))
|
||||
except (FileNotFoundError, ValueError):
|
||||
pass
|
||||
return ventana, cada_cuantos
|
||||
|
||||
|
||||
def read_playlist():
|
||||
try:
|
||||
with open(PLAYLIST_FILE) as f:
|
||||
return [l.strip() for l in f if l.strip()]
|
||||
except FileNotFoundError:
|
||||
return []
|
||||
|
||||
|
||||
def write_playlist(lines):
|
||||
with open(PLAYLIST_FILE, "w") as f:
|
||||
for line in lines:
|
||||
f.write(line + "\n")
|
||||
|
||||
|
||||
def is_locution(entry):
|
||||
ruta = entry.split("\t")[0]
|
||||
return "intro_" in os.path.basename(ruta)
|
||||
|
||||
|
||||
def get_song_path(entry):
|
||||
return entry.split("\t")[0]
|
||||
|
||||
|
||||
MAX_DECISIONES = 300 # tope para que el archivo no crezca sin límite
|
||||
|
||||
|
||||
def load_contador():
|
||||
try:
|
||||
with open(CONTADOR_FILE) as f:
|
||||
state = json.load(f)
|
||||
state.setdefault("count", 0)
|
||||
state.setdefault("decisiones", {})
|
||||
return state
|
||||
except Exception:
|
||||
return {"count": 0, "decisiones": {}}
|
||||
|
||||
|
||||
def save_contador(state):
|
||||
try:
|
||||
with open(CONTADOR_FILE, "w") as f:
|
||||
json.dump(state, f)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def corresponde_narrar(song_path, cada_cuantos):
|
||||
"""Decide si a este tema le toca locución, contando temas distintos vistos.
|
||||
|
||||
Recuerda la decisión de CADA tema ya evaluado (no solo el último): el
|
||||
escaneo de ensure_window() reinicia desde el principio en cada ciclo de
|
||||
poll, así que un tema ya decidido puede volver a aparecer como "primer
|
||||
candidato" en ciclos siguientes — si solo se recordara el último tema
|
||||
visto, ese tema se recontaría como si fuera nuevo cada vez, corrompiendo
|
||||
el conteo de "cada N temas".
|
||||
"""
|
||||
state = load_contador()
|
||||
decisiones = state["decisiones"]
|
||||
if song_path not in decisiones:
|
||||
state["count"] += 1
|
||||
decisiones[song_path] = (state["count"] % cada_cuantos == 0)
|
||||
# Podar entradas más viejas si crece demasiado (dict preserva orden de inserción)
|
||||
while len(decisiones) > MAX_DECISIONES:
|
||||
decisiones.pop(next(iter(decisiones)))
|
||||
save_contador(state)
|
||||
return decisiones[song_path]
|
||||
|
||||
|
||||
def generate_locution(song_path):
|
||||
if not os.path.exists(GENERAR_SCRIPT):
|
||||
log(f"ERROR: no existe {GENERAR_SCRIPT}")
|
||||
return None, None
|
||||
if not os.path.exists(song_path):
|
||||
log(f"ERROR: no existe {song_path}")
|
||||
return None, None
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["bash", GENERAR_SCRIPT, song_path],
|
||||
capture_output=True, text=True, timeout=300,
|
||||
cwd=SCRIPT_DIR
|
||||
)
|
||||
if result.returncode == 0:
|
||||
rel_path = result.stdout.strip().split("\n")[-1]
|
||||
mp3_path = rel_path if os.path.isabs(rel_path) else os.path.join(SCRIPT_DIR, rel_path)
|
||||
if mp3_path and mp3_path.endswith(".mp3") and os.path.exists(mp3_path):
|
||||
dur = get_duration(mp3_path)
|
||||
return mp3_path, dur
|
||||
err = result.stderr.strip() or "(sin stderr)"
|
||||
log(f"ERROR generar-intro (rc={result.returncode}): {err[:150]}")
|
||||
return None, None
|
||||
except subprocess.TimeoutExpired:
|
||||
log("ERROR: generar-intro TIMEOUT (>300s)")
|
||||
return None, None
|
||||
except Exception as e:
|
||||
log(f"ERROR: {e}")
|
||||
return None, None
|
||||
|
||||
|
||||
def get_duration(mp3_path):
|
||||
try:
|
||||
r = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration",
|
||||
"-of", "default=noprint_wrappers=1:nokey=1", mp3_path],
|
||||
capture_output=True, text=True, timeout=5)
|
||||
secs = float(r.stdout.strip())
|
||||
mins = int(secs // 60)
|
||||
secs = secs % 60
|
||||
return f"{mins:02d}:{secs:05.3f}"[:11]
|
||||
except Exception:
|
||||
pass
|
||||
return "00:00:16.000"
|
||||
|
||||
|
||||
LOCK_MAX_AGE = 300 # 5 min — la generación real tarda como mucho ~2 min (CPU)
|
||||
|
||||
|
||||
def ensure_window():
|
||||
"""Mantiene `ventana_previas` locuciones al inicio de la playlist, solo
|
||||
para los temas que correspondan según `cada_cuantos_temas`. Genera UNA
|
||||
a la vez para evitar saturar el motor de TTS.
|
||||
"""
|
||||
if os.path.exists(LOCK_FILE):
|
||||
# Si el lock es viejo, quedó de una generación que murió sin pasar
|
||||
# por el `finally` (kill -9, crash) — lo consideramos obsoleto y lo
|
||||
# limpiamos, sino el monitor queda trabado para siempre sin avisar.
|
||||
edad = time.time() - os.path.getmtime(LOCK_FILE)
|
||||
if edad > LOCK_MAX_AGE:
|
||||
log(f"Lock de generación obsoleto ({edad:.0f}s) — se limpia y se continúa")
|
||||
os.unlink(LOCK_FILE)
|
||||
else:
|
||||
return # ya hay una generación en curso
|
||||
|
||||
ventana, cada_cuantos = leer_locutor_config()
|
||||
|
||||
pl = read_playlist()
|
||||
if len(pl) < 1:
|
||||
return
|
||||
|
||||
pairs = 0
|
||||
start_i = 1 if (len(pl) > 0 and not is_locution(pl[0])) else 0
|
||||
i = start_i
|
||||
while i < len(pl) - 1:
|
||||
if is_locution(pl[i]) and not is_locution(pl[i + 1]):
|
||||
pairs += 1
|
||||
i += 2
|
||||
else:
|
||||
break
|
||||
|
||||
if pairs >= ventana:
|
||||
return # ventana completa
|
||||
|
||||
# El primer item de la playlist es el que está sonando (o por sonar).
|
||||
# Si es un song sin locución, su intro ya fue consumida → omitir (evita loop).
|
||||
start = 1 if (len(pl) > 0 and not is_locution(pl[0])) else 0
|
||||
for i in range(start, len(pl)):
|
||||
entry = pl[i]
|
||||
if is_locution(entry):
|
||||
continue
|
||||
if i > 0 and is_locution(pl[i - 1]):
|
||||
continue
|
||||
|
||||
song_path = get_song_path(entry)
|
||||
|
||||
if not corresponde_narrar(song_path, cada_cuantos):
|
||||
continue # a este tema no le toca, seguir buscando el próximo candidato
|
||||
|
||||
song_name = os.path.basename(song_path)
|
||||
log(f"Generando locución para: {song_name} ({pairs+1}/{ventana})")
|
||||
|
||||
open(LOCK_FILE, "w").close()
|
||||
try:
|
||||
mp3, dur = generate_locution(song_path)
|
||||
finally:
|
||||
os.unlink(LOCK_FILE)
|
||||
|
||||
if not mp3:
|
||||
log(" ✗ falló generación, reintentando después")
|
||||
return
|
||||
|
||||
loc_entry = f"{mp3}\t{dur}"
|
||||
|
||||
# Re-leer playlist actual para evitar race condition: el player pudo
|
||||
# haber consumido items durante la generación (decenas de segundos).
|
||||
current_pl = read_playlist()
|
||||
if not current_pl:
|
||||
current_pl = [loc_entry]
|
||||
log(" ✓ playlist vacía, insertando única línea")
|
||||
else:
|
||||
target_path = get_song_path(entry)
|
||||
inserted = False
|
||||
for j, line in enumerate(current_pl):
|
||||
if get_song_path(line) == target_path:
|
||||
if j > 0 and is_locution(current_pl[j - 1]):
|
||||
log(" ✗ la canción ya tiene locución, omitiendo")
|
||||
return
|
||||
current_pl.insert(j, loc_entry)
|
||||
inserted = True
|
||||
break
|
||||
if not inserted:
|
||||
current_pl.append(loc_entry)
|
||||
log(" ✓ canción ya consumida, insertada al final")
|
||||
else:
|
||||
log(f" ✓ insertada en playlist: {len(current_pl)} líneas")
|
||||
|
||||
write_playlist(current_pl)
|
||||
return # solo una por ciclo
|
||||
|
||||
log("(no se encontraron temas pendientes de narrar)")
|
||||
|
||||
|
||||
def locutor_habilitado():
|
||||
try:
|
||||
with open(LOCUTOR_AUTO_FILE) as f:
|
||||
return f.read().strip() == "1"
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
log("=== Monitor de Locuciones (gr-locutor) iniciado ===")
|
||||
|
||||
# Un lock encontrado al arrancar es, por definición, de una instancia
|
||||
# anterior (este proceso recién empieza, no pudo haber creado ninguno
|
||||
# todavía) — por ejemplo si el daemon anterior murió por SIGTERM en
|
||||
# medio de una generación, sin pasar por el `finally` que lo libera.
|
||||
if os.path.exists(LOCK_FILE):
|
||||
log("Lock de generación de una instancia anterior encontrado al iniciar — se limpia")
|
||||
os.unlink(LOCK_FILE)
|
||||
|
||||
if not locutor_habilitado():
|
||||
log("tmp/locutor_auto no está en '1' — el monitor no generará nada.")
|
||||
log("Activalo en GRP → Configuración → Locutor Automático.")
|
||||
|
||||
ventana, cada_cuantos = leer_locutor_config()
|
||||
log(f"Ventana de {ventana} locuciones · narra 1 de cada {cada_cuantos} temas")
|
||||
log("playlist-refill sigue activo — no se pausa (conviven sin conflicto)")
|
||||
|
||||
last_track = ""
|
||||
|
||||
while True:
|
||||
try:
|
||||
if locutor_habilitado():
|
||||
ensure_window()
|
||||
except Exception as e:
|
||||
log(f"Error: {e}")
|
||||
|
||||
try:
|
||||
with open(ESTADO_FILE) as f:
|
||||
estado = json.load(f)
|
||||
except (FileNotFoundError, json.JSONDecodeError):
|
||||
estado = None
|
||||
|
||||
if estado:
|
||||
track = estado.get("track_actual", "")
|
||||
reproduciendo = estado.get("reproduciendo", False)
|
||||
if track and track != last_track and reproduciendo:
|
||||
name = os.path.basename(track)
|
||||
if "intro_" in name:
|
||||
log(f"▶ Locución: {name}")
|
||||
else:
|
||||
log(f"▶ Tema: {name}")
|
||||
last_track = track
|
||||
|
||||
time.sleep(POLL_INTERVAL)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except KeyboardInterrupt:
|
||||
log("Monitor detenido por el usuario")
|
||||
sys.exit(0)
|
||||
@@ -0,0 +1,54 @@
|
||||
#!/bin/bash
|
||||
# Inicia/detiene el monitor de locuciones (gr-locutor)
|
||||
# Uso: start-monitor.sh [start|stop|status]
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
PIDFILE="$HOME/.gradio/data/tmp/pids/locuciones-monitor.pid"
|
||||
MONITOR="$SCRIPT_DIR/monitor-locuciones.py"
|
||||
LOCUTOR_VENV="$HOME/.gradio/locutor-venv"
|
||||
|
||||
case "${1:-start}" in
|
||||
start)
|
||||
if [ -f "$PIDFILE" ]; then
|
||||
pid=$(cat "$PIDFILE" 2>/dev/null)
|
||||
if kill -0 "$pid" 2>/dev/null; then
|
||||
echo "Monitor ya está corriendo (PID $pid)"
|
||||
exit 0
|
||||
fi
|
||||
rm -f "$PIDFILE"
|
||||
fi
|
||||
mkdir -p "$(dirname "$PIDFILE")"
|
||||
# shellcheck source=/dev/null
|
||||
source "$LOCUTOR_VENV/bin/activate"
|
||||
nohup python3 "$MONITOR" > "$HOME/.gradio/data/tmp/locuciones-monitor.log" 2>&1 &
|
||||
echo $! > "$PIDFILE"
|
||||
echo "Monitor de locuciones iniciado (PID $!)"
|
||||
;;
|
||||
stop)
|
||||
if [ -f "$PIDFILE" ]; then
|
||||
pid=$(cat "$PIDFILE" 2>/dev/null)
|
||||
kill "$pid" 2>/dev/null && echo "Monitor detenido" || echo "No se pudo detener"
|
||||
rm -f "$PIDFILE"
|
||||
else
|
||||
pkill -f "monitor-locuciones.py" 2>/dev/null && echo "Monitor detenido" || echo "Monitor no estaba corriendo"
|
||||
fi
|
||||
;;
|
||||
status)
|
||||
if [ -f "$PIDFILE" ]; then
|
||||
pid=$(cat "$PIDFILE" 2>/dev/null)
|
||||
if kill -0 "$pid" 2>/dev/null; then
|
||||
echo "Monitor ACTIVO (PID $pid)"
|
||||
tail -3 "$HOME/.gradio/data/tmp/locuciones-monitor.log" 2>/dev/null
|
||||
else
|
||||
echo "Monitor INACTIVO (PID file obsoleto)"
|
||||
rm -f "$PIDFILE"
|
||||
fi
|
||||
else
|
||||
echo "Monitor INACTIVO"
|
||||
fi
|
||||
;;
|
||||
*)
|
||||
echo "Uso: $0 [start|stop|status]"
|
||||
exit 1
|
||||
;;
|
||||
esac
|
||||
Reference in New Issue
Block a user