Upload 2 files

app.py

@@ -17,13 +17,13 @@ from ovos_tts_plugin_matxa_multispeaker_cat import MatxaCatalanTTSPlugin
 
 # Utilidades SRT/AD (tu script original)
 # - parsea SRT y construye pistas AD y mezcla, usando ffmpeg/pydub
-from tts_ad_from_srt import (
+from generate_tts import (
     parse_srt_ad_only,
     mix_segments_on_timeline,
     build_ad_track_from_srt,
     ffmpeg_extract_audio_mp4_to_mp3,
     mix_two_audios_simultaneous,
-    ffmpeg_mux_video_with_audio
+    ffmpeg_mux_video_with_audio,
 )
 
 APP_STARTED_AT = time.strftime("%Y-%m-%d %H:%M:%S")
@@ -219,6 +219,31 @@ def tts_from_long_text(
 
 # ---------- SRT → AD (+ mezcla y opcional MP4) ----------
 
+@app.post("/tts/srt_ad_audio")
+def tts_ad_audio_from_srt(
+    srt: UploadFile = File(..., description="Archivo .srt con líneas (AD): ..."),
+    voice: str = Form("central/grau"),
+    ad_format: str = Form("mp3"),
+):
+    """Genera solo la pista de audiodescripción (une_ad.mp3/une_ad.wav) a partir de un SRT."""
+    ad_format = _ensure_fmt(ad_format)
+
+    with tempfile.TemporaryDirectory(prefix="matxa_srt_ad_only_") as td:
+        td = Path(td)
+        srt_path = td / "input.srt"
+        srt_bytes = srt.file.read()
+        srt_path.write_bytes(srt_bytes)
+
+        ad_out = td / f"une_ad.{ad_format}"
+        _ = build_ad_track_from_srt(str(srt_path), output_path=str(ad_out), voice=voice)
+
+        data = ad_out.read_bytes()
+        content_type = "audio/wav" if ad_format == "wav" else "audio/mpeg"
+        filename = f"une_ad.{ad_format}"
+        headers = {"Content-Disposition": f'inline; filename="{filename}"'}
+        return Response(content=data, media_type=content_type, headers=headers)
+
+
 @app.post("/tts/srt")
 def tts_from_srt(
     srt: UploadFile = File(..., description="Archivo .srt con líneas (AD): ..."),

generate_tts.py

@@ -0,0 +1,293 @@
+import os
+import re
+import math
+import tempfile
+import subprocess
+from dataclasses import dataclass
+from typing import List, Optional, Tuple
+
+import numpy as np
+import soundfile as sf
+
+# TTS plugin Matxa (ONNX/OVOS)
+from ovos_tts_plugin_matxa_multispeaker_cat import MatxaCatalanTTSPlugin
+
+# MP3 (vía ffmpeg) con pydub
+from pydub import AudioSegment
+
+
+@dataclass
+class Segment:
+    idx: int
+    start_s: float
+    end_s: float
+    text: str  # ya sin "(AD): "
+
+
+SRT_TS = re.compile(
+    r"(?P<h1>\d{2}):(?P<m1>\d{2}):(?P<s1>\d{2}),(?P<ms1>\d{3})\s*-->\s*"
+    r"(?P<h2>\d{2}):(?P<m2>\d{2}):(?P<s2>\d{2}),(?P<ms2>\d{3})"
+)
+
+
+def _ts_to_seconds(h: str, m: str, s: str, ms: str) -> float:
+    return int(h) * 3600 + int(m) * 60 + int(s) + int(ms) / 1000.0
+
+
+def _is_empty_ad_text(t: str) -> bool:
+    """Devuelve True si el texto de AD está vacío de contenido (solo espacios/puntuación)."""
+    cleaned = re.sub(r"[^\wÀ-ÿ]", "", t, flags=re.UNICODE)
+    return len(cleaned.strip()) == 0
+
+
+def parse_srt_ad_only(path: str) -> List[Segment]:
+    """Devuelve sólo segmentos cuyo bloque contiene líneas de AD con contenido."""
+    with open(path, "r", encoding="utf-8") as f:
+        content = f.read()
+
+    content = content.replace("\r\n", "\n").replace("\r", "\n")
+    blocks = [b.strip() for b in re.split(r"\n\s*\n", content) if b.strip()]
+
+    segs: List[Segment] = []
+    for block in blocks:
+        lines = block.split("\n")
+        if len(lines) < 2:
+            continue
+        try:
+            idx = int(lines[0].strip())
+            ts_line = lines[1].strip()
+            m = SRT_TS.match(ts_line)
+            if not m:
+                continue
+            start_s = _ts_to_seconds(m["h1"], m["m1"], m["s1"], m["ms1"])
+            end_s = _ts_to_seconds(m["h2"], m["m2"], m["s2"], m["ms2"])
+
+            ad_texts = []
+            for t in lines[2:]:
+                t = t.strip()
+                if (t.startswith("(AD):") or t.startswith("[AD]:") or
+                    t.startswith("(AD)") and not t.startswith("(AD):") or
+                    t.startswith("[AD]") and not t.startswith("[AD]:")):
+
+                    if t.startswith("(AD):"):
+                        t = t[len("(AD):"):].lstrip()
+                    elif t.startswith("[AD]:"):
+                        t = t[len("[AD]:"):].lstrip()
+                    elif t.startswith("(AD)"):
+                        t = t[len("(AD)"):].lstrip()
+                    elif t.startswith("[AD]"):
+                        t = t[len("[AD]"):].lstrip()
+
+                    if t and not _is_empty_ad_text(t):
+                        ad_texts.append(t)
+
+            if not ad_texts:
+                continue
+
+            text = " ".join(ad_texts)
+            segs.append(Segment(idx=idx, start_s=start_s, end_s=end_s, text=text))
+        except Exception:
+            continue
+
+    segs.sort(key=lambda s: (s.start_s, s.idx))
+    return segs
+
+
+def tts_to_wav(
+    text: str,
+    out_path: str,
+    voice: str = "central/grau",
+    tts: Optional[MatxaCatalanTTSPlugin] = None,
+) -> Tuple[int, np.ndarray]:
+    created_tts = tts is None
+    if tts is None:
+        tts = MatxaCatalanTTSPlugin()
+
+    tts.get_tts(text, out_path, voice=voice)
+
+    data, sr = sf.read(out_path, dtype="float32", always_2d=False)
+    if created_tts:
+        del tts
+    if data.ndim == 2:
+        data = data.mean(axis=1)
+    return sr, data
+
+
+def trim_or_pad_to_duration(data: np.ndarray, sr: int, target_sec: float) -> np.ndarray:
+    target_len = int(round(target_sec * sr))
+    cur_len = len(data)
+    if cur_len > target_len:
+        return data[:target_len]
+    elif cur_len < target_len:
+        pad = np.zeros(target_len - cur_len, dtype=data.dtype)
+        return np.concatenate([data, pad])
+    return data
+
+
+def _resample_np(x: np.ndarray, sr_from: int, sr_to: int) -> np.ndarray:
+    if sr_from == sr_to:
+        return x
+    ratio = sr_to / sr_from
+    new_len = int(round(len(x) * ratio))
+    xp = np.linspace(0, 1, num=len(x), endpoint=False)
+    fp = x
+    xq = np.linspace(0, 1, num=new_len, endpoint=False)
+    yq = np.interp(xq, xp, fp).astype(np.float32)
+    return yq
+
+
+def mix_segments_on_timeline(
+    segments: List[Segment],
+    voice: str,
+    out_final: str,
+    target_sr: Optional[int] = None,
+) -> str:
+    """Genera un master de todos los segmentos AD, colocándolos en su timestamp SRT."""
+    if not segments:
+        raise ValueError("No hay segmentos (AD) con contenido en el SRT.")
+
+    total_dur = max(s.end_s for s in segments)
+
+    tts = MatxaCatalanTTSPlugin()
+
+    tmpdir = tempfile.mkdtemp(prefix="matxa_ad_")
+    tmp_clips: List[Tuple[int, np.ndarray, float, float]] = []
+
+    for seg in segments:
+        seg_wav = os.path.join(tmpdir, f"ad_{seg.idx}.wav")
+        sr, data = tts_to_wav(seg.text, seg_wav, voice=voice, tts=tts)
+        seg_dur = seg.end_s - seg.start_s
+        data = trim_or_pad_to_duration(data, sr, seg_dur)
+        tmp_clips.append((sr, data, seg.start_s, seg.end_s))
+
+    master_sr = target_sr or tmp_clips[0][0]
+    master_len = int(round(total_dur * master_sr))
+    master = np.zeros(master_len, dtype=np.float32)
+
+    for sr, data, start_s, _ in tmp_clips:
+        d = _resample_np(data, sr, master_sr)
+        start_i = int(round(start_s * master_sr))
+        end_i = start_i + len(d)
+        if end_i > len(master):
+            end_i = len(master)
+            d = d[: end_i - start_i]
+        master[start_i:end_i] += d
+
+    peak = np.max(np.abs(master)) if master.size else 0.0
+    if peak > 0.999:
+        master = (master / peak * 0.98).astype(np.float32)
+
+    base, ext = os.path.splitext(out_final)
+    if ext.lower() == ".mp3":
+        tmp_wav = base + ".__tmp_master__.wav"
+        sf.write(tmp_wav, master, master_sr, subtype="PCM_16")
+        au = AudioSegment.from_wav(tmp_wav)
+        au.export(out_final, format="mp3")
+        os.remove(tmp_wav)
+        return out_final
+    else:
+        out_wav = base + ".wav" if ext.lower() != ".wav" else out_final
+        sf.write(out_wav, master, master_sr, subtype="PCM_16")
+        return out_wav
+
+
+def ffmpeg_extract_audio_mp4_to_mp3(mp4_path: str, out_mp3_path: str, bitrate: str = "192k") -> str:
+    """Extrae el audio del MP4 y lo guarda como MP3 (requiere ffmpeg)."""
+    cmd = [
+        "ffmpeg",
+        "-y",
+        "-i",
+        mp4_path,
+        "-vn",
+        "-acodec",
+        "libmp3lame",
+        "-b:a",
+        bitrate,
+        out_mp3_path,
+    ]
+    subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    return out_mp3_path
+
+
+def mix_two_audios_simultaneous(
+    mp3_a_path: str,
+    mp3_b_path: str,
+    out_mp3_path: str,
+    normalise: bool = True,
+) -> str:
+    """Mezcla simultáneamente dos MP3 (p.ej., audio original + AD) y exporta un MP3."""
+    a = AudioSegment.from_file(mp3_a_path)
+    b = AudioSegment.from_file(mp3_b_path)
+
+    max_len = max(len(a), len(b))
+    if len(a) < max_len:
+        a = a.append(AudioSegment.silent(duration=max_len - len(a)), crossfade=0)
+    if len(b) < max_len:
+        b = b.append(AudioSegment.silent(duration=max_len - len(b)), crossfade=0)
+
+    mixed = a.overlay(b)
+
+    if normalise:
+        peak = mixed.max_dBFS
+        headroom = -1.0
+        gain = headroom - peak
+        mixed = mixed.apply_gain(gain)
+
+    mixed.export(out_mp3_path, format="mp3")
+    return out_mp3_path
+
+
+def ffmpeg_mux_video_with_audio(video_mp4: str, audio_mp3: str, out_mp4: str) -> str:
+    """Crea un MP4 con el vídeo mudo del original y la pista de audio proporcionada."""
+    cmd = [
+        "ffmpeg",
+        "-y",
+        "-i",
+        video_mp4,
+        "-i",
+        audio_mp3,
+        "-map",
+        "0:v:0",
+        "-map",
+        "1:a:0",
+        "-c:v",
+        "copy",
+        "-shortest",
+        out_mp4,
+    ]
+    subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    return out_mp4
+
+
+def build_ad_track_from_srt(srt_path: str, output_path: str = "ad_master.mp3", voice: str = "central/grau") -> str:
+    segs = parse_srt_ad_only(srt_path)
+    if not segs:
+        from pydub import AudioSegment as _AS
+
+        silence = _AS.silent(duration=1000)
+        silence.export(output_path, format="mp3" if output_path.endswith(".mp3") else "wav")
+        return output_path
+
+    result = mix_segments_on_timeline(segs, voice=voice, out_final=output_path)
+    return result
+
+
+def make_final_assets_from_video_and_srt(
+    video_mp4: str,
+    srt_path: str,
+    out_ad_mp3: str = "ad_master.mp3",
+    out_mix_mp3: str = "mix_original_plus_ad.mp3",
+    out_final_mp4: str = "video_con_ad.mp4",
+    voice: str = "upc_ona-medium",
+) -> Tuple[str, str, str]:
+    """Pipeline completo: AD, extracción de audio, mezcla y MP4 final."""
+    ad_mp3 = build_ad_track_from_srt(srt_path, output_path=out_ad_mp3, voice=voice)
+
+    ori_mp3 = ffmpeg_extract_audio_mp4_to_mp3(
+        video_mp4,
+        out_mp3_path=os.path.splitext(out_ad_mp3)[0] + "_original.mp3",
+    )
+    mix_mp3 = mix_two_audios_simultaneous(ori_mp3, ad_mp3, out_mix_mp3)
+
+    final_mp4 = ffmpeg_mux_video_with_audio(video_mp4, mix_mp3, out_final_mp4)
+    return ad_mp3, mix_mp3, final_mp4