app.py

#!/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"<div class='ad-update'>🆕 Hay una versión nueva ({v}). <a href='{DESKTOP_URL}'>Descárgala aquí</a> "
            "e instálala encima: tus ajustes se conservan.</div>")


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("</div>") + len("</div>") if out.startswith('<div class="ad-banner') else 0
    out = out[:first_end] + extra + out[first_end:]
    out += ui_html.deliverables_table(profiles.deliverables(prof, durs))
    return out


def _pairs_html(prof, paths) -> str:
    """Comprobación de parejas de pistas, solo en perfiles de diálogo a dos pistas."""
    if not prof.get("pistas_pareadas") or len(paths) < 2:
        return ""
    import pairs

    found, alone = pairs.find_pairs(paths)
    results = []
    for a, b in found:
        try:
            results.append(pairs.check_pair(a, b))
        except Exception as e:
            print(f"[PAREJAS] {a} / {b}: {e}")
    return ui_html.pairs_table(results, alone)


def _clear():
    """Borra los resultados anteriores para que no se confundan con los de los audios nuevos."""
    hidden = gr.update(visible=False, value=None)
    return ui_html.empty("Aquí aparecerán los resultados."), "", hidden, hidden


# ── Corregir ─────────────────────────────────────────────────────────────────

def _file_url(path: str) -> str:
    return "/gradio_api/file=" + Path(path).as_posix()


def _eta(seconds: float) -> str:
    if seconds < 60:
        return "menos de 1 min"
    return f"~{round(seconds / 60)} min"


def _correction_verdict(r, ok: bool, untouched: bool, modo: str) -> str:
    """Veredicto de una línea: ¿se puede mandar? ¿qué se ha cambiado?"""
    stuck = profiles.failed(r.after.rows) if r.after else []
    if not ok:
        fixable = [f for f in stuck if pipeline.FIX_FOR.get(f)]
        detail = ("Con el nivel elegido no se toca; con 🤖 Automático o 🥷 Ninja se intentaría arreglar."
                  if fixable and modo in ("formato", "permitido") else
                  "Esto no tiene arreglo automático: conviene volver a grabarlo.")
        return ui_html.verdict("bad", f"❌ No queda bien: {', '.join(stuck)}", detail)
    if untouched:
        return ui_html.verdict("ok", "✅ Listo para enviar · ya cumplía, no se ha tocado",
                               "El fichero descargado es idéntico a tu original.")
    words = list(dict.fromkeys(pipeline.FIX_WORDS.get(f, f) for f in sorted(r.fixes)))
    lv = pipeline.level_used(r.fixes)
    detail = f"Nivel que ha hecho falta: {pipeline.MODE_LABEL.get(lv, lv)}" if lv else ""
    if r.pulido:
        detail += (" · " if detail else "") + "con pulido automático (suena mejor)"
    return ui_html.verdict("ok", f"✅ Listo para enviar · se ha cambiado: {', '.join(words) or 'formato'}", detail)


def cb_correct(files, pid, preset, hp, ultra, escalate, use_dnsmos, use_gpu, mode, mono,
               cut_pauses, pause_max, events=False, progress=gr.Progress()):
    hidden = gr.update(visible=False)
    paths = _paths(files)
    if not paths:
        yield ui_html.empty("Primero sube uno o varios audios (paso 2)."), "", hidden, hidden
        return
    prof = _prof(pid, pause_max)
    settings = _settings(prof, preset, hp, ultra, escalate, use_dnsmos, use_gpu, mono, mode, events)
    settings.acortar_pausas = bool(cut_pauses)
    skey = repr(settings)
    modo = settings.modo
    yield ui_html.banner("info", f"⏳ Corrigiendo {len(paths)} audio(s)…"), "", hidden, hidden
    job = _new_job_dir()
    out_dir = job / "corregidos"
    cards, durs, items, outputs, all_rows = [], [], [], [], []
    n_ok = n_fixed = n_err = 0
    # Tiempo estimado: segundos de proceso por segundo de audio, medido con los audios ya hechos
    lens = [(m.duration if (m := pipeline.quick_meta(p)) else 0.0) for p in paths]
    t0, done_audio = time.time(), 0.0

    for i, p in enumerate(paths):
        name = Path(p).name
        rate = (time.time() - t0) / done_audio if done_audio > 0 else None
        left = f" · quedan {_eta(rate * sum(lens[i:]))}" if rate else ""

        def prog(f, desc, i=i, name=name, left=left):
            progress((i + f) / len(paths), desc=f"{name} · {desc}{left}")

        progress(i / len(paths), desc=f"{name} · revisando{left}")
        pre = _analyze(p, prof, pid)
        if pre.error:
            n_err += 1
            cards.append(ui_html.card(name, "", ("bad", "✗ ERROR"), [], error=pre.error))
            items.append({"nombre": name, "estado_inicial": "ERROR", "estado_final": "ERROR", "notas": [pre.error]})
        else:
            key = ("co", *_file_key(p), pid, float(pause_max), skey)
            r = _CACHE.get(key)
            if r is None or not (r.out_path and Path(r.out_path).is_file()):
                r = pipeline.correct(p, prof, settings, out_dir, progress=prog, before=pre)
                r.spec = None
                if not r.error and r.after is not None and r.after is not r.before and lens[i] <= 3600:
                    prog(0.97, "dibujando espectrogramas")
                    r.spec = spectro.make_png(p, r.out_path, job / f"espectro-{i + 1}.png")
                _CACHE[key] = r
                _cached(key, lambda: r)  # recorta la caché si se ha pasado del máximo
            if r.error or r.after is None or r.after.error:
                n_err += 1
                err = r.error or (r.after.error if r.after else "error desconocido")
                cards.append(ui_html.card(name, pre.meta.describe(), ("bad", "✗ ERROR"), [], error=err))
                items.append({"nombre": name, "estado_inicial": "NO CUMPLÍA", "estado_final": "ERROR", "notas": [err]})
            else:
                ok = r.after.ok
                durs.append((Path(r.out_path).name, r.after.meta.duration))
                outputs.append(r.out_path)
                untouched = r.after is r.before
                if ok and untouched:
                    n_ok += 1
                    badge = ("ok", "✓ YA CUMPLE · sin tocar")
                elif ok:
                    n_fixed += 1
                    badge = ("ok", "✓ CORREGIDO") if not r.before.ok else ("ok", "✓ PROCESADO")
                else:
                    n_err += 1
                    badge = ("bad", "✗ NO SE HA PODIDO DEJAR BIEN")
                all_rows += r.before.rows
                cards.append(ui_html.card(name, r.before.meta.describe() if untouched else
                                          f"{r.before.meta.describe()}  →  {r.after.meta.describe()}", badge,
                                          r.before.rows, None if untouched else r.after.rows, notes=r.notes,
                                          download=_file_url(r.out_path), m_before=r.before.m,
                                          m_after=None if untouched else r.after.m,
                                          meta_before=r.before.meta, meta_after=r.after.meta,
                                          pause_max=profiles.pause_max(prof),
                                          verdict_html=_correction_verdict(r, ok, untouched, modo),
                                          original=_file_url(p), events=r.eventos,
                                          spec_url=_file_url(r.spec) if getattr(r, "spec", None) else None))
                lv = pipeline.level_used(r.fixes) if not untouched else None
                items.append({
                    "nombre": Path(r.out_path).name, "duracion_s": r.after.meta.duration,
                    "estado_inicial": "CUMPLÍA" if r.before.ok else "NO CUMPLÍA",
                    "estado_final": ("YA CUMPLÍA (sin tocar)" if untouched else "CORREGIDO (ya cumple)") if ok
                    else "SIGUE SIN CUMPLIR",
                    "formato": r.before.meta.describe(), "antes": r.before.m, "despues": r.after.m,
                    "fallos": profiles.failed(r.before.rows), "notas": r.notes,
                    "nivel": pipeline.MODE_LABEL.get(lv, "sin tocar") if lv else "sin tocar",
                    "cambios": list(dict.fromkeys(pipeline.FIX_WORDS.get(f, f) for f in sorted(r.fixes)))
                    if not untouched else [],
                })
        done_audio += lens[i]
        files_update = gr.update(value=list(outputs), visible=bool(outputs)) if outputs else hidden
        rate = (time.time() - t0) / done_audio if done_audio > 0 else None
        left = f" · quedan {_eta(rate * sum(lens[i + 1:]))}" if rate and i + 1 < len(paths) else ""
        yield (_summary(prof, len(paths), n_ok, n_fixed, n_err, durs, "Corregidos", final=False, left=left),
               ui_html.glossary_all(all_rows) + "".join(reversed(cards)), hidden, files_update)

    # Informe + ZIP
    progress(0.99, desc="Preparando la descarga")
    rows = report.build_rows(items)
    resumen = [("Fecha", time.strftime("%d/%m/%Y %H:%M")), ("Archivos", str(len(paths))),
               ("Cumplen / corregidos", str(n_ok + n_fixed)), ("Con errores", str(n_err)),
               ("Duración total", fmt_duration(sum(d for _, d in durs)))]
    stamp = time.strftime("%Y%m%d-%H%M")
    xlsx = report.write_xlsx(job / f"informe-{stamp}.xlsx", rows, PROFILES[pid].get("nombre", pid), resumen)
    csvf = report.write_csv(job / f"informe-{stamp}.csv", rows)
    zip_path = job / f"audios-corregidos-{stamp}.zip"
    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as z:
        for f in outputs:
            z.write(f, Path(f).name)
        z.write(xlsx, xlsx.name)
        z.write(csvf, csvf.name)
    yield (_summary(prof, len(paths), n_ok, n_fixed, n_err, durs, "Corregidos", final=True,
                    extra=_pairs_html(prof, outputs) + ui_html.report_table(items)),
           ui_html.glossary_all(all_rows) + "".join(reversed(cards)),
           gr.update(value=str(zip_path), visible=True),
           gr.update(value=outputs + [str(xlsx), str(csvf)], visible=True))


# ── Probar un fragmento ──────────────────────────────────────────────────────

def _mel_power(mono, sr):
    import librosa  # carga diferida: solo hace falta en «Probar» y tarda en importarse

    return librosa.feature.melspectrogram(y=mono, sr=sr, n_fft=2048, hop_length=512, n_mels=128)


def _spec_png(powers, sr, path: Path, titles) -> str:
    """Espectrogramas mel con la misma referencia (el máximo del primero) para poder compararlos."""
    import librosa
    import librosa.display
    import matplotlib

    matplotlib.use("Agg")
    import matplotlib.pyplot as plt

    ref = max(float(np.max(powers[0])), 1e-12)
    fig, axes = plt.subplots(len(powers), 1, figsize=(11, 2.5 * len(powers)), squeeze=False)
    fig.patch.set_facecolor("#14141f")
    for ax, P, title in zip(axes[:, 0], powers, titles):
        S = librosa.power_to_db(np.maximum(P, 1e-20), ref=ref, top_db=None)
        img = librosa.display.specshow(S, sr=sr, hop_length=512, x_axis="time", y_axis="mel", ax=ax,
                                       cmap="magma", vmin=-90, vmax=0)
        ax.set_title(title, color="white", fontsize=10, loc="left")
        ax.set_xlabel("Tiempo", color="#ddd")
        ax.set_ylabel("Hz", color="#ddd")
        ax.tick_params(colors="#ddd", labelsize=8)
        cb = fig.colorbar(img, ax=ax, format="%+2.0f dB", pad=0.01)
        cb.ax.tick_params(colors="#ddd", labelsize=8)
    fig.tight_layout()
    fig.savefig(path, dpi=110, facecolor=fig.get_facecolor())
    plt.close(fig)
    return str(path)


def cb_load_preview(path):
    if not path:
        return None, gr.update(maximum=0, value=0), ""
    meta = read_meta(path)
    max_s = min(meta.duration, 600)
    data, sr, _ = read_audio(path, stop=max_s)
    mono = dsp.to_mono(data)[:, 0]
    if sr > 48000:  # el resumen visual no necesita más
        mono = dsp.resample(mono[:, None], sr, 48000)[:, 0]
        sr = 48000
    title = f"Espectrograma {'completo' if max_s >= meta.duration else 'de los primeros 10 min'} — elige el tramo a probar"
    png = _spec_png([_mel_power(mono, sr)], sr, _new_job_dir() / "vista.png", [title])
    return png, gr.update(maximum=max(0, int(meta.duration - 1)), value=0), f"**{meta.name}** · {meta.describe()}"


def cb_preview(path, start, length, preset, hp, ultra, use_dnsmos, use_gpu, progress=gr.Progress()):
    if not path:
        raise gr.Error("Sube primero un audio.")
    progress(0.05, desc="Leyendo")
    seg, sr, meta = read_audio(path, start=start, stop=start + length)
    if seg.shape[0] < sr * 0.5:
        raise gr.Error("El tramo elegido es demasiado corto.")
    x = dsp.highpass(seg, sr, float(hp) or None)
    progress(0.2, desc="Reduciendo ruido")
    y, info = dsp.denoise(x, sr, PRESETS[preset], keep_ultrasonic=ultra, use_gpu=use_gpu)
    progress(0.7, desc="Midiendo")
    job = _new_job_dir()
    from audio_io import write_audio

    orig_p, proc_p = job / "original.wav", job / "procesado.wav"
    write_audio(orig_p, seg, sr, "WAV", "FLOAT")
    write_audio(proc_p, y, sr, "WAV", "FLOAT")
    ma, mb = metrics.measure(seg, sr), metrics.measure(y, sr)
    if use_dnsmos:
        import dnsmos

        ma["dnsmos"] = dnsmos.score(dsp.to_mono(seg)[:, 0], sr)
        mb["dnsmos"] = dnsmos.score(dsp.to_mono(y)[:, 0], sr)
    rows_a = profiles.check(meta, ma, {})
    rows_b = profiles.check(meta, mb, {})
    notes = [f"Cuánto ruido quitar: {preset} · Retumbes: {dict((v, k) for k, v in HP_CHOICES).get(int(hp), hp)}"
             f" · Procesado en {info['dispositivo']}"] + info["avisos"]
    card = ui_html.card(f"Tramo {fmt_duration(start)} – {fmt_duration(start + seg.shape[0] / sr)}", meta.describe(),
                        ("info", "Prueba"), rows_a, rows_b, notes=notes)
    progress(0.85, desc="Dibujando espectrogramas")
    pa, pb = _mel_power(dsp.to_mono(seg)[:, 0], sr), _mel_power(dsp.to_mono(y)[:, 0], sr)
    png = _spec_png([pa, pb, np.maximum(pa - pb, 0)], sr, job / "comparativa.png",
                    ["Original", "Procesado", "Lo que se ha quitado"])
    return str(orig_p), str(proc_p), png, card


# ── Flujo automático, ajustes guardados ───────────────────────────────────────

def cb_upload(files, auto, pid, preset, hp, ultra, escalate, use_dnsmos, use_gpu, mode, mono, cut_pauses, pause_max,
              events, progress=gr.Progress()):
    """Al subir audios (o cambiar de proyecto): en Automático, todo de una vez; si no, solo la revisión."""
    hidden = gr.update(visible=False)
    if not _paths(files):
        yield ui_html.empty("Aquí aparecerán los resultados."), "", hidden, hidden
        return
    if auto:
        yield from cb_correct(files, pid, preset, hp, ultra, escalate, True, use_gpu, AUTO_MODE, mono, False,
                              pause_max, False, progress=progress)
    else:
        for summ, res in cb_analyze(files, pid, pause_max, progress=progress):
            yield summ, res, hidden, hidden


def _auto_view(auto: bool):
    on = bool(auto)
    return (gr.update(visible=not on), gr.update(visible=not on), gr.update(visible=on),
            STEPS_AUTO if on else STEPS)


SETTINGS_DEFAULT = {"perfil": DEFAULT_PROFILE, "auto": True, "dnsmos": True, "pausa": 2.0, "cortar": False,
                    "eventos": False}


def _restore(st):
    st = {**SETTINGS_DEFAULT, **(st or {})}
    pid = st["perfil"] if st["perfil"] in PROFILES else DEFAULT_PROFILE
    return (gr.update(value=pid), gr.update(value=bool(st["auto"])), gr.update(value=bool(st["dnsmos"])),
            gr.update(value=float(st["pausa"])), gr.update(value=bool(st["cortar"])),
            gr.update(value=bool(st["eventos"])), *_auto_view(st["auto"]), *_profile_info(pid), _update_banner())


def _save(pid, auto, dnsmos_on, pausa, cortar, eventos):
    return {"perfil": pid, "auto": bool(auto), "dnsmos": bool(dnsmos_on), "pausa": float(pausa),
            "cortar": bool(cortar), "eventos": bool(eventos)}


# ── Interfaz ─────────────────────────────────────────────────────────────────

STEPS = """
<div class="ad-steps">
  <div><b>1</b> Elige el proyecto</div>
  <div><b>2</b> Sube los audios</div>
  <div><b>3</b> Pulsa <i>Revisar</i></div>
  <div><b>4</b> Pulsa <i>Corregir y descargar</i></div>
</div>
"""
STEPS_AUTO = """
<div class="ad-steps">
  <div><b>1</b> Elige el proyecto</div>
  <div><b>2</b> Sube los audios</div>
  <div><b>3</b> ¡Listo! Mira el resultado y descarga</div>
</div>
"""

DESKTOP_HELP = f"""
## 💻 Versión para Windows

Si la usas a menudo, instálala en tu ordenador: **no hay que subir los audios a internet, no hay límites diarios** y funciona sin conexión.

1. Descarga el instalador: **[Instalar-Limpieza-de-voz.exe]({DESKTOP_URL})** (unos cientos de MB).
2. Ábrelo. Si Windows avisa de que es una aplicación desconocida, pulsa **Más información** → **Ejecutar de todas formas**.
3. Pulsa **Siguiente** hasta terminar. No hace falta ser administradora.
4. Abre **Limpieza de voz** desde el escritorio. Se abrirá una ventana negra (no la cierres mientras la uses) y, enseguida, la app en tu navegador, igual que esta.

¿No te deja instalar? Descarga la **[versión portable (ZIP)]({DESKTOP_ZIP_URL})**, descomprímela y haz doble clic en **Iniciar.bat**.
"""

LOGIN_HELP = ("**Antes de empezar:** entra con tu cuenta de Hugging Face (botón *Log in*, arriba en la página). "
              "Si no entras también funciona, pero puede ir más despacio.\n\n")

HELP = """
## Paso a paso (modo 🤖 Automático, el normal)

{login}1. **Elige el proyecto** en el desplegable de arriba. Debajo verás qué pide ese cliente.
2. **Sube los audios** en la pestaña *✅ Corregir*: arrástralos o pulsa para buscarlos. Puedes subir varios a la vez.
3. **Ya está.** La app los revisa y arregla **solo lo que haga falta** (lo mínimo). Cada audio aparece en una tarjeta con una frase arriba:
   - ✅ **Listo para enviar · ya cumplía, no se ha tocado**: el fichero es idéntico a tu original.
   - ✅ **Listo para enviar · se ha cambiado: …**: te dice qué se ha tocado y qué nivel ha hecho falta.
   - ❌ **No queda bien: …**: lo que falla y si tiene arreglo.
4. Escúchalo en la propia tarjeta (**🎧 Original** y **✨ Corregido**; el botón **⇄** cambia de uno a otro en el mismo punto) y descárgalo: botón de cada tarjeta o **⬇️ Descargar todo** (ZIP con los audios, **con el mismo nombre que tenían**, y un informe en Excel).

Si cambias de proyecto con los audios ya subidos, se vuelven a procesar solos. **🔄 Volver a procesar** repite todo.

**¿Quieres decidir tú?** Desmarca **🤖 Automático**: aparecen el nivel de procesado, las demás opciones y los botones **🔍 Revisar** y **✨ Corregir y descargar**. Al subir audios, entonces solo se revisan (no se cambia nada hasta que pulses Corregir).

La app recuerda el proyecto y tus ajustes para la próxima vez.

## Cómo se leen las tarjetas

Cada fila es una comprobación. Las columnas son **Antes** (tu audio), **Después** (el corregido) y **Requisito** (lo que pide el cliente). **OK** significa que cumple y **Falla** que no. Las filas grises con **info** son solo informativas: no cuentan para aprobar.

| Fila | Qué significa |
|---|---|
| Formato / Codificación / Frecuencia / Canales | El tipo de fichero. Se arregla solo al corregir. |
| Pico máximo | El momento más fuerte del audio. Si se pasa, puede sonar saturado. |
| RMS de voz | El volumen medio de la voz. |
| Ruido relativo · SNR | Cuánto ruido de fondo hay comparado con la voz. |
| Silencio inicial / final | Lo que dura el silencio antes de la primera palabra y después de la última. |
| Saturación | Muestras que «revientan» (distorsión). Tiene que ser 0. |
| Calidad percibida | Nota de 1 a 5 de cómo sonaría a un oyente. Por encima de 3,5 suena muy bien. |

Debajo de cada tarjeta, **«Qué se ha hecho»** resume los arreglos aplicados a ese audio. El botón **⬇️ Descargar este audio** de cada tarjeta descarga ese audio suelto; el botón grande **⬇️ Descargar todo** baja el ZIP completo cuando terminan todos.

## Nivel de procesado (📏 🎚️ 🥷), con Automático desmarcado

Tú eliges cuánto se toca el audio. Cada proyecto arranca con un nivel, pero puedes cambiarlo antes de pulsar «Corregir». **🤖 Automático** equivale a usar el nivel más bajo que lo arregle, sin pulido.

- **📏 Solo formato:** bits, canales y frecuencia. **El sonido no cambia.**
- **🎚️ Permitido:** lo anterior y, además, el volumen, los picos y los silencios del principio y del final.
- **🥷 Ninja:** todo lo anterior y, además, limpieza sutil de ruido de fondo y retumbes. Incluye el **pulido automático**: la app prueba a limpiar y **solo se queda con el resultado si suena mejor** (lo mide la nota de calidad). Si no mejora, se descarta.

Lo que falle y no se pueda arreglar con el nivel elegido aparece en «Qué se ha hecho», con la indicación de que se arreglaría subiendo el nivel.

## Quitar ruidos sueltos en las pausas (🔇)

Para cláxones, golpes, puertas… que suenan **mientras nadie habla**. Un detector de voz con inteligencia artificial distingue la voz de esos ruidos; cada ruido se sustituye por el silencio de fondo de la propia grabación, y la voz no se toca. En «Qué se ha hecho» verás en qué minuto estaba cada uno, para que lo escuches.

- Viene **desmarcado**. Márcalo solo cuando lo necesites.
- **No lo uses con risas, llantos o muletillas** («mmm», «eh»): el detector no los considera voz y los quitaría.
- Si el ruido suena **a la vez que la voz**, esto no lo quita (en «Ninja» se atenúa con la limpieza de ruido).

## Espectrogramas (🔬)

En cada audio corregido, **«🔬 Espectrogramas»** enseña cómo era el sonido antes, cómo ha quedado y la diferencia (azul = lo que se ha quitado, rojo = lo que se ha añadido). Es para curiosos: no hace falta mirarlo.

## Nota de calidad (⭐)

Viene **marcada**. Cada tarjeta muestra una nota de 1 a 5 **antes** y **después**: cómo valoraría el audio una persona escuchándolo (por encima de 3,5 suena muy bien). Se separa en **voz** (¿suena natural, sin distorsión?) y **fondo** (¿se oye ruido?). Hace que cada audio tarde más; desmárcala si tienes prisa.

## Acortar pausas largas (✂️)

Al revisar, cada tarjeta indica si el audio tiene **pausas largas** entre frases, por ejemplo «2 · la más larga 4,3 s». Si quieres acortarlas, marca **«Acortar pausas largas»** y elige la **pausa máxima** (2 segundos por defecto) antes de pulsar «Corregir». Solo se acortan las pausas que pasan de esa duración: se quita el trozo del medio y se conservan el principio y el final de la pausa, con un fundido para que el corte no se note. Las pausas normales no se tocan.

## Antes de corregir muchos audios, prueba uno

En la pestaña **🎧 Probar** subes un audio, eliges un trozo y pulsas **▶ Probar este tramo**. Oirás el original y el limpio, uno al lado del otro.

- Si notas la voz apagada o rara, abre **Opciones avanzadas** y elige una reducción de ruido **Suave**.
- Si todavía se oye ruido, elige **Fuerte**.

Esa elección se usa después al corregir.

## Opciones avanzadas

Normalmente **no hace falta tocarlas**: cada proyecto ya trae las suyas.

- **Cuánto ruido quitar**: *Natural* es lo recomendado. *Suave* deja algo de ambiente; *Fuerte* y *Máximo* son para grabaciones con mucho ruido.
- **Quitar retumbes y golpes graves**: elimina golpes de vecinos, de la mesa o del pie del micro.
- **Convertir a mono**: si no lo marcas, el audio sale con los mismos canales con los que entró.

## Si algo no va bien

- **Tarda mucho:** es normal con audios largos. No cierres la página; la barra indica el avance.
- **«No se ha podido dejar bien»:** revisa en la tarjeta qué fila falla. Suele ser ruido muy fuerte en la grabación. Prueba con **Fuerte** o **Máximo** en Opciones avanzadas o, si no basta, conviene volver a grabar esa parte.
- **ERROR al abrir un audio:** expórtalo de nuevo desde Audacity como **WAV, 24 bits, 48000 Hz**.

## Consejos para grabar en Audacity

- Pon la frecuencia del proyecto en **48000 Hz**.
- Al exportar elige **WAV** y **PCM 24 bits**.
- Deja **un segundo de silencio** antes de empezar a hablar y otro al terminar.
- Que el volumen no llegue nunca al rojo del medidor.
"""

with gr.Blocks(title="Limpieza y control de voz", delete_cache=(3600, 86400)) as app:
    with gr.Row(equal_height=True, elem_id="cabecera-fila"):
        gr.HTML("<div id='cabecera'><h1>🎙️ Limpieza y control de voz</h1>"
                "<p>Quita el ruido de tus grabaciones y comprueba que cumplen lo que pide el cliente.</p></div>")
        theme_btn = gr.Button("🌓", elem_id="tema-btn", scale=0, min_width=52, variant="secondary")
    theme_btn.click(None, js=ui_html.TOGGLE_JS)
    update_html = gr.HTML()
    steps = gr.HTML(STEPS_AUTO)
    if ON_SPACE:
        gr.HTML(f"<div class='ad-desktop'>💻 ¿Lo usas a menudo? <a href='{DESKTOP_URL}'>Descarga la versión para Windows</a>: "
                "va en tu ordenador, sin esperas ni límites y sin subir los audios a internet.</div>")

    with gr.Row():
        profile_dd = gr.Dropdown(PROFILE_CHOICES, value=DEFAULT_PROFILE, label="1 · Proyecto", scale=3)
    profile_md = gr.Markdown(elem_classes="ad-profile")
    auto = gr.Checkbox(value=True, label=AUTO_LABEL, info=AUTO_INFO, elem_id="auto-toggle")
    settings_state = gr.BrowserState(SETTINGS_DEFAULT, storage_key="limpieza-voz-ajustes",
                                     secret="limpieza-de-voz-ajustes-v1")

    with gr.Accordion("⚙️ Opciones avanzadas (normalmente no hace falta tocarlas)", open=False,
                      visible=False) as adv:
        preset = gr.Radio(list(PRESETS), value=DEFAULT_PRESET, label="Cuánto ruido quitar",
                          info="Natural deja la voz limpia sin cambiar su timbre.")
        with gr.Row():
            hp = gr.Dropdown(HP_CHOICES, value=70, label="Quitar retumbes y golpes graves",
                             info="Golpes de vecinos, de la mesa, del pie del micro…")
            mono = gr.Checkbox(value=False, label="Convertir a mono",
                               info="Si no lo marcas, sale con los mismos canales que entró.")
        escalate = gr.Checkbox(value=True, label="Si sigue habiendo ruido, insistir automáticamente")
        use_gpu = gr.Checkbox(value=True, label="Usar la tarjeta gráfica si está libre (solo audios largos)",
                              visible=gpu_available())
        ultra = gr.Checkbox(value=True, label="Conservar agudos extremos (solo si grabas a 96 kHz)")

    with gr.Tabs():
        with gr.Tab("✅ Corregir"):
            files = gr.File(label="2 · Sube aquí los audios (puedes arrastrar varios a la vez)",
                            file_count="multiple", file_types=AUDIO_EXTENSIONS)
            rerun_btn = gr.Button("🔄 Volver a procesar", variant="secondary", size="sm")
            with gr.Column(visible=False) as manual:
              with gr.Row(elem_classes="ad-quality"):
                mode = gr.Radio(list(MODE_CHOICES), value="🥷 Ninja", label="Nivel de procesado", info=MODE_INFO,
                                scale=2)
                use_dnsmos = gr.Checkbox(value=True, label=QUALITY_LABEL, info=QUALITY_INFO)
              with gr.Row(elem_classes="ad-quality"):
                cut_pauses = gr.Checkbox(value=False, label="✂️ Acortar pausas largas", scale=2,
                                         info="Recorta las pausas entre frases que pasen de la duración máxima. "
                                              "Conserva el principio y el final de cada pausa para que suene natural.")
                pause_max = gr.Slider(0.5, 5.0, value=2.0, step=0.1, label="Pausa máxima (segundos)", scale=1)
              with gr.Row(elem_classes="ad-quality"):
                events = gr.Checkbox(value=False, label=EVENTS_LABEL, info=EVENTS_INFO)
              with gr.Row():
                analyze_btn = gr.Button("🔍 3 · Revisar", variant="secondary")
                correct_btn = gr.Button("✨ 4 · Corregir y descargar", variant="primary")
            summary = gr.HTML(ui_html.empty("Aquí aparecerán los resultados."))
            results = gr.HTML()
            zip_btn = gr.DownloadButton("⬇️ Descargar todo (ZIP con los audios y el informe Excel)", variant="primary",
                                        size="lg", visible=False, elem_id="zip-btn")
            downloads = gr.File(label="Descargas sueltas: cada audio corregido y el informe",
                                file_count="multiple", interactive=False, visible=False)

        with gr.Tab("🎧 Probar"):
            gr.Markdown("Escucha cómo queda un trozo antes de corregir todo. **No gasta nada ni cambia tus audios.**")
            p_audio = gr.Audio(label="Audio para probar", type="filepath", sources=["upload", "microphone"])
            p_info = gr.Markdown()
            p_overview = gr.Image(label="Vista del audio", type="filepath", interactive=False, buttons=["fullscreen"],
                                  elem_classes="ad-img")
            with gr.Row():
                p_start = gr.Slider(0, 600, value=0, step=1, label="Empezar en el segundo…")
                p_len = gr.Slider(5, 60, value=15, step=5, label="Durante (segundos)")
            p_dnsmos = gr.Checkbox(value=False, label="⭐ Poner nota de calidad (más lento)",
                                   info="Puntúa de 1 a 5 el antes y el después. Tarda bastante más.")
            p_btn = gr.Button("▶ Probar este tramo", variant="primary")
            with gr.Row():
                p_orig = gr.Audio(label="Antes", interactive=False)
                p_proc = gr.Audio(label="Después", interactive=False)
            p_card = gr.HTML()
            with gr.Accordion("Ver espectrogramas (para curiosos)", open=False):
                p_spec = gr.Image(label="Espectrogramas", type="filepath", interactive=False,
                                  buttons=["fullscreen", "download"], elem_classes="ad-img")

        with gr.Tab("❓ Ayuda"):
            gr.Markdown(HELP.format(login=LOGIN_HELP if ON_SPACE else ""))
            if ON_SPACE:
                gr.Markdown(DESKTOP_HELP)

    gr.HTML(f"<div class='ad-version'>Versión {VERSION}</div>")

    upload_inputs = [files, auto, profile_dd, preset, hp, ultra, escalate, use_dnsmos, use_gpu, mode, mono, cut_pauses,
                     pause_max, events]
    run_outputs = [summary, results, zip_btn, downloads]
    profile_dd.change(_profile_info, profile_dd, [profile_md, preset, hp, ultra, mode, mono]).then(
        cb_upload, upload_inputs, run_outputs)
    auto.change(_auto_view, auto, [manual, adv, rerun_btn, steps])
    app.load(_restore, settings_state,
             [profile_dd, auto, use_dnsmos, pause_max, cut_pauses, events, manual, adv, rerun_btn, steps,
              profile_md, preset, hp, ultra, mode, mono, update_html])
    gr.on([profile_dd.change, auto.change, use_dnsmos.change, pause_max.change, cut_pauses.change, events.change],
          _save, [profile_dd, auto, use_dnsmos, pause_max, cut_pauses, events], settings_state)
    files.change(_clear, None, run_outputs).then(cb_upload, upload_inputs, run_outputs)
    rerun_btn.click(cb_upload, upload_inputs, run_outputs)
    analyze_btn.click(cb_analyze, [files, profile_dd, pause_max], [summary, results])
    correct_btn.click(cb_correct, [files, profile_dd, preset, hp, ultra, escalate, use_dnsmos, use_gpu, mode, mono,
                       cut_pauses, pause_max, events], run_outputs)
    if not ON_SPACE:
        upd_timer = gr.Timer(8)
        upd_timer.tick(lambda: (_update_banner(), gr.Timer(active=False)), None, [update_html, upd_timer])
    p_audio.change(cb_load_preview, p_audio, [p_overview, p_start, p_info])
    p_btn.click(cb_preview, [p_audio, p_start, p_len, preset, hp, ultra, p_dnsmos, use_gpu],
                [p_orig, p_proc, p_spec, p_card])

app.queue(default_concurrency_limit=1)


def main(**launch_kwargs):
    WORK_DIR.mkdir(parents=True, exist_ok=True)
    app.launch(
        theme=gr.themes.Soft(primary_hue="teal", secondary_hue="slate"),
        css=ui_html.CSS,
        head=ui_html.HEAD,
        show_error=True,
        allowed_paths=[str(WORK_DIR)],  # para los botones «Descargar este audio»
        **launch_kwargs,
    )


if __name__ == "__main__":
    main()