Compare commits

...
5 Commits
Author SHA1 Message Date
Claude 4b8fbe6d8e fix(voix SRT): prix et unités reconnus, coupes dans les vraies pauses
- Contrôle de lecture : les nombres et l'unité qui les suit ne sont plus
  comparés lettre à lettre. Whisper écrit « 11 € 99 » ou « 40 cm » là où le
  texte dit « 11,99 € » ou « 40 centimètres » : une lecture correcte passait
  pour des mots sautés (88–91 %) et relançait jusqu'à 3 générations.
- Découpe de la lecture SRT : la coupe entre deux sous-titres se fait dans
  la pause mesurée la plus proche de la borne estimée (fins de phrases
  préférées aux virgules), et non plus à la borne Whisper elle-même, fausse
  quand un prix non reconnu termine la phrase : la coupe tombait alors au
  milieu d'une phrase.
- Whisper reçoit 0,5 s de silence avant la voix : il laissait passer le
  premier mot (« Alerte ») d'une voix qui démarre aussitôt.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01X3ywVwJUjjUXuvM6PwSzSi
2026-09-28 16:44:41 +00:00
Claude 7c38ec459f fix(voix): refuser les lectures Gemini qui sautent des mots
Le contrôle lettre à lettre acceptait une lecture dès 85 %, alors qu'une
lecture complète atteint ~97 % (petites erreurs de Whisper) : une lecture à
86 % (mots sautés) passait sans nouvelle génération. Seuil porté à 92 %.

Les mots non entendus sont nommés dans les journaux et l'avertissement,
pour savoir ce qui manquait. Le modèle Whisper, déjà dans l'image, est
chargé sans interroger Hugging Face (HF_HUB_OFFLINE).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018Ze4bs7tpF1KGWUk6ZZSZ4
2026-09-28 09:25:43 +00:00
Claude d4c544e467 fix(voix): plus de fins de phrases coupées ni de bruits parasites
Fins de phrases coupées :
- Le fondu final ne commence plus avant la fin de la voix (il rognait la
  dernière phrase quand la vidéo finissait peu après) ; il se raccourcit au
  besoin. Même règle côté Python, aperçu et rendu Remotion
- SRT : découpe au milieu de la zone la plus calme de chaque pause, mesurée
  sur l'enveloppe sonore (tranches de 10 ms), avec un seuil relatif au
  niveau de la voix qui respecte les fins douces ; 300 ms gardées après le
  dernier son

Bruits parasites :
- Normalisation de la voix en deux passes, en mode linéaire : le mode
  dynamique relevait souffles et bruits entre les phrases (mesuré : -64 dB
  avant, -75 dB après dans les pauses)
- Réduction de bruit douce (afftdn), compresseur sans gain ajouté, fondus
  de 5 ms aux extrémités ; mix final moins « pompant » (LRA 20)
- Voix Gemini vérifiée : les bruits ajoutés avant le premier ou après le
  dernier mot sont retirés

Vérification de lecture comparée lettre à lettre : les petites erreurs de
Whisper ne déclenchent plus de régénérations inutiles.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018Ze4bs7tpF1KGWUk6ZZSZ4
2026-09-28 09:21:18 +00:00
Claude 44899055b2 fix(voix): intonation stable, lecture vérifiée et SRT sans mots coupés
Intonation :
- Gemini reçoit une température basse et une graine fixe (réglables) : la
  lecture ne change plus de ton d'une génération à l'autre
- SRT : tout le texte est lu d'une seule traite puis découpé phrase par
  phrase, au lieu d'une génération (et d'une intonation) par sous-titre ;
  un seul traitement de niveau, plus de sauts de volume entre les phrases

Mots coupés :
- La voix est découpée dans ses vrais silences (silencedetect), plus aux
  bornes des mots estimées par Whisper qui rognaient les syllabes ; fondus
  de 10 ms aux coupes
- Whisper n'est plus amorcé avec le texte attendu : il hallucinait des
  répétitions qui masquaient les fins tronquées

Fiabilité :
- Chaque voix Gemini est réécoutée : mots sautés, fin tronquée ou consigne
  lue à voix haute déclenchent une nouvelle génération (3 au plus), puis
  repli sur Edge avec avertissement
- Consigne de lecture intégrale ajoutée au prompt

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018Ze4bs7tpF1KGWUk6ZZSZ4
2026-09-28 09:04:08 +00:00
Claude f3113a0f5a fix(reels): garder le son d'origine de la vidéo quand aucune musique n'est choisie
Sans musique (bouton « Passer »), la bande son de la vidéo était supprimée
dès que la voix était activée. Elle sert désormais de fond sonore : conservée
telle quelle, baissée automatiquement sous la voix, puis normalisée avec le
reste. Une musique choisie la remplace toujours.

Même comportement dans l'aperçu, le rendu Remotion et le rendu FFmpeg de
secours. Le bouton indique « Passer (garder le son de la vidéo) ».

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018Ze4bs7tpF1KGWUk6ZZSZ4
2026-09-28 08:57:38 +00:00
23 changed files with 910 additions and 205 deletions

No files matched your search

+7
View File
@@ -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,
+2 -2
View File
@@ -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>
+22 -8
View File
@@ -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",
});
+2
View File
@@ -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
+19 -6
View File
@@ -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)
+3 -1
View File
@@ -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
View File
@@ -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:
+6
View File
@@ -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")
+4 -3
View File
@@ -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
+243
View File
@@ -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)
+50 -36
View File
@@ -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
View File
@@ -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]:
+91 -13
View File
@@ -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)
+35 -15
View File
@@ -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(
[
+1
View File
@@ -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
+105
View File
@@ -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
+59 -1
View File
@@ -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))
+35 -5
View File
@@ -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():
+7
View File
@@ -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,
+2 -2
View File
@@ -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
View File
@@ -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
View File
@@ -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,
};
}