#!/usr/bin/env python3
"""
app.py — Limpieza de voz y control de calidad de entregas de audio.
Ejecutar en local: python app.py (http://localhost:7860)
En Hugging Face (ZeroGPU) se usa la GPU si hay cuota y, si no, la CPU.
"""
try:
import spaces # noqa: F401 (en ZeroGPU debe importarse antes que torch/gradio)
except ImportError: # versión de escritorio
spaces = None
import os
import shutil
import subprocess
import tempfile
import threading
import time
import urllib.request
import zipfile
from collections import OrderedDict
from pathlib import Path
import gradio as gr
import numpy as np
import dsp
import metrics
import pipeline
import profiles
import report
import spectro
import ui_html
from audio_io import AUDIO_EXTENSIONS, fmt_duration, read_audio, read_meta
from denoise_core import gpu_available
PROFILES = profiles.load_profiles()
PROFILE_CHOICES = [(p.get("nombre", pid), pid) for pid, p in PROFILES.items()]
DEFAULT_PROFILE = next(iter(PROFILES))
PRESETS = {"Suave": 12.0, "Natural (recomendado)": 24.0, "Fuerte": 36.0, "Máximo": None}
DEFAULT_PRESET = "Natural (recomendado)"
HP_CHOICES = [("No", 0), ("Muy suave (40 Hz)", 40), ("Suave (60 Hz)", 60), ("Normal (70 Hz, recomendado)", 70),
("Fuerte (80 Hz)", 80), ("Muy fuerte (100 Hz)", 100)]
WORK_DIR = Path(tempfile.gettempdir()) / "audio-denoise-trabajos"
KEEP_HOURS = 6
ON_SPACE = bool(os.environ.get("SPACE_ID"))
DESKTOP_REPO = "Azuredraco/limpieza-de-voz-windows"
DESKTOP_URL = f"https://huggingface.co/{DESKTOP_REPO}/resolve/main/Instalar-Limpieza-de-voz.exe"
DESKTOP_ZIP_URL = f"https://huggingface.co/{DESKTOP_REPO}/resolve/main/Limpieza-de-voz-portable.zip"
def _read_version() -> str:
"""Versión: fichero VERSION (lo escriben los despliegues) o, en desarrollo, el commit de git."""
f = Path(__file__).with_name("VERSION")
if f.is_file():
return f.read_text(encoding="utf-8-sig").strip()
try:
out = subprocess.run(["git", "log", "-1", "--format=%cd-%h", "--date=format:%Y.%m.%d"],
cwd=Path(__file__).parent, capture_output=True, text=True, timeout=3)
return (out.stdout.strip() + "-dev") if out.returncode == 0 and out.stdout.strip() else "dev"
except Exception:
return "dev"
VERSION = _read_version()
VERSION_URL = f"https://huggingface.co/{DESKTOP_REPO}/resolve/main/version.txt"
_latest = {"v": None}
def _check_update():
"""En la versión de escritorio, mira (en segundo plano) si hay una versión nueva publicada."""
try:
req = urllib.request.Request(VERSION_URL, headers={"User-Agent": "audio-denoise"})
with urllib.request.urlopen(req, timeout=8) as r:
_latest["v"] = r.read().decode("utf-8").strip() or None
except Exception:
pass
if not ON_SPACE and VERSION != "dev" and not VERSION.endswith("-dev"):
threading.Thread(target=_check_update, daemon=True).start()
def _update_banner() -> str:
v = _latest["v"]
if not v or v == VERSION or ON_SPACE:
return ""
return (f"
🆕 Hay una versión nueva ({v}).
Descárgala aquí "
"e instálala encima: tus ajustes se conservan.
")
QUALITY_LABEL = "⭐ Nota de calidad (antes y después)"
QUALITY_INFO = ("Puntúa de 1 a 5 cómo suena cada audio antes y después, como lo valoraría un oyente. "
"Hace que cada audio tarde más: desmárcala solo si tienes prisa.")
MODE_CHOICES = {
"📏 Solo formato": "formato",
"🎚️ Permitido": "permitido",
"🥷 Ninja": "ninja",
}
MODE_INFO = ("Solo formato: bits, canales y frecuencia; el sonido no cambia. · Permitido: además volumen, picos y "
"silencios del principio y del final. · Ninja: además quita ruido de fondo y retumbes de forma sutil "
"(pulido automático: solo se queda si suena mejor).")
AUTO_LABEL = "🤖 AUTOMÁTICO · la app lo hace todo"
AUTO_INFO = ("Sube los audios y te los devuelve listos: los revisa, arregla solo lo que haga falta (lo mínimo) y te "
"enseña qué ha cambiado. Si ya cumplen, no se tocan. Desmárcalo para elegir tú cómo procesar.")
AUTO_MODE = "auto"
EVENTS_LABEL = "🔇 Quitar ruidos sueltos en las pausas"
EVENTS_INFO = ("Cláxones, golpes, puertas… que suenan mientras nadie habla se cambian por el silencio de la sala; "
"la voz no se toca. En cada audio verás en qué minuto estaba cada uno. Ojo con risas, llantos o "
"muletillas («mmm»): el detector no las considera voz y las quitaría.")
# ── Utilidades ───────────────────────────────────────────────────────────────
def _new_job_dir() -> Path:
WORK_DIR.mkdir(parents=True, exist_ok=True)
limit = time.time() - KEEP_HOURS * 3600
for d in WORK_DIR.iterdir():
try:
if d.stat().st_mtime < limit:
shutil.rmtree(d, ignore_errors=True)
except OSError:
pass
return Path(tempfile.mkdtemp(prefix=time.strftime("%Y%m%d-%H%M%S-"), dir=WORK_DIR))
def _preset_label(db) -> str:
for k, v in PRESETS.items():
if v == db:
return k
return DEFAULT_PRESET
def _settings(prof, preset, hp, ultra, escalate, use_dnsmos, use_gpu, mono=None, mode=None,
events=False) -> pipeline.Settings:
modo = mode if mode == AUTO_MODE else MODE_CHOICES.get(mode, pipeline.default_mode(prof))
return pipeline.Settings.from_profile(
prof, reduccion_db=PRESETS[preset], paso_alto_hz=float(hp) or None,
conservar_ultrasonidos=ultra, escalar_reduccion=escalate, dnsmos=use_dnsmos, usar_gpu=use_gpu,
forzar_mono=mono, pulir=modo == "ninja", modo=modo, quitar_eventos=bool(events),
)
def _mode_label(modo: str) -> str:
return next(k for k, v in MODE_CHOICES.items() if v == modo)
def _paths(files) -> list[str]:
if not files:
return []
return [f if isinstance(f, str) else f.name for f in files]
def _profile_info(pid: str):
prof = PROFILES[pid]
proc = prof.get("procesado") or {}
desc = prof.get("explicacion") or f"**{prof.get('nombre', pid)}** — {prof.get('descripcion', '')}"
return (
desc,
gr.update(value=_preset_label(proc.get("reduccion_db", 24.0))),
gr.update(value=int(proc.get("paso_alto_hz") or 0)),
gr.update(value=bool(proc.get("conservar_ultrasonidos", True))),
gr.update(value=_mode_label(pipeline.default_mode(prof))),
gr.update(value=(prof.get("formato") or {}).get("canales") == 1),
)
# ── Cachés: no repetir el análisis ni la corrección de un audio que no ha cambiado ─────
_CACHE: "OrderedDict[tuple, object]" = OrderedDict()
CACHE_MAX = 64
def _file_key(path: str) -> tuple:
st = os.stat(path)
return path, st.st_size, st.st_mtime_ns
def _cached(key: tuple, make):
if key in _CACHE:
_CACHE.move_to_end(key)
return _CACHE[key]
val = make()
_CACHE[key] = val
while len(_CACHE) > CACHE_MAX:
_CACHE.popitem(last=False)
return val
def _analyze(path: str, prof: dict, pid: str):
return _cached(("an", *_file_key(path), pid, prof.get("_pausa_max_s")), lambda: pipeline.analyze(path, prof))
# ── Comprobar ────────────────────────────────────────────────────────────────
def _prof(pid: str, pause_max: float) -> dict:
return {**PROFILES[pid], "_pausa_max_s": float(pause_max)}
def cb_analyze(files, pid, pause_max, progress=gr.Progress()):
paths = _paths(files)
if not paths:
yield ui_html.empty("Primero sube uno o varios audios (paso 2)."), ""
return
prof = _prof(pid, pause_max)
cards, durs, all_rows = [], [], []
n_ok = n_fix = n_err = n_pauses = 0
yield ui_html.banner("info", f"⏳ Revisando {len(paths)} audio(s)…"), ""
for i, p in enumerate(paths):
progress(i / len(paths), desc=f"Analizando {Path(p).name}")
a = _analyze(p, prof, pid)
name = Path(p).name
if a.error:
n_err += 1
cards.append(ui_html.card(name, "", ("bad", "✗ ERROR"), [], error=a.error))
else:
durs.append((name, a.meta.duration))
if a.ok:
n_ok += 1
badge = ("ok", "✓ CUMPLE")
vd = ui_html.verdict("ok", "✅ Cumple: listo para enviar tal cual")
else:
n_fix += 1
badge = ("fix", "⚡ HAY QUE CORREGIR")
vd = ui_html.verdict("fix", f"⚠️ No cumple: {', '.join(profiles.failed(a.rows))}",
"Pulsa «✨ Corregir y descargar» (o activa 🤖 Automático) para arreglarlo.")
n_pauses += bool(profiles.long_pauses(a.m, prof))
all_rows += a.rows
cards.append(ui_html.card(name, a.meta.describe(), badge, a.rows, m_before=a.m, verdict_html=vd,
original=_file_url(p)))
extra = ""
if n_pauses and i == len(paths) - 1:
extra = ui_html.banner("info", f"✂️ Además, {n_pauses} audio(s) tienen pausas de más de {pause_max:.1f} s. "
"Si quieres acortarlas, marca «Acortar pausas largas» antes de corregir.")
if i == len(paths) - 1:
extra += _pairs_html(prof, [p for p in paths])
summary = _summary(prof, len(paths), n_ok, n_fix, n_err, durs, "Por corregir", extra=extra)
yield summary, ui_html.glossary_all(all_rows) + "".join(reversed(cards))
progress(1.0)
def _summary(prof, total, n_ok, n_fix, n_err, durs, fix_label, final=False, extra: str = "", left: str = "") -> str:
target = (prof.get("entregables") or {}).get("objetivo_total_s")
out = ui_html.stats(total, n_ok, n_fix, n_err, sum(d for _, d in durs), target, fix_label)
done = final or (n_ok + n_fix + n_err) == total
reviewing = fix_label == "Por corregir"
if not done:
out = ui_html.banner("info", f"⏳ {'Revisando' if reviewing else 'Corrigiendo'}… "
f"{n_ok + n_fix + n_err} de {total} audio(s) terminados{left}") + out
elif n_err:
out = ui_html.banner("bad", "❌ Algún audio no se ha podido leer o sigue sin cumplir. "
"Mira su tarjeta, aquí debajo, antes de entregar.") + out
elif reviewing and n_fix:
out = ui_html.banner("fix", f"⚡ {n_fix} de {total} audio(s) NO cumplen. En cada tarjeta, aquí debajo, tienes "
"qué falla (filas «Falla»). Pulsa «✨ Corregir y descargar» para arreglarlos.") + out
elif reviewing:
out = ui_html.banner("ok", f"✅ Revisados: los {total} audio(s) cumplen ya. No hace falta corregir nada.") + out
else:
partes = []
if n_fix:
partes.append(f"{n_fix} corregido(s)")
if n_ok:
partes.append(f"{n_ok} que ya cumplía(n) y no se ha(n) tocado")
out = ui_html.banner("ok", f"✅ Hecho: {' y '.join(partes)}. Ahora todos cumplen. En cada tarjeta, aquí debajo, "
"ves el antes y el después de cada medida, cuánto ha cambiado y qué significa; las descargas están al final.") + out
# el aviso secundario (p. ej. pausas largas) va detrás del principal
first_end = out.find("") + len("") if out.startswith('