Merge commit 'd2293f8' into claude/intelligent-wright-pr07e9

# Conflicts:
#	server/src/services/ffmpeg.rs
This commit is contained in:
Claude committed 2026-10-12 17:37:13 +00:00
commit 2bba946f34
15 files changed
+6624 -16

No files matched your search

+3
View File
@@ -0,0 +1,3 @@
node_modules
# Bundle construit par `node server.mjs --prepare`
bundle
+66
View File
@@ -0,0 +1,66 @@
# Service de rendu Remotion des Reels (sous-titres animés, logo, effet de fin).
# Contexte de build : la racine du dépôt (les compositions vivent dans
# client/src/remotion et importent shared/captions.ts).
# docker build -f remotion-service/Dockerfile .
# Debian slim — glibc requis par le Chrome headless shell de Remotion
FROM node:20-slim
# Dépendances du Chrome headless shell et polices de secours
RUN apt-get update && apt-get install -y --no-install-recommends \
ca-certificates \
fonts-dejavu-core \
fontconfig \
libasound2 \
libatk-bridge2.0-0 \
libatk1.0-0 \
libcups2 \
libdbus-1-3 \
libdrm2 \
libgbm1 \
libglib2.0-0 \
libgtk-3-0 \
libnspr4 \
libnss3 \
libpango-1.0-0 \
libx11-6 \
libx11-xcb1 \
libxcb1 \
libxcomposite1 \
libxdamage1 \
libxext6 \
libxfixes3 \
libxkbcommon0 \
libxrandr2 \
libxshmfence1 \
libxtst6 \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
# Dépendances (versions épinglées, identiques au package.json racine)
COPY remotion-service/package.json remotion-service/package-lock.json ./
RUN npm ci --omit=dev --no-audit --no-fund
# Compositions et module partagé qu'elles importent (@shared/captions).
# La police Montserrat vient du paquet @fontsource/montserrat.
COPY client/src/remotion ./client/src/remotion
COPY shared/captions.ts ./shared/captions.ts
COPY remotion-service/server.mjs ./server.mjs
# Pré-bundle des compositions et téléchargement du Chrome headless shell
# (équivalent de `npx remotion browser ensure`) : aucun accès réseau au
# premier rendu.
RUN node server.mjs --prepare && rm -rf /tmp/remotion-*
ENV NODE_ENV=production \
PORT=8100 \
OUTPUT_DIR=/tmp/remotion-renders
# Utilisateur non privilégié ; seuls /tmp et le cache Chrome sont utilisés
RUN chown -R node:node /app
USER node
EXPOSE 8100
CMD ["node", "server.mjs"]
+2931
View File
File diff suppressed because it is too large. Load diff
+24
View File
@@ -0,0 +1,24 @@
{
"name": "socialflow-remotion-service",
"version": "1.0.0",
"private": true,
"description": "Service de rendu Remotion des Reels SocialFlow (sous-titres animés, logo, effet de fin)",
"type": "module",
"license": "MIT",
"main": "server.mjs",
"scripts": {
"start": "node server.mjs",
"prepare-render": "node server.mjs --prepare"
},
"engines": {
"node": ">=20"
},
"dependencies": {
"@fontsource/montserrat": "5.3.0",
"@remotion/bundler": "4.0.438",
"@remotion/renderer": "4.0.438",
"react": "18.3.1",
"react-dom": "18.3.1",
"remotion": "4.0.438"
}
}
+349
View File
@@ -0,0 +1,349 @@
#!/usr/bin/env node
/**
* Service de rendu Remotion des Reels SocialFlow.
*
* Le serveur principal (Rust) ne peut pas exécuter Remotion (React +
* Chromium) : il lui confie la composition « ReelVideo » (sous-titres animés,
* logo, effet de fin) avec la vidéo et le son déjà préparés par le service
* FFmpeg, puis récupère le MP4.
*
* API (en-tête X-API-Key obligatoire, variable API_KEY) :
* POST /renders { compositionId, inputProps } → { id }
* GET /renders/:id → { status, progress (0..1), error? }
* GET /renders/:id/file → le MP4
* DELETE /renders/:id → annule le rendu et supprime le fichier
* GET /health → { status: "ok" }
*
* `node server.mjs --prepare` construit le bundle et télécharge le Chrome
* headless shell (étape de build Docker), puis s'arrête.
*/
import crypto from "node:crypto";
import fs from "node:fs";
import http from "node:http";
import os from "node:os";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { bundle } from "@remotion/bundler";
import { ensureBrowser, makeCancelSignal, renderMedia, selectComposition } from "@remotion/renderer";
const HERE = path.dirname(fileURLToPath(import.meta.url));
/** Racine contenant client/src/remotion et shared/ : l'image Docker, sinon le dépôt. */
const SOURCE_ROOT =
process.env.REMOTION_SOURCE_ROOT ||
(fs.existsSync(path.join(HERE, "client/src/remotion")) ? HERE : path.resolve(HERE, ".."));
const ENTRY_POINT = path.join(SOURCE_ROOT, "client/src/remotion/index.ts");
const BUNDLE_DIR = process.env.BUNDLE_DIR || path.join(HERE, "bundle");
const OUTPUT_DIR = process.env.OUTPUT_DIR || path.join(os.tmpdir(), "remotion-renders");
const PORT = Number(process.env.PORT) || 8100;
const API_KEY = process.env.API_KEY || "";
/** Rendus simultanés (chacun occupe déjà plusieurs onglets Chromium). */
const MAX_PARALLEL_RENDERS = Math.max(1, Number(process.env.MAX_PARALLEL_RENDERS) || 1);
/** Les props contiennent le logo en data URL. */
const MAX_BODY_BYTES = 32 * 1024 * 1024;
/** Rendus terminés conservés au plus une heure s'ils ne sont pas supprimés. */
const RESULT_TTL_MS = 60 * 60_000;
const log = (...args) => console.log(new Date().toISOString(), ...args);
const warn = (...args) => console.warn(new Date().toISOString(), ...args);
// ---------------------------------------------------------------------------
// Bundle des compositions
// ---------------------------------------------------------------------------
function bundleOptions(outDir) {
return {
entryPoint: ENTRY_POINT,
outDir,
// Mêmes alias que Vite : les compositions importent @shared/*
webpackOverride: (config) => ({
...config,
resolve: {
...config.resolve,
alias: {
...(config.resolve?.alias ?? {}),
"@shared": path.join(SOURCE_ROOT, "shared"),
},
// Dépendances (react, remotion, polices) installées à côté de ce service
modules: [...(config.resolve?.modules ?? ["node_modules"]), path.join(HERE, "node_modules")],
},
}),
};
}
let bundlePromise = null;
/** Bundle construit au build Docker, sinon construit une fois par processus. */
function getBundle() {
if (fs.existsSync(path.join(BUNDLE_DIR, "index.html"))) return Promise.resolve(BUNDLE_DIR);
if (!bundlePromise) {
log("📦 [Remotion] Bundle de", ENTRY_POINT);
bundlePromise = bundle(bundleOptions(null)).catch((error) => {
bundlePromise = null; // réessayer au prochain rendu
throw error;
});
}
return bundlePromise;
}
/** Onglets Chromium en parallèle : par défaut la moitié des cœurs, 1 au minimum. */
function renderConcurrency() {
const configured = Number(process.env.RENDER_CONCURRENCY);
if (Number.isFinite(configured) && configured > 0) return Math.floor(configured);
return Math.max(1, Math.floor(os.cpus().length / 2));
}
// ---------------------------------------------------------------------------
// Rendus
// ---------------------------------------------------------------------------
/** @type {Map<string, {id: string, compositionId: string, inputProps: object, status: string, progress: number, error?: string, output: string, finishedAt?: number, cancel?: () => void}>} */
const renders = new Map();
const waiting = [];
let running = 0;
function enqueue(compositionId, inputProps) {
const id = crypto.randomUUID();
const render = {
id,
compositionId,
inputProps,
status: "pending",
progress: 0,
output: path.join(OUTPUT_DIR, `${id}.mp4`),
};
renders.set(id, render);
waiting.push(render);
pump();
return render;
}
function pump() {
while (running < MAX_PARALLEL_RENDERS && waiting.length) {
const render = waiting.shift();
if (!renders.has(render.id)) continue; // supprimé avant son tour
running++;
run(render).finally(() => {
running--;
pump();
});
}
}
async function run(render) {
render.status = "rendering";
const startedAt = Date.now();
const { cancelSignal, cancel } = makeCancelSignal();
render.cancel = cancel;
try {
const serveUrl = await getBundle();
const composition = await selectComposition({
serveUrl,
id: render.compositionId,
inputProps: render.inputProps,
});
await renderMedia({
composition,
serveUrl,
inputProps: render.inputProps,
outputLocation: render.output,
codec: "h264",
// Qualité visuelle élevée, conforme aux recommandations Instagram/TikTok
crf: 18,
x264Preset: "medium",
pixelFormat: "yuv420p",
colorSpace: "bt709",
audioCodec: "aac",
audioBitrate: "192k",
concurrency: renderConcurrency(),
cancelSignal,
onProgress: ({ progress }) => {
render.progress = progress;
},
// Une image lente (décodage vidéo sur une petite machine) ne doit pas faire échouer le rendu
timeoutInMilliseconds: 60_000,
onBrowserLog: (entry) => {
if (entry.type === "error") warn("⚠️ [Remotion] Navigateur :", entry.text.slice(0, 300));
},
});
render.status = "done";
render.progress = 1;
log(`🎞️ [Remotion] ${render.compositionId} ${render.id} rendu en ${((Date.now() - startedAt) / 1000).toFixed(1)} s`);
} catch (error) {
render.status = "failed";
render.error = error instanceof Error ? error.message : String(error);
warn(`❌ [Remotion] Rendu ${render.id} en échec :`, render.error);
fs.promises.rm(render.output, { force: true }).catch(() => {});
} finally {
render.cancel = undefined;
render.finishedAt = Date.now();
// Les props (logo en data URL) ne servent plus
render.inputProps = null;
}
}
function remove(id) {
const render = renders.get(id);
if (!render) return;
renders.delete(id);
render.cancel?.();
fs.promises.rm(render.output, { force: true }).catch(() => {});
}
// Purge des rendus oubliés
setInterval(() => {
const now = Date.now();
for (const render of renders.values()) {
if (render.finishedAt && now - render.finishedAt > RESULT_TTL_MS) remove(render.id);
}
}, 10 * 60_000).unref();
// ---------------------------------------------------------------------------
// HTTP
// ---------------------------------------------------------------------------
function send(res, status, body) {
const data = JSON.stringify(body);
res.writeHead(status, { "Content-Type": "application/json; charset=utf-8", "Content-Length": Buffer.byteLength(data) });
res.end(data);
}
function authorized(req) {
const given = Buffer.from(String(req.headers["x-api-key"] ?? ""));
const expected = Buffer.from(API_KEY);
return given.length === expected.length && crypto.timingSafeEqual(given, expected);
}
function readJson(req) {
return new Promise((resolve, reject) => {
const chunks = [];
let size = 0;
req.on("data", (chunk) => {
size += chunk.length;
if (size > MAX_BODY_BYTES) {
reject(Object.assign(new Error("Corps de requête trop volumineux"), { status: 413 }));
req.destroy();
return;
}
chunks.push(chunk);
});
req.on("end", () => {
try {
resolve(chunks.length ? JSON.parse(Buffer.concat(chunks).toString("utf8")) : {});
} catch {
reject(Object.assign(new Error("JSON invalide"), { status: 400 }));
}
});
req.on("error", reject);
});
}
async function handle(req, res) {
const url = new URL(req.url ?? "/", "http://localhost");
const parts = url.pathname.split("/").filter(Boolean);
if (!authorized(req)) return send(res, 401, { error: "Invalid API Key" });
if (req.method === "GET" && url.pathname === "/health") {
return send(res, 200, { status: "ok", renders: renders.size, running });
}
if (parts[0] !== "renders") return send(res, 404, { error: "Route introuvable" });
if (parts.length === 1 && req.method === "POST") {
const body = await readJson(req);
if (typeof body?.compositionId !== "string" || !body.compositionId) {
return send(res, 400, { error: "compositionId requis" });
}
if (body.inputProps == null || typeof body.inputProps !== "object" || Array.isArray(body.inputProps)) {
return send(res, 400, { error: "inputProps doit être un objet" });
}
const render = enqueue(body.compositionId, body.inputProps);
log(`🎬 [Remotion] Rendu ${render.id} (${render.compositionId}) en file`);
return send(res, 202, { id: render.id });
}
const render = parts[1] ? renders.get(parts[1]) : undefined;
if (parts.length === 2 && req.method === "DELETE") {
remove(parts[1]);
return send(res, 200, { success: true });
}
if (!render) return send(res, 404, { error: "Rendu introuvable ou expiré" });
if (parts.length === 2 && req.method === "GET") {
return send(res, 200, {
status: render.status,
progress: render.progress,
...(render.error ? { error: render.error } : {}),
});
}
if (parts.length === 3 && parts[2] === "file" && req.method === "GET") {
if (render.status !== "done") return send(res, 409, { error: "Rendu non terminé" });
let stat;
try {
stat = await fs.promises.stat(render.output);
} catch {
return send(res, 404, { error: "Fichier introuvable ou expiré" });
}
res.writeHead(200, { "Content-Type": "video/mp4", "Content-Length": stat.size });
fs.createReadStream(render.output)
.on("error", () => res.destroy())
.pipe(res);
return;
}
return send(res, 405, { error: "Méthode non autorisée" });
}
// ---------------------------------------------------------------------------
// Démarrage
// ---------------------------------------------------------------------------
async function prepare() {
log("📦 [Remotion] Pré-bundle de", ENTRY_POINT, "→", BUNDLE_DIR);
await fs.promises.rm(BUNDLE_DIR, { recursive: true, force: true });
await bundle(bundleOptions(BUNDLE_DIR));
log("🌐 [Remotion] Téléchargement du Chrome headless shell…");
await ensureBrowser();
log("✅ [Remotion] Bundle et navigateur prêts");
}
if (process.argv.includes("--prepare")) {
prepare().catch((error) => {
console.error("❌ [Remotion] Préparation impossible :", error);
process.exit(1);
});
} else {
if (!API_KEY) {
console.error("API_KEY doit être défini : le service refuse de démarrer sans clé.");
process.exit(1);
}
fs.mkdirSync(OUTPUT_DIR, { recursive: true });
const server = http.createServer((req, res) => {
handle(req, res).catch((error) => {
if (res.headersSent) return res.destroy();
send(res, error?.status ?? 500, { error: error instanceof Error ? error.message : String(error) });
});
});
server.listen(PORT, "0.0.0.0", () => {
log(`🎬 [Remotion] Service de rendu à l'écoute sur le port ${PORT} (rendus simultanés : ${MAX_PARALLEL_RENDERS})`);
});
// Bundle et navigateur prêts avant le premier rendu (sans bloquer le démarrage)
getBundle().catch((error) => warn("⚠️ [Remotion] Bundle impossible :", error.message));
ensureBrowser().catch((error) => warn("⚠️ [Remotion] Navigateur indisponible :", error.message));
const shutdown = () => {
log("Arrêt du service de rendu");
for (const render of renders.values()) render.cancel?.();
server.close(() => process.exit(0));
setTimeout(() => process.exit(0), 5_000).unref();
};
process.on("SIGTERM", shutdown);
process.on("SIGINT", shutdown);
}
+464 -2
View File
@@ -1,9 +1,471 @@
//! STUB — routes à porter (voir `server/routes/reels.ts`).
//! Routes des Reels et de la musique (portage de `server/routes/reels.ts`).
//! Toutes exigent une session.
use axum::Router;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::routing::{delete, get, post};
use axum::{Json, Router};
use base64::Engine;
use serde_json::{json, Value};
use crate::auth::AuthUser;
use crate::db::{NewMusicFavorite, NewPost, PostStatus};
use crate::error::{ApiResult, AppError, OrApiError};
use crate::extract::{is_truthy, Body, QueryMap};
use crate::services::ffmpeg::{self, VoicePreviewOptions};
use crate::services::openrouter;
use crate::services::reels::assets::{
qwen_target_from_config, resolve_gemini_api_key, resolve_logo_path, resolve_qwen_target,
resolve_store_name,
};
use crate::services::reels::params::{parse_tts_preview, parse_video_reel_params, srt_text};
use crate::services::reels::pipeline::effective_renderer;
use crate::services::reels::queue::{count_active_reel_jobs, enqueue_reel_job};
use crate::services::reels::tts_sync::estimate_voice_timing;
use crate::state::AppState;
use crate::time::Ts;
pub fn router() -> Router<AppState> {
Router::new()
// Musique (Jamendo désactivé : listes vides)
.route("/api/music/search", get(music_search))
.route("/api/music/more", get(music_more))
.route("/api/music/popular", get(music_popular))
.route(
"/api/music/favorites",
get(list_favorites).post(add_favorite),
)
.route("/api/music/favorites/check/{track_id}", get(check_favorite))
.route("/api/music/favorites/{track_id}", delete(remove_favorite))
.route("/api/music/{track_id}", get(music_details))
// Reels
.route("/api/reels/generate-text", post(generate_text))
.route("/api/reels/voices", get(voices))
.route("/api/reels/tts-preview", post(tts_preview))
.route("/api/reels/sync-info", post(sync_info))
.route("/api/reels", post(create_reel))
.route("/api/reels/config", get(reels_config))
.route("/api/reels/ongoing", get(ongoing))
.route("/api/reels/{id}", delete(delete_reel))
}
// ---------------------------------------------------------------------------
// Musique
// ---------------------------------------------------------------------------
/// `parseInt(value)` : chiffres en tête (signe et espaces tolérés), sinon NaN (null en JSON).
fn js_parse_int(value: &str) -> Value {
let trimmed = value.trim_start();
let (sign, digits) = match trimmed.strip_prefix('-') {
Some(rest) => (-1, rest),
None => (1, trimmed.strip_prefix('+').unwrap_or(trimmed)),
};
let numeric: String = digits.chars().take_while(|c| c.is_ascii_digit()).collect();
match numeric.parse::<i64>() {
Ok(n) => json!(sign * n),
Err(_) => Value::Null,
}
}
fn empty_music_page(query: &QueryMap) -> Json<Value> {
Json(json!({
"tracks": [],
"pagination": {
"offset": js_parse_int(query.get("offset").unwrap_or("0")),
"limit": js_parse_int(query.get("limit").unwrap_or("10")),
"hasMore": false,
},
}))
}
async fn music_search(_user: AuthUser, query: QueryMap) -> Json<Value> {
empty_music_page(&query)
}
async fn music_more(_user: AuthUser, query: QueryMap) -> Json<Value> {
empty_music_page(&query)
}
async fn music_popular(_user: AuthUser) -> Json<Value> {
Json(json!({ "tracks": [] }))
}
async fn list_favorites(State(state): State<AppState>, user: AuthUser) -> ApiResult<Json<Value>> {
let favorites = state
.db
.get_music_favorites(&user.id)
.await
.or_500("Erreur lors de la récupération des favoris")?;
let tracks: Vec<Value> = favorites
.into_iter()
.map(|f| {
json!({
"id": f.track_id,
"title": f.title,
"artist": f.artist,
"albumName": f.album_name.unwrap_or_default(),
"duration": f.duration,
"previewUrl": f.preview_url,
"downloadUrl": f.download_url,
"imageUrl": f.image_url.unwrap_or_default(),
"license": f.license.unwrap_or_default(),
})
})
.collect();
Ok(Json(json!({ "tracks": tracks, "isFavorites": true })))
}
async fn check_favorite(
State(state): State<AppState>,
user: AuthUser,
Path(track_id): Path<String>,
) -> ApiResult<Json<Value>> {
let is_favorite = state
.db
.is_music_favorite(&user.id, &track_id)
.await
.or_500("Erreur lors de la vérification du favori")?;
Ok(Json(json!({ "isFavorite": is_favorite })))
}
/// Valeur texte d'un champ JSON « truthy » (les nombres sont convertis comme en JS).
fn text_field(body: &Value, key: &str) -> Option<String> {
let value = body.get(key);
if !is_truthy(value) {
return None;
}
match value? {
Value::String(s) => Some(s.clone()),
Value::Number(n) => Some(n.to_string()),
Value::Bool(b) => Some(b.to_string()),
other => Some(other.to_string()),
}
}
async fn add_favorite(
State(state): State<AppState>,
user: AuthUser,
Body(body): Body<Value>,
) -> ApiResult<Json<Value>> {
const ERROR: &str = "Erreur lors de l'ajout du favori";
let (Some(track_id), Some(title)) = (text_field(&body, "trackId"), text_field(&body, "title"))
else {
return Err(AppError::bad_request("trackId et title requis"));
};
if state
.db
.is_music_favorite(&user.id, &track_id)
.await
.or_500(ERROR)?
{
return Ok(Json(
json!({ "success": true, "message": "Déjà en favori" }),
));
}
let duration = body
.get("duration")
.and_then(Value::as_f64)
.map(|d| d as i32)
.unwrap_or(0);
state
.db
.add_music_favorite(NewMusicFavorite {
user_id: user.id.clone(),
track_id,
title,
artist: text_field(&body, "artist").unwrap_or_default(),
album_name: text_field(&body, "albumName"),
duration,
preview_url: text_field(&body, "previewUrl").unwrap_or_default(),
download_url: text_field(&body, "downloadUrl").unwrap_or_default(),
image_url: text_field(&body, "imageUrl"),
license: text_field(&body, "license"),
})
.await
.or_500(ERROR)?;
Ok(Json(json!({ "success": true })))
}
async fn remove_favorite(
State(state): State<AppState>,
user: AuthUser,
Path(track_id): Path<String>,
) -> ApiResult<Json<Value>> {
state
.db
.remove_music_favorite(&user.id, &track_id)
.await
.or_500("Erreur lors de la suppression du favori")?;
Ok(Json(json!({ "success": true })))
}
async fn music_details(_user: AuthUser, Path(_track_id): Path<String>) -> AppError {
AppError::not_found("Musique non trouvée")
}
// ---------------------------------------------------------------------------
// Reels
// ---------------------------------------------------------------------------
/// Textes IA pour un Reel (même service que les posts).
async fn generate_text(
State(state): State<AppState>,
user: AuthUser,
Body(body): Body<Value>,
) -> ApiResult<Json<Value>> {
let product_info = body.get("productInfo");
if !is_truthy(product_info) {
return Err(AppError::bad_request("Informations produit requises"));
}
// Une simple chaîne sert de nom et de description du produit
let product_info = match product_info {
Some(Value::String(text)) => {
let text = text.trim();
json!({ "name": text, "description": text })
}
Some(other) => other.clone(),
None => Value::Null,
};
let model = body.get("model").and_then(Value::as_str);
let fail = |error: anyhow::Error| {
tracing::error!("❌ Error generating Reel text: {error:#}");
let (status, message) = openrouter::describe_generation_error(&error);
AppError::new(status, message)
};
let texts = openrouter::generate_post_text(&state, &product_info, &user.id, model)
.await
.map_err(fail)?;
let generated = Value::Array(texts);
state
.db
.create_ai_generation(&user.id, &product_info, &generated)
.await
.map_err(|e| fail(e.into()))?;
Ok(Json(json!({ "variants": generated })))
}
/// Voix disponibles (dont les voix clonées Qwen) et disponibilité du moteur local.
async fn voices(State(state): State<AppState>, _user: AuthUser) -> Json<Value> {
let result = async {
let config = state.db.get_app_config().await?;
ffmpeg::list_voices(&state, qwen_target_from_config(config.as_ref())).await
}
.await;
match result {
Ok(catalog) => Json(json!({
"qwenAvailable": catalog.qwen_available,
"qwen": catalog.qwen,
"qwenStatus": catalog.qwen_status,
})),
Err(error) => {
tracing::error!("❌ Error listing voices: {error}");
Json(json!({ "qwenAvailable": false, "qwen": [], "qwenStatus": null }))
}
}
}
/// Aperçu de la voix.
async fn tts_preview(
State(state): State<AppState>,
_user: AuthUser,
Body(body): Body<Value>,
) -> ApiResult<Json<Value>> {
let params = parse_tts_preview(&body).map_err(AppError::bad_request)?;
let engine = params.tts_engine.as_deref();
let options = async {
Ok::<_, anyhow::Error>(VoicePreviewOptions {
voice: params.tts_voice.clone(),
engine: params.tts_engine.clone(),
style: params.tts_style.clone(),
gemini_api_key: resolve_gemini_api_key(&state, engine).await?,
qwen: resolve_qwen_target(&state, engine).await?,
})
}
.await
.map_err(|error| {
tracing::error!("❌ Error generating TTS preview: {error}");
AppError::new(StatusCode::BAD_GATEWAY, error.to_string())
})?;
let preview = ffmpeg::preview_voice(&state, &params.text, &options)
.await
.map_err(|error| {
tracing::error!("❌ Error generating TTS preview: {error}");
let status = if error.status == Some(400) {
StatusCode::BAD_REQUEST
} else {
StatusCode::BAD_GATEWAY
};
AppError::new(status, error.message)
})?;
Ok(Json(json!({
"success": true,
"audioBase64": base64::engine::general_purpose::STANDARD.encode(&preview.audio),
"duration": preview.duration,
"words": preview.words,
"engine": preview.engine,
"voice": preview.voice,
"warnings": preview.warnings,
})))
}
/// Estimation locale de la durée de la voix : ne déclenche aucune synthèse.
async fn sync_info(_user: AuthUser, Body(body): Body<Value>) -> ApiResult<Json<Value>> {
let text = body.get("text");
if !is_truthy(text) {
return Err(AppError::bad_request("Texte requis"));
}
let Some(text) = text.and_then(Value::as_str) else {
return Err(AppError::internal("Erreur de calcul de synchronisation"));
};
Ok(Json(json!(estimate_voice_timing(text))))
}
/// Crée un Reel : le rendu part dans la file persistante.
async fn create_reel(
State(state): State<AppState>,
user: AuthUser,
Body(body): Body<Value>,
) -> ApiResult<Json<Value>> {
let mut params = parse_video_reel_params(&body).map_err(AppError::bad_request)?;
let fail = |message: String| {
tracing::error!("❌ Error initiating Reel: {message}");
AppError::json(
StatusCode::INTERNAL_SERVER_ERROR,
json!({ "success": false, "error": message }),
)
};
// Avec un SRT, le texte libre n'est plus utilisé : la voix lit les sous-titres
if let Some(cues) = &params.srt_cues {
params.overlay_text = None;
if params.description.is_none() {
params.description = Some(srt_text(cues)).filter(|d| !d.is_empty());
}
}
params.store_name = resolve_store_name(
&state,
&user.id,
params.page_ids.first().map(String::as_str),
)
.await;
let waiting = count_active_reel_jobs(&state)
.await
.map_err(|e| fail(e.to_string()))?;
let scheduled_for = match &params.scheduled_for {
Some(date) => Some(Ts::parse(date).ok_or_else(|| fail("Invalid time value".into()))?),
None => None,
};
let mut new_post = NewPost::new(
user.id.clone(),
params
.description
.clone()
.or_else(|| params.overlay_text.clone())
.unwrap_or_default(),
);
new_post.status = if scheduled_for.is_some() {
PostStatus::Scheduled
} else {
PostStatus::Draft
};
new_post.scheduled_for = scheduled_for;
new_post.generation_status = Some("pending".into());
new_post.generation_progress = Some(0);
let post = state
.db
.create_post(new_post)
.await
.map_err(|e| fail(e.to_string()))?;
let stored = serde_json::to_value(&params).map_err(|e| fail(e.to_string()))?;
enqueue_reel_job(&state, "video", &user.id, Some(&post.id), stored)
.await
.map_err(|e| fail(e.to_string()))?;
tracing::info!(
"✨ Reel en file. Post {}, {waiting} job(s) avant lui.",
post.id
);
Ok(Json(json!({
"success": true,
"postId": post.id,
"queued": waiting > 0,
"message": if waiting > 0 {
"File d'attente occupée. Votre vidéo sera traitée dès que possible."
} else {
"Traitement démarré en arrière-plan."
},
"results": [],
"videoUrl": "",
})))
}
/// Configuration des services Reels.
async fn reels_config(State(state): State<AppState>, _user: AuthUser) -> Json<Value> {
let (ffmpeg_configured, logo_url) =
tokio::join!(ffmpeg::health_check(&state), resolve_logo_path(&state));
Json(json!({
"jamendo": { "configured": false },
"freesound": { "configured": false },
"ffmpeg": { "configured": ffmpeg_configured },
// Aperçu : logo affiché par le lecteur, moteur de rendu utilisé
"logoUrl": logo_url,
"renderer": effective_renderer(&state),
}))
}
/// Reels en cours de génération.
async fn ongoing(State(state): State<AppState>, user: AuthUser) -> ApiResult<Json<Value>> {
let posts = state
.db
.get_ongoing_reel_posts(&user.id)
.await
.or_500("Erreur lors de la récupération des reels en cours")?;
Ok(Json(Value::Array(posts)))
}
/// Supprime un Reel (le post et, en cascade, ses publications et liens médias).
async fn delete_reel(
State(state): State<AppState>,
user: AuthUser,
Path(id): Path<String>,
) -> ApiResult<Json<Value>> {
const ERROR: &str = "Erreur lors de la suppression du Reel";
let post = state
.db
.get_post(&id)
.await
.or_500(ERROR)?
.ok_or_else(|| AppError::not_found("Reel non trouvé"))?;
if !user.is_admin() && post.user_id != user.id {
return Err(AppError::forbidden("Non autorisé à supprimer ce Reel"));
}
state.db.delete_post(&id).await.or_500(ERROR)?;
Ok(Json(
json!({ "success": true, "message": "Reel supprimé avec succès" }),
))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parse_int_comme_javascript() {
assert_eq!(js_parse_int("10"), json!(10));
assert_eq!(js_parse_int("12abc"), json!(12));
assert_eq!(js_parse_int(" -3"), json!(-3));
assert_eq!(js_parse_int("abc"), Value::Null);
assert_eq!(js_parse_int(""), Value::Null);
}
}
+608 -8
View File
@@ -1,13 +1,33 @@
//! Client du service FFmpeg (conteneur Python `ffmpeg-service`).
//!
//! STUB — à porter depuis `server/services/ffmpeg.ts`.
#![allow(unused_variables)]
//! Le service rend le Reel puis expose le MP4 en téléchargement
//! (GET /files/{job}/output.mp4) : la vidéo ne transite pas en base64 dans du
//! JSON. Adresse et clé viennent de `state.ffmpeg` (modifiables à chaud depuis
//! les paramètres).
use std::path::{Path, PathBuf};
use std::time::Duration;
use base64::Engine;
use bytes::Bytes;
use futures::StreamExt;
use reqwest::Method;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use serde_json::{Map, Value};
use tokio::io::AsyncWriteExt;
use crate::services::reels::params::SrtCue;
use crate::state::AppState;
/// Un rendu long (stabilisation + encodage) peut dépasser plusieurs minutes.
const PROCESS_TIMEOUT: Duration = Duration::from_secs(20 * 60);
const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(5 * 60);
/// Qwen (local, CPU) met plusieurs fois la durée de la voix à la générer.
const TTS_TIMEOUT: Duration = Duration::from_secs(15 * 60);
const HEALTH_TIMEOUT: Duration = Duration::from_secs(5);
const VOICES_TIMEOUT: Duration = Duration::from_secs(30);
/// Service Qwen TTS (local ou serveur GPU) réglé dans les paramètres.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
@@ -33,15 +53,595 @@ pub struct VoiceCatalog {
pub edge: Vec<Value>,
}
/// Le service répond-il ?
pub async fn health_check(state: &AppState) -> bool {
false
/// Erreur renvoyée par le service, avec son message lisible et son statut HTTP.
#[derive(Debug, Clone, thiserror::Error)]
#[error("{message}")]
pub struct FFmpegServiceError {
pub message: String,
pub status: Option<u16>,
}
/// Voix disponibles ; `qwen` remplace la cible Qwen configurée côté service.
impl FFmpegServiceError {
pub fn new(message: impl Into<String>, status: Option<u16>) -> Self {
Self {
message: message.into(),
status,
}
}
}
/// Mot affiché et son intervalle, en secondes depuis le début de la vidéo.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TimedWord {
pub text: String,
pub start: f64,
pub end: f64,
}
/// Options d'un rendu (ou d'une préparation Remotion). `None` = valeur par
/// défaut du service : le champ n'est pas envoyé, comme `undefined` en JSON.
#[derive(Debug, Clone, Default)]
pub struct ReelRenderOptions {
pub text: Option<String>,
/// Sous-titres SRT : remplacent `text`, la voix lit chacun à son instant.
pub srt_cues: Option<Vec<SrtCue>>,
pub music_url: Option<String>,
pub tts_enabled: Option<bool>,
pub tts_voice: Option<String>,
pub tts_engine: Option<String>,
pub tts_style: Option<String>,
pub gemini_api_key: Option<String>,
pub qwen: Option<QwenTarget>,
pub font_size: Option<i64>,
pub music_volume: Option<f64>,
pub draw_text: Option<bool>,
pub stabilize: Option<bool>,
pub watermark_url: Option<String>,
pub store_name: Option<String>,
pub enable_ending_effect: Option<bool>,
/// Petit logo pendant la vidéo (le grand logo de fin dépend de `enable_ending_effect`).
pub show_logo: Option<bool>,
}
#[derive(Debug, Clone)]
pub struct ReelRenderResult {
pub video: Bytes,
pub duration: f64,
pub tts_engine: Option<String>,
pub tts_voice: Option<String>,
pub warnings: Vec<String>,
}
/// Rendu préparé pour Remotion : fichiers déjà écrits sur le disque.
#[derive(Debug, Clone)]
pub struct PreparedReel {
pub video_path: PathBuf,
pub audio_path: Option<PathBuf>,
pub total_duration: f64,
pub video_duration: f64,
pub logo_start: Option<f64>,
/// Fondu final, calculé par le service pour ne jamais rogner la voix.
pub fade_start: Option<f64>,
pub fade_duration: Option<f64>,
pub words: Vec<TimedWord>,
pub tts_engine: Option<String>,
pub tts_voice: Option<String>,
pub warnings: Vec<String>,
}
#[derive(Debug, Clone, Default)]
pub struct VoicePreviewOptions {
pub voice: Option<String>,
pub engine: Option<String>,
pub style: Option<String>,
pub gemini_api_key: Option<String>,
pub qwen: Option<QwenTarget>,
}
#[derive(Debug, Clone)]
pub struct VoicePreview {
pub audio: Vec<u8>,
pub duration: f64,
pub words: Vec<TimedWord>,
pub engine: String,
pub voice: String,
pub warnings: Vec<String>,
}
// ---------------------------------------------------------------------------
// Appels HTTP
// ---------------------------------------------------------------------------
struct Endpoint {
url: String,
key: String,
}
async fn endpoint(state: &AppState) -> Result<Endpoint, FFmpegServiceError> {
let settings = state.ffmpeg.read().await;
match (&settings.api_url, &settings.api_key) {
(Some(url), Some(key)) if !url.is_empty() && !key.is_empty() => Ok(Endpoint {
url: url.trim_end_matches('/').to_string(),
key: key.clone(),
}),
_ => Err(FFmpegServiceError::new(
"Le service FFmpeg n'est pas configuré (FFMPEG_API_URL / FFMPEG_API_KEY).",
None,
)),
}
}
fn network_error(path: &str, error: reqwest::Error) -> FFmpegServiceError {
if error.is_timeout() {
FFmpegServiceError::new(
format!("Le service FFmpeg n'a pas répondu à temps ({path})"),
None,
)
} else {
FFmpegServiceError::new(format!("Service FFmpeg injoignable : {error}"), None)
}
}
/// Message lisible d'une réponse en erreur : `detail` du JSON FastAPI, sinon le corps.
fn error_detail(body: &str) -> String {
let detail = match serde_json::from_str::<Value>(body) {
Ok(Value::Object(map)) => match map.get("detail") {
Some(Value::String(s)) => s.clone(),
Some(Value::Null) | None => body.to_string(),
Some(other) => other.to_string(),
},
_ => body.to_string(),
};
detail.chars().take(2000).collect()
}
async fn call(
state: &AppState,
method: Method,
path: &str,
body: Option<&Value>,
headers: &[(&str, String)],
timeout: Duration,
) -> Result<reqwest::Response, FFmpegServiceError> {
let endpoint = endpoint(state).await?;
let mut request = state
.http
.request(method, format!("{}{path}", endpoint.url))
.header("X-API-Key", &endpoint.key)
.timeout(timeout);
for (name, value) in headers {
request = request.header(*name, value);
}
if let Some(body) = body {
request = request.json(body);
}
let response = request.send().await.map_err(|e| network_error(path, e))?;
let status = response.status();
if !status.is_success() {
let text = response.text().await.unwrap_or_default();
return Err(FFmpegServiceError::new(
error_detail(&text),
Some(status.as_u16()),
));
}
Ok(response)
}
async fn read_json<T: DeserializeOwned>(
response: reqwest::Response,
path: &str,
) -> Result<T, FFmpegServiceError> {
let bytes = response.bytes().await.map_err(|e| network_error(path, e))?;
serde_json::from_slice(&bytes).map_err(|e| {
FFmpegServiceError::new(
format!("Réponse invalide du service FFmpeg ({path}) : {e}"),
None,
)
})
}
/// Le fichier n'est plus utile au service une fois récupéré (sinon purgé plus tard).
fn delete_job_later(state: &AppState, job_id: &str) {
let state = state.clone();
let path = format!("/jobs/{job_id}");
tokio::spawn(async move {
let _ = call(&state, Method::DELETE, &path, None, &[], HEALTH_TIMEOUT).await;
});
}
fn put<T: Into<Value>>(map: &mut Map<String, Value>, key: &str, value: Option<T>) {
if let Some(value) = value {
map.insert(key.to_string(), value.into());
}
}
/// Champs communs à /process-reel et /prepare-reel.
fn base_body(video_url: &str, options: &ReelRenderOptions) -> Map<String, Value> {
let mut body = Map::new();
body.insert("video_url".into(), video_url.into());
put(&mut body, "text", options.text.clone());
if let Some(cues) = &options.srt_cues {
body.insert(
"srt_cues".into(),
serde_json::to_value(cues).unwrap_or(Value::Null),
);
}
put(&mut body, "music_url", options.music_url.clone());
body.insert(
"tts_enabled".into(),
options.tts_enabled.unwrap_or(false).into(),
);
put(&mut body, "tts_voice", options.tts_voice.clone());
put(&mut body, "tts_engine", options.tts_engine.clone());
put(&mut body, "tts_style", options.tts_style.clone());
put(&mut body, "gemini_api_key", options.gemini_api_key.clone());
put(
&mut body,
"qwen_tts_url",
options.qwen.as_ref().map(|q| q.url.clone()),
);
put(
&mut body,
"qwen_tts_api_key",
options.qwen.as_ref().and_then(|q| q.api_key.clone()),
);
body
}
/// Rend un Reel à partir de l'URL d'une vidéo et renvoie le MP4 produit.
pub async fn render_reel(
state: &AppState,
video_url: &str,
options: &ReelRenderOptions,
) -> Result<ReelRenderResult, FFmpegServiceError> {
let mut body = base_body(video_url, options);
body.insert("font_size".into(), options.font_size.unwrap_or(64).into());
body.insert(
"music_volume".into(),
options.music_volume.unwrap_or(0.25).into(),
);
body.insert("draw_text".into(), options.draw_text.unwrap_or(true).into());
body.insert(
"stabilize".into(),
options.stabilize.unwrap_or(false).into(),
);
put(&mut body, "watermark_url", options.watermark_url.clone());
put(&mut body, "store_name", options.store_name.clone());
body.insert(
"enable_ending_effect".into(),
options.enable_ending_effect.unwrap_or(true).into(),
);
body.insert(
"show_watermark".into(),
options.show_logo.unwrap_or(true).into(),
);
let tts = if options.tts_enabled.unwrap_or(false) {
format!(
"{}/{}/{}",
options.tts_engine.as_deref().unwrap_or("undefined"),
options.tts_voice.as_deref().unwrap_or("undefined"),
options.tts_style.as_deref().unwrap_or("neutral")
)
} else {
"false".into()
};
tracing::info!(
"🎬 Rendu du Reel : vidéo {video_url}, texte {} car., musique {}, voix {tts}",
options
.text
.as_deref()
.map(|t| t.chars().count())
.unwrap_or(0),
options.music_url.is_some()
);
#[derive(Deserialize)]
struct ProcessResponse {
job_id: String,
output_path: String,
#[serde(default)]
duration: f64,
#[serde(default)]
tts_engine: Option<String>,
#[serde(default)]
tts_voice: Option<String>,
#[serde(default)]
warnings: Option<Vec<String>>,
}
let response = call(
state,
Method::POST,
"/process-reel",
Some(&Value::Object(body)),
&[],
PROCESS_TIMEOUT,
)
.await?;
let data: ProcessResponse = read_json(response, "/process-reel").await?;
let result = async {
let file = call(
state,
Method::GET,
&data.output_path,
None,
&[],
DOWNLOAD_TIMEOUT,
)
.await?;
let video = file
.bytes()
.await
.map_err(|e| network_error(&data.output_path, e))?;
let warnings = data.warnings.clone().unwrap_or_default();
for warning in &warnings {
tracing::warn!("⚠️ [FFmpeg] {warning}");
}
Ok(ReelRenderResult {
video,
duration: data.duration,
tts_engine: data.tts_engine.clone(),
tts_voice: data.tts_voice.clone(),
warnings,
})
}
.await;
delete_job_later(state, &data.job_id);
result
}
/// Prépare un rendu Remotion : image recadrée à la durée finale et piste son
/// finale, écrites dans `target_dir`, plus le minutage des mots.
pub async fn prepare_reel(
state: &AppState,
video_url: &str,
options: &ReelRenderOptions,
has_logo: bool,
target_dir: &Path,
) -> Result<PreparedReel, FFmpegServiceError> {
let mut body = base_body(video_url, options);
body.insert(
"music_volume".into(),
options.music_volume.unwrap_or(0.25).into(),
);
body.insert("draw_text".into(), options.draw_text.unwrap_or(true).into());
body.insert(
"stabilize".into(),
options.stabilize.unwrap_or(false).into(),
);
put(&mut body, "store_name", options.store_name.clone());
body.insert(
"enable_ending_effect".into(),
options.enable_ending_effect.unwrap_or(true).into(),
);
body.insert("has_logo".into(), has_logo.into());
#[derive(Deserialize)]
struct PrepareResponse {
job_id: String,
video_path: String,
#[serde(default)]
audio_path: Option<String>,
#[serde(default)]
total_duration: f64,
#[serde(default)]
video_duration: f64,
#[serde(default)]
logo_start: Option<f64>,
#[serde(default)]
fade_start: Option<f64>,
#[serde(default)]
fade_duration: Option<f64>,
#[serde(default)]
words: Option<Vec<TimedWord>>,
#[serde(default)]
tts_engine: Option<String>,
#[serde(default)]
tts_voice: Option<String>,
#[serde(default)]
warnings: Option<Vec<String>>,
}
let response = call(
state,
Method::POST,
"/prepare-reel",
Some(&Value::Object(body)),
&[],
PROCESS_TIMEOUT,
)
.await?;
let data: PrepareResponse = read_json(response, "/prepare-reel").await?;
let result = async {
tokio::fs::create_dir_all(target_dir).await.map_err(|e| {
FFmpegServiceError::new(format!("Dossier de rendu inaccessible : {e}"), None)
})?;
let video_path = target_dir.join("video.mp4");
download_to(state, &data.video_path, &video_path).await?;
let audio_path = match &data.audio_path {
Some(path) => {
let target = target_dir.join("audio.wav");
download_to(state, path, &target).await?;
Some(target)
}
None => None,
};
let warnings = data.warnings.clone().unwrap_or_default();
for warning in &warnings {
tracing::warn!("⚠️ [FFmpeg] {warning}");
}
Ok(PreparedReel {
video_path,
audio_path,
total_duration: data.total_duration,
video_duration: data.video_duration,
logo_start: data.logo_start,
fade_start: data.fade_start,
fade_duration: data.fade_duration,
words: data.words.clone().unwrap_or_default(),
tts_engine: data.tts_engine.clone(),
tts_voice: data.tts_voice.clone(),
warnings,
})
}
.await;
delete_job_later(state, &data.job_id);
result
}
/// Télécharge un fichier du service en flux, sans le charger en mémoire.
async fn download_to(
state: &AppState,
service_path: &str,
target: &Path,
) -> Result<(), FFmpegServiceError> {
let response = call(
state,
Method::GET,
service_path,
None,
&[],
DOWNLOAD_TIMEOUT,
)
.await?;
let write_error = |e: std::io::Error| {
FFmpegServiceError::new(
format!("Écriture de {} impossible : {e}", target.display()),
None,
)
};
let mut file = tokio::fs::File::create(target).await.map_err(write_error)?;
let mut stream = response.bytes_stream();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|e| network_error(service_path, e))?;
file.write_all(&chunk).await.map_err(write_error)?;
}
file.flush().await.map_err(write_error)?;
Ok(())
}
/// Le service répond-il ?
pub async fn health_check(state: &AppState) -> bool {
call(state, Method::GET, "/health", None, &[], HEALTH_TIMEOUT)
.await
.is_ok()
}
/// Voix proposées par moteur ; `qwen` remplace la cible Qwen configurée côté service.
pub async fn list_voices(
state: &AppState,
qwen: Option<QwenTarget>,
) -> anyhow::Result<VoiceCatalog> {
anyhow::bail!("service FFmpeg non implémenté")
let mut headers: Vec<(&str, String)> = Vec::new();
if let Some(qwen) = &qwen {
if !qwen.url.is_empty() {
headers.push(("X-Qwen-Url", qwen.url.clone()));
}
if let Some(key) = qwen.api_key.as_ref().filter(|k| !k.is_empty()) {
headers.push(("X-Qwen-Key", key.clone()));
}
}
let response = call(
state,
Method::GET,
"/voices",
None,
&headers,
VOICES_TIMEOUT,
)
.await?;
Ok(read_json(response, "/voices").await?)
}
/// Génère la voix seule (aperçu), avec le minutage de chaque mot.
pub async fn preview_voice(
state: &AppState,
text: &str,
options: &VoicePreviewOptions,
) -> Result<VoicePreview, FFmpegServiceError> {
let mut body = Map::new();
body.insert("text".into(), text.into());
put(&mut body, "tts_voice", options.voice.clone());
put(&mut body, "tts_engine", options.engine.clone());
put(&mut body, "tts_style", options.style.clone());
put(&mut body, "gemini_api_key", options.gemini_api_key.clone());
put(
&mut body,
"qwen_tts_url",
options.qwen.as_ref().map(|q| q.url.clone()),
);
put(
&mut body,
"qwen_tts_api_key",
options.qwen.as_ref().and_then(|q| q.api_key.clone()),
);
#[derive(Deserialize)]
struct PreviewResponse {
#[serde(default)]
audio_base64: String,
#[serde(default)]
duration: f64,
#[serde(default)]
words: Option<Vec<TimedWord>>,
#[serde(default)]
engine: String,
#[serde(default)]
voice: String,
#[serde(default)]
warnings: Option<Vec<String>>,
}
let response = call(
state,
Method::POST,
"/preview-tts",
Some(&Value::Object(body)),
&[],
TTS_TIMEOUT,
)
.await?;
let data: PreviewResponse = read_json(response, "/preview-tts").await?;
let audio = base64::engine::general_purpose::STANDARD
.decode(data.audio_base64.trim())
.map_err(|e| FFmpegServiceError::new(format!("Audio de l'aperçu illisible : {e}"), None))?;
Ok(VoicePreview {
audio,
duration: data.duration,
words: data.words.unwrap_or_default(),
engine: data.engine,
voice: data.voice,
warnings: data.warnings.unwrap_or_default(),
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn detail_des_erreurs() {
assert_eq!(
error_detail(r#"{"detail":"Vidéo illisible"}"#),
"Vidéo illisible"
);
assert_eq!(
error_detail("Internal Server Error"),
"Internal Server Error"
);
assert_eq!(error_detail(r#"{"other":1}"#), r#"{"other":1}"#);
assert_eq!(error_detail(&"x".repeat(3000)).len(), 2000);
}
#[test]
fn champs_absents_non_envoyes() {
let body = base_body("http://app/v.mp4", &ReelRenderOptions::default());
assert_eq!(body.get("tts_enabled"), Some(&Value::Bool(false)));
// tts_engine absent : le service prend sa valeur par défaut (gemini)
assert!(!body.contains_key("tts_engine"));
assert!(!body.contains_key("text"));
}
}
+103
View File
@@ -0,0 +1,103 @@
//! Résolution des ressources d'un Reel : musique, logo, clé Gemini, service
//! Qwen et nom du magasin.
use crate::db::AppConfig;
use crate::services::ffmpeg::QwenTarget;
use crate::services::media_store;
use crate::state::AppState;
/// URL de la musique téléchargeable par le service FFmpeg (réseau Docker
/// interne : l'URL publique HTTPS n'y est pas joignable).
pub async fn resolve_music_url(
state: &AppState,
music_track_id: Option<&str>,
music_url: Option<&str>,
) -> Option<String> {
if let Some(url) = music_url.filter(|u| !u.is_empty()) {
return Some(media_store::resolve_internal_url(state, url));
}
let id = music_track_id?.strip_prefix("internal_")?;
match state.db.get_audio_track(id).await {
Ok(track) => track.map(|t| media_store::resolve_internal_url(state, &t.url)),
Err(error) => {
tracing::error!("⚠️ [Reels] Piste audio introuvable : {error}");
None
}
}
}
/// Chemin relatif (/uploads/...) du logo configuré, s'il existe.
pub async fn resolve_logo_path(state: &AppState) -> Option<String> {
match state.db.get_cloudinary_config().await {
Ok(config) => config
.and_then(|c| c.logo_public_id)
.filter(|id| !id.is_empty())
.map(|id| media_store::build_url(&id)),
Err(error) => {
tracing::error!("⚠️ [Reels] Configuration du logo illisible : {error}");
None
}
}
}
/// Clé Gemini : configuration de l'application, sinon variable d'environnement.
/// Aussi transmise avec Qwen, qui se replie sur Gemini si le service local échoue.
pub async fn resolve_gemini_api_key(
state: &AppState,
tts_engine: Option<&str>,
) -> anyhow::Result<Option<String>> {
if !matches!(tts_engine, Some("gemini") | Some("qwen")) {
return Ok(None);
}
let app_config = state.db.get_app_config().await?;
Ok(app_config
.and_then(|c| c.gemini_api_key)
.or_else(|| state.config.gemini_api_key.clone()))
}
/// Service Qwen TTS réglé dans les paramètres (sinon ffmpeg-api prend QWEN_TTS_URL).
pub async fn resolve_qwen_target(
state: &AppState,
tts_engine: Option<&str>,
) -> anyhow::Result<Option<QwenTarget>> {
if tts_engine != Some("qwen") {
return Ok(None);
}
Ok(qwen_target_from_config(
state.db.get_app_config().await?.as_ref(),
))
}
pub fn qwen_target_from_config(app_config: Option<&AppConfig>) -> Option<QwenTarget> {
let config = app_config?;
let url = config.qwen_tts_url.as_ref().filter(|u| !u.is_empty())?;
Some(QwenTarget {
url: url.clone(),
api_key: config.qwen_tts_api_key.clone(),
})
}
/// Nom affiché dans l'outro : celui de la page choisie, sinon de la première page de l'utilisateur.
pub async fn resolve_store_name(
state: &AppState,
user_id: &str,
preferred_page_id: Option<&str>,
) -> Option<String> {
let result: Result<Option<String>, sqlx::Error> = async {
if let Some(page_id) = preferred_page_id.filter(|p| !p.is_empty()) {
if let Some(page) = state.db.get_social_page(page_id).await? {
return Ok(Some(page.page_name));
}
}
let pages = state.db.get_social_pages(user_id).await?;
Ok(pages.into_iter().next().map(|p| p.page_name))
}
.await;
match result {
Ok(name) => name,
Err(error) => {
tracing::error!("⚠️ [Reels] Nom du magasin introuvable : {error}");
None
}
}
}
+33 -6
View File
@@ -1,14 +1,24 @@
//! Reels : file de rendus persistante, montage, publication.
//!
//! STUB — à porter depuis `server/services/reels/*`.
#![allow(unused_variables)]
//! Reels : file de rendus persistante, montage (service FFmpeg + service
//! Remotion), publication.
pub mod assets;
pub mod params;
pub mod pipeline;
pub mod publish;
pub mod queue;
pub mod remotion;
pub mod tts_sync;
use serde_json::Value;
use crate::state::AppState;
/// Démarre le worker de la file `reel_jobs`.
pub fn start(state: AppState) {}
/// Démarre le worker de la file `reel_jobs` : récupération des jobs
/// orphelins, puis traitement de la file (réveil à l'ajout d'un job et
/// toutes les 10 s).
pub fn start(state: AppState) {
queue::start(state);
}
/// Adresse http(s) du service Qwen saisie dans les paramètres, sans « / » final ; None si invalide.
pub fn normalize_qwen_url(value: Option<&Value>) -> Option<String> {
@@ -22,3 +32,20 @@ pub fn normalize_qwen_url(value: Option<&Value>) -> Option<String> {
}
Some(url.to_string().trim_end_matches('/').to_string())
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn adresse_qwen() {
assert_eq!(
normalize_qwen_url(Some(&json!(" http://gpu:8001/ "))).as_deref(),
Some("http://gpu:8001")
);
assert_eq!(normalize_qwen_url(Some(&json!("ftp://gpu"))), None);
assert_eq!(normalize_qwen_url(Some(&json!(""))), None);
assert_eq!(normalize_qwen_url(None), None);
}
}
+591
View File
@@ -0,0 +1,591 @@
//! Paramètres des rendus de Reels (portage des schémas zod de `shared/reel.ts`
//! et `shared/srt.ts`).
//!
//! Validés à l'entrée des routes et stockés tels quels (JSON camelCase) dans
//! `reel_jobs.params` : un job repris après redémarrage — ou mis en file par
//! l'ancien serveur — retrouve exactement ce que l'utilisateur avait demandé.
//! Les messages reproduisent ceux de zod : seul le premier problème est
//! renvoyé au client (`{ error }`, statut 400).
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
pub const TTS_ENGINES: [&str; 3] = ["qwen", "gemini", "edge"];
pub const TTS_STYLES: [&str; 5] = ["neutral", "dynamic", "warm", "calm", "promo"];
pub const CAPTION_STYLES: [&str; 3] = ["bold", "neon", "minimal"];
pub const DEFAULT_CAPTION_STYLE: &str = "bold";
pub const MAX_SRT_CUES: usize = 300;
/// Sous-titre d'un fichier SRT (secondes).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SrtCue {
pub start: f64,
pub end: f64,
pub text: String,
}
/// Reel construit à partir d'une vidéo de la médiathèque.
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct VideoReelParams {
pub video_media_id: String,
pub page_ids: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub music_track_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub music_url: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub overlay_text: Option<String>,
/// Sous-titres SRT : remplacent le texte libre, la voix lit chacun à son instant.
#[serde(skip_serializing_if = "Option::is_none")]
pub srt_cues: Option<Vec<SrtCue>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
pub tts_enabled: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub tts_voice: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tts_engine: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tts_style: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub scheduled_for: Option<String>,
/// Publie aussi la vidéo en story sur les pages Facebook (sans effet sur TikTok).
pub also_story: bool,
/// Ancien réglage, ignoré : le minutage vient désormais de la voix elle-même.
#[serde(skip_serializing_if = "Option::is_none")]
pub word_duration: Option<f64>,
pub font_size: i64,
pub music_volume: f64,
pub draw_text: bool,
pub caption_style: String,
pub stabilize: bool,
/// Petit logo pendant la vidéo ; le grand logo de fin dépend de `enable_ending_effect`.
pub show_logo: bool,
pub enable_ending_effect: bool,
/// Déterminé par le serveur à partir de la première page, jamais par le client.
#[serde(skip_serializing_if = "Option::is_none")]
pub store_name: Option<String>,
}
/// Aperçu de la voix.
#[derive(Debug, Clone, PartialEq)]
pub struct TtsPreviewParams {
pub text: String,
pub tts_voice: Option<String>,
pub tts_engine: Option<String>,
pub tts_style: Option<String>,
}
/// Texte complet du SRT (description de la publication, aperçu de la voix).
pub fn srt_text(cues: &[SrtCue]) -> String {
cues.iter()
.map(|c| c.text.as_str())
.collect::<Vec<_>>()
.join(" ")
}
// ---------------------------------------------------------------------------
// Petit équivalent de zod : chaque fonction renvoie le premier problème
// rencontré, avec le message par défaut de zod v3 (ou le message personnalisé).
// ---------------------------------------------------------------------------
type Issue = String;
/// Type reçu, tel que le nomme zod (`getParsedType`).
fn parsed_type(value: Option<&Value>) -> &'static str {
match value {
None => "undefined",
Some(Value::Null) => "null",
Some(Value::Bool(_)) => "boolean",
Some(Value::Number(_)) => "number",
Some(Value::String(_)) => "string",
Some(Value::Array(_)) => "array",
Some(Value::Object(_)) => "object",
}
}
fn invalid_type(expected: &str, value: Option<&Value>) -> Issue {
match value {
None => "Required".into(),
Some(_) => format!("Expected {expected}, received {}", parsed_type(value)),
}
}
/// `String.prototype.trim()` : espaces Unicode et BOM.
pub fn js_trim(text: &str) -> &str {
text.trim_matches(|c: char| (c.is_whitespace() && c != '\u{85}') || c == '\u{FEFF}')
}
/// Longueur JavaScript (unités UTF-16).
fn js_len(text: &str) -> usize {
text.encode_utf16().count()
}
fn join_values(options: &[&str]) -> String {
options
.iter()
.map(|o| format!("'{o}'"))
.collect::<Vec<_>>()
.join(" | ")
}
/// `z.string().trim().optional().transform(v => v || undefined)`.
fn optional_text(obj: &Map<String, Value>, key: &str) -> Result<Option<String>, Issue> {
match obj.get(key) {
None => Ok(None),
Some(Value::String(s)) => {
let trimmed = js_trim(s);
Ok((!trimmed.is_empty()).then(|| trimmed.to_string()))
}
other => Err(invalid_type("string", other)),
}
}
/// `z.string().optional()` (sans trim).
fn optional_string(obj: &Map<String, Value>, key: &str) -> Result<Option<String>, Issue> {
match obj.get(key) {
None => Ok(None),
Some(Value::String(s)) => Ok(Some(s.clone())),
other => Err(invalid_type("string", other)),
}
}
/// `z.boolean().default(default)`.
fn bool_or(obj: &Map<String, Value>, key: &str, default: bool) -> Result<bool, Issue> {
match obj.get(key) {
None => Ok(default),
Some(Value::Bool(b)) => Ok(*b),
other => Err(invalid_type("boolean", other)),
}
}
fn enum_value(value: Option<&Value>, options: &[&str]) -> Result<String, Issue> {
match value {
Some(Value::String(s)) if options.contains(&s.as_str()) => Ok(s.clone()),
Some(Value::String(s)) => Err(format!(
"Invalid enum value. Expected {}, received '{s}'",
join_values(options)
)),
other => Err(invalid_type(&join_values(options), other)),
}
}
/// `z.enum(options).optional()`.
fn optional_enum(
obj: &Map<String, Value>,
key: &str,
options: &[&str],
) -> Result<Option<String>, Issue> {
match obj.get(key) {
None => Ok(None),
value => enum_value(value, options).map(Some),
}
}
fn number(value: Option<&Value>) -> Result<f64, Issue> {
match value {
Some(Value::Number(n)) => n.as_f64().ok_or_else(|| invalid_type("number", value)),
other => Err(invalid_type("number", other)),
}
}
fn check_min(n: f64, min: f64, label: &str) -> Result<(), Issue> {
if n < min {
return Err(format!("Number must be greater than or equal to {label}"));
}
Ok(())
}
fn check_max(n: f64, max: f64, label: &str) -> Result<(), Issue> {
if n > max {
return Err(format!("Number must be less than or equal to {label}"));
}
Ok(())
}
fn as_object(value: &Value) -> Result<&Map<String, Value>, Issue> {
match value {
Value::Object(map) => Ok(map),
other => Err(invalid_type("object", Some(other))),
}
}
/// `srtCueSchema` : bornes, texte de 1 à 500 caractères, fin après le début.
fn srt_cue(value: &Value) -> Result<SrtCue, Issue> {
let obj = as_object(value)?;
let start = number(obj.get("start"))?;
check_min(start, 0.0, "0")?;
let end = number(obj.get("end"))?;
check_min(end, 0.0, "0")?;
let text = match obj.get("text") {
Some(Value::String(s)) => js_trim(s).to_string(),
other => return Err(invalid_type("string", other)),
};
if js_len(&text) < 1 {
return Err("String must contain at least 1 character(s)".into());
}
if js_len(&text) > 500 {
return Err("String must contain at most 500 character(s)".into());
}
if end <= start {
return Err("Sous-titre de durée nulle".into());
}
Ok(SrtCue { start, end, text })
}
/// `srtCuesSchema.optional()`.
fn srt_cues(obj: &Map<String, Value>, key: &str) -> Result<Option<Vec<SrtCue>>, Issue> {
let items = match obj.get(key) {
None => return Ok(None),
Some(Value::Array(items)) => items,
other => return Err(invalid_type("array", other)),
};
if items.is_empty() {
return Err("Fichier SRT vide".into());
}
if items.len() > MAX_SRT_CUES {
return Err(format!(
"Fichier SRT trop long ({MAX_SRT_CUES} sous-titres au plus)"
));
}
items
.iter()
.map(srt_cue)
.collect::<Result<Vec<_>, _>>()
.map(Some)
}
/// Valide les paramètres d'un Reel vidéo (`videoReelParamsSchema`).
/// Les clés inconnues sont ignorées, les valeurs par défaut appliquées.
pub fn parse_video_reel_params(value: &Value) -> Result<VideoReelParams, Issue> {
let obj = as_object(value)?;
let video_media_id = match obj.get("videoMediaId") {
None => return Err("Vidéo requise".into()),
Some(Value::String(s)) if s.is_empty() => return Err("Vidéo requise".into()),
Some(Value::String(s)) => s.clone(),
other => return Err(invalid_type("string", other)),
};
let page_ids = match obj.get("pageIds") {
None => return Err("Au moins une page requise".into()),
Some(Value::Array(items)) => {
if items.is_empty() {
return Err("Au moins une page requise".into());
}
let mut ids = Vec::with_capacity(items.len());
for item in items {
match item {
Value::String(s) if s.is_empty() => {
return Err("String must contain at least 1 character(s)".into())
}
Value::String(s) => ids.push(s.clone()),
other => return Err(invalid_type("string", Some(other))),
}
}
ids
}
other => return Err(invalid_type("array", other)),
};
let music_track_id = optional_text(obj, "musicTrackId")?;
let music_url = optional_text(obj, "musicUrl")?;
let overlay_text = optional_text(obj, "overlayText")?;
let srt_cues = srt_cues(obj, "srtCues")?;
let description = optional_text(obj, "description")?;
let tts_enabled = bool_or(obj, "ttsEnabled", false)?;
let tts_voice = optional_text(obj, "ttsVoice")?;
let tts_engine = optional_enum(obj, "ttsEngine", &TTS_ENGINES)?;
let tts_style = optional_enum(obj, "ttsStyle", &TTS_STYLES)?;
let scheduled_for = optional_text(obj, "scheduledFor")?;
let also_story = bool_or(obj, "alsoStory", false)?;
let word_duration = match obj.get("wordDuration") {
None => None,
value => Some(number(value)?),
};
let font_size = match obj.get("fontSize") {
None => 64,
value => {
let n = number(value)?;
if n.fract() != 0.0 || !n.is_finite() {
return Err("Expected integer, received float".into());
}
check_min(n, 16.0, "16")?;
check_max(n, 200.0, "200")?;
n as i64
}
};
let music_volume = match obj.get("musicVolume") {
None => 0.25,
value => {
let n = number(value)?;
check_min(n, 0.0, "0")?;
check_max(n, 2.0, "2")?;
n
}
};
let draw_text = bool_or(obj, "drawText", true)?;
let caption_style = match obj.get("captionStyle") {
None => DEFAULT_CAPTION_STYLE.to_string(),
value => enum_value(value, &CAPTION_STYLES)?,
};
let stabilize = bool_or(obj, "stabilize", false)?;
let show_logo = bool_or(obj, "showLogo", true)?;
let enable_ending_effect = bool_or(obj, "enableEndingEffect", true)?;
let store_name = optional_string(obj, "storeName")?;
Ok(VideoReelParams {
video_media_id,
page_ids,
music_track_id,
music_url,
overlay_text,
srt_cues,
description,
tts_enabled,
tts_voice,
tts_engine,
tts_style,
scheduled_for,
also_story,
word_duration,
font_size,
music_volume,
draw_text,
caption_style,
stabilize,
show_logo,
enable_ending_effect,
store_name,
})
}
/// Valide une demande d'aperçu de la voix (`ttsPreviewSchema`).
pub fn parse_tts_preview(value: &Value) -> Result<TtsPreviewParams, Issue> {
let obj = as_object(value)?;
let text = match obj.get("text") {
None => return Err("Texte requis".into()),
Some(Value::String(s)) => js_trim(s).to_string(),
other => return Err(invalid_type("string", other)),
};
if js_len(&text) < 1 {
return Err("Texte requis".into());
}
if js_len(&text) > 2000 {
return Err("String must contain at most 2000 character(s)".into());
}
Ok(TtsPreviewParams {
text,
tts_voice: optional_text(obj, "ttsVoice")?,
tts_engine: optional_enum(obj, "ttsEngine", &TTS_ENGINES)?,
tts_style: optional_enum(obj, "ttsStyle", &TTS_STYLES)?,
})
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn base() -> Value {
json!({ "videoMediaId": "v", "pageIds": ["p"] })
}
fn with(extra: Value) -> Value {
let mut value = base();
for (k, v) in extra.as_object().unwrap() {
value[k] = v.clone();
}
value
}
#[test]
fn exige_une_video_et_une_page() {
assert_eq!(
parse_video_reel_params(&json!({ "pageIds": [] })).unwrap_err(),
"Vidéo requise"
);
assert_eq!(
parse_video_reel_params(&json!({ "videoMediaId": "" })).unwrap_err(),
"Vidéo requise"
);
assert_eq!(
parse_video_reel_params(&json!({ "videoMediaId": "v", "pageIds": [] })).unwrap_err(),
"Au moins une page requise"
);
assert_eq!(
parse_video_reel_params(&json!({ "videoMediaId": "v" })).unwrap_err(),
"Au moins une page requise"
);
assert_eq!(
parse_video_reel_params(&json!({ "videoMediaId": 3, "pageIds": ["p"] })).unwrap_err(),
"Expected string, received number"
);
assert_eq!(
parse_video_reel_params(&json!([])).unwrap_err(),
"Expected object, received array"
);
}
#[test]
fn applique_les_valeurs_par_defaut() {
let params = parse_video_reel_params(&base()).unwrap();
assert_eq!(params.caption_style, "bold");
assert!(!params.stabilize);
assert!(params.draw_text);
assert_eq!(params.music_volume, 0.25);
assert!(params.show_logo);
assert!(!params.also_story);
assert!(!params.tts_enabled);
assert!(params.enable_ending_effect);
assert_eq!(params.font_size, 64);
}
#[test]
fn json_stocke_en_camel_case_sans_champs_vides() {
let params = parse_video_reel_params(&with(json!({
"overlayText": " Bonjour ",
"description": " ",
"inconnu": 1,
})))
.unwrap();
let stored = serde_json::to_value(&params).unwrap();
assert_eq!(stored["overlayText"], "Bonjour");
assert!(stored.get("description").is_none());
assert!(stored.get("inconnu").is_none());
assert_eq!(stored["captionStyle"], "bold");
assert_eq!(stored["enableEndingEffect"], true);
// Relu par le worker : identique
assert_eq!(parse_video_reel_params(&stored).unwrap(), params);
}
#[test]
fn accepte_les_anciens_parametres() {
assert!(parse_video_reel_params(&with(json!({ "wordDuration": 0.6 }))).is_ok());
}
#[test]
fn refuse_un_moteur_de_voix_inconnu() {
assert_eq!(
parse_video_reel_params(&with(json!({ "ttsEngine": "autre" }))).unwrap_err(),
"Invalid enum value. Expected 'qwen' | 'gemini' | 'edge', received 'autre'"
);
assert_eq!(
parse_video_reel_params(&with(json!({ "ttsStyle": null }))).unwrap_err(),
"Expected 'neutral' | 'dynamic' | 'warm' | 'calm' | 'promo', received null"
);
}
#[test]
fn bornes_numeriques() {
assert_eq!(
parse_video_reel_params(&with(json!({ "fontSize": 10 }))).unwrap_err(),
"Number must be greater than or equal to 16"
);
assert_eq!(
parse_video_reel_params(&with(json!({ "fontSize": 300 }))).unwrap_err(),
"Number must be less than or equal to 200"
);
assert_eq!(
parse_video_reel_params(&with(json!({ "fontSize": 20.5 }))).unwrap_err(),
"Expected integer, received float"
);
assert_eq!(
parse_video_reel_params(&with(json!({ "fontSize": 72.0 })))
.unwrap()
.font_size,
72
);
assert_eq!(
parse_video_reel_params(&with(json!({ "musicVolume": 3 }))).unwrap_err(),
"Number must be less than or equal to 2"
);
assert_eq!(
parse_video_reel_params(&with(json!({ "musicVolume": -1 }))).unwrap_err(),
"Number must be greater than or equal to 0"
);
assert_eq!(
parse_video_reel_params(&with(json!({ "musicTrackId": null }))).unwrap_err(),
"Expected string, received null"
);
assert_eq!(
parse_video_reel_params(&with(json!({ "stabilize": "oui" }))).unwrap_err(),
"Expected boolean, received string"
);
}
#[test]
fn sous_titres_srt() {
let ok = parse_video_reel_params(&with(
json!({ "srtCues": [{ "start": 0, "end": 1, "text": " a " }] }),
))
.unwrap();
assert_eq!(ok.srt_cues.unwrap()[0].text, "a");
assert_eq!(
parse_video_reel_params(&with(
json!({ "srtCues": [{ "start": 1, "end": 1, "text": "a" }] })
))
.unwrap_err(),
"Sous-titre de durée nulle"
);
assert_eq!(
parse_video_reel_params(&with(json!({ "srtCues": [] }))).unwrap_err(),
"Fichier SRT vide"
);
let many: Vec<Value> = (0..301)
.map(|i| json!({ "start": i, "end": i + 1, "text": "x" }))
.collect();
assert_eq!(
parse_video_reel_params(&with(json!({ "srtCues": many }))).unwrap_err(),
"Fichier SRT trop long (300 sous-titres au plus)"
);
assert_eq!(
parse_video_reel_params(&with(json!({ "srtCues": [{ "end": 1, "text": "a" }] })))
.unwrap_err(),
"Required"
);
}
#[test]
fn texte_du_srt() {
let cues = vec![
SrtCue {
start: 1.0,
end: 3.5,
text: "Bonjour à tous !".into(),
},
SrtCue {
start: 4.0,
end: 6.25,
text: "Découvrez nos nouveautés en magasin".into(),
},
];
assert_eq!(
srt_text(&cues),
"Bonjour à tous ! Découvrez nos nouveautés en magasin"
);
}
#[test]
fn apercu_de_la_voix() {
assert_eq!(
parse_tts_preview(&json!({ "text": " " })).unwrap_err(),
"Texte requis"
);
assert_eq!(parse_tts_preview(&json!({})).unwrap_err(), "Texte requis");
assert_eq!(
parse_tts_preview(&json!({ "text": "x".repeat(2001) })).unwrap_err(),
"String must contain at most 2000 character(s)"
);
let ok =
parse_tts_preview(&json!({ "text": " Salut ", "ttsEngine": "edge", "ttsVoice": "" }))
.unwrap();
assert_eq!(ok.text, "Salut");
assert_eq!(ok.tts_engine.as_deref(), Some("edge"));
assert_eq!(ok.tts_voice, None);
}
}
+256
View File
@@ -0,0 +1,256 @@
//! Rendu d'un Reel à partir d'une vidéo, puis médiathèque et publication.
//! Exécuté par la file `reel_jobs` (voir `queue.rs`).
//!
//! Deux moteurs :
//! - « remotion » (défaut) : le service Python prépare l'image (recadrage, HDR,
//! stabilisation) et la piste son finale, le service Remotion compose les
//! sous-titres animés, le logo et l'effet de fin — le même rendu que l'aperçu ;
//! - « ffmpeg » : tout est fait par le service Python (sous-titres ASS),
//! plus rapide, conservé en secours (`REEL_RENDERER=ffmpeg`).
use std::time::Instant;
use anyhow::{anyhow, bail};
use bytes::Bytes;
use serde_json::{json, Map, Value};
use super::assets::{
resolve_gemini_api_key, resolve_logo_path, resolve_music_url, resolve_qwen_target,
};
use super::params::{parse_video_reel_params, VideoReelParams};
use super::publish::{
describe_publish_failure, publish_reel_to_pages, store_rendered_video, PublishOptions,
};
use super::queue::JobContext;
use super::remotion::{render_composition, temp_file_url, to_data_url};
use crate::db::MediaType;
use crate::services::ffmpeg::{self, ReelRenderOptions};
use crate::services::media_store;
use crate::state::AppState;
/// Moteur demandé par la configuration (`REEL_RENDERER`).
pub fn configured_renderer(state: &AppState) -> &'static str {
if state.config.reel_renderer == "ffmpeg" {
"ffmpeg"
} else {
"remotion"
}
}
/// Moteur réellement utilisé : Remotion exige l'adresse de son service.
pub fn effective_renderer(state: &AppState) -> &'static str {
match configured_renderer(state) {
"remotion" if state.config.remotion_url.is_some() => "remotion",
_ => "ffmpeg",
}
}
pub async fn run_video_reel_job(ctx: &JobContext) -> anyhow::Result<Value> {
let state = &ctx.state;
let job = &ctx.job;
let post_id = job
.post_id
.as_deref()
.ok_or_else(|| anyhow!("Job vidéo sans post associé"))?;
let params = parse_video_reel_params(&job.params).map_err(|issue| anyhow!(issue))?;
ctx.progress(5, "prepare").await?;
let media = state.db.get_media_by_id(&params.video_media_id).await?;
let Some(media) = media.filter(|m| m.media_type == MediaType::Video) else {
bail!("Vidéo source introuvable ou invalide");
};
let engine = params.tts_engine.as_deref();
let (music_url, logo_path, gemini_api_key, qwen) = tokio::join!(
resolve_music_url(
state,
params.music_track_id.as_deref(),
params.music_url.as_deref()
),
resolve_logo_path(state),
resolve_gemini_api_key(state, engine),
resolve_qwen_target(state, engine),
);
let options = ReelRenderOptions {
text: if params.srt_cues.is_some() {
None
} else {
params.overlay_text.clone()
},
srt_cues: params.srt_cues.clone(),
music_url,
tts_enabled: Some(params.tts_enabled),
tts_voice: params.tts_voice.clone(),
tts_engine: params.tts_engine.clone(),
tts_style: params.tts_style.clone(),
gemini_api_key: gemini_api_key?,
qwen: qwen?,
font_size: Some(params.font_size),
music_volume: Some(params.music_volume),
draw_text: Some(params.draw_text),
stabilize: Some(params.stabilize),
watermark_url: None,
store_name: params.store_name.clone(),
enable_ending_effect: Some(params.enable_ending_effect),
show_logo: Some(params.show_logo),
};
let renderer = match (configured_renderer(state), &state.config.remotion_url) {
("remotion", None) => {
tracing::warn!(
"⚠️ [Reels] REMOTION_RENDERER_URL non défini : rendu par le service FFmpeg à la place de Remotion"
);
"ffmpeg"
}
(renderer, _) => renderer,
};
let video_url = media_store::resolve_internal_url(state, &media.original_url);
let started = Instant::now();
let video = match (renderer, state.config.remotion_url.as_deref()) {
("remotion", Some(base_url)) => {
render_with_remotion(
ctx,
base_url,
&video_url,
&options,
&params,
logo_path.as_deref(),
)
.await?
}
_ => render_with_ffmpeg(ctx, &video_url, options, logo_path.as_deref()).await?,
};
tracing::info!(
"⏱️ [Reels] Rendu {renderer} : {:.1} s",
started.elapsed().as_secs_f64()
);
ctx.progress(88, "store").await?;
let file_name = format!("reel-{}.mp4", chrono::Utc::now().timestamp_millis());
let processed = store_rendered_video(state, &job.user_id, &video, &file_name).await?;
state
.db
.update_post_media(post_id, std::slice::from_ref(&processed.id))
.await?;
ctx.progress(93, "publish").await?;
let description = params
.description
.clone()
.or_else(|| params.overlay_text.clone())
.unwrap_or_default();
let results = publish_reel_to_pages(
state,
PublishOptions {
post_id,
page_ids: &params.page_ids,
video,
description: &description,
scheduled_for: params.scheduled_for.as_deref(),
also_story: params.also_story,
},
)
.await?;
if !results.iter().any(|r| r.success) {
bail!(describe_publish_failure(&results));
}
Ok(json!({ "mediaId": processed.id, "renderer": renderer, "results": results }))
}
async fn render_with_ffmpeg(
ctx: &JobContext,
video_url: &str,
mut options: ReelRenderOptions,
logo_path: Option<&str>,
) -> anyhow::Result<Bytes> {
ctx.progress(15, "render").await?;
// Le logo sert au coin de la vidéo et/ou à l'effet de fin : inutile de l'envoyer si aucun des deux
let logo_used = options.show_logo != Some(false) || options.enable_ending_effect != Some(false);
options.watermark_url = logo_path
.filter(|_| logo_used)
.map(|path| media_store::resolve_internal_url(&ctx.state, path));
let rendered = ffmpeg::render_reel(&ctx.state, video_url, &options).await?;
Ok(rendered.video)
}
async fn render_with_remotion(
ctx: &JobContext,
base_url: &str,
video_url: &str,
options: &ReelRenderOptions,
params: &VideoReelParams,
logo_path: Option<&str>,
) -> anyhow::Result<Bytes> {
let state = &ctx.state;
let workdir = state
.config
.temp_upload_dir
.join(format!("reel-{}", ctx.job.id));
let result = async {
// 1. Image recadrée à la durée finale, piste son finale, mots minutés
ctx.progress(10, "voice").await?;
let prepared =
ffmpeg::prepare_reel(state, video_url, options, logo_path.is_some(), &workdir).await?;
// 2. Composition : sous-titres animés, logo, effet de fin
ctx.progress(40, "render").await?;
let mut props = Map::new();
props.insert(
"videoUrl".into(),
temp_file_url(state, &prepared.video_path).into(),
);
props.insert("videoDuration".into(), prepared.total_duration.into());
props.insert("totalDuration".into(), prepared.total_duration.into());
props.insert("words".into(), serde_json::to_value(&prepared.words)?);
props.insert("captionStyle".into(), params.caption_style.clone().into());
if let Some(logo) = logo_path {
props.insert("logoUrl".into(), to_data_url(state, logo).await?.into());
}
props.insert("showWatermark".into(), params.show_logo.into());
if params.enable_ending_effect {
if let Some(name) = &params.store_name {
props.insert("storeName".into(), name.clone().into());
}
}
props.insert("logoStart".into(), prepared.logo_start.into());
props.insert("fadeStart".into(), prepared.fade_start.into());
props.insert("fadeDuration".into(), prepared.fade_duration.into());
if let Some(audio) = &prepared.audio_path {
props.insert("mixedAudioUrl".into(), temp_file_url(state, audio).into());
}
let output = workdir.join("reel.mp4");
let mut last_reported = 40;
render_composition(
state,
base_url,
"ReelVideo",
Value::Object(props),
&output,
|ratio| {
let pct = 40 + (ratio * 45.0).floor() as i32;
let report = pct >= last_reported + 5;
if report {
last_reported = pct;
}
async move {
if report {
// Non bloquant : un échec d'écriture n'interrompt pas le rendu
let _ = ctx.progress(pct, "render").await;
}
}
},
)
.await?;
Ok(Bytes::from(tokio::fs::read(&output).await?))
}
.await;
// Dossier toujours nettoyé, rendu réussi ou non
let _ = tokio::fs::remove_dir_all(&workdir).await;
result
}
+350
View File
@@ -0,0 +1,350 @@
//! Enregistrement et publication d'un Reel rendu.
//!
//! La vidéo est toujours envoyée en binaire : les fichiers vivent sous
//! /uploads, une URL relative que Facebook ne peut pas télécharger.
use bytes::Bytes;
use serde::Serialize;
use crate::db::{
Media, MediaType, NewMedia, NewScheduledPost, Patch, Platform, PostStatus, PostType, SocialPage,
};
use crate::services::media_store::{self, MediaSource};
use crate::services::thumbnail::{self, ThumbnailSource};
use crate::services::{facebook, tiktok};
use crate::state::AppState;
use crate::time::Ts;
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct PagePublishResult {
pub page_id: String,
pub success: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub reel_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
/// Échec de la story associée : n'empêche pas le Reel d'être publié.
#[serde(skip_serializing_if = "Option::is_none")]
pub story_error: Option<String>,
}
impl PagePublishResult {
fn failed(page_id: &str, error: impl Into<String>) -> Self {
Self {
page_id: page_id.to_string(),
success: false,
reel_id: None,
error: Some(error.into()),
story_error: None,
}
}
fn ok(page_id: &str, reel_id: impl Into<String>, story_error: Option<String>) -> Self {
Self {
page_id: page_id.to_string(),
success: true,
reel_id: Some(reel_id.into()),
error: None,
story_error,
}
}
}
/// Range la vidéo rendue dans la médiathèque, avec sa vignette.
pub async fn store_rendered_video(
state: &AppState,
user_id: &str,
video: &[u8],
file_name: &str,
) -> anyhow::Result<Media> {
let uploaded = media_store::upload_media(state, MediaSource::Bytes(video), file_name).await?;
// Vignette extraite maintenant : la vidéo sera supprimée du disque après
// publication, mais l'historique doit rester illustré.
let thumbnail_url =
thumbnail::create_video_thumbnail(state, ThumbnailSource::Bytes(video)).await;
Ok(state
.db
.create_media(NewMedia {
user_id: user_id.to_string(),
media_type: Some(MediaType::Video),
cloudinary_public_id: Some(uploaded.public_id),
original_url: uploaded.original_url,
facebook_landscape_url: None,
facebook_feed_url: uploaded.facebook_feed_url,
instagram_feed_url: uploaded.instagram_feed_url,
instagram_story_url: uploaded.instagram_story_url,
thumbnail_url,
file_name: file_name.to_string(),
file_size: i32::try_from(video.len()).unwrap_or(i32::MAX),
})
.await?)
}
pub struct PublishOptions<'a> {
pub post_id: &'a str,
pub page_ids: &'a [String],
pub video: Bytes,
pub description: &'a str,
pub scheduled_for: Option<&'a str>,
pub also_story: bool,
}
/// `new Date(value)` ; une date invalide fait échouer l'écriture comme avec Drizzle.
fn parse_date(value: &str) -> anyhow::Result<Ts> {
Ts::parse(value).ok_or_else(|| anyhow::anyhow!("Invalid time value"))
}
/// Publie (ou planifie) la vidéo sur chaque page, crée les entrées
/// `scheduled_posts` et met à jour le statut du post.
///
/// Avec `also_story`, chaque page Facebook reçoit aussi la vidéo en story
/// (TikTok n'a pas de stories).
pub async fn publish_reel_to_pages(
state: &AppState,
options: PublishOptions<'_>,
) -> anyhow::Result<Vec<PagePublishResult>> {
let mut results = Vec::with_capacity(options.page_ids.len());
for page_id in options.page_ids {
match publish_to_page(state, &options, page_id).await {
Ok(result) => results.push(result),
Err(error) => {
tracing::error!(
"❌ [Reels] Publication sur la page {page_id} impossible : {error:#}"
);
results.push(PagePublishResult::failed(page_id, error.to_string()));
}
}
}
let succeeded = results.iter().filter(|r| r.success).count();
let status = if options.scheduled_for.is_some() && succeeded > 0 {
PostStatus::Scheduled
} else if succeeded > 0 {
PostStatus::Published
} else {
PostStatus::Failed
};
state
.db
.update_post(options.post_id, Patch::new().set("status", status))
.await?;
Ok(results)
}
async fn publish_to_page(
state: &AppState,
options: &PublishOptions<'_>,
page_id: &str,
) -> anyhow::Result<PagePublishResult> {
let Some(page) = state.db.get_social_page(page_id).await? else {
return Ok(PagePublishResult::failed(page_id, "Page non trouvée"));
};
if page.platform != Platform::Facebook && page.platform != Platform::Tiktok {
return Ok(PagePublishResult::failed(
page_id,
format!(
"Plateforme non supportée pour les reels : {}",
page.platform
),
));
}
let scheduled_at = match options.scheduled_for {
Some(date) => parse_date(date)?,
None => Ts::now(),
};
let scheduled_post = state
.db
.create_scheduled_post(NewScheduledPost::new(
options.post_id,
&page.id,
PostType::Reel,
scheduled_at,
))
.await?;
let with_story = options.also_story && page.platform == Platform::Facebook;
if let Some(date) = options.scheduled_for {
// Le planificateur publiera à l'heure dite
if with_story {
state
.db
.create_scheduled_post(NewScheduledPost::new(
options.post_id,
&page.id,
PostType::Story,
parse_date(date)?,
))
.await?;
}
return Ok(PagePublishResult::ok(page_id, "scheduled", None));
}
tracing::info!(
"🚀 [Reels] Publication sur {} {}...",
page.platform,
page.page_name
);
if page.platform == Platform::Facebook {
let reel_id = facebook::publish_video_from_buffer(
state,
&page,
options.video.clone(),
options.description,
)
.await?;
state
.db
.update_scheduled_post(
&scheduled_post.id,
Patch::new()
.set("published_at", Ts::now())
.set("external_post_id", reel_id.clone()),
)
.await?;
let story_error = if with_story {
publish_story(state, options.post_id, &page, options.video.clone()).await
} else {
None
};
Ok(PagePublishResult::ok(page_id, reel_id, story_error))
} else {
// TikTok finalise la publication de son côté : l'envoi est marqué comme
// effectué pour ne pas republier, et le suivi de statut renseignera
// l'identifiant définitif du post.
let publish_id = tiktok::publish_video_from_buffer(
state,
&page,
options.video.clone(),
options.description,
)
.await?;
state
.db
.update_scheduled_post(
&scheduled_post.id,
Patch::new()
.set("published_at", Ts::now())
.set("publish_id", publish_id.clone())
.set("publish_status", "PROCESSING_UPLOAD"),
)
.await?;
Ok(PagePublishResult::ok(page_id, publish_id, None))
}
}
/// Publie la story d'une page juste après son Reel : un échec est consigné sur
/// son entrée planifiée (visible dans l'historique) et renvoyé, sans faire
/// échouer le Reel.
async fn publish_story(
state: &AppState,
post_id: &str,
page: &SocialPage,
video: Bytes,
) -> Option<String> {
// Créée seulement maintenant : en attente pendant l'envoi du Reel, le
// planificateur aurait pu la publier une seconde fois.
let mut scheduled_story_id: Option<String> = None;
let result: anyhow::Result<()> = async {
let scheduled = state
.db
.create_scheduled_post(NewScheduledPost::new(
post_id,
&page.id,
PostType::Story,
Ts::now(),
))
.await?;
scheduled_story_id = Some(scheduled.id.clone());
let story_id = facebook::publish_video_story_from_buffer(state, page, video).await?;
state
.db
.update_scheduled_post(
&scheduled.id,
Patch::new()
.set("published_at", Ts::now())
.set("external_post_id", story_id),
)
.await?;
Ok(())
}
.await;
match result {
Ok(()) => None,
Err(error) => {
let message = error.to_string();
tracing::error!(
"❌ [Reels] Story sur {} impossible : {error:#}",
page.page_name
);
if let Some(id) = scheduled_story_id {
// Marquée comme traitée pour que le planificateur ne la republie pas en boucle
if let Err(e) = state
.db
.update_scheduled_post(
&id,
Patch::new()
.set("published_at", Ts::now())
.set("error", message.clone()),
)
.await
{
tracing::error!("⚠️ [Reels] Échec de la story non enregistré : {e}");
}
}
Some(message)
}
}
}
/// Message d'erreur lisible lorsque toutes les publications ont échoué.
pub fn describe_publish_failure(results: &[PagePublishResult]) -> String {
let details: Vec<&str> = results
.iter()
.filter_map(|r| r.error.as_deref())
.filter(|e| !e.is_empty())
.collect();
if details.is_empty() {
"Publication impossible sur toutes les pages".into()
} else {
format!(
"Publication impossible sur toutes les pages : {}",
details.join(" ; ")
)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn message_d_echec_global() {
let results = vec![
PagePublishResult::failed("a", "Page non trouvée"),
PagePublishResult::failed("b", "Jeton expiré"),
];
assert_eq!(
describe_publish_failure(&results),
"Publication impossible sur toutes les pages : Page non trouvée ; Jeton expiré"
);
assert_eq!(
describe_publish_failure(&[]),
"Publication impossible sur toutes les pages"
);
}
#[test]
fn resultat_serialise_sans_champs_vides() {
let value = serde_json::to_value(PagePublishResult::ok("p", "scheduled", None)).unwrap();
assert_eq!(
value,
serde_json::json!({ "pageId": "p", "success": true, "reelId": "scheduled" })
);
}
}
+391
View File
@@ -0,0 +1,391 @@
//! File d'attente persistante des rendus de Reels.
//!
//! Chaque job est une ligne de `reel_jobs` :
//! - le worker la réserve atomiquement (`FOR UPDATE SKIP LOCKED`), ce qui
//! exclut tout double traitement, même avec plusieurs instances ;
//! - un battement de cœur est écrit pendant le rendu ; au démarrage, les jobs
//! dont le cœur ne bat plus sont remis en file (ou échouent après
//! `MAX_ATTEMPTS` tentatives).
use std::time::Duration;
use serde_json::Value;
use crate::db::ReelJob;
use crate::state::AppState;
use crate::time::Ts;
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);
/// Au-delà, un job « processing » est considéré comme orphelin.
const STALE_AFTER_MS: i64 = 2 * 60_000;
const POLL_INTERVAL: Duration = Duration::from_secs(10);
const MAX_ATTEMPTS: i32 = 2;
/// Outils fournis à un traitement pour rendre compte de son avancement.
pub struct JobContext {
pub state: AppState,
pub job: ReelJob,
}
impl JobContext {
pub async fn progress(&self, progress: i32, step: &str) -> anyhow::Result<()> {
update_job(
&self.state,
&self.job,
JobPatch {
progress: Some(progress),
step: Some(step.to_string()),
..Default::default()
},
)
.await
}
}
/// Champs modifiés d'un job (les `None` ne sont pas écrits).
#[derive(Debug, Default)]
struct JobPatch {
status: Option<&'static str>,
progress: Option<i32>,
step: Option<String>,
result: Option<Value>,
error: Option<String>,
}
/// Ajoute un job à la file et réveille le worker.
pub async fn enqueue_reel_job(
state: &AppState,
kind: &str,
user_id: &str,
post_id: Option<&str>,
params: Value,
) -> Result<ReelJob, sqlx::Error> {
let job: ReelJob = sqlx::query_as(
"INSERT INTO reel_jobs (kind, user_id, post_id, params) VALUES ($1, $2, $3, $4) RETURNING *",
)
.bind(kind)
.bind(user_id)
.bind(post_id)
.bind(params)
.fetch_one(&state.db.pool)
.await?;
if let Some(post_id) = &job.post_id {
sync_post(state, post_id, "pending", Some(0), None).await?;
}
state.reel_queue.notify_one();
Ok(job)
}
pub async fn get_reel_job(state: &AppState, id: &str) -> Result<Option<ReelJob>, sqlx::Error> {
sqlx::query_as("SELECT * FROM reel_jobs WHERE id = $1")
.bind(id)
.fetch_optional(&state.db.pool)
.await
}
/// Nombre de jobs qui passeront avant un nouveau job (pour informer l'utilisateur).
pub async fn count_active_reel_jobs(state: &AppState) -> Result<i64, sqlx::Error> {
sqlx::query_scalar(
"SELECT count(*)::bigint FROM reel_jobs WHERE status IN ('pending', 'processing')",
)
.fetch_one(&state.db.pool)
.await
}
/// Démarre le worker : récupère les jobs orphelins puis traite la file.
pub fn start(state: AppState) {
tokio::spawn(async move {
if let Err(error) = recover(&state).await {
tracing::error!("❌ [ReelQueue] Récupération des jobs orphelins impossible : {error}");
}
tracing::info!("🎬 [ReelQueue] Worker démarré (traitements : video)");
loop {
if let Err(error) = drain(&state).await {
tracing::error!("❌ [ReelQueue] Erreur de la boucle : {error}");
}
// Réveil à l'ajout d'un job, ou toutes les 10 s
tokio::select! {
_ = state.reel_queue.notified() => {}
_ = tokio::time::sleep(POLL_INTERVAL) => {}
}
}
});
}
async fn recover(state: &AppState) -> anyhow::Result<()> {
recover_stale_jobs(state).await?;
fail_legacy_stuck_posts(state).await?;
Ok(())
}
async fn drain(state: &AppState) -> anyhow::Result<()> {
while let Some(job) = claim_next_job(state).await? {
run_job(state, job).await;
}
Ok(())
}
async fn claim_next_job(state: &AppState) -> Result<Option<ReelJob>, sqlx::Error> {
let id: Option<String> = sqlx::query_scalar(
r#"UPDATE reel_jobs
SET status = 'processing',
attempts = attempts + 1,
heartbeat_at = now(),
updated_at = now(),
error = NULL
WHERE id = (
SELECT id FROM reel_jobs
WHERE status = 'pending'
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT 1
)
RETURNING id"#,
)
.fetch_optional(&state.db.pool)
.await?;
match id {
Some(id) => get_reel_job(state, &id).await,
None => Ok(None),
}
}
/// Message lisible d'une panique du traitement.
fn panic_message(error: tokio::task::JoinError) -> String {
match error.try_into_panic() {
Ok(payload) => payload
.downcast_ref::<String>()
.cloned()
.or_else(|| payload.downcast_ref::<&str>().map(|s| s.to_string()))
.unwrap_or_else(|| "Erreur inattendue pendant le rendu".into()),
Err(error) => error.to_string(),
}
}
async fn run_job(state: &AppState, job: ReelJob) {
tracing::info!(
"🎬 [ReelQueue] Job {} ({}), tentative {}",
job.id,
job.kind,
job.attempts
);
let heartbeat = {
let state = state.clone();
let id = job.id.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(HEARTBEAT_INTERVAL);
interval.tick().await;
loop {
interval.tick().await;
if let Err(error) =
sqlx::query("UPDATE reel_jobs SET heartbeat_at = $2 WHERE id = $1")
.bind(&id)
.bind(Ts::now())
.execute(&state.db.pool)
.await
{
tracing::warn!("⚠️ [ReelQueue] Heartbeat {id} : {error}");
}
}
})
};
// Le traitement tourne dans sa propre tâche : une panique fait échouer le
// job sans arrêter le worker.
let ctx = JobContext {
state: state.clone(),
job: job.clone(),
};
let outcome = tokio::spawn(async move {
if ctx.job.kind != "video" {
anyhow::bail!(
"Aucun traitement enregistré pour les jobs « {} »",
ctx.job.kind
);
}
if let Some(post_id) = &ctx.job.post_id {
sync_post(&ctx.state, post_id, "processing", Some(0), None).await?;
}
super::pipeline::run_video_reel_job(&ctx).await
})
.await;
let outcome = match outcome {
Ok(result) => result.map_err(|e| e.to_string()),
Err(join_error) => Err(panic_message(join_error)),
};
match outcome {
Ok(result) => {
let patch = JobPatch {
status: Some("completed"),
progress: Some(100),
step: Some("done".into()),
result: Some(result),
error: None,
};
match update_job(state, &job, patch).await {
Ok(()) => tracing::info!("✅ [ReelQueue] Job {} terminé", job.id),
Err(error) => tracing::error!(
"⚠️ [ReelQueue] Statut final non enregistré pour {} : {error}",
job.id
),
}
}
Err(message) => {
tracing::error!("❌ [ReelQueue] Job {} en échec : {message}", job.id);
let patch = JobPatch {
status: Some("failed"),
error: Some(message),
..Default::default()
};
if let Err(error) = update_job(state, &job, patch).await {
tracing::error!(
"⚠️ [ReelQueue] Statut d'échec non enregistré pour {} : {error}",
job.id
);
}
}
}
heartbeat.abort();
}
async fn update_job(state: &AppState, job: &ReelJob, patch: JobPatch) -> anyhow::Result<()> {
let now = Ts::now();
let mut qb = sqlx::QueryBuilder::<sqlx::Postgres>::new("UPDATE reel_jobs SET heartbeat_at = ");
qb.push_bind(now).push(", updated_at = ").push_bind(now);
if let Some(status) = patch.status {
qb.push(", status = ").push_bind(status);
}
if let Some(progress) = patch.progress {
qb.push(", progress = ").push_bind(progress);
}
if let Some(step) = &patch.step {
qb.push(", step = ").push_bind(step.clone());
}
if let Some(result) = &patch.result {
qb.push(", result = ").push_bind(result.clone());
}
if let Some(error) = &patch.error {
qb.push(", error = ").push_bind(error.clone());
}
qb.push(" WHERE id = ").push_bind(job.id.clone());
qb.build().execute(&state.db.pool).await?;
if let Some(post_id) = &job.post_id {
let status = patch.status.unwrap_or("processing");
let progress = if patch.status == Some("failed") {
Some(0)
} else {
patch.progress
};
sync_post(state, post_id, status, progress, patch.error.as_deref()).await?;
}
Ok(())
}
/// Reflète l'état du job sur le post : l'interface (`/reels/ongoing`) lit les
/// colonnes `generation_*` du post.
async fn sync_post(
state: &AppState,
post_id: &str,
status: &str,
progress: Option<i32>,
error: Option<&str>,
) -> Result<(), sqlx::Error> {
let mut qb = sqlx::QueryBuilder::<sqlx::Postgres>::new("UPDATE posts SET generation_status = ");
qb.push_bind(status.to_string())
.push(", updated_at = ")
.push_bind(Ts::now());
if let Some(progress) = progress {
qb.push(", generation_progress = ").push_bind(progress);
}
if let Some(error) = error {
qb.push(", generation_error = ")
.push_bind(error.to_string());
}
if status == "failed" {
qb.push(", status = 'failed'::post_status");
}
qb.push(" WHERE id = ").push_bind(post_id.to_string());
qb.build().execute(&state.db.pool).await?;
Ok(())
}
/// Remet en file les jobs interrompus (serveur arrêté pendant un rendu), ou
/// les fait échouer s'ils ont déjà épuisé leurs tentatives : un job qui fait
/// tomber le serveur ne doit pas le faire tomber en boucle.
async fn recover_stale_jobs(state: &AppState) -> anyhow::Result<()> {
let stale_before = Ts::now().add(chrono::Duration::milliseconds(-STALE_AFTER_MS));
let stale: Vec<ReelJob> = sqlx::query_as(
"SELECT * FROM reel_jobs WHERE status = 'processing' AND (heartbeat_at IS NULL OR heartbeat_at < $1)",
)
.bind(stale_before)
.fetch_all(&state.db.pool)
.await?;
for job in stale {
if job.attempts >= MAX_ATTEMPTS {
tracing::warn!(
"⚠️ [ReelQueue] Job {} abandonné après {} tentatives",
job.id,
job.attempts
);
update_job(
state,
&job,
JobPatch {
status: Some("failed"),
error: Some(
"Le rendu a été interrompu à plusieurs reprises (redémarrage du serveur)."
.into(),
),
..Default::default()
},
)
.await?;
} else {
tracing::warn!("♻️ [ReelQueue] Job {} interrompu, remis en file", job.id);
sqlx::query(
"UPDATE reel_jobs SET status = 'pending', step = NULL, progress = 0, updated_at = $2 WHERE id = $1",
)
.bind(&job.id)
.bind(Ts::now())
.execute(&state.db.pool)
.await?;
if let Some(post_id) = &job.post_id {
sync_post(state, post_id, "pending", Some(0), None).await?;
}
}
}
Ok(())
}
/// Posts laissés « processing »/« pending » par l'ancienne file (avant
/// reel_jobs) : aucun job ne peut les reprendre. On les marque en échec pour
/// que l'utilisateur relance.
async fn fail_legacy_stuck_posts(state: &AppState) -> Result<(), sqlx::Error> {
let result = sqlx::query(
r#"UPDATE posts
SET generation_status = 'failed',
generation_progress = 0,
generation_error = 'Génération interrompue, veuillez relancer le Reel.',
status = 'failed',
updated_at = now()
WHERE generation_status IN ('processing', 'pending')
AND NOT EXISTS (SELECT 1 FROM reel_jobs j WHERE j.post_id = posts.id)"#,
)
.execute(&state.db.pool)
.await?;
if result.rows_affected() > 0 {
tracing::warn!(
"⚠️ [ReelQueue] {} Reel(s) de l'ancienne file marqués en échec",
result.rows_affected()
);
}
Ok(())
}
+346
View File
@@ -0,0 +1,346 @@
//! Rendu Remotion : ressources fournies aux compositions et client du service
//! de rendu (conteneur Node `remotion-service`, voir ce dossier).
//!
//! Le serveur Rust ne peut pas exécuter Remotion (React + Chromium) : il
//! confie la composition au service, qui lit la vidéo et le son préparés via
//! les URL `/uploads/temp/...` servies par l'application.
use std::future::Future;
use std::path::{Component, Path, PathBuf};
use std::time::{Duration, Instant};
use anyhow::{anyhow, bail};
use base64::Engine;
use futures::StreamExt;
use serde::Deserialize;
use serde_json::{json, Value};
use tokio::io::AsyncWriteExt;
use crate::state::AppState;
const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(5 * 60);
const POLL_INTERVAL: Duration = Duration::from_secs(1);
/// Garde-fou : un rendu bloqué ne doit pas immobiliser la file indéfiniment.
const MAX_RENDER_TIME: Duration = Duration::from_secs(60 * 60);
/// Erreurs réseau consécutives tolérées pendant le suivi (redémarrage bref, etc.).
const MAX_POLL_ERRORS: u32 = 10;
/// Base des URL internes par lesquelles le service de rendu joint l'application.
fn internal_base(state: &AppState) -> String {
state
.config
.internal_app_url
.clone()
.unwrap_or_else(|| format!("http://localhost:{}", state.config.port))
}
/// URL servie (`/uploads/temp/...`) d'un fichier du dossier temporaire,
/// joignable par le service de rendu.
pub fn temp_file_url(state: &AppState, absolute_path: &Path) -> String {
let relative = absolute_path
.strip_prefix(&state.config.temp_upload_dir)
.unwrap_or(absolute_path)
.components()
.filter_map(|c| match c {
Component::Normal(part) => Some(part.to_string_lossy().into_owned()),
_ => None,
})
.collect::<Vec<_>>()
.join("/");
format!("{}/uploads/temp/{relative}", internal_base(state))
}
/// Normalisation lexicale (`path.resolve`) : résout `.` et `..` sans toucher au disque.
fn normalize(path: &Path) -> PathBuf {
let mut out = PathBuf::new();
for component in path.components() {
match component {
Component::ParentDir => {
out.pop();
}
Component::CurDir => {}
other => out.push(other.as_os_str()),
}
}
out
}
fn absolute(path: &Path) -> PathBuf {
normalize(&std::path::absolute(path).unwrap_or_else(|_| path.to_path_buf()))
}
fn mime_for(path: &Path) -> &'static str {
match path
.extension()
.and_then(|e| e.to_str())
.map(str::to_ascii_lowercase)
.as_deref()
{
Some("png") => "image/png",
Some("gif") => "image/gif",
Some("webp") => "image/webp",
Some("svg") => "image/svg+xml",
_ => "image/jpeg",
}
}
/// Convertit un chemin local /uploads/... en data URL : les images sont
/// affichées par Chromium dans le service de rendu. Refuse tout chemin
/// sortant du dossier uploads.
pub async fn to_data_url(state: &AppState, relative_url: &str) -> anyhow::Result<String> {
to_data_url_in(
&state.config.uploads_dir,
&state.config.temp_upload_dir,
relative_url,
)
.await
}
pub async fn to_data_url_in(
uploads_dir: &Path,
temp_dir: &Path,
relative_url: &str,
) -> anyhow::Result<String> {
if !relative_url.starts_with("/uploads/") {
return Ok(relative_url.to_string());
}
let uploads_root = absolute(uploads_dir);
let temp_root = absolute(temp_dir);
let file_path = match relative_url.strip_prefix("/uploads/temp/") {
Some(rest) => normalize(&temp_root.join(rest)),
None => normalize(&uploads_root.join(&relative_url["/uploads/".len()..])),
};
let inside = |root: &Path| file_path.starts_with(root) && file_path != root;
if !inside(&uploads_root) && !inside(&temp_root) {
bail!("Chemin d'image refusé : {relative_url}");
}
match tokio::fs::read(&file_path).await {
Ok(data) => Ok(format!(
"data:{};base64,{}",
mime_for(&file_path),
base64::engine::general_purpose::STANDARD.encode(data)
)),
Err(_) => {
tracing::warn!("⚠️ [Remotion] Image introuvable : {}", file_path.display());
Ok(relative_url.to_string())
}
}
}
/// Clé du service de rendu : la même que celle du service FFmpeg.
async fn api_key(state: &AppState) -> String {
state
.ffmpeg
.read()
.await
.api_key
.clone()
.unwrap_or_default()
}
/// Message d'erreur d'une réponse du service (`{ error }`), sinon le corps.
async fn response_error(response: reqwest::Response) -> String {
let status = response.status();
let text = response.text().await.unwrap_or_default();
let message = serde_json::from_str::<Value>(&text)
.ok()
.and_then(|v| v.get("error").and_then(Value::as_str).map(str::to_string))
.unwrap_or(text);
format!(
"Service de rendu Remotion ({}) : {}",
status.as_u16(),
message.chars().take(2000).collect::<String>()
)
}
#[derive(Debug, Deserialize)]
struct RenderStatus {
status: String,
#[serde(default)]
progress: f64,
#[serde(default)]
error: Option<String>,
}
/// Rend une composition dans le service Remotion et écrit le MP4 dans
/// `output`. `on_progress` reçoit l'avancement (0..1). Le job du service est
/// toujours supprimé à la fin, réussi ou non.
pub async fn render_composition<F, Fut>(
state: &AppState,
base_url: &str,
composition_id: &str,
input_props: Value,
output: &Path,
mut on_progress: F,
) -> anyhow::Result<()>
where
F: FnMut(f64) -> Fut,
Fut: Future<Output = ()>,
{
let base = base_url.trim_end_matches('/');
let key = api_key(state).await;
let unreachable = |e: reqwest::Error| anyhow!("Service de rendu Remotion injoignable : {e}");
let response = state
.http
.post(format!("{base}/renders"))
.header("X-API-Key", &key)
.timeout(REQUEST_TIMEOUT)
.json(&json!({ "compositionId": composition_id, "inputProps": input_props }))
.send()
.await
.map_err(unreachable)?;
if !response.status().is_success() {
bail!(response_error(response).await);
}
#[derive(Deserialize)]
struct Created {
id: String,
}
let created: Created = response.json().await.map_err(unreachable)?;
let render_url = format!("{base}/renders/{}", created.id);
let started = Instant::now();
let result = async {
let mut errors = 0;
loop {
if started.elapsed() > MAX_RENDER_TIME {
bail!(
"Rendu Remotion trop long (plus de {} min)",
MAX_RENDER_TIME.as_secs() / 60
);
}
tokio::time::sleep(POLL_INTERVAL).await;
let response = match state
.http
.get(&render_url)
.header("X-API-Key", &key)
.timeout(REQUEST_TIMEOUT)
.send()
.await
{
Ok(response) => response,
Err(error) => {
errors += 1;
if errors >= MAX_POLL_ERRORS {
return Err(unreachable(error));
}
continue;
}
};
if !response.status().is_success() {
bail!(response_error(response).await);
}
let status: RenderStatus = response.json().await.map_err(unreachable)?;
errors = 0;
match status.status.as_str() {
"done" => break,
"failed" => bail!(status
.error
.unwrap_or_else(|| "Rendu Remotion en échec".into())),
_ => on_progress(status.progress.clamp(0.0, 1.0)).await,
}
}
// Téléchargement en flux du MP4 produit
let response = state
.http
.get(format!("{render_url}/file"))
.header("X-API-Key", &key)
.timeout(DOWNLOAD_TIMEOUT)
.send()
.await
.map_err(unreachable)?;
if !response.status().is_success() {
bail!(response_error(response).await);
}
let mut file = tokio::fs::File::create(output).await?;
let mut stream = response.bytes_stream();
while let Some(chunk) = stream.next().await {
file.write_all(&chunk.map_err(unreachable)?).await?;
}
file.flush().await?;
tracing::info!(
"🎞️ [Remotion] {composition_id} rendu en {:.1} s",
started.elapsed().as_secs_f64()
);
Ok(())
}
.await;
// Le fichier n'est plus utile au service (sinon purgé plus tard)
let _ = state
.http
.delete(&render_url)
.header("X-API-Key", &key)
.timeout(Duration::from_secs(5))
.send()
.await;
result
}
#[cfg(test)]
mod tests {
use super::*;
fn dirs() -> (PathBuf, PathBuf) {
let root = std::env::temp_dir().join(format!("sf-remotion-test-{}", std::process::id()));
(root.join("uploads"), root.join("uploads").join("temp"))
}
#[tokio::test]
async fn refuse_un_chemin_qui_sort_du_dossier_uploads() {
let (uploads, temp) = dirs();
let error = to_data_url_in(&uploads, &temp, "/uploads/../../etc/passwd")
.await
.unwrap_err();
assert!(error.to_string().contains("refusé"));
let error = to_data_url_in(&uploads, &temp, "/uploads/temp/../../../etc/passwd")
.await
.unwrap_err();
assert!(error.to_string().contains("refusé"));
}
#[tokio::test]
async fn laisse_les_url_externes_telles_quelles() {
let (uploads, temp) = dirs();
assert_eq!(
to_data_url_in(&uploads, &temp, "https://exemple.fr/logo.png")
.await
.unwrap(),
"https://exemple.fr/logo.png"
);
}
#[tokio::test]
async fn convertit_un_logo_en_data_url() {
let (uploads, temp) = dirs();
let logos = uploads.join("logos");
std::fs::create_dir_all(&logos).unwrap();
std::fs::write(logos.join("logo.png"), b"abc").unwrap();
assert_eq!(
to_data_url_in(&uploads, &temp, "/uploads/logos/logo.png")
.await
.unwrap(),
"data:image/png;base64,YWJj"
);
// Fichier absent : l'URL est rendue telle quelle
assert_eq!(
to_data_url_in(&uploads, &temp, "/uploads/logos/absent.png")
.await
.unwrap(),
"/uploads/logos/absent.png"
);
let _ = std::fs::remove_dir_all(uploads.parent().unwrap());
}
#[test]
fn normalisation_lexicale() {
assert_eq!(
normalize(Path::new("/a/b/../c/./d")),
PathBuf::from("/a/c/d")
);
assert_eq!(normalize(Path::new("/a/../../etc")), PathBuf::from("/etc"));
}
}
+109
View File
@@ -0,0 +1,109 @@
//! Estimation de la durée de lecture d'un texte, affichée pendant la saisie
//! (portage de `server/services/ttsSync.ts`).
//!
//! Le minutage réel des mots vient de la voix au moment du rendu ; ici, une
//! estimation suffit et ne déclenche aucune synthèse (payante avec Gemini).
use once_cell::sync::Lazy;
use regex::Regex;
use serde::Serialize;
/// Débit moyen d'une voix de synthèse française, en mots par seconde.
const WORDS_PER_SECOND: f64 = 2.6;
/// Au-delà, la voix dépasse la durée confortable d'un Reel.
const MAX_COMFORTABLE_SECONDS: f64 = 45.0;
const PUNCTUATION_PAUSE: f64 = 0.35;
static HASHTAG: Lazy<Regex> =
Lazy::new(|| Regex::new(r"#[A-Za-z0-9_\x{C0}-\x{FF}]+").expect("regex"));
static LINK: Lazy<Regex> = Lazy::new(|| Regex::new(r"https?://\S+").expect("regex"));
static WORDLIKE: Lazy<Regex> =
Lazy::new(|| Regex::new(r"[A-Za-z0-9\x{C0}-\x{FF}]").expect("regex"));
static PAUSE: Lazy<Regex> = Lazy::new(|| Regex::new(r"[.!?;:]").expect("regex"));
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SyncTiming {
pub word_duration: f64,
pub audio_duration: f64,
pub word_count: usize,
pub punctuation_pause: f64,
pub is_healthy: bool,
pub warnings: Vec<String>,
}
/// `Math.round` de JavaScript (les demis arrondis vers le haut).
fn js_round(value: f64) -> f64 {
(value + 0.5).floor()
}
pub fn estimate_voice_timing(text: &str) -> SyncTiming {
let without_tags = HASHTAG.replace_all(text, "");
let cleaned = LINK.replace_all(&without_tags, "");
let word_count = cleaned
.split(char::is_whitespace)
.filter(|w| WORDLIKE.is_match(w))
.count();
let pauses = PAUSE.find_iter(text).count();
let audio_duration = word_count as f64 / WORDS_PER_SECOND + pauses as f64 * PUNCTUATION_PAUSE;
let mut warnings = Vec::new();
if audio_duration > MAX_COMFORTABLE_SECONDS {
warnings.push(format!(
"Texte long : environ {} s de voix. Visez moins de {} s.",
js_round(audio_duration),
MAX_COMFORTABLE_SECONDS
));
}
if word_count > 0 && word_count < 4 {
warnings.push("Texte très court : la voix ne durera que quelques secondes.".into());
}
SyncTiming {
word_duration: if word_count > 0 {
audio_duration / word_count as f64
} else {
0.0
},
audio_duration,
word_count,
punctuation_pause: PUNCTUATION_PAUSE,
is_healthy: warnings.is_empty(),
warnings,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ignore_hashtags_et_liens() {
assert_eq!(
estimate_voice_timing("Venez vite #promo https://x.fr").word_count,
2
);
}
#[test]
fn previent_quand_la_voix_sera_trop_longue() {
let timing = estimate_voice_timing(&"mot ".repeat(200));
assert!(!timing.is_healthy);
assert!(timing.warnings[0].starts_with("Texte long"));
assert_eq!(
timing.warnings[0],
"Texte long : environ 77 s de voix. Visez moins de 45 s."
);
}
#[test]
fn texte_court() {
let timing = estimate_voice_timing("Bonjour !");
assert_eq!(timing.word_count, 1);
assert_eq!(
timing.warnings,
vec!["Texte très court : la voix ne durera que quelques secondes."]
);
assert!((timing.audio_duration - (1.0 / 2.6 + 0.35)).abs() < 1e-9);
}
}