#!/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('