ayrisdev/screenclipper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 9:16 crop + dinamik altyazı render motoru (Faz 2, Modül 4)

Browse Source 
Her aday klip transkribe olduktan hemen sonra otomatik olarak
video-render kuyruğuna giriyor. Yeni Python worker (video_render.py):

- Klibi ffmpeg ile keser
- OpenCV + MediaPipe Face Detector (Tasks API — yeni mediapipe
  sürümlerinde eski solutions API'si kalktı, .tflite model dosyası
  Docker build'de indiriliyor) ile örneklenmiş karelerde yüz merkezi
  tespit eder, yumuşatır (moving average); hiç yüz bulunamazsa sabit
  merkez crop'a düşer, hata saymaz
- Kırpılmış kareleri OpenCV'den ffmpeg'e pipe'layıp tek geçişte orijinal
  sesi map'ler ve transcript_json'dan ürettiği .ass altyazıyı yakar
- Sonucu /shared-media/shorts/<candidate_id>.mp4'e yazar

Yeni ShortVideo modeli (PENDING/RENDERING/READY/FAILED). Panelde
(segments + kanal detay sayfaları) render durumu + READY olunca video
önizleme/indirme. VIDEO_RENDER_CONCURRENCY ayrı ve düşük (varsayılan 2)
— bu iş diğerlerinden çok daha ağır.

Kapsam: sadece tekil konuşmacı şablonu (çoklu konuşmacı/split-screen
sonraki bir iterasyona bırakıldı — aktif-konuşmacı tespiti gerektiriyor).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
ayrisdev
2026-09-01 15:26:22 +03:00
co-authored by Claude Sonnet 5
parent 5fef7460cf
commit f00fc73b0b
17 changed files with 519 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/frontend/app/api/segments/[id]/video/route.ts
+2 -38
View File
@@ -1,7 +1,6 @@
import { createReadStream, statSync } from "node:fs";
import { Readable } from "node:stream";
import type { NextRequest } from "next/server";
import { prisma } from "@streamclipper/db";
import { streamVideoFile } from "../../../../../lib/videoStream";
export async function GET(req: NextRequest, { params }: { params: Promise<{ id: string }> }) {
const { id } = await params;
@@ -9,40 +8,5 @@ export async function GET(req: NextRequest, { params }: { params: Promise<{ id:
const segment = await prisma.rawSegment.findUnique({ where: { id } });
if (!segment) return new Response("Not found", { status: 404 });
let size: number;
try {
size = statSync(segment.filePath).size;
} catch {
return new Response("File not found on disk", { status: 404 });
}
const headers: Record<string, string> = {
"Content-Type": "video/mp4",
"Accept-Ranges": "bytes",
};
if (req.nextUrl.searchParams.get("download") === "1") {
headers["Content-Disposition"] = `attachment; filename="${id}.mp4"`;
}
const range = req.headers.get("range");
const match = range ? /bytes=(\d+)-(\d*)/.exec(range) : null;
if (match) {
const start = Number(match[1]);
const end = match[2] ? Number(match[2]) : size - 1;
const stream = createReadStream(segment.filePath, { start, end });
return new Response(Readable.toWeb(stream) as ReadableStream, {
status: 206,
headers: {
...headers,
"Content-Range": `bytes ${start}-${end}/${size}`,
"Content-Length": String(end - start + 1),
},
});
}
const stream = createReadStream(segment.filePath);
return new Response(Readable.toWeb(stream) as ReadableStream, {
headers: { ...headers, "Content-Length": String(size) },
});
return streamVideoFile(req, segment.filePath, `${id}.mp4`);
}
apps/frontend/app/api/shorts/[id]/video/route.ts
+12
View File
@@ -0,0 +1,12 @@
import type { NextRequest } from "next/server";
import { prisma } from "@streamclipper/db";
import { streamVideoFile } from "../../../../../lib/videoStream";
export async function GET(req: NextRequest, { params }: { params: Promise<{ id: string }> }) {
const { id } = await params;
const short = await prisma.shortVideo.findUnique({ where: { id } });
if (!short?.filePath) return new Response("Not found", { status: 404 });
return streamVideoFile(req, short.filePath, `${id}.mp4`);
}
apps/frontend/app/channels/[id]/page.tsx
+25 -1
View File
@@ -13,6 +13,8 @@ const STATUS_BADGE: Record<string, string> = {
DISCARDED: "err",
PENDING_STT: "warn",
TRANSCRIBED: "ok",
RENDERING: "warn",
READY: "ok",
FAILED: "err",
};
@@ -34,7 +36,7 @@ export default async function ChannelDetailPage({ params }: { params: Promise<{
include: {
segments: {
orderBy: { createdAt: "desc" },
include: { candidates: { orderBy: { createdAt: "desc" } } },
include: { candidates: { orderBy: { createdAt: "desc" }, include: { short: true } } },
},
},
},
@@ -132,6 +134,28 @@ export default async function ChannelDetailPage({ params }: { params: Promise<{
{transcriptPreview(c.transcriptJson) && (
<p className="transcript-preview">{transcriptPreview(c.transcriptJson)}</p>
)}
{c.short && (
<div style={{ marginTop: "0.5rem" }}>
<span className={`badge ${STATUS_BADGE[c.short.status] ?? "muted"}`}>
9:16: {c.short.status}
</span>
{c.short.status === "READY" && (
<>
<video controls preload="metadata" style={{ width: "220px", marginTop: "0.4rem", borderRadius: "6px", display: "block" }}>
<source src={`/api/shorts/${c.short.id}/video`} type="video/mp4" />
</video>
<a className="badge muted" style={{ marginTop: "0.3rem", display: "inline-block" }} href={`/api/shorts/${c.short.id}/video?download=1`}>
İndir
</a>
</>
)}
{c.short.status === "FAILED" && c.short.errorMessage && (
<p className="mono" style={{ fontSize: "0.75rem", marginTop: "0.3rem" }}>{c.short.errorMessage}</p>
)}
</div>
)}
<form action={deleteCandidateSegment.bind(null, c.id)} style={{ marginTop: "0.35rem" }}>
<DeleteButton confirmText="Bu aday klibi silmek istediğine emin misin?" />
</form>
apps/frontend/app/segments/page.tsx
+25 -1
View File
@@ -10,6 +10,8 @@ const STATUS_BADGE: Record<string, string> = {
DISCARDED: "err",
PENDING_STT: "warn",
TRANSCRIBED: "ok",
RENDERING: "warn",
READY: "ok",
FAILED: "err",
};
@@ -23,7 +25,7 @@ export default async function SegmentsPage() {
const segments = await prisma.rawSegment.findMany({
orderBy: { createdAt: "desc" },
take: 50,
include: { candidates: { orderBy: { createdAt: "desc" } } },
include: { candidates: { orderBy: { createdAt: "desc" }, include: { short: true } } },
});
return (
@@ -71,6 +73,28 @@ export default async function SegmentsPage() {
{transcriptPreview(c.transcriptJson) && (
<p className="transcript-preview">{transcriptPreview(c.transcriptJson)}</p>
)}
{c.short && (
<div style={{ marginTop: "0.5rem" }}>
<span className={`badge ${STATUS_BADGE[c.short.status] ?? "muted"}`}>
9:16: {c.short.status}
</span>
{c.short.status === "READY" && (
<>
<video controls preload="metadata" style={{ width: "220px", marginTop: "0.4rem", borderRadius: "6px", display: "block" }}>
<source src={`/api/shorts/${c.short.id}/video`} type="video/mp4" />
</video>
<a className="badge muted" style={{ marginTop: "0.3rem", display: "inline-block" }} href={`/api/shorts/${c.short.id}/video?download=1`}>
İndir
</a>
</>
)}
{c.short.status === "FAILED" && c.short.errorMessage && (
<p className="mono" style={{ fontSize: "0.75rem", marginTop: "0.3rem" }}>{c.short.errorMessage}</p>
)}
</div>
)}
<form action={deleteCandidateSegment.bind(null, c.id)} style={{ marginTop: "0.35rem" }}>
<DeleteButton confirmText="Bu aday klibi silmek istediğine emin misin?" />
</form>
apps/frontend/lib/videoStream.ts
+42
View File
@@ -0,0 +1,42 @@
import { createReadStream, statSync } from "node:fs";
import { Readable } from "node:stream";
import type { NextRequest } from "next/server";
export function streamVideoFile(req: NextRequest, filePath: string, downloadName: string): Response {
let size: number;
try {
size = statSync(filePath).size;
} catch {
return new Response("File not found on disk", { status: 404 });
}
const headers: Record<string, string> = {
"Content-Type": "video/mp4",
"Accept-Ranges": "bytes",
};
if (req.nextUrl.searchParams.get("download") === "1") {
headers["Content-Disposition"] = `attachment; filename="${downloadName}"`;
}
const range = req.headers.get("range");
const match = range ? /bytes=(\d+)-(\d*)/.exec(range) : null;
if (match) {
const start = Number(match[1]);
const end = match[2] ? Number(match[2]) : size - 1;
const stream = createReadStream(filePath, { start, end });
return new Response(Readable.toWeb(stream) as ReadableStream, {
status: 206,
headers: {
...headers,
"Content-Range": `bytes ${start}-${end}/${size}`,
"Content-Length": String(end - start + 1),
},
});
}
const stream = createReadStream(filePath);
return new Response(Readable.toWeb(stream) as ReadableStream, {
headers: { ...headers, "Content-Length": String(size) },
});
}
apps/worker/Dockerfile
+7 -1
View File
@@ -1,7 +1,7 @@
FROM python:3.12-slim
RUN apt-get update && apt-get install -y --no-install-recommends \
ffmpeg libsndfile1 \
ffmpeg libsndfile1 libgl1 libglib2.0-0 curl \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
@@ -9,5 +9,11 @@ COPY apps/worker/requirements.txt ./apps/worker/requirements.txt
RUN pip install --no-cache-dir -r apps/worker/requirements.txt
COPY apps/worker ./apps/worker
# MediaPipe's Tasks API needs the face detector model asset on disk — not
# bundled with the pip package, downloaded once at build time instead of on
# every worker startup.
RUN mkdir -p /app/models && curl -L -o /app/models/blaze_face_short_range.tflite \
https://storage.googleapis.com/mediapipe-models/face_detector/blaze_face_short_range/float16/latest/blaze_face_short_range.tflite
WORKDIR /app/apps/worker
CMD ["python", "-m", "worker.main"]
apps/worker/requirements.txt
+2
View File
@@ -6,3 +6,5 @@ openai>=1.50.0
httpx>=0.27.0
chat-downloader>=0.2.7
python-dotenv>=1.0.0
opencv-python-headless>=4.10.0
mediapipe>=0.10.14
apps/worker/worker/db.py
+58
View File
@@ -93,3 +93,61 @@ async def update_candidate_transcript(candidate_id: str, transcript: dict, statu
async def get_active_sessions() -> list[asyncpg.Record]:
pool = await get_pool()
return await pool.fetch("SELECT id, live_video_id FROM stream_sessions WHERE ended_at IS NULL")
async def get_candidate_for_render(candidate_id: str) -> asyncpg.Record | None:
pool = await get_pool()
return await pool.fetchrow(
"""
SELECT cs.id, cs.start_sec, cs.end_sec, cs.transcript_json, rs.file_path, c.name AS channel_name
FROM candidate_segments cs
JOIN raw_segments rs ON rs.id = cs.raw_segment_id
JOIN stream_sessions ss ON ss.id = rs.session_id
JOIN channels c ON c.id = ss.channel_id
WHERE cs.id = $1
""",
candidate_id,
)
async def upsert_short_video(candidate_id: str, status: str) -> str:
pool = await get_pool()
existing = await pool.fetchrow(
"SELECT id FROM short_videos WHERE candidate_id = $1", candidate_id
)
if existing:
await pool.execute(
"UPDATE short_videos SET status = $2, error_message = NULL, updated_at = now() WHERE id = $1",
existing["id"],
status,
)
return existing["id"]
short_id = str(uuid.uuid4())
await pool.execute(
"""
INSERT INTO short_videos (id, candidate_id, status, created_at, updated_at)
VALUES ($1, $2, $3, now(), now())
""",
short_id,
candidate_id,
status,
)
return short_id
async def update_short_video(
short_id: str, status: str, file_path: str | None = None, error_message: str | None = None
) -> None:
pool = await get_pool()
await pool.execute(
"""
UPDATE short_videos
SET status = $2, file_path = COALESCE($3, file_path), error_message = $4, updated_at = now()
WHERE id = $1
""",
short_id,
status,
file_path,
error_message,
)
apps/worker/worker/main.py
+4 -1
View File
@@ -4,6 +4,7 @@ from . import db
from .chat_capture import start_chat_capture
from .signal_detection import start_signal_detection_worker
from .stt_scoring import start_stt_scoring_worker
from .video_render import start_video_render_worker
SESSION_WATCH_INTERVAL_SEC = 10
@@ -22,13 +23,15 @@ async def watch_sessions_for_chat_capture():
async def main():
signal_worker = start_signal_detection_worker()
stt_worker = start_stt_scoring_worker()
print("[main] worker started: signal-detection + stt-scoring queues")
render_worker = start_video_render_worker()
print("[main] worker started: signal-detection + stt-scoring + video-render queues")
try:
await watch_sessions_for_chat_capture()
finally:
await signal_worker.close()
await stt_worker.close()
await render_worker.close()
if __name__ == "__main__":
apps/worker/worker/queues.py
+5
View File
@@ -21,3 +21,8 @@ SHARED_MEDIA_ROOT = os.environ.get("SHARED_MEDIA_ROOT", "./shared-media")
# producing segments/candidates around the same time would otherwise process
# strictly one at a time and build up a backlog.
WORKER_CONCURRENCY = int(os.environ.get("WORKER_CONCURRENCY", "5"))
# Video rendering (face detection across sampled frames + two ffmpeg encodes)
# is far heavier per job than signal-detection/stt-scoring — kept separate
# and lower by default so several renders at once don't starve the container.
VIDEO_RENDER_CONCURRENCY = int(os.environ.get("VIDEO_RENDER_CONCURRENCY", "2"))
apps/worker/worker/stt_scoring.py
+4 -1
View File
@@ -3,7 +3,7 @@ import os
import subprocess
import tempfile
from bullmq import Worker
from bullmq import Queue, Worker
from openai import OpenAI
from . import db
@@ -11,6 +11,7 @@ from .queues import QUEUE_NAMES, REDIS_URL, WORKER_CONCURRENCY
from .telegram import send_telegram_message
_openai_client: OpenAI | None = None
video_render_queue = Queue(QUEUE_NAMES["VIDEO_RENDER"], {"connection": REDIS_URL})
def _client() -> OpenAI:
@@ -67,6 +68,8 @@ async def process_stt_scoring(job, job_token=None):
f"({candidate['start_sec']}-{candidate['end_sec']}s):\n_{preview}_"
)
await video_render_queue.add("render-short", {"candidateSegmentId": candidate_id})
print(f"[stt-scoring] candidate {candidate_id} transcribed")
apps/worker/worker/video_render.py
+260
View File
@@ -0,0 +1,260 @@
"""
9:16 crop (yüz takipli) + dinamik altyazı render motoru.
İki geçişli işleme: (1) OpenCV ile örneklenmiş karelerde MediaPipe yüz
tespiti yapıp bir merkez-yol haritası çıkarır, yumuşatır; (2) OpenCV ile
kareleri tekrar okuyup o an için hesaplanan crop penceresini uygulayıp
ffmpeg'e pipe'lar — aynı ffmpeg çağrısı orijinal klibin sesini de
map'leyip .ass altyazısını yakar. Hiç yüz bulunamazsa sabit merkez crop'a
düşer (hata değil, sadece daha az "akıllı" bir sonuç).
"""
import asyncio
import json
import os
import subprocess
import tempfile
from pathlib import Path
import cv2
import numpy as np
from bullmq import Worker
from mediapipe.tasks.python import BaseOptions
from mediapipe.tasks.python.vision import FaceDetector, FaceDetectorOptions
import mediapipe as mp
from . import db
from .queues import QUEUE_NAMES, REDIS_URL, SHARED_MEDIA_ROOT, VIDEO_RENDER_CONCURRENCY
from .telegram import send_telegram_message
FACE_MODEL_PATH = os.environ.get(
"FACE_DETECTOR_MODEL_PATH", "/app/models/blaze_face_short_range.tflite"
)
SAMPLE_INTERVAL_SEC = 0.5
OUTPUT_WIDTH = 1080
OUTPUT_HEIGHT = 1920
_detector: FaceDetector | None = None
def _get_detector() -> FaceDetector:
global _detector
if _detector is None:
options = FaceDetectorOptions(base_options=BaseOptions(model_asset_path=FACE_MODEL_PATH))
_detector = FaceDetector.create_from_options(options)
return _detector
def _detect_face_centers(video_path: str) -> tuple[list[tuple[float, float]], int, int, float]:
"""Örneklenmiş karelerde en büyük yüzün merkezini (x_frac, y_frac) döner."""
cap = cv2.VideoCapture(video_path)
fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
frame_interval = max(1, int(fps * SAMPLE_INTERVAL_SEC))
detector = _get_detector()
points: list[tuple[float, float]] = [] # (time_sec, x_frac)
frame_idx = 0
while True:
ok, frame = cap.read()
if not ok:
break
if frame_idx % frame_interval == 0:
rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=rgb)
result = detector.detect(mp_image)
if result.detections:
best = max(
result.detections,
key=lambda d: d.bounding_box.width * d.bounding_box.height,
)
bbox = best.bounding_box
cx = (bbox.origin_x + bbox.width / 2) / width
points.append((frame_idx / fps, cx))
frame_idx += 1
cap.release()
return points, width, height, fps
def _smooth_x_path(points: list[tuple[float, float]], total_frames: int, fps: float) -> np.ndarray | None:
"""Kare başına yumuşatılmış crop merkezi (x_frac). Hiç yüz yoksa None (sabit merkez crop kullanılır)."""
if not points or total_frames <= 0:
return None
times = np.array([p[0] for p in points])
xs = np.array([p[1] for p in points])
frame_times = np.arange(total_frames) / fps
interp_x = np.interp(frame_times, times, xs)
window = min(max(1, int(fps)), len(interp_x))
kernel = np.ones(window) / window
return np.convolve(interp_x, kernel, mode="same")
def _seconds_to_ass_time(t: float) -> str:
t = max(0.0, t)
h = int(t // 3600)
m = int((t % 3600) // 60)
s = t % 60
return f"{h:d}:{m:02d}:{s:05.2f}"
def _build_ass_subtitles(transcript: dict, path: str, words_per_caption: int = 3) -> None:
words = transcript.get("words") or []
header = (
"[Script Info]\n"
"ScriptType: v4.00+\n"
f"PlayResX: {OUTPUT_WIDTH}\n"
f"PlayResY: {OUTPUT_HEIGHT}\n\n"
"[V4+ Styles]\n"
"Format: Name, Fontname, Fontsize, PrimaryColour, OutlineColour, BackColour, Bold, "
"BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV\n"
"Style: Default,Arial,64,&H00FFFFFF,&H00000000,&H80000000,1,1,3,2,2,60,60,300\n\n"
"[Events]\n"
"Format: Layer, Start, End, Style, Text\n"
)
lines = [header]
for i in range(0, len(words), words_per_caption):
chunk = words[i : i + words_per_caption]
if not chunk:
continue
start = chunk[0].get("start", 0.0)
end = chunk[-1].get("end", start)
text = " ".join(str(w.get("word", "")).strip() for w in chunk).strip()
if not text:
continue
lines.append(f"Dialogue: 0,{_seconds_to_ass_time(start)},{_seconds_to_ass_time(end)},Default,{text}\n")
with open(path, "w", encoding="utf-8") as f:
f.writelines(lines)
def _render_final(
clip_path: str,
smoothed_x: np.ndarray | None,
width: int,
height: int,
fps: float,
ass_path: str,
out_path: str,
) -> None:
crop_w = min(width, max(2, int(height * OUTPUT_WIDTH / OUTPUT_HEIGHT) // 2 * 2))
ffmpeg = subprocess.Popen(
[
"ffmpeg", "-y",
"-f", "rawvideo", "-pix_fmt", "bgr24",
"-s", f"{OUTPUT_WIDTH}x{OUTPUT_HEIGHT}",
"-r", str(fps),
"-i", "-",
"-i", clip_path,
"-vf", f"ass={ass_path}",
"-map", "0:v:0", "-map", "1:a:0?",
"-c:v", "libx264", "-preset", "veryfast", "-pix_fmt", "yuv420p",
"-c:a", "aac",
"-shortest",
out_path,
],
stdin=subprocess.PIPE,
stderr=subprocess.DEVNULL,
)
cap = cv2.VideoCapture(clip_path)
frame_idx = 0
try:
while True:
ok, frame = cap.read()
if not ok:
break
cx_frac = 0.5
if smoothed_x is not None and frame_idx < len(smoothed_x):
cx_frac = float(smoothed_x[frame_idx])
cx_px = int(cx_frac * width)
x0 = max(0, min(width - crop_w, cx_px - crop_w // 2))
cropped = frame[0:height, x0 : x0 + crop_w]
resized = cv2.resize(cropped, (OUTPUT_WIDTH, OUTPUT_HEIGHT))
assert ffmpeg.stdin is not None
ffmpeg.stdin.write(resized.tobytes())
frame_idx += 1
finally:
cap.release()
if ffmpeg.stdin:
ffmpeg.stdin.close()
ffmpeg.wait()
def _render_sync(file_path: str, start_sec: int, end_sec: int, transcript: dict, out_path: str) -> None:
with tempfile.TemporaryDirectory() as tmp_dir:
clip_path = os.path.join(tmp_dir, "clip.mp4")
subprocess.run(
[
"ffmpeg", "-y",
"-ss", str(start_sec),
"-to", str(end_sec),
"-i", file_path,
"-c", "copy",
clip_path,
],
check=True,
capture_output=True,
)
points, width, height, fps = _detect_face_centers(clip_path)
total_frames = int(cv2.VideoCapture(clip_path).get(cv2.CAP_PROP_FRAME_COUNT))
smoothed = _smooth_x_path(points, total_frames, fps)
ass_path = os.path.join(tmp_dir, "subs.ass")
_build_ass_subtitles(transcript, ass_path)
_render_final(clip_path, smoothed, width, height, fps, ass_path, out_path)
async def process_video_render(job, job_token=None):
candidate_id = job.data["candidateSegmentId"]
candidate = await db.get_candidate_for_render(candidate_id)
if candidate is None:
print(f"[video-render] candidate {candidate_id} not found, skipping")
return
short_id = await db.upsert_short_video(candidate_id, status="RENDERING")
out_dir = Path(SHARED_MEDIA_ROOT) / "shorts"
out_dir.mkdir(parents=True, exist_ok=True)
out_path = str(out_dir / f"{candidate_id}.mp4")
try:
transcript = candidate["transcript_json"] or {}
if isinstance(transcript, str):
transcript = json.loads(transcript)
await asyncio.to_thread(
_render_sync,
candidate["file_path"],
candidate["start_sec"],
candidate["end_sec"],
transcript,
out_path,
)
await db.update_short_video(short_id, status="READY", file_path=out_path)
await send_telegram_message(f"🎬 *{candidate['channel_name']}* için 9:16 kısa video hazır!")
print(f"[video-render] short {short_id} ready: {out_path}")
except Exception as exc:
print(f"[video-render] failed for candidate {candidate_id}: {exc}")
await db.update_short_video(short_id, status="FAILED", error_message=str(exc)[:500])
def start_video_render_worker() -> Worker:
return Worker(
QUEUE_NAMES["VIDEO_RENDER"],
process_video_render,
{"connection": REDIS_URL, "concurrency": VIDEO_RENDER_CONCURRENCY},
)