mirror of
https://github.com/R0m1k3/Socialflow.git
synced 2026-10-22 22:48:09 +02:00
Compare commits
5
Commits
1d47763b29
...
4b8fbe6d8e
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4b8fbe6d8e | ||
|
|
7c38ec459f | ||
|
|
d4c544e467 | ||
|
|
44899055b2 | ||
|
|
f3113a0f5a |
No files matched your search
@@ -120,6 +120,13 @@ GEMINI_API_KEY=
|
||||
# sur celui-ci si le modèle choisi n'existe pas).
|
||||
# GEMINI_TTS_MODEL=
|
||||
#
|
||||
# Stabilité de la voix Gemini : température basse et graine fixe donnent la
|
||||
# même intonation d'une génération à l'autre ; chaque lecture est réécoutée et
|
||||
# régénérée (jusqu'à GEMINI_TTS_ATTEMPTS fois) si des mots manquent.
|
||||
# GEMINI_TTS_TEMPERATURE=0.3
|
||||
# GEMINI_TTS_SEED=20260928
|
||||
# GEMINI_TTS_ATTEMPTS=3
|
||||
#
|
||||
# Modèle Whisper utilisé pour caler les sous-titres mot à mot sur la voix
|
||||
# (tiny, base, small). Choisi à la construction de l'image : changer la valeur
|
||||
# demande de reconstruire ffmpeg-api.
|
||||
|
||||
@@ -144,6 +144,7 @@ export function ReelPreview(props: ReelPreviewProps) {
|
||||
storeName,
|
||||
logoStart: timing.logoStart,
|
||||
fadeStart: timing.fadeStart,
|
||||
fadeDuration: timing.fadeDuration,
|
||||
voiceUrl: srtCues ? undefined : voice?.audioUrl,
|
||||
voiceDelay: VOICE_DELAY,
|
||||
musicUrl,
|
||||
|
||||
@@ -694,7 +694,7 @@ export default function NewReel() {
|
||||
setSelectedTrack(null);
|
||||
setCurrentStep('text');
|
||||
}}>
|
||||
Passer
|
||||
Passer (garder le son de la vidéo)
|
||||
</Button>
|
||||
<Button onClick={() => setCurrentStep('text')}>
|
||||
Continuer
|
||||
@@ -1052,7 +1052,7 @@ export default function NewReel() {
|
||||
</div>
|
||||
<div className="flex justify-between">
|
||||
<span className="text-muted-foreground">Musique</span>
|
||||
<span>{selectedTrack ? selectedTrack.title : 'Aucune'}</span>
|
||||
<span>{selectedTrack ? selectedTrack.title : 'Son de la vidéo'}</span>
|
||||
</div>
|
||||
<div className="flex justify-between">
|
||||
<span className="text-muted-foreground">Texte</span>
|
||||
|
||||
@@ -17,6 +17,7 @@ export type ReelVideoProps = {
|
||||
storeName?: string;
|
||||
logoStart?: number | null;
|
||||
fadeStart?: number | null;
|
||||
fadeDuration?: number | null;
|
||||
/** Rendu final : piste son déjà mixée (voix, musique baissée sous la voix, niveau normalisé). */
|
||||
mixedAudioUrl?: string;
|
||||
/** Aperçu : pistes séparées, mixées dans le navigateur. */
|
||||
@@ -31,14 +32,21 @@ export const ReelVideo: React.FC<ReelVideoProps> = (props) => {
|
||||
const { fps } = useVideoConfig();
|
||||
const {
|
||||
videoUrl, videoDuration, totalDuration, words, captionStyle, logoUrl, showWatermark = true, storeName,
|
||||
logoStart = null, fadeStart = null, mixedAudioUrl, voiceUrl, voiceDelay = 2, musicUrl, musicVolume = 0.25,
|
||||
logoStart = null, fadeStart = null, fadeDuration = FADE_SECONDS, mixedAudioUrl, voiceUrl, voiceDelay = 2, musicUrl, musicVolume = 0.25,
|
||||
} = props;
|
||||
|
||||
const videoFrames = Math.max(1, Math.floor(videoDuration * fps));
|
||||
const totalFrames = Math.max(1, Math.round(totalDuration * fps));
|
||||
const hasAddedAudio = Boolean(mixedAudioUrl || voiceUrl || musicUrl);
|
||||
// Le son d'origine est remplacé par la musique choisie ; sans musique, il reste
|
||||
// en fond (baissé sous la voix). Au rendu final, il est déjà dans la piste mixée.
|
||||
const keepVideoSound = !mixedAudioUrl && !musicUrl;
|
||||
const video = (
|
||||
<OffthreadVideo src={videoUrl} muted={hasAddedAudio} style={{ width: "100%", height: "100%", objectFit: "cover" }} />
|
||||
<OffthreadVideo
|
||||
src={videoUrl}
|
||||
muted={!keepVideoSound}
|
||||
volume={keepVideoSound && voiceUrl ? (frame) => previewMusicVolume(frame / fps, words, 1, fadeStart, fadeDuration ?? FADE_SECONDS) : 1}
|
||||
style={{ width: "100%", height: "100%", objectFit: "cover" }}
|
||||
/>
|
||||
);
|
||||
|
||||
return (
|
||||
@@ -64,7 +72,7 @@ export const ReelVideo: React.FC<ReelVideoProps> = (props) => {
|
||||
<Html5Audio
|
||||
src={musicUrl}
|
||||
loop
|
||||
volume={(frame) => previewMusicVolume(frame / fps, words, musicVolume, fadeStart)}
|
||||
volume={(frame) => previewMusicVolume(frame / fps, words, musicVolume, fadeStart, fadeDuration ?? FADE_SECONDS)}
|
||||
/>
|
||||
)}
|
||||
|
||||
@@ -73,18 +81,24 @@ export const ReelVideo: React.FC<ReelVideoProps> = (props) => {
|
||||
{logoStart != null && (logoUrl || storeName) && (
|
||||
<Outro start={logoStart} logoUrl={logoUrl} storeName={storeName} />
|
||||
)}
|
||||
{fadeStart != null && <FadeOut start={fadeStart} duration={FADE_SECONDS} />}
|
||||
{fadeStart != null && <FadeOut start={fadeStart} duration={fadeDuration ?? FADE_SECONDS} />}
|
||||
</AbsoluteFill>
|
||||
);
|
||||
};
|
||||
|
||||
/** Aperçu : la musique baisse pendant la parole et s'éteint avec le fondu final. */
|
||||
function previewMusicVolume(time: number, words: TimedWord[], base: number, fadeStart: number | null): number {
|
||||
/** Aperçu : le fond sonore (musique ou son d'origine) baisse pendant la parole et s'éteint avec le fondu final. */
|
||||
function previewMusicVolume(
|
||||
time: number,
|
||||
words: TimedWord[],
|
||||
base: number,
|
||||
fadeStart: number | null,
|
||||
fadeDuration: number,
|
||||
): number {
|
||||
const speaking = words.some((w) => time >= w.start - 0.2 && time <= w.end + 0.3);
|
||||
const fade =
|
||||
fadeStart == null
|
||||
? 1
|
||||
: interpolate(time, [fadeStart, fadeStart + FADE_SECONDS], [1, 0], {
|
||||
: interpolate(time, [fadeStart, fadeStart + fadeDuration], [1, 0], {
|
||||
extrapolateLeft: "clamp",
|
||||
extrapolateRight: "clamp",
|
||||
});
|
||||
|
||||
@@ -30,6 +30,8 @@ ENV WHISPER_MODEL=${WHISPER_MODEL} \
|
||||
HOME=/tmp \
|
||||
XDG_CACHE_HOME=/tmp/.cache
|
||||
RUN python -c "from faster_whisper import WhisperModel; WhisperModel('${WHISPER_MODEL}', device='cpu', compute_type='int8')"
|
||||
# Modèle déjà présent : plus aucune requête vers Hugging Face au démarrage
|
||||
ENV HF_HUB_OFFLINE=1
|
||||
|
||||
COPY main.py .
|
||||
COPY app ./app
|
||||
|
||||
@@ -116,23 +116,36 @@ def _model():
|
||||
return WhisperModel(config.WHISPER_MODEL, device="cpu", compute_type=config.WHISPER_COMPUTE_TYPE)
|
||||
|
||||
|
||||
def _transcribe_sync(audio: Path, hint: str | None) -> list[Word]:
|
||||
# Silence ajouté avant la voix pour Whisper : sur une voix qui démarre
|
||||
# aussitôt, il laissait souvent passer le premier mot (« Alerte »).
|
||||
LEAD_SILENCE = 0.5
|
||||
|
||||
|
||||
def _transcribe_sync(audio: Path) -> list[Word]:
|
||||
import numpy as np
|
||||
from faster_whisper.audio import decode_audio
|
||||
|
||||
samples = decode_audio(str(audio), sampling_rate=16000)
|
||||
padded = np.concatenate([np.zeros(int(16000 * LEAD_SILENCE), dtype=samples.dtype), samples])
|
||||
# Pas de texte attendu en amorce (initial_prompt) ni d'enchaînement sur le
|
||||
# segment précédent : les deux font halluciner Whisper, qui répète la phrase
|
||||
# et masque une fin tronquée. On veut entendre ce qui est réellement dit.
|
||||
segments, _ = _model().transcribe(
|
||||
str(audio),
|
||||
padded,
|
||||
language="fr",
|
||||
word_timestamps=True,
|
||||
initial_prompt=hint or None,
|
||||
condition_on_previous_text=False,
|
||||
vad_filter=False,
|
||||
beam_size=5,
|
||||
)
|
||||
return [
|
||||
Word(w.word.strip(), float(w.start), float(w.end))
|
||||
Word(w.word.strip(), max(0.0, w.start - LEAD_SILENCE), max(0.0, w.end - LEAD_SILENCE))
|
||||
for segment in segments
|
||||
for w in (segment.words or [])
|
||||
if w.word.strip()
|
||||
]
|
||||
|
||||
|
||||
async def transcribe(audio: Path, hint: str | None = None) -> list[Word]:
|
||||
async def transcribe(audio: Path) -> list[Word]:
|
||||
"""Mots prononcés et leurs instants (Whisper, exécuté hors de la boucle async)."""
|
||||
return await asyncio.to_thread(_transcribe_sync, audio, hint)
|
||||
return await asyncio.to_thread(_transcribe_sync, audio)
|
||||
@@ -407,6 +407,8 @@ async def _prepare(request: ReelRequest, job_id: str, workdir: Path, started: fl
|
||||
"total_duration": plan.total_duration,
|
||||
"video_duration": info.duration,
|
||||
"logo_start": plan.logo_start if plan.has_outro else None,
|
||||
"fade_start": plan.fade_start if plan.ending_effect else None,
|
||||
"fade_duration": plan.fade_duration if plan.ending_effect else None,
|
||||
"words": [w.to_dict() for w in words],
|
||||
**_voice_info(track),
|
||||
"processing_stats": clock.summary(),
|
||||
@@ -418,7 +420,7 @@ async def _caption_words_without_voice(display: str, video: Path, info, plan: re
|
||||
sinon texte réparti sur la durée (hors effet de fin)."""
|
||||
if info.has_audio:
|
||||
try:
|
||||
spoken = await align.transcribe(video, hint=display)
|
||||
spoken = await align.transcribe(video)
|
||||
if len(spoken) >= max(2, len(display.split()) // 3):
|
||||
return align.align_words(display, spoken, info.duration)
|
||||
except Exception as error: # noqa: BLE001 — on retombe sur la répartition
|
||||
|
||||
+59
-25
@@ -1,42 +1,76 @@
|
||||
"""Traitement de la voix : filtrage, compression et niveau sonore constant."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from . import proc
|
||||
|
||||
# Coupe les basses inutiles, lisse la dynamique et amène la voix à -16 LUFS :
|
||||
# elle reste intelligible par-dessus la musique sans saturer.
|
||||
VOICE_CHAIN = (
|
||||
# Nettoyage de la voix : coupe les basses inutiles, retire le souffle de fond
|
||||
# (réduction douce, sans gate qui hacherait les fins de phrases) et arrondit
|
||||
# les pics. Pas de gain ajouté : c'est la normalisation qui règle le niveau.
|
||||
CLEAN_CHAIN = (
|
||||
"highpass=f=80,"
|
||||
"acompressor=threshold=0.1:ratio=3:attack=5:release=120:makeup=2,"
|
||||
"loudnorm=I=-16:TP=-1.5:LRA=7,"
|
||||
"aresample=48000"
|
||||
"afftdn=nr=10:nf=-50:tn=1,"
|
||||
"acompressor=threshold=0.125:ratio=2.5:attack=5:release=150:makeup=1"
|
||||
)
|
||||
# Niveau cible de la voix : intelligible par-dessus la musique sans saturer
|
||||
TARGET = "I=-16:TP=-1.5:LRA=11"
|
||||
# Fondu de 5 ms aux extrémités : pas de clic au début ni à la fin
|
||||
EDGE_FADE = 0.005
|
||||
|
||||
|
||||
async def process_voice(source: Path, target: Path) -> None:
|
||||
"""Produit un WAV 48 kHz mono prêt à mixer."""
|
||||
await proc.run(
|
||||
"""Produit un WAV 48 kHz mono prêt à mixer.
|
||||
|
||||
Normalisation en deux passes, en mode linéaire : un seul gain appliqué à
|
||||
toute la voix. Le mode dynamique (une passe) relevait les passages calmes
|
||||
et amplifiait souffles et bruits entre les phrases."""
|
||||
measure = await proc.run(
|
||||
[
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-hide_banner",
|
||||
"-loglevel",
|
||||
"error",
|
||||
"-i",
|
||||
str(source),
|
||||
"-af",
|
||||
VOICE_CHAIN,
|
||||
"-ac",
|
||||
"1",
|
||||
"-ar",
|
||||
"48000",
|
||||
"-c:a",
|
||||
"pcm_s16le",
|
||||
str(target),
|
||||
"ffmpeg", "-hide_banner", "-nostats", "-i", str(source),
|
||||
"-af", f"{CLEAN_CHAIN},loudnorm={TARGET}:print_format=json", "-f", "null", "-",
|
||||
],
|
||||
timeout=120,
|
||||
)
|
||||
stderr_output=True,
|
||||
) # fmt: skip
|
||||
stats = _loudnorm_stats(measure)
|
||||
if stats:
|
||||
loudnorm = (
|
||||
f"loudnorm={TARGET}:linear=true:"
|
||||
f"measured_I={stats['input_i']}:measured_TP={stats['input_tp']}:"
|
||||
f"measured_LRA={stats['input_lra']}:measured_thresh={stats['input_thresh']}:"
|
||||
f"offset={stats['target_offset']}"
|
||||
)
|
||||
else:
|
||||
loudnorm = f"loudnorm={TARGET}"
|
||||
duration = (await proc.probe(source)).duration
|
||||
fades = f"afade=t=in:d={EDGE_FADE},afade=t=out:st={max(0.0, duration - EDGE_FADE):.3f}:d={EDGE_FADE}"
|
||||
await proc.run(
|
||||
[
|
||||
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-i", str(source),
|
||||
"-af", f"{CLEAN_CHAIN},{loudnorm},aresample=48000,{fades}",
|
||||
"-ac", "1", "-ar", "48000", "-c:a", "pcm_s16le", str(target),
|
||||
],
|
||||
timeout=120,
|
||||
) # fmt: skip
|
||||
|
||||
|
||||
def _loudnorm_stats(ffmpeg_log: str) -> dict[str, str] | None:
|
||||
"""Mesures de la première passe (bloc JSON écrit par loudnorm)."""
|
||||
start, end = ffmpeg_log.rfind("{"), ffmpeg_log.rfind("}")
|
||||
if start < 0 or end <= start:
|
||||
return None
|
||||
try:
|
||||
stats = json.loads(ffmpeg_log[start : end + 1])
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
keys = ("input_i", "input_tp", "input_lra", "input_thresh", "target_offset")
|
||||
if not all(k in stats for k in keys):
|
||||
return None
|
||||
# Voix quasi muette : les mesures valent -inf, la normalisation simple suffit
|
||||
if any("inf" in str(stats[k]) for k in keys):
|
||||
return None
|
||||
return {k: stats[k] for k in keys}
|
||||
|
||||
|
||||
async def encode_preview(source: Path, target: Path) -> None:
|
||||
|
||||
@@ -14,6 +14,12 @@ TEMP_DIR = Path(os.environ.get("TEMP_DIR", "/tmp/ffmpeg_processing"))
|
||||
FILE_TTL_SECONDS = int(os.environ.get("FILE_TTL_SECONDS", "3600"))
|
||||
|
||||
GEMINI_TTS_MODEL = os.environ.get("GEMINI_TTS_MODEL", "gemini-2.5-flash-preview-tts")
|
||||
# Gemini tire sa lecture au hasard : une température basse et une graine fixe
|
||||
# donnent une intonation stable d'une génération à l'autre.
|
||||
GEMINI_TTS_TEMPERATURE = float(os.environ.get("GEMINI_TTS_TEMPERATURE", "0.3"))
|
||||
GEMINI_TTS_SEED = int(os.environ.get("GEMINI_TTS_SEED", "20260928"))
|
||||
# Générations tentées quand la lecture ne correspond pas au texte
|
||||
GEMINI_TTS_ATTEMPTS = int(os.environ.get("GEMINI_TTS_ATTEMPTS", "3"))
|
||||
|
||||
# Modèle Whisper utilisé pour caler les sous-titres sur la voix.
|
||||
WHISPER_MODEL = os.environ.get("WHISPER_MODEL", "base")
|
||||
|
||||
@@ -20,8 +20,9 @@ class CommandError(RuntimeError):
|
||||
self.stderr = stderr
|
||||
|
||||
|
||||
async def run(cmd: list[str], timeout: float = 900) -> str:
|
||||
"""Lance une commande et renvoie sa sortie standard."""
|
||||
async def run(cmd: list[str], timeout: float = 900, *, stderr_output: bool = False) -> str:
|
||||
"""Lance une commande et renvoie sa sortie standard (ou d'erreur, où FFmpeg
|
||||
écrit les mesures de ses filtres d'analyse)."""
|
||||
log.debug("exec: %s", " ".join(cmd))
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
|
||||
@@ -34,7 +35,7 @@ async def run(cmd: list[str], timeout: float = 900) -> str:
|
||||
raise CommandError(cmd, -1, f"délai de {timeout:.0f} s dépassé") from error
|
||||
if process.returncode != 0:
|
||||
raise CommandError(cmd, process.returncode, stderr.decode(errors="replace"))
|
||||
return stdout.decode(errors="replace")
|
||||
return (stderr if stderr_output else stdout).decode(errors="replace")
|
||||
|
||||
|
||||
@dataclass
|
||||
|
||||
@@ -0,0 +1,243 @@
|
||||
"""Contrôles de la voix générée : texte réellement lu et silences.
|
||||
|
||||
Gemini lit parfois mal : mots sautés, fin tronquée, consigne de style lue à
|
||||
voix haute. On compare donc ce que Whisper entend au texte attendu, et on
|
||||
découpe la voix d'après son niveau sonore réel (mesuré toutes les 10 ms)
|
||||
plutôt qu'aux bornes des mots estimées par Whisper, trop imprécises pour ne
|
||||
pas rogner une syllabe.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import difflib
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
from . import proc
|
||||
from .align import Word, normalize
|
||||
|
||||
MIN_COVERAGE = 0.92
|
||||
|
||||
|
||||
@dataclass
|
||||
class ReadingCheck:
|
||||
coverage: float # part des mots attendus effectivement entendus
|
||||
extra: float # mots entendus en trop (consigne lue, ajouts), rapportés au texte
|
||||
ending_ok: bool # la fin du texte est bien lue
|
||||
missing: list[str] = field(default_factory=list) # mots attendus non entendus (diagnostic)
|
||||
|
||||
@property
|
||||
def acceptable(self) -> bool:
|
||||
# Comparaison lettre à lettre : une lecture complète atteint ~97 %
|
||||
# (petites erreurs de Whisper) ; sous 92 %, des mots ont été sautés.
|
||||
return self.coverage >= MIN_COVERAGE and self.extra <= 0.25 and self.ending_ok
|
||||
|
||||
def describe(self) -> str:
|
||||
parts = [f"{self.coverage:.0%} des mots lus"]
|
||||
if not self.ending_ok:
|
||||
parts.append("fin tronquée")
|
||||
if self.extra > 0.25:
|
||||
parts.append("mots ajoutés")
|
||||
if self.missing and self.coverage < MIN_COVERAGE:
|
||||
parts.append("non entendus : " + " ".join(self.missing[:12]))
|
||||
return ", ".join(parts)
|
||||
|
||||
|
||||
def check_reading(expected_text: str, spoken: list[Word]) -> ReadingCheck:
|
||||
"""Compare le texte attendu à ce qui a été entendu (Whisper), lettre à
|
||||
lettre : les petites erreurs de reconnaissance (« petit » pour « petits »,
|
||||
« week -end » découpé) ne comptent pas comme des mots manquants, alors
|
||||
qu'un mot sauté, une fin tronquée ou une phrase ajoutée se voient."""
|
||||
expected = "".join(comparable_tokens(expected_text.split()))
|
||||
heard = "".join(comparable_tokens([w.text for w in spoken]))
|
||||
if not expected:
|
||||
return ReadingCheck(1.0, 0.0, True)
|
||||
if not heard:
|
||||
return ReadingCheck(0.0, 0.0, False)
|
||||
|
||||
matcher = difflib.SequenceMatcher(a=expected, b=heard, autojunk=False)
|
||||
blocks = [b for b in matcher.get_matching_blocks() if b.size]
|
||||
matched = sum(b.size for b in blocks)
|
||||
# Fin lue : une bonne partie des dernières lettres du texte est reconnue
|
||||
tail_start = max(0, len(expected) - 8)
|
||||
tail_matched = sum(max(0, min(b.a + b.size, len(expected)) - max(b.a, tail_start)) for b in blocks)
|
||||
return ReadingCheck(
|
||||
coverage=matched / len(expected),
|
||||
extra=max(0, len(heard) - matched) / len(expected),
|
||||
ending_ok=tail_matched >= min(4, len(expected) - tail_start),
|
||||
missing=_missing_words(expected_text, spoken),
|
||||
)
|
||||
|
||||
|
||||
# Unités écrites de plusieurs façons par Whisper (« 40 cm » pour « 40 centimètres »)
|
||||
UNITS = {
|
||||
"e", "eur", "euro", "euros", "centime", "centimes", "cm", "centimetre", "centimetres",
|
||||
"mm", "millimetre", "millimetres", "m", "metre", "metres", "km", "kilometre", "kilometres",
|
||||
"g", "gramme", "grammes", "kg", "kilo", "kilos", "kilogramme", "kilogrammes",
|
||||
"l", "litre", "litres", "cl", "centilitre", "centilitres", "ml", "millilitre", "millilitres",
|
||||
"pourcent", "pourcents", "x", "fois",
|
||||
} # fmt: skip
|
||||
|
||||
|
||||
def comparable_tokens(words: list[str]) -> list[str]:
|
||||
"""Mots normalisés, sans les nombres ni l'unité qui les suit : Whisper écrit
|
||||
un prix ou une mesure à sa façon (« 11 € 99 », « 11,99 euros », « 40 cm »),
|
||||
ce qui passait pour des mots sautés et relançait la génération."""
|
||||
tokens: list[str] = []
|
||||
after_number = False
|
||||
for word in words:
|
||||
token = normalize(word)
|
||||
if not token:
|
||||
continue
|
||||
if any(c.isdigit() for c in token):
|
||||
after_number = True
|
||||
continue
|
||||
if after_number and token in UNITS:
|
||||
continue
|
||||
after_number = False
|
||||
tokens.append(token)
|
||||
return tokens
|
||||
|
||||
|
||||
def _missing_words(expected_text: str, spoken: list[Word]) -> list[str]:
|
||||
"""Mots attendus sans équivalent proche dans ce qui a été entendu."""
|
||||
heard = set(comparable_tokens([w.text for w in spoken]))
|
||||
kept = set(comparable_tokens(expected_text.split()))
|
||||
missing = []
|
||||
for word in expected_text.split():
|
||||
token = normalize(word)
|
||||
if token in kept and not any(difflib.SequenceMatcher(a=token, b=h).ratio() >= 0.7 for h in heard):
|
||||
missing.append(word.strip(".,!?;:…"))
|
||||
return missing
|
||||
|
||||
|
||||
# Enveloppe sonore : niveau de la voix toutes les 10 ms
|
||||
FRAME = 0.01
|
||||
# Est « audible » ce qui dépasse le niveau de crête moins 38 dB : un seuil
|
||||
# relatif respecte les fins de phrases douces (« s » final, voix qui retombe),
|
||||
# qu'un seuil absolu prenait pour du silence.
|
||||
ACTIVE_RANGE_DB = 38.0
|
||||
# Marges gardées autour de la parole
|
||||
LEAD_PAD = 0.08
|
||||
TAIL_PAD = 0.30
|
||||
# Pause entre deux phrases : au moins 80 ms de silence, cherchée jusqu'à 1,2 s
|
||||
# de la borne estimée par Whisper
|
||||
MIN_PAUSE = 0.08
|
||||
PAUSE_SEARCH = 1.2
|
||||
|
||||
|
||||
@dataclass
|
||||
class Envelope:
|
||||
db: np.ndarray # niveau (dBFS) de chaque tranche de 10 ms
|
||||
|
||||
@property
|
||||
def duration(self) -> float:
|
||||
return len(self.db) * FRAME
|
||||
|
||||
@property
|
||||
def threshold(self) -> float:
|
||||
if not len(self.db):
|
||||
return 0.0
|
||||
return float(np.percentile(self.db, 95)) - ACTIVE_RANGE_DB
|
||||
|
||||
def _frames(self, start: float, end: float) -> tuple[int, int]:
|
||||
first = max(0, int(start / FRAME))
|
||||
last = min(len(self.db), int(np.ceil(end / FRAME)))
|
||||
return first, max(first, last)
|
||||
|
||||
def quietest(self, start: float, end: float) -> float:
|
||||
"""Milieu de la zone la plus calme d'un intervalle : la coupe tombe au
|
||||
cœur de la pause, loin des deux phrases qu'elle sépare."""
|
||||
first, last = self._frames(start, end)
|
||||
if last - first < 1:
|
||||
return (start + end) / 2
|
||||
smooth = np.convolve(self.db, np.ones(5) / 5, mode="same")[first:last]
|
||||
quiet = smooth <= smooth.min() + 3.0 # à 3 dB du minimum
|
||||
best_start, best_length, run_start = 0, 0, None
|
||||
for i, is_quiet in enumerate([*quiet, False]):
|
||||
if is_quiet and run_start is None:
|
||||
run_start = i
|
||||
elif not is_quiet and run_start is not None:
|
||||
if i - run_start > best_length:
|
||||
best_start, best_length = run_start, i - run_start
|
||||
run_start = None
|
||||
return (first + best_start + best_length / 2) * FRAME
|
||||
|
||||
def first_sound(self, start: float, end: float) -> float | None:
|
||||
first, last = self._frames(start, end)
|
||||
active = np.nonzero(self.db[first:last] > self.threshold)[0]
|
||||
return (first + int(active[0])) * FRAME if len(active) else None
|
||||
|
||||
def last_sound(self, start: float, end: float) -> float | None:
|
||||
first, last = self._frames(start, end)
|
||||
active = np.nonzero(self.db[first:last] > self.threshold)[0]
|
||||
return (first + int(active[-1]) + 1) * FRAME if len(active) else None
|
||||
|
||||
def pauses(self, min_length: float = MIN_PAUSE) -> list[tuple[float, float]]:
|
||||
"""Silences (sous le seuil d'audibilité) d'au moins min_length secondes."""
|
||||
quiet = self.db <= self.threshold
|
||||
runs, run_start = [], None
|
||||
for i, is_quiet in enumerate([*quiet, False]):
|
||||
if is_quiet and run_start is None:
|
||||
run_start = i
|
||||
elif not is_quiet and run_start is not None:
|
||||
if (i - run_start) * FRAME >= min_length:
|
||||
runs.append((run_start * FRAME, i * FRAME))
|
||||
run_start = None
|
||||
return runs
|
||||
|
||||
def speech_span(self, start: float, end: float) -> tuple[float, float]:
|
||||
"""Parole contenue dans [start, end], avec ses marges, sans en sortir."""
|
||||
onset, offset = self.first_sound(start, end), self.last_sound(start, end)
|
||||
if onset is None or offset is None:
|
||||
return start, end
|
||||
return max(start, onset - LEAD_PAD), min(end, offset + TAIL_PAD)
|
||||
|
||||
|
||||
def envelope_from_samples(samples: np.ndarray, rate: int) -> Envelope:
|
||||
per_frame = int(rate * FRAME)
|
||||
count = len(samples) // per_frame
|
||||
if count == 0:
|
||||
return Envelope(np.array([-120.0]))
|
||||
frames = samples[: count * per_frame].reshape(count, per_frame).astype(np.float64)
|
||||
rms = np.sqrt(np.mean(frames**2, axis=1))
|
||||
return Envelope(20 * np.log10(np.maximum(rms, 1e-6)))
|
||||
|
||||
|
||||
async def load_envelope(path: Path) -> Envelope:
|
||||
"""Décode l'audio (mono 16 kHz) et mesure son niveau toutes les 10 ms."""
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
"ffmpeg", "-hide_banner", "-loglevel", "error", "-i", str(path),
|
||||
"-ac", "1", "-ar", "16000", "-f", "f32le", "-",
|
||||
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
|
||||
) # fmt: skip
|
||||
raw, err = await process.communicate()
|
||||
if process.returncode != 0:
|
||||
raise proc.CommandError(["ffmpeg"], process.returncode, err.decode(errors="replace"))
|
||||
return envelope_from_samples(np.frombuffer(raw, dtype=np.float32), 16000)
|
||||
|
||||
|
||||
def sentence_cut(envelope: Envelope, previous_end: float, next_start: float, after: float = 0.0) -> float:
|
||||
"""Coupe entre deux phrases, au cœur d'une vraie pause.
|
||||
|
||||
Les bornes Whisper sont imprécises, et fausses quand un mot n'a pas été
|
||||
reconnu (un prix en fin de phrase : sa place est alors estimée) ; couper
|
||||
à la borne estimée tombait parfois au milieu d'une phrase. On choisit donc
|
||||
la pause mesurée la plus proche de la borne estimée, les longues pauses
|
||||
(fin de phrase) étant préférées aux courtes (virgule)."""
|
||||
low = max(after, min(previous_end, next_start) - 0.1)
|
||||
high = max(low + 0.05, max(previous_end, next_start) + 0.2)
|
||||
best: tuple[float, float, float] | None = None
|
||||
for start, end in envelope.pauses():
|
||||
if end <= after + 0.05 or start <= 0.0 or end >= envelope.duration:
|
||||
continue # silences avant la première et après la dernière phrase exclus
|
||||
distance = max(0.0, low - end, start - high)
|
||||
if distance > PAUSE_SEARCH:
|
||||
continue
|
||||
score = distance - (end - start)
|
||||
if best is None or score < best[0]:
|
||||
best = (score, max(start, after), end)
|
||||
if best:
|
||||
return envelope.quietest(best[1], best[2])
|
||||
return envelope.quietest(low, high)
|
||||
@@ -14,6 +14,8 @@ from . import config
|
||||
from .subtitles import filter_path
|
||||
|
||||
FADE_SECONDS = 2.0
|
||||
# Fondu raccourci quand la voix finit tard, mais jamais sec
|
||||
MIN_FADE = 0.3
|
||||
LOGO_SECONDS = 5.0
|
||||
# Silence laissé après la dernière phrase avant la fin de la vidéo
|
||||
VOICE_TAIL = 0.8
|
||||
@@ -76,7 +78,16 @@ class RenderPlan:
|
||||
|
||||
@property
|
||||
def fade_start(self) -> float:
|
||||
return max(0.0, self.total_duration - FADE_SECONDS)
|
||||
"""Le fondu final ne commence jamais avant la fin de la voix : il
|
||||
rognait la dernière phrase quand la vidéo finissait peu après."""
|
||||
start = max(0.0, self.total_duration - FADE_SECONDS)
|
||||
if self.voice:
|
||||
start = max(start, min(self.speech_end + 0.15, self.total_duration - MIN_FADE))
|
||||
return round(start, 3)
|
||||
|
||||
@property
|
||||
def fade_duration(self) -> float:
|
||||
return round(max(MIN_FADE, self.total_duration - self.fade_start), 3)
|
||||
|
||||
@property
|
||||
def freeze_duration(self) -> float:
|
||||
@@ -102,34 +113,47 @@ def video_filters(plan: RenderPlan) -> list[str]:
|
||||
return chain
|
||||
|
||||
|
||||
def uses_original_audio(plan: RenderPlan) -> bool:
|
||||
"""Sans musique choisie, la bande son d'origine de la vidéo sert de fond sonore."""
|
||||
return plan.music is None and plan.keep_original_audio
|
||||
|
||||
|
||||
def audio_graph(
|
||||
plan: RenderPlan, music_idx: int | None, voice_idx: int | None
|
||||
plan: RenderPlan, music_idx: int | None, voice_idx: int | None, original_idx: int | None = None
|
||||
) -> tuple[list[str], str | None]:
|
||||
"""Graphe audio : voix décalée, musique bouclée et baissée sous la voix,
|
||||
niveau final ~ -14 LUFS. Renvoie les filtres et l'étiquette de sortie."""
|
||||
"""Graphe audio : voix décalée sur un fond sonore (musique bouclée, ou à
|
||||
défaut bande son d'origine) baissé sous la voix, niveau final ~ -14 LUFS.
|
||||
Renvoie les filtres et l'étiquette de sortie."""
|
||||
graph: list[str] = []
|
||||
if voice_idx is not None:
|
||||
delay_ms = int(plan.voice_delay * 1000)
|
||||
graph.append(f"[{voice_idx}:a]aresample=48000,adelay={delay_ms}:all=1,apad[voice]")
|
||||
if music_idx is not None:
|
||||
graph.append(f"[{music_idx}:a]aresample=48000,volume={plan.music_volume:.3f}[music]")
|
||||
|
||||
if voice_idx is not None and music_idx is not None:
|
||||
# La musique baisse automatiquement quand la voix parle (ducking)
|
||||
bed = None
|
||||
if music_idx is not None:
|
||||
graph.append(f"[{music_idx}:a]aresample=48000,volume={plan.music_volume:.3f}[bed]")
|
||||
bed = "bed"
|
||||
elif original_idx is not None:
|
||||
# Son d'origine conservé à son niveau ; il se tait après la fin de la vidéo
|
||||
graph.append(f"[{original_idx}:a:0]aresample=48000,apad[bed]")
|
||||
bed = "bed"
|
||||
|
||||
if voice_idx is not None and bed:
|
||||
# Le fond sonore baisse automatiquement quand la voix parle (ducking)
|
||||
graph.append("[voice]asplit=2[vmix][vkey]")
|
||||
graph.append("[music][vkey]sidechaincompress=threshold=0.02:ratio=8:attack=20:release=400[ducked]")
|
||||
graph.append(f"[{bed}][vkey]sidechaincompress=threshold=0.02:ratio=8:attack=20:release=400[ducked]")
|
||||
graph.append("[ducked][vmix]amix=inputs=2:duration=longest:normalize=0[mix]")
|
||||
mix = "mix"
|
||||
elif voice_idx is not None:
|
||||
mix = "voice"
|
||||
elif music_idx is not None:
|
||||
mix = "music"
|
||||
elif bed:
|
||||
mix = bed
|
||||
else:
|
||||
return graph, None
|
||||
|
||||
tail = ["loudnorm=I=-14:TP=-1.5:LRA=11", "aresample=48000"]
|
||||
tail = ["loudnorm=I=-14:TP=-1.5:LRA=20", "aresample=48000"]
|
||||
if plan.ending_effect:
|
||||
tail.append(f"afade=t=out:st={plan.fade_start:.3f}:d={FADE_SECONDS}")
|
||||
tail.append(f"afade=t=out:st={plan.fade_start:.3f}:d={plan.fade_duration}")
|
||||
graph.append(f"[{mix}]{','.join(tail)}[aout]")
|
||||
return graph, "aout"
|
||||
|
||||
@@ -202,18 +226,15 @@ def build_command(plan: RenderPlan) -> list[str]:
|
||||
|
||||
tail = []
|
||||
if plan.ending_effect:
|
||||
tail.append(f"fade=t=out:st={plan.fade_start:.3f}:d={FADE_SECONDS}")
|
||||
tail.append(f"fade=t=out:st={plan.fade_start:.3f}:d={plan.fade_duration}")
|
||||
tail.append("format=yuv420p")
|
||||
graph.append(f"[{current}]{','.join(tail)}[vout]")
|
||||
|
||||
audio_filters, audio_out = audio_graph(plan, music_idx, voice_idx)
|
||||
# La vidéo (entrée 0) fournit la bande son d'origine si aucune musique n'est choisie
|
||||
original_idx = 0 if uses_original_audio(plan) else None
|
||||
audio_filters, audio_out = audio_graph(plan, music_idx, voice_idx, original_idx)
|
||||
graph += audio_filters
|
||||
if audio_out:
|
||||
audio_map = ["-map", f"[{audio_out}]"]
|
||||
elif plan.keep_original_audio:
|
||||
audio_map = ["-map", "0:a:0"]
|
||||
else:
|
||||
audio_map = []
|
||||
audio_map = ["-map", f"[{audio_out}]"] if audio_out else []
|
||||
|
||||
cmd += ["-filter_complex", ";".join(graph), "-map", "[vout]", *audio_map]
|
||||
cmd += ["-t", f"{total:.3f}", *_video_encoding(19, "medium")]
|
||||
@@ -236,21 +257,14 @@ def build_prepared_video_command(plan: RenderPlan, output: Path) -> list[str]:
|
||||
def build_audio_mix_command(plan: RenderPlan, output: Path) -> list[str] | None:
|
||||
"""Piste son finale (WAV 48 kHz stéréo), ou None s'il n'y a aucun son à produire."""
|
||||
total = f"{plan.total_duration:.3f}"
|
||||
if not plan.music and not plan.voice:
|
||||
if not plan.keep_original_audio:
|
||||
return None
|
||||
# Son d'origine seul : même niveau final que les autres Reels
|
||||
chain = ["aresample=48000", "loudnorm=I=-14:TP=-1.5:LRA=11", "apad"]
|
||||
if plan.ending_effect:
|
||||
chain.append(f"afade=t=out:st={plan.fade_start:.3f}:d={FADE_SECONDS}")
|
||||
return [
|
||||
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-i", str(plan.video),
|
||||
"-map", "0:a:0", "-af", ",".join(chain), "-t", total,
|
||||
"-ar", "48000", "-ac", "2", "-c:a", "pcm_s16le", str(output),
|
||||
] # fmt: skip
|
||||
|
||||
audio_args, music_idx, voice_idx, _ = _audio_inputs(plan, 0)
|
||||
graph, audio_out = audio_graph(plan, music_idx, voice_idx)
|
||||
audio_args, music_idx, voice_idx, index = _audio_inputs(plan, 0)
|
||||
original_idx = None
|
||||
if uses_original_audio(plan):
|
||||
audio_args += ["-i", str(plan.video)]
|
||||
original_idx = index
|
||||
graph, audio_out = audio_graph(plan, music_idx, voice_idx, original_idx)
|
||||
if not audio_out:
|
||||
return None
|
||||
return [
|
||||
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error", *audio_args,
|
||||
"-filter_complex", ";".join(graph), "-map", f"[{audio_out}]", "-t", total,
|
||||
|
||||
+138
-86
@@ -1,29 +1,38 @@
|
||||
"""Voix calée sur un fichier SRT.
|
||||
|
||||
Chaque sous-titre est lu séparément puis posé à son instant de début. Si la
|
||||
lecture est plus longue que le sous-titre, elle est accélérée (sans changer
|
||||
la hauteur de la voix) pour finir à temps : la voix respecte le minutage du
|
||||
fichier, au lieu d'un texte lu d'une traite.
|
||||
Tout le texte est lu d'une seule traite, puis la voix est découpée phrase par
|
||||
phrase au point le plus calme de chaque pause, et chaque morceau est posé à l'instant de son
|
||||
sous-titre. Une lecture unique garde la même intonation, le même rythme et le
|
||||
même niveau d'un bout à l'autre ; lire chaque sous-titre séparément donnait
|
||||
une voix différente à chaque phrase, et le découpage aux bornes estimées par
|
||||
Whisper rognait des syllabes.
|
||||
|
||||
Si la lecture unique ne peut pas être répartie entre les sous-titres, on lit
|
||||
chaque sous-titre séparément (même voix, même graine), en découpant là aussi
|
||||
au niveau sonore mesuré.
|
||||
|
||||
Quand un morceau est plus long que son sous-titre, il est accéléré (sans
|
||||
changer la hauteur de la voix) pour finir à temps.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from . import proc, tts
|
||||
from .align import Word
|
||||
from . import proc, quality, tts
|
||||
from .align import Word, display_tokens
|
||||
from .text import clean_text
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# Débordement toléré sur le silence qui suit un sous-titre avant d'accélérer
|
||||
OVERFLOW_TOLERANCE = 0.25
|
||||
# Marges gardées autour des mots lors du retrait des silences de la voix
|
||||
TRIM_LEAD = 0.05
|
||||
TRIM_TAIL = 0.12
|
||||
# Au-delà, l'accélération s'entend nettement : l'utilisateur est prévenu
|
||||
AUDIBLE_SPEEDUP = 1.35
|
||||
|
||||
_SENTENCE_END = re.compile(r"[.!?…]$")
|
||||
|
||||
|
||||
@dataclass
|
||||
class Cue:
|
||||
@@ -32,6 +41,15 @@ class Cue:
|
||||
text: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class Segment:
|
||||
"""Morceau de la voix lue, à poser à l'instant d'un sous-titre."""
|
||||
|
||||
source_start: float
|
||||
source_end: float
|
||||
words: list[Word]
|
||||
|
||||
|
||||
def cue_windows(cues: list[Cue]) -> list[float]:
|
||||
"""Durée disponible pour lire chaque sous-titre sans empiéter sur le suivant."""
|
||||
windows = []
|
||||
@@ -59,13 +77,42 @@ def atempo_chain(factor: float) -> str:
|
||||
return ",".join(f"atempo={s:.4f}" for s in steps)
|
||||
|
||||
|
||||
def spoken_span(words: list[Word], duration: float) -> tuple[float, float]:
|
||||
"""Partie parlée de la voix : les silences de début et de fin sont retirés."""
|
||||
if not words:
|
||||
return 0.0, duration
|
||||
start = max(0.0, words[0].start - TRIM_LEAD)
|
||||
end = min(duration, words[-1].end + TRIM_TAIL)
|
||||
return (start, end) if end > start else (0.0, duration)
|
||||
def as_sentence(text: str) -> str:
|
||||
"""Chaque sous-titre finit par une ponctuation : la voix y marque une pause,
|
||||
ce qui laisse un silence où couper."""
|
||||
text = clean_text(text)
|
||||
return text if _SENTENCE_END.search(text) else f"{text}."
|
||||
|
||||
|
||||
def split_by_cue(words: list[Word], cue_texts: list[str]) -> list[list[Word]] | None:
|
||||
"""Répartit les mots de la lecture complète entre les sous-titres
|
||||
(les mots affichés suivent l'ordre du texte). None si les comptes divergent."""
|
||||
counts = [len(display_tokens(t)) for t in cue_texts]
|
||||
if sum(counts) != len(words) or 0 in counts:
|
||||
return None
|
||||
groups, index = [], 0
|
||||
for count in counts:
|
||||
groups.append(words[index : index + count])
|
||||
index += count
|
||||
return groups
|
||||
|
||||
|
||||
def segments_from_reading(groups: list[list[Word]], envelope: quality.Envelope) -> list[Segment] | None:
|
||||
"""Découpe la lecture complète entre les phrases, au point le plus calme de
|
||||
chaque pause, puis retire le silence autour de chaque phrase en gardant une
|
||||
marge (300 ms après le dernier son : les fins de phrases ne sont pas rognées)."""
|
||||
cuts = [0.0]
|
||||
for previous, following in zip(groups, groups[1:]):
|
||||
cuts.append(quality.sentence_cut(envelope, previous[-1].end, following[0].start, after=cuts[-1]))
|
||||
cuts.append(envelope.duration)
|
||||
if any(b <= a for a, b in zip(cuts, cuts[1:])):
|
||||
return None
|
||||
|
||||
segments = []
|
||||
for i, group in enumerate(groups):
|
||||
seg_start, seg_end = envelope.speech_span(cuts[i], cuts[i + 1])
|
||||
segments.append(Segment(seg_start, seg_end, group))
|
||||
return segments
|
||||
|
||||
|
||||
async def synthesize_cues(
|
||||
@@ -82,75 +129,89 @@ async def synthesize_cues(
|
||||
if not cues:
|
||||
raise ValueError("Aucun sous-titre lisible dans le fichier SRT")
|
||||
|
||||
warnings: list[str] = []
|
||||
windows = cue_windows(cues)
|
||||
segments: list[tuple[Path, float]] = []
|
||||
options = dict(engine=engine, voice=voice, style=style, gemini_api_key=gemini_api_key)
|
||||
texts = [as_sentence(c.text) for c in cues]
|
||||
|
||||
# 1. Lecture d'une seule traite, découpée dans les silences
|
||||
script = " ".join(texts)
|
||||
reading = await tts.synthesize(text=script, display_text=script, workdir=workdir, **options)
|
||||
envelope = await quality.load_envelope(reading.path)
|
||||
groups = split_by_cue(reading.words, texts)
|
||||
segments = segments_from_reading(groups, envelope) if groups else None
|
||||
|
||||
if segments:
|
||||
sources = [(reading.path, seg) for seg in segments]
|
||||
warnings = list(reading.warnings)
|
||||
used_engine, used_voice = reading.engine, reading.voice
|
||||
else:
|
||||
# 2. Repli : chaque sous-titre lu séparément, avec la voix de la lecture complète
|
||||
log.warning("Lecture SRT non répartie entre les sous-titres : lecture phrase par phrase")
|
||||
sources, warnings = [], list(reading.warnings)
|
||||
used_engine, used_voice = reading.engine, reading.voice
|
||||
for index, text in enumerate(texts):
|
||||
cue_dir = workdir / f"cue_{index:03d}"
|
||||
cue_dir.mkdir(exist_ok=True)
|
||||
track = await tts.synthesize(
|
||||
text=text,
|
||||
display_text=text,
|
||||
workdir=cue_dir,
|
||||
**{**options, "engine": used_engine, "voice": used_voice or voice},
|
||||
)
|
||||
warnings += [w for w in track.warnings if w not in warnings]
|
||||
cue_envelope = await quality.load_envelope(track.path)
|
||||
span = cue_envelope.speech_span(0.0, cue_envelope.duration)
|
||||
sources.append((track.path, Segment(span[0], span[1], track.words)))
|
||||
|
||||
placed: list[tuple[Path, float]] = []
|
||||
words: list[Word] = []
|
||||
used_engine, used_voice = engine or "gemini", voice or ""
|
||||
|
||||
for index, (cue, window) in enumerate(zip(cues, windows)):
|
||||
cue_dir = workdir / f"cue_{index:03d}"
|
||||
cue_dir.mkdir(exist_ok=True)
|
||||
text = clean_text(cue.text)
|
||||
track = await tts.synthesize(
|
||||
text=text,
|
||||
display_text=text,
|
||||
engine=used_engine,
|
||||
voice=voice,
|
||||
style=style,
|
||||
gemini_api_key=gemini_api_key,
|
||||
workdir=cue_dir,
|
||||
)
|
||||
for warning in track.warnings:
|
||||
if warning not in warnings:
|
||||
warnings.append(warning)
|
||||
# Après un repli sur Edge, on reste sur Edge : une seule voix du début à la fin
|
||||
used_engine, used_voice = track.engine, track.voice
|
||||
|
||||
span_start, span_end = spoken_span(track.words, track.duration)
|
||||
factor = speed_factor(span_end - span_start, window)
|
||||
for index, (cue, window, (source, segment)) in enumerate(zip(cues, cue_windows(cues), sources)):
|
||||
factor = speed_factor(segment.source_end - segment.source_start, window)
|
||||
if factor > AUDIBLE_SPEEDUP:
|
||||
warnings.append(
|
||||
f"Sous-titre {index + 1} (« {cue.text[:40]} ») lu {factor:.1f}× plus vite "
|
||||
"pour tenir dans son minutage : raccourcissez-le ou allongez sa durée."
|
||||
)
|
||||
fitted = workdir / f"segment_{index:03d}.wav"
|
||||
await proc.run(extract_command(source, segment, factor, fitted), timeout=120)
|
||||
placed.append((fitted, cue.start))
|
||||
added_period = not _SENTENCE_END.search(clean_text(cue.text))
|
||||
for position, w in enumerate(segment.words):
|
||||
text = w.text
|
||||
if added_period and position == len(segment.words) - 1:
|
||||
text = text.rstrip(".") # point ajouté pour la lecture, pas à afficher
|
||||
start = cue.start + max(0.0, w.start - segment.source_start) / factor
|
||||
end = cue.start + max(0.0, w.end - segment.source_start) / factor
|
||||
words.append(Word(text or w.text, start, min(max(end, start + 0.05), cue.start + window)))
|
||||
|
||||
fitted = cue_dir / "fitted.wav"
|
||||
trim = f"atrim=start={span_start:.3f}:end={span_end:.3f}"
|
||||
await proc.run(
|
||||
[
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-hide_banner",
|
||||
"-loglevel",
|
||||
"error",
|
||||
"-i",
|
||||
str(track.path),
|
||||
"-af",
|
||||
f"{trim},asetpts=PTS-STARTPTS,{atempo_chain(factor)}",
|
||||
"-ac",
|
||||
"1",
|
||||
"-ar",
|
||||
"48000",
|
||||
"-c:a",
|
||||
"pcm_s16le",
|
||||
str(fitted),
|
||||
],
|
||||
timeout=120,
|
||||
)
|
||||
segments.append((fitted, cue.start))
|
||||
for w in track.words:
|
||||
start = cue.start + max(0.0, w.start - span_start) / factor
|
||||
end = cue.start + max(0.0, w.end - span_start) / factor
|
||||
words.append(Word(w.text, start, min(max(end, start + 0.05), cue.start + window)))
|
||||
|
||||
output = workdir / "voice.wav"
|
||||
await proc.run(mix_command(segments, output), timeout=300)
|
||||
output = workdir / "voice_srt.wav"
|
||||
await proc.run(mix_command(placed, output), timeout=300)
|
||||
duration = (await proc.probe(output)).duration
|
||||
log.info("Voix SRT prête : %s/%s, %d sous-titres, %.1f s", used_engine, used_voice, len(cues), duration)
|
||||
log.info(
|
||||
"Voix SRT prête : %s/%s, %d sous-titres, %.1f s (%s)",
|
||||
used_engine,
|
||||
used_voice,
|
||||
len(cues),
|
||||
duration,
|
||||
"lecture unique" if segments else "phrase par phrase",
|
||||
)
|
||||
return tts.VoiceTrack(output, duration, _monotonic(words), used_engine, used_voice, warnings)
|
||||
|
||||
|
||||
def extract_command(source: Path, segment: Segment, factor: float, output: Path) -> list[str]:
|
||||
"""Extrait un morceau de voix, avec un fondu de 10 ms aux bords (pas de clic)
|
||||
et l'accélération éventuelle."""
|
||||
length = segment.source_end - segment.source_start
|
||||
fade_out = max(0.0, length - 0.01)
|
||||
filters = (
|
||||
f"atrim=start={segment.source_start:.3f}:end={segment.source_end:.3f},asetpts=PTS-STARTPTS,"
|
||||
f"afade=t=in:d=0.01,afade=t=out:st={fade_out:.3f}:d=0.01,{atempo_chain(factor)}"
|
||||
)
|
||||
return [
|
||||
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-i", str(source),
|
||||
"-af", filters, "-ac", "1", "-ar", "48000", "-c:a", "pcm_s16le", str(output),
|
||||
] # fmt: skip
|
||||
|
||||
|
||||
def mix_command(segments: list[tuple[Path, float]], output: Path) -> list[str]:
|
||||
"""Pose chaque segment de voix à son instant, sur une seule piste."""
|
||||
command = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error"]
|
||||
@@ -162,18 +223,9 @@ def mix_command(segments: list[tuple[Path, float]], output: Path) -> list[str]:
|
||||
f"{labels}amix=inputs={len(segments)}:duration=longest:dropout_transition=0:normalize=0[voice]"
|
||||
)
|
||||
return command + [
|
||||
"-filter_complex",
|
||||
";".join(graph),
|
||||
"-map",
|
||||
"[voice]",
|
||||
"-ac",
|
||||
"1",
|
||||
"-ar",
|
||||
"48000",
|
||||
"-c:a",
|
||||
"pcm_s16le",
|
||||
str(output),
|
||||
]
|
||||
"-filter_complex", ";".join(graph), "-map", "[voice]",
|
||||
"-ac", "1", "-ar", "48000", "-c:a", "pcm_s16le", str(output),
|
||||
] # fmt: skip
|
||||
|
||||
|
||||
def _monotonic(words: list[Word]) -> list[Word]:
|
||||
|
||||
@@ -4,7 +4,7 @@ import logging
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from .. import align, audio, proc
|
||||
from .. import align, audio, config, proc, quality
|
||||
from ..align import Word
|
||||
from . import edge, gemini
|
||||
|
||||
@@ -32,7 +32,7 @@ async def synthesize(
|
||||
workdir: Path,
|
||||
) -> VoiceTrack:
|
||||
warnings: list[str] = []
|
||||
raw: Path | None = None
|
||||
processed: Path | None = None
|
||||
spoken: list[Word] = []
|
||||
used_engine, used_voice = "edge", ""
|
||||
|
||||
@@ -40,24 +40,102 @@ async def synthesize(
|
||||
if not gemini_api_key:
|
||||
warnings.append("Clé Gemini absente : voix Edge utilisée à la place.")
|
||||
else:
|
||||
try:
|
||||
raw, used_voice = await gemini.synthesize(text, voice, style, gemini_api_key, workdir)
|
||||
result = await _gemini_checked(text, voice, style, gemini_api_key, workdir, warnings)
|
||||
if result:
|
||||
processed, spoken, used_voice, reading_ok = result
|
||||
used_engine = "gemini"
|
||||
except Exception as error: # noqa: BLE001 — repli sur Edge
|
||||
log.warning("Gemini TTS en échec, repli sur Edge : %s", error)
|
||||
warnings.append(f"Gemini indisponible ({error}) : voix Edge utilisée à la place.")
|
||||
if reading_ok:
|
||||
processed, spoken = await _trim_stray_sounds(processed, spoken, workdir)
|
||||
|
||||
if raw is None:
|
||||
if processed is None:
|
||||
raw, spoken, used_voice = await edge.synthesize(text, voice, style, workdir)
|
||||
processed = workdir / "voice.wav"
|
||||
await audio.process_voice(raw, processed)
|
||||
|
||||
processed = workdir / "voice.wav"
|
||||
await audio.process_voice(raw, processed)
|
||||
duration = (await proc.probe(processed)).duration
|
||||
|
||||
if not spoken:
|
||||
# Gemini ne donne pas le minutage : Whisper le retrouve dans l'audio
|
||||
spoken = await align.transcribe(processed, hint=text)
|
||||
spoken = await align.transcribe(processed)
|
||||
|
||||
words = align.align_words(display_text, spoken, duration)
|
||||
log.info("Voix prête : %s/%s, %.1f s, %d mots", used_engine, used_voice, duration, len(words))
|
||||
return VoiceTrack(processed, duration, words, used_engine, used_voice, warnings)
|
||||
|
||||
|
||||
async def _gemini_checked(
|
||||
text: str, voice: str | None, style: str | None, api_key: str, workdir: Path, warnings: list[str]
|
||||
) -> tuple[Path, list[Word], str, bool] | None:
|
||||
"""Voix Gemini dont la lecture a été vérifiée : Whisper réécoute chaque
|
||||
génération ; mots sautés, fin tronquée ou consigne lue à voix haute
|
||||
déclenchent une nouvelle génération. Renvoie None pour se replier sur Edge."""
|
||||
best: tuple[quality.ReadingCheck, Path, list[Word], str] | None = None
|
||||
last_error: Exception | None = None
|
||||
|
||||
for attempt in range(max(1, config.GEMINI_TTS_ATTEMPTS)):
|
||||
try:
|
||||
raw, gemini_voice = await gemini.synthesize(text, voice, style, api_key, workdir, attempt)
|
||||
except Exception as error: # noqa: BLE001 — nouvelle tentative, puis repli sur Edge
|
||||
log.warning("Gemini TTS en échec (tentative %d) : %s", attempt + 1, error)
|
||||
last_error = error
|
||||
continue
|
||||
|
||||
processed = workdir / f"voice_gemini_{attempt}.wav"
|
||||
await audio.process_voice(raw, processed)
|
||||
spoken = await align.transcribe(processed)
|
||||
check = quality.check_reading(text, spoken)
|
||||
log.info("Lecture Gemini (tentative %d) : %s", attempt + 1, check.describe())
|
||||
|
||||
if best is None or _score(check) > _score(best[0]):
|
||||
best = (check, processed, spoken, gemini_voice)
|
||||
if check.acceptable:
|
||||
break
|
||||
|
||||
if best is None:
|
||||
warnings.append(f"Gemini indisponible ({last_error}) : voix Edge utilisée à la place.")
|
||||
return None
|
||||
|
||||
check, processed, spoken, gemini_voice = best
|
||||
if not check.acceptable:
|
||||
if check.coverage < 0.7:
|
||||
warnings.append(f"Gemini a mal lu le texte ({check.describe()}) : voix Edge utilisée à la place.")
|
||||
return None
|
||||
warnings.append(f"Lecture Gemini imparfaite ({check.describe()}) : écoutez le résultat.")
|
||||
return processed, spoken, gemini_voice, check.acceptable
|
||||
|
||||
|
||||
# Au-delà de la fin du dernier mot, ce n'est plus de la parole
|
||||
STRAY_AFTER_LAST_WORD = 0.6
|
||||
STRAY_BEFORE_FIRST_WORD = 0.4
|
||||
|
||||
|
||||
async def _trim_stray_sounds(path: Path, spoken: list[Word], workdir: Path) -> tuple[Path, list[Word]]:
|
||||
"""Retire les bruits que Gemini ajoute parfois avant le premier ou après le
|
||||
dernier mot. Appelé seulement quand la lecture a été vérifiée : le dernier
|
||||
mot est bien reconnu, on ne coupe donc jamais de parole."""
|
||||
if not spoken:
|
||||
return path, spoken
|
||||
envelope = await quality.load_envelope(path)
|
||||
duration = envelope.duration
|
||||
window_start = max(0.0, spoken[0].start - STRAY_BEFORE_FIRST_WORD)
|
||||
window_end = min(duration, spoken[-1].end + STRAY_AFTER_LAST_WORD)
|
||||
start, end = envelope.speech_span(window_start, window_end)
|
||||
if start < 0.02 and end > duration - 0.02:
|
||||
return path, spoken
|
||||
|
||||
trimmed = workdir / "voice_trimmed.wav"
|
||||
fade_out = max(0.0, end - start - 0.01)
|
||||
await proc.run(
|
||||
[
|
||||
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-i", str(path),
|
||||
"-af", f"atrim=start={start:.3f}:end={end:.3f},asetpts=PTS-STARTPTS,"
|
||||
f"afade=t=in:d=0.01,afade=t=out:st={fade_out:.3f}:d=0.01",
|
||||
"-ac", "1", "-ar", "48000", "-c:a", "pcm_s16le", str(trimmed),
|
||||
],
|
||||
timeout=60,
|
||||
) # fmt: skip
|
||||
log.info("Voix recadrée sur la parole : %.2f–%.2f s (sur %.2f s)", start, end, duration)
|
||||
shifted = [Word(w.text, w.start - start, w.end - start) for w in spoken if w.start < end]
|
||||
return trimmed, shifted
|
||||
|
||||
|
||||
def _score(check: quality.ReadingCheck) -> float:
|
||||
return check.coverage - check.extra + (0.2 if check.ending_ok else 0.0)
|
||||
@@ -21,19 +21,29 @@ class GeminiError(RuntimeError):
|
||||
|
||||
|
||||
def build_prompt(text: str, style: str | None) -> str:
|
||||
"""Texte envoyé au modèle : la consigne de style précède le texte à lire."""
|
||||
"""Texte envoyé au modèle : la consigne de style précède le texte à lire.
|
||||
La consigne demande une lecture intégrale, sans ajout : le modèle sautait
|
||||
parfois des mots ou tronquait la fin."""
|
||||
instruction = style_instruction(style)
|
||||
return f"{instruction} :\n{text}" if instruction else text
|
||||
if not instruction:
|
||||
return text
|
||||
return f"{instruction}. Lis tout le texte, mot pour mot, sans rien ajouter :\n{text}"
|
||||
|
||||
|
||||
async def _request(model: str, prompt: str, voice: str, api_key: str) -> dict:
|
||||
payload = {
|
||||
"contents": [{"parts": [{"text": prompt}]}],
|
||||
"generationConfig": {
|
||||
"responseModalities": ["AUDIO"],
|
||||
"speechConfig": {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": voice}}},
|
||||
},
|
||||
# Paramètres de tirage refusés par un modèle : on ne les renvoie plus
|
||||
_unsupported_sampling: set[str] = set()
|
||||
|
||||
|
||||
async def _request(model: str, prompt: str, voice: str, api_key: str, seed: int) -> dict:
|
||||
generation: dict = {
|
||||
"responseModalities": ["AUDIO"],
|
||||
"speechConfig": {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": voice}}},
|
||||
}
|
||||
if model not in _unsupported_sampling:
|
||||
# Intonation stable d'une génération à l'autre
|
||||
generation["temperature"] = config.GEMINI_TTS_TEMPERATURE
|
||||
generation["seed"] = seed
|
||||
payload = {"contents": [{"parts": [{"text": prompt}]}], "generationConfig": generation}
|
||||
async with httpx.AsyncClient(timeout=120) as client:
|
||||
# Clé en en-tête : dans l'URL, elle finissait dans les journaux
|
||||
response = await client.post(
|
||||
@@ -41,27 +51,37 @@ async def _request(model: str, prompt: str, voice: str, api_key: str) -> dict:
|
||||
json=payload,
|
||||
headers={"x-goog-api-key": api_key},
|
||||
)
|
||||
rejected_sampling = response.status_code == 400 and any(
|
||||
word in response.text.lower() for word in ("seed", "temperature", "unknown name", "invalid json")
|
||||
)
|
||||
if rejected_sampling and "temperature" in generation:
|
||||
# Modèle qui refuse température ou graine : même requête sans eux
|
||||
log.warning("Gemini %s refuse les paramètres de tirage : %s", model, response.text[:200])
|
||||
_unsupported_sampling.add(model)
|
||||
return await _request(model, prompt, voice, api_key, seed)
|
||||
if response.status_code != 200:
|
||||
raise GeminiError(f"Gemini {model} : HTTP {response.status_code} {response.text[:300]}")
|
||||
return response.json()
|
||||
|
||||
|
||||
async def synthesize(
|
||||
text: str, voice: str | None, style: str | None, api_key: str, workdir: Path
|
||||
text: str, voice: str | None, style: str | None, api_key: str, workdir: Path, attempt: int = 0
|
||||
) -> tuple[Path, str]:
|
||||
"""Génère la voix ; renvoie un WAV brut et le nom de la voix utilisée."""
|
||||
"""Génère la voix ; renvoie un WAV brut et le nom de la voix utilisée.
|
||||
Chaque nouvelle tentative change de graine pour obtenir une autre lecture."""
|
||||
seed = config.GEMINI_TTS_SEED + attempt
|
||||
gemini_voice = resolve_gemini_voice(voice).id
|
||||
prompt = build_prompt(text, style)
|
||||
model = config.GEMINI_TTS_MODEL
|
||||
log.info("Gemini TTS : modèle=%s voix=%s style=%s (%d car.)", model, gemini_voice, style, len(text))
|
||||
|
||||
try:
|
||||
data = await _request(model, prompt, gemini_voice, api_key)
|
||||
data = await _request(model, prompt, gemini_voice, api_key, seed)
|
||||
except GeminiError as error:
|
||||
if model == FALLBACK_MODEL or "HTTP 404" not in str(error):
|
||||
raise
|
||||
log.warning("Modèle %s indisponible, repli sur %s", model, FALLBACK_MODEL)
|
||||
data = await _request(FALLBACK_MODEL, prompt, gemini_voice, api_key)
|
||||
data = await _request(FALLBACK_MODEL, prompt, gemini_voice, api_key, seed)
|
||||
|
||||
try:
|
||||
part = next(p for p in data["candidates"][0]["content"]["parts"] if "inlineData" in p)["inlineData"]
|
||||
@@ -70,14 +90,14 @@ async def synthesize(
|
||||
|
||||
audio = base64.b64decode(part["data"])
|
||||
mime = part.get("mimeType", "")
|
||||
raw = workdir / "gemini_raw.wav"
|
||||
raw = workdir / f"gemini_raw_{attempt}.wav"
|
||||
|
||||
if "wav" in mime:
|
||||
raw.write_bytes(audio)
|
||||
else:
|
||||
# PCM brut 16 bits (« audio/L16;codec=pcm;rate=24000 »)
|
||||
rate_match = re.search(r"rate=(\d+)", mime)
|
||||
pcm = workdir / "gemini.pcm"
|
||||
pcm = workdir / f"gemini_{attempt}.pcm"
|
||||
pcm.write_bytes(audio)
|
||||
await proc.run(
|
||||
[
|
||||
|
||||
@@ -5,3 +5,4 @@ httpx==0.28.1
|
||||
edge-tts==7.2.8
|
||||
emoji==2.14.1
|
||||
faster-whisper==1.2.1
|
||||
numpy>=1.26
|
||||
@@ -0,0 +1,105 @@
|
||||
import numpy as np
|
||||
|
||||
from app.align import Word
|
||||
from app.quality import FRAME, Envelope, check_reading, envelope_from_samples, sentence_cut
|
||||
|
||||
|
||||
def _heard(text: str) -> list[Word]:
|
||||
return [Word(t, i * 0.3, i * 0.3 + 0.25) for i, t in enumerate(text.split())]
|
||||
|
||||
|
||||
def test_complete_reading_is_accepted():
|
||||
check = check_reading(
|
||||
"Découvrez nos nouveautés en magasin", _heard("découvrez nos nouveautés en magasin")
|
||||
)
|
||||
assert check.acceptable and check.coverage == 1.0
|
||||
|
||||
|
||||
def test_truncated_ending_is_rejected():
|
||||
check = check_reading("Découvrez nos nouveautés en magasin", _heard("découvrez nos nouveautés"))
|
||||
assert not check.ending_ok and not check.acceptable
|
||||
|
||||
|
||||
def test_instruction_read_aloud_is_rejected():
|
||||
heard = _heard("lis ce texte sur un ton dynamique découvrez nos nouveautés en magasin")
|
||||
assert not check_reading("Découvrez nos nouveautés en magasin", heard).acceptable
|
||||
|
||||
|
||||
def _envelope(levels):
|
||||
"""Enveloppe synthétique : liste de (durée en s, niveau en dB)."""
|
||||
db = []
|
||||
for seconds, level in levels:
|
||||
db += [level] * int(round(seconds / FRAME))
|
||||
return Envelope(np.array(db, dtype=float))
|
||||
|
||||
|
||||
def test_soft_sentence_endings_are_kept():
|
||||
# Parole à -20 dB, fin de phrase douce à -50 dB (relatif : -30 dB, audible),
|
||||
# puis vrai silence à -90 dB
|
||||
env = _envelope([(0.2, -90), (1.0, -20), (0.3, -50), (0.6, -90)])
|
||||
start, end = env.speech_span(0.0, env.duration)
|
||||
assert round(start, 2) == 0.12 # 80 ms avant la parole
|
||||
assert round(end, 2) == 1.8 # 300 ms après la fin douce (à 1,5 s)
|
||||
|
||||
|
||||
def test_cut_falls_in_the_quietest_part_of_the_pause():
|
||||
env = _envelope([(1.0, -20), (0.15, -60), (0.2, -85), (0.15, -60), (1.0, -20)])
|
||||
cut = sentence_cut(env, previous_end=0.95, next_start=1.55)
|
||||
assert 1.15 <= cut <= 1.35
|
||||
|
||||
|
||||
def test_envelope_from_samples_measures_level():
|
||||
rate = 16000
|
||||
samples = np.concatenate([np.zeros(rate // 2), 0.5 * np.ones(rate // 2)]).astype(np.float32)
|
||||
env = envelope_from_samples(samples, rate)
|
||||
assert env.db[0] < -100 and abs(env.db[-1] - 20 * np.log10(0.5)) < 0.1
|
||||
|
||||
|
||||
def test_small_recognition_errors_are_not_missing_words():
|
||||
expected = "Découvrez nos plantes de saison à petits prix. On vous attend ce week-end."
|
||||
heard = _heard("Découvrez nos plans de saison à petit prix. On vous attend ce week -end.")
|
||||
check = check_reading(expected, heard)
|
||||
assert check.acceptable, check
|
||||
|
||||
|
||||
def test_skipped_words_are_detected_and_named():
|
||||
expected = (
|
||||
"Pour Halloween découvrez nos costumes nos bonbons et toute notre décoration effrayante en magasin."
|
||||
)
|
||||
heard = _heard("Pour Halloween découvrez nos costumes et toute notre décoration en magasin.")
|
||||
check = check_reading(expected, heard)
|
||||
assert not check.acceptable
|
||||
assert "bonbons" in check.missing and "effrayante" in check.missing
|
||||
assert "non entendus" in check.describe()
|
||||
|
||||
|
||||
def test_prices_and_units_written_differently_are_not_missing_words():
|
||||
expected = "Alerte léopard ! Le coussin de 40 centimètres à 11,99 €. Le cabas à 13,99 €."
|
||||
heard = _heard("Alerte léopard ! Le coussin de 40 cm à 11 € 99. Le cabas à 13,99 euros.")
|
||||
check = check_reading(expected, heard)
|
||||
assert check.acceptable, check
|
||||
assert check.missing == []
|
||||
|
||||
|
||||
def test_skipped_word_next_to_a_price_is_still_detected():
|
||||
expected = "Le coussin léopard à 11,99 €. Le cabas doux et chaud à 13,99 €."
|
||||
heard = _heard("Le coussin à 11,99 €. Le cabas à 13,99 €.")
|
||||
check = check_reading(expected, heard)
|
||||
assert not check.acceptable
|
||||
assert "léopard" in check.missing
|
||||
|
||||
|
||||
def test_cut_uses_the_real_pause_when_word_timing_is_estimated():
|
||||
# Phrase 1 jusqu'à 2,0 s (dont un prix non reconnu), virgule à 1,0 s,
|
||||
# pause de fin de phrase 2,0–2,4 s. Whisper place la fin de la phrase
|
||||
# et le début de la suivante vers 2,9 s (mots interpolés).
|
||||
env = _envelope([(0.3, -90), (0.7, -20), (0.1, -90), (0.9, -20), (0.4, -90), (1.5, -20), (0.3, -90)])
|
||||
cut = sentence_cut(env, previous_end=2.9, next_start=2.95)
|
||||
assert 2.0 <= cut <= 2.4
|
||||
|
||||
|
||||
def test_cut_prefers_the_sentence_pause_over_a_comma():
|
||||
env = _envelope([(0.3, -90), (1.0, -20), (0.12, -90), (0.8, -20), (0.45, -90), (1.0, -20), (0.3, -90)])
|
||||
# Borne estimée entre la virgule (1,3 s) et la fin de phrase (2,22 s)
|
||||
cut = sentence_cut(env, previous_end=1.7, next_start=1.75)
|
||||
assert 2.22 <= cut <= 2.67
|
||||
@@ -51,7 +51,50 @@ def test_original_audio_kept_when_nothing_added():
|
||||
video=Path("in.mp4"), video_duration=5.0, output=Path("out.mp4"), keep_original_audio=True
|
||||
)
|
||||
cmd = build_command(plan)
|
||||
assert cmd[cmd.index("-map", cmd.index("[vout]")) + 1] == "0:a:0"
|
||||
assert "[0:a:0]aresample=48000,apad[bed]" in _graph(cmd)
|
||||
assert cmd[cmd.index("-map", cmd.index("[vout]")) + 1] == "[aout]"
|
||||
|
||||
|
||||
def test_original_audio_stays_under_the_voice_when_no_music_is_chosen():
|
||||
plan = RenderPlan(
|
||||
video=Path("in.mp4"),
|
||||
video_duration=10.0,
|
||||
output=Path("out.mp4"),
|
||||
voice=Path("v.wav"),
|
||||
voice_duration=4.0,
|
||||
keep_original_audio=True,
|
||||
)
|
||||
graph = _graph(build_command(plan))
|
||||
assert "[0:a:0]aresample=48000,apad[bed]" in graph
|
||||
assert "[bed][vkey]sidechaincompress" in graph
|
||||
|
||||
|
||||
def test_chosen_music_replaces_the_original_audio():
|
||||
plan = RenderPlan(
|
||||
video=Path("in.mp4"),
|
||||
video_duration=10.0,
|
||||
output=Path("out.mp4"),
|
||||
music=Path("m.mp3"),
|
||||
keep_original_audio=True,
|
||||
)
|
||||
assert "0:a:0" not in _graph(build_command(plan))
|
||||
|
||||
|
||||
def test_prepared_audio_mix_keeps_original_audio_under_voice():
|
||||
from app.render import build_audio_mix_command
|
||||
|
||||
plan = RenderPlan(
|
||||
video=Path("in.mp4"),
|
||||
video_duration=10.0,
|
||||
output=Path("out.mp4"),
|
||||
voice=Path("v.wav"),
|
||||
voice_duration=4.0,
|
||||
keep_original_audio=True,
|
||||
)
|
||||
cmd = build_audio_mix_command(plan, Path("a.wav"))
|
||||
# Entrées : voix (0) puis vidéo d'origine (1)
|
||||
assert cmd[cmd.index("in.mp4") - 1] == "-i"
|
||||
assert "[1:a:0]aresample=48000,apad[bed]" in " ".join(cmd)
|
||||
|
||||
|
||||
def test_encoding_targets_social_networks():
|
||||
@@ -111,3 +154,18 @@ def test_audio_mix_absent_without_any_sound():
|
||||
assert build_audio_mix_command(plan, Path("a.wav")) is None
|
||||
plan.keep_original_audio = True
|
||||
assert "loudnorm=I=-14" in " ".join(build_audio_mix_command(plan, Path("a.wav")))
|
||||
|
||||
|
||||
def test_final_fade_never_starts_before_the_voice_ends():
|
||||
# Vidéo courte sans logo : 2 s + 5 s de voix, la vidéo finit 0,8 s après.
|
||||
plan = RenderPlan(
|
||||
video=Path("in.mp4"),
|
||||
video_duration=4.0,
|
||||
output=Path("out.mp4"),
|
||||
voice=Path("v.wav"),
|
||||
voice_duration=5.0,
|
||||
)
|
||||
assert plan.total_duration == 7.8
|
||||
assert plan.fade_start >= plan.speech_end
|
||||
assert plan.fade_duration >= 0.3
|
||||
assert f"afade=t=out:st={plan.fade_start:.3f}" in _graph(build_command(plan))
|
||||
@@ -1,8 +1,20 @@
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
from app.align import Word, spread_words
|
||||
from app.quality import Envelope
|
||||
from app.render import RenderPlan
|
||||
from app.srt_voice import Cue, atempo_chain, cue_windows, mix_command, speed_factor, spoken_span
|
||||
from app.srt_voice import (
|
||||
Cue,
|
||||
as_sentence,
|
||||
atempo_chain,
|
||||
cue_windows,
|
||||
mix_command,
|
||||
segments_from_reading,
|
||||
speed_factor,
|
||||
split_by_cue,
|
||||
)
|
||||
|
||||
|
||||
def test_window_stops_at_next_cue_and_tolerates_a_short_overflow():
|
||||
@@ -20,10 +32,28 @@ def test_atempo_is_split_into_supported_steps():
|
||||
assert atempo_chain(3.0) == "atempo=2.0000,atempo=1.5000"
|
||||
|
||||
|
||||
def test_spoken_span_trims_silences_around_words():
|
||||
words = [Word("Bonjour", 0.4, 0.9), Word("!", 0.9, 1.2)]
|
||||
assert [round(t, 3) for t in spoken_span(words, 2.0)] == [0.35, 1.32]
|
||||
assert spoken_span([], 2.0) == (0.0, 2.0)
|
||||
def test_each_cue_becomes_a_sentence_for_the_reading():
|
||||
assert as_sentence("Venez vite") == "Venez vite."
|
||||
assert as_sentence("Promo !") == "Promo !"
|
||||
|
||||
|
||||
def test_words_of_the_single_reading_are_split_between_cues():
|
||||
words = [Word(t, i, i + 0.5) for i, t in enumerate(["Bonjour.", "Venez", "vite."])]
|
||||
groups = split_by_cue(words, ["Bonjour.", "Venez vite."])
|
||||
assert [[w.text for w in g] for g in groups] == [["Bonjour."], ["Venez", "vite."]]
|
||||
assert split_by_cue(words, ["Bonjour."]) is None
|
||||
|
||||
|
||||
def test_reading_is_cut_in_the_pause_without_clipping_sentence_endings():
|
||||
# Whisper situe mal la fin de « Bonjour » (0,9 s) : la voix retombe
|
||||
# doucement jusqu'à 1,05 s, puis vraie pause jusqu'à 1,6 s.
|
||||
db = [-90.0] * 15 + [-20.0] * 75 + [-50.0] * 15 + [-90.0] * 55 + [-20.0] * 100 + [-90.0] * 40
|
||||
env = Envelope(np.array(db))
|
||||
groups = [[Word("Bonjour.", 0.2, 0.9)], [Word("Venez", 1.7, 2.1), Word("vite.", 2.1, 2.6)]]
|
||||
first, second = segments_from_reading(groups, env)
|
||||
assert first.source_end >= 1.05 + 0.25 # fin douce gardée, avec sa marge
|
||||
assert first.source_end <= second.source_start
|
||||
assert 1.5 <= second.source_start < 1.6
|
||||
|
||||
|
||||
def test_each_segment_is_placed_at_its_cue_start():
|
||||
|
||||
@@ -55,6 +55,9 @@ export interface PreparedReel {
|
||||
totalDuration: number;
|
||||
videoDuration: number;
|
||||
logoStart: number | null;
|
||||
/** Fondu final, calculé par le service pour ne jamais rogner la voix. */
|
||||
fadeStart: number | null;
|
||||
fadeDuration: number | null;
|
||||
words: TimedWord[];
|
||||
ttsEngine?: string | null;
|
||||
ttsVoice?: string | null;
|
||||
@@ -226,6 +229,8 @@ export class FFmpegService {
|
||||
total_duration: number;
|
||||
video_duration: number;
|
||||
logo_start: number | null;
|
||||
fade_start?: number | null;
|
||||
fade_duration?: number | null;
|
||||
words: TimedWord[];
|
||||
tts_engine?: string | null;
|
||||
tts_voice?: string | null;
|
||||
@@ -248,6 +253,8 @@ export class FFmpegService {
|
||||
totalDuration: data.total_duration,
|
||||
videoDuration: data.video_duration,
|
||||
logoStart: data.logo_start,
|
||||
fadeStart: data.fade_start ?? null,
|
||||
fadeDuration: data.fade_duration ?? null,
|
||||
words: data.words ?? [],
|
||||
ttsEngine: data.tts_engine,
|
||||
ttsVoice: data.tts_voice,
|
||||
|
||||
@@ -12,7 +12,6 @@
|
||||
|
||||
import fs from "fs";
|
||||
import path from "path";
|
||||
import { FADE_SECONDS } from "@shared/captions";
|
||||
import { videoReelParamsSchema, type VideoReelParams } from "@shared/reel";
|
||||
import { storage } from "../../storage";
|
||||
import { ffmpegService, type ReelRenderOptions } from "../ffmpeg";
|
||||
@@ -139,7 +138,8 @@ async function renderWithRemotion(
|
||||
showWatermark: params.showLogo,
|
||||
storeName: params.enableEndingEffect ? params.storeName : undefined,
|
||||
logoStart: prepared.logoStart,
|
||||
fadeStart: params.enableEndingEffect ? Math.max(0, prepared.totalDuration - FADE_SECONDS) : null,
|
||||
fadeStart: prepared.fadeStart,
|
||||
fadeDuration: prepared.fadeDuration,
|
||||
mixedAudioUrl: prepared.audioPath ? tempFileUrl(prepared.audioPath) : undefined,
|
||||
},
|
||||
onProgress: (ratio) => {
|
||||
|
||||
+10
-1
@@ -24,7 +24,16 @@ describe("computeReelTiming", () => {
|
||||
|
||||
it("n'a ni logo ni fondu sans effet de fin", () => {
|
||||
const timing = computeReelTiming({ videoDuration: 12, hasOutro: true, endingEffect: false });
|
||||
expect(timing).toEqual({ total: 12, logoStart: null, fadeStart: null });
|
||||
expect(timing).toEqual({ total: 12, logoStart: null, fadeStart: null, fadeDuration: null });
|
||||
});
|
||||
});
|
||||
|
||||
describe("fondu final", () => {
|
||||
it("ne commence jamais avant la fin de la voix (mêmes valeurs que Python)", () => {
|
||||
const timing = computeReelTiming({ videoDuration: 4, voiceDuration: 5, hasOutro: false, endingEffect: true });
|
||||
expect(timing.total).toBe(7.8);
|
||||
expect(timing.fadeStart).toBeGreaterThanOrEqual(7);
|
||||
expect(timing.fadeDuration).toBeGreaterThanOrEqual(0.3);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
+9
-1
@@ -32,12 +32,15 @@ const VOICE_TAIL = 0.8;
|
||||
const OUTRO_MIN = 2.5;
|
||||
const LOGO_SECONDS = 5;
|
||||
export const FADE_SECONDS = 2;
|
||||
/** Fondu raccourci quand la voix finit tard, mais jamais sec. */
|
||||
const MIN_FADE = 0.3;
|
||||
|
||||
export interface ReelTiming {
|
||||
total: number;
|
||||
/** Début de l'effet de fin (grand logo), ou null sans effet de fin. */
|
||||
logoStart: number | null;
|
||||
fadeStart: number | null;
|
||||
fadeDuration: number | null;
|
||||
}
|
||||
|
||||
/** Durée finale : la vidéo s'allonge (dernière image figée) si la voix dépasse. */
|
||||
@@ -57,10 +60,15 @@ export function computeReelTiming(input: {
|
||||
if (hasOutro) total = Math.max(total, speechEnd + OUTRO_MIN);
|
||||
}
|
||||
total = Math.round(total * 1000) / 1000;
|
||||
// Le fondu final ne commence jamais avant la fin de la voix
|
||||
let fadeStart = Math.max(0, total - FADE_SECONDS);
|
||||
if (input.voiceDuration) fadeStart = Math.max(fadeStart, Math.min(speechEnd + 0.15, total - MIN_FADE));
|
||||
fadeStart = Math.round(fadeStart * 1000) / 1000;
|
||||
return {
|
||||
total,
|
||||
logoStart: hasOutro ? Math.max(0, total - LOGO_SECONDS, speechEnd) : null,
|
||||
fadeStart: input.endingEffect ? Math.max(0, total - FADE_SECONDS) : null,
|
||||
fadeStart: input.endingEffect ? fadeStart : null,
|
||||
fadeDuration: input.endingEffect ? Math.max(MIN_FADE, total - fadeStart) : null,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user