"""Estudio de voz local: grabaciones, importaciones y Piper opcional. Los filtros reproducen el preset de generate-parasite-voice.py. Ninguna frase ni grabación se envía a un servicio externo. FFmpeg recibe argumentos fijos. """ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse import argparse import importlib.util import json import math import os import shutil import subprocess import sys import tempfile import threading import uuid import webbrowser # Rutas y límites. Los audios de trabajo desaparecen al cerrar el estudio. HERE = Path(__file__).resolve().parent TOOLS = Path.home() / "AppData/Local/CodexTools" VOICE_ROOT = TOOLS / "RapsVoice" FFMPEG = TOOLS / "ffmpeg/ffmpeg-9.0-full_build-shared/bin/ffmpeg.exe" FFPROBE = FFMPEG.with_name("ffprobe.exe") ENGINE = HERE / "generate-parasite-voice.py" if not ENGINE.exists(): ENGINE = HERE.parent / "generate-parasite-voice.py" MAX_UPLOAD = 32 * 1024 * 1024 MAX_SECONDS = 180 NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0) # .opus usa el contenedor Ogg; se permite su demultiplexor y no protocolos de red. AUDIO_FORMATS = "wav,mp3,ogg,mov,mp4,m4a,3gp,3g2,mj2,matroska,webm,flac,aac" def run_ffmpeg(*args, timeout=150): subprocess.run([str(FFMPEG), "-nostdin", "-y", "-v", "error", "-threads", "1", "-filter_threads", "1", "-filter_complex_threads", "1", *map(str, args)], check=True, capture_output=True, timeout=timeout, creationflags=NO_WINDOW) def duration(path): result = subprocess.run([str(FFPROBE), "-v", "error", "-protocol_whitelist", "file,pipe", "-format_whitelist", AUDIO_FORMATS, "-show_entries", "format=duration", "-of", "json", str(path)], check=True, capture_output=True, text=True, timeout=12, creationflags=NO_WINDOW) value = float(json.loads(result.stdout)["format"]["duration"]) if not math.isfinite(value) or value <= 0 or value > MAX_SECONDS: raise ValueError("Usá un audio de hasta tres minutos.") return value # Tres capas: voz grave, segunda garganta y aire. El preset inicial conserva # las proporciones del audio de la criatura que ya utiliza el mod. def creature_filter(pitch=.77, texture=1.0, echo=1.0): throat_pitch = .67 * pitch / .77 return ( f"[0:a]rubberband=pitch={pitch:.6f}:formant=shifted," "highpass=f=62,lowpass=f=6800,asplit=3[main][throat][air];" "[main]acompressor=threshold=.18:ratio=3:attack=12:release=100,volume=.87[m];" f"[throat]rubberband=pitch={throat_pitch:.6f}:formant=shifted,lowpass=f=820," f"volume={.22 * texture:.6f},adelay=21:all=1[t];" "[air]highpass=f=1300,lowpass=f=4600,tremolo=f=31:d=.55," f"volume={.14 * texture:.6f},adelay=47:all=1[a];" "[m][t][a]amix=inputs=3:normalize=0," f"aecho=.8:.6:127|269:{.12 * echo:.6f}|{.055 * echo:.6f}," "alimiter=limit=.92:level=false" ) def settings(raw): if not isinstance(raw, dict) or set(raw) - {"pitch", "texture", "echo"}: raise ValueError("Ajustes inválidos.") limits = {"pitch": (.60, .95, .77), "texture": (0, 1.5, 1), "echo": (0, 1.5, 1)} result = {} for key, (low, high, default) in limits.items(): value = raw.get(key, default) if isinstance(value, bool) or not isinstance(value, (int, float)): raise ValueError("Ajustes inválidos.") if not math.isfinite(value) or not low <= value <= high: raise ValueError("El ajuste está fuera de su rango.") result[key] = value return result def render(source, folder, options): # Las tomas WebM del micrófono suelen omitir su duración. Decodificar primero # produce un WAV medible; el segundo adicional detecta tomas demasiado largas. decoded = folder / "entrada-normalizada.wav" run_ffmpeg("-protocol_whitelist", "file,pipe", "-format_whitelist", AUDIO_FORMATS, "-i", source, "-t", MAX_SECONDS + 1, "-ac", "1", "-ar", "32000", "-c:a", "pcm_s16le", decoded) seconds = duration(decoded) # WAV conserva calidad para edición; OGG entra en Minecraft; MP3 se comparte. wav = folder / "criatura.wav" run_ffmpeg("-protocol_whitelist", "file,pipe", "-format_whitelist", AUDIO_FORMATS, "-i", decoded, "-filter_complex", creature_filter(**options), "-ac", "1", "-ar", "32000", "-c:a", "pcm_s16le", wav) run_ffmpeg("-i", wav, "-c:a", "libvorbis", "-q:a", "5", folder / "criatura.ogg") run_ffmpeg("-i", wav, "-c:a", "libmp3lame", "-q:a", "3", folder / "criatura.mp3") return {"seconds": round(seconds, 2), "settings": options} # Piper se carga sólo al pedir texto, en un proceso que termina tras la frase. def synthesis_available(): return ENGINE.exists() and (VOICE_ROOT / "es_ES-carlfm-x_low.onnx").exists() \ and importlib.util.find_spec("piper") is not None def synthesize_file(text_file, target): # Evitar que la carga inicial de Piper abra un grupo de hilos por cada núcleo. import onnxruntime original_session = onnxruntime.InferenceSession def single_thread_session(*args, **kwargs): if kwargs.get("sess_options") is None: options = onnxruntime.SessionOptions() options.intra_op_num_threads = 1 options.inter_op_num_threads = 1 kwargs["sess_options"] = options return original_session(*args, **kwargs) onnxruntime.InferenceSession = single_thread_session spec = importlib.util.spec_from_file_location("creature_voice_engine", ENGINE) engine = importlib.util.module_from_spec(spec) spec.loader.exec_module(engine) engine.synthesize(Path(text_file).read_text(encoding="utf-8"), Path(target), 1.12) class Studio(ThreadingHTTPServer): daemon_threads = True request_queue_size = 4 def __init__(self, address): self.temp = tempfile.TemporaryDirectory(prefix="raps-voz-") self.work = Path(self.temp.name) self.busy = threading.Lock() self.synthesis = synthesis_available() super().__init__(address, Handler) def server_close(self): super().server_close() self.temp.cleanup() class Handler(BaseHTTPRequestHandler): def send_bytes(self, data, content_type, status=200, download=None): self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-store") self.send_header("X-Content-Type-Options", "nosniff") if download: self.send_header("Content-Disposition", 'attachment; filename="' + download + '"') self.end_headers() self.wfile.write(data) def send_json(self, value, status=200): self.send_bytes(json.dumps(value, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8", status) def same_origin(self): port = self.server.server_port return self.headers.get("Origin") in (None, f"http://127.0.0.1:{port}", f"http://localhost:{port}") def do_GET(self): route = urlparse(self.path).path if route == "/api/health": self.send_json({"studio": "raps-voice", "synthesis": self.server.synthesis}) return files = {"/": ("index.html", "text/html; charset=utf-8"), "/app.js": ("app.js", "text/javascript; charset=utf-8")} if route in files: name, mime = files[route] self.send_bytes((HERE / name).read_bytes(), mime) return parts = route.strip("/").split("/") if len(parts) == 3 and parts[0] == "audio" and len(parts[1]) == 32 \ and all(c in "0123456789abcdef" for c in parts[1]) \ and parts[2] in {"criatura.wav", "criatura.ogg", "criatura.mp3"}: path = self.server.work / parts[1] / parts[2] if path.is_file(): types = {"wav": "audio/wav", "ogg": "audio/ogg", "mp3": "audio/mpeg"} self.send_bytes(path.read_bytes(), types[path.suffix[1:]], download=path.name if urlparse(self.path).query == "download" else None) return self.send_json({"error": "No se encontró el archivo."}, 404) def do_POST(self): if not self.same_origin(): self.send_json({"error": "Origen inválido."}, 403) return route = urlparse(self.path).path if route == "/api/shutdown": self.send_json({"closed": True}) threading.Thread(target=self.server.shutdown, daemon=True).start() return if route not in {"/api/process", "/api/synthesize"}: self.send_json({"error": "Ruta inválida."}, 404) return if not self.server.busy.acquire(blocking=False): self.send_json({"error": "Hay otro audio en preparación. Esperá a que termine."}, 409) return folder = None completed = False try: self.connection.settimeout(15) size = int(self.headers.get("Content-Length", "0")) if size <= 0 or size > MAX_UPLOAD: raise ValueError("Usá un archivo de hasta 32 MB.") folder_id = uuid.uuid4().hex folder = self.server.work / folder_id folder.mkdir() if route == "/api/process": options = settings(json.loads(self.headers.get("X-Voice-Settings", "{}"))) source = folder / "entrada.audio" source.write_bytes(self.rfile.read(size)) else: if not self.server.synthesis: raise ValueError("La síntesis local no está instalada.") if size > 4096: raise ValueError("La frase es demasiado larga.") body = json.loads(self.rfile.read(size)) text = body.get("text") if not isinstance(text, str) or not text.strip() or len(text) > 400 \ or any(ord(c) < 32 and c not in "\n\t" for c in text): raise ValueError("Escribí una frase de hasta 400 caracteres.") options = settings(body.get("settings", {})) text_file = folder / "frase.txt" text_file.write_text(text, encoding="utf-8") source = folder / "entrada.wav" subprocess.run([sys.executable, "-X", "utf8", str(HERE / "studio.py"), "--synthesize", str(text_file), str(source)], check=True, timeout=180, capture_output=True, creationflags=NO_WINDOW, env={**os.environ, "OMP_NUM_THREADS": "1", "OPENBLAS_NUM_THREADS": "1"}) result = render(source, folder, options) # No se acumulan tomas indefinidamente: conservar las últimas ocho. old = sorted((p for p in self.server.work.iterdir() if p.is_dir()), key=lambda p: p.stat().st_mtime) for stale in old[:-8]: shutil.rmtree(stale) result["files"] = {ext: f"/audio/{folder_id}/criatura.{ext}" for ext in ("wav", "ogg", "mp3")} completed = True self.send_json(result) except (ValueError, TypeError, KeyError, json.JSONDecodeError): self.send_json({"error": "Revisá el archivo, la frase y los ajustes. Usá audio de hasta tres minutos."}, 400) except subprocess.TimeoutExpired: self.send_json({"error": "El audio tardó demasiado. Probá una toma más corta."}, 504) except (subprocess.CalledProcessError, OSError): self.send_json({"error": "No se pudo preparar este audio. Probá WAV, MP3, OGG, OPUS o WebM."}, 422) finally: if folder is not None and not completed: shutil.rmtree(folder, ignore_errors=True) self.server.busy.release() def log_message(self, fmt, *args): # No registrar frases, grabaciones ni contenido de archivos. pass def main(): if len(sys.argv) == 4 and sys.argv[1] == "--synthesize": synthesize_file(sys.argv[2], sys.argv[3]) return parser = argparse.ArgumentParser() parser.add_argument("--port", type=int, default=8778) parser.add_argument("--open", action="store_true") args = parser.parse_args() if not FFMPEG.is_file() or not FFPROBE.is_file(): raise SystemExit("No se encontró FFmpeg instalado.") server = Studio(("127.0.0.1", args.port)) if args.open: webbrowser.open(f"http://127.0.0.1:{server.server_port}/") try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() if __name__ == "__main__": main()