From f00fc73b0bfc4f7c11722c5b901bedd61783cb42 Mon Sep 17 00:00:00 2001 From: ayrisdev Date: Tue, 1 Sep 2026 15:26:22 +0300 Subject: [PATCH] =?UTF-8?q?feat:=209:16=20crop=20+=20dinamik=20altyaz?= =?UTF-8?q?=C4=B1=20render=20motoru=20(Faz=202,=20Mod=C3=BCl=204)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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/.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 --- .env.example | 5 + README.md | 39 ++- .../app/api/segments/[id]/video/route.ts | 40 +-- .../app/api/shorts/[id]/video/route.ts | 12 + apps/frontend/app/channels/[id]/page.tsx | 26 +- apps/frontend/app/segments/page.tsx | 26 +- apps/frontend/lib/videoStream.ts | 42 +++ apps/worker/Dockerfile | 8 +- apps/worker/requirements.txt | 2 + apps/worker/worker/db.py | 58 ++++ apps/worker/worker/main.py | 5 +- apps/worker/worker/queues.py | 5 + apps/worker/worker/stt_scoring.py | 5 +- apps/worker/worker/video_render.py | 260 ++++++++++++++++++ docker-compose.yml | 1 + .../20260901010000_short_videos/migration.sql | 21 ++ packages/db/prisma/schema.prisma | 22 ++ 17 files changed, 519 insertions(+), 58 deletions(-) create mode 100644 apps/frontend/app/api/shorts/[id]/video/route.ts create mode 100644 apps/frontend/lib/videoStream.ts create mode 100644 apps/worker/worker/video_render.py create mode 100644 packages/db/prisma/migrations/20260901010000_short_videos/migration.sql diff --git a/.env.example b/.env.example index 94c4415..cf958ec 100644 --- a/.env.example +++ b/.env.example @@ -50,3 +50,8 @@ PROXY_URL= # (DB row + file) — PRD's auto-purge lifecycle. Only segments already fully # analyzed (no candidate still waiting on STT) are eligible. RAW_SEGMENT_TTL_HOURS=24 + +# Video render (yüz takibi + 9:16 crop + altyazı) her segment/stt-scoring +# işinden çok daha ağır — düşük tutulur ki aynı anda birden fazla render +# worker container'ını boğmasın. +VIDEO_RENDER_CONCURRENCY=2 diff --git a/README.md b/README.md index 98fcd38..2db5e12 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,14 @@ -# StreamClipper AI — Faz 1 +# StreamClipper AI — Faz 1 + Faz 2 (render) -`prd.md` içindeki tam kapsamlı otonom pipeline'ın ilk fazı: ingestion (yt-dlp -headless capture + 15dk segmentleme), sinyal analizi (ses peak + chat -velocity), Whisper STT (OpenAI API, kelime zaman damgalı), BullMQ/Redis/ -Postgres altyapısı ve durum/transkript gösteren bir Next.js panel. LLM -virality skorlama, render/crop/altyazı ve multi-platform dağıtım bu fazda -**yok** — bkz. `prd.md` Modül 3 (skorlama kısmı), 4 ve 5. +`prd.md` içindeki tam kapsamlı otonom pipeline'ın Faz 1+2'si: ingestion +(yt-dlp headless capture + 15dk segmentleme), sinyal analizi (ses peak + +chat velocity), Whisper STT (OpenAI API, kelime zaman damgalı), BullMQ/ +Redis/Postgres altyapısı, auth korumalı bir Next.js panel — ve Faz 2 olarak +transkribe edilen her aday klibi otomatik yüz-takipli 9:16'ya crop'layıp +dinamik altyazı basan bir render motoru (`apps/worker/worker/video_render.py`). +LLM virality skorlama, çoklu-konuşmacı/split-screen şablonu ve +multi-platform dağıtım hâlâ **yok** — bkz. `prd.md` Modül 3 (skorlama +kısmı) ve Modül 5. ## Klasör Yapısı @@ -77,8 +80,8 @@ Panel: http://localhost:3000 · API daemon health: http://localhost:4001/health Panelin tüm sayfaları giriş gerektirir (middleware ile korunuyor). İlk açılışta henüz hiç kullanıcı yoksa `/login` otomatik olarak "İlk Yönetici Hesabını Oluştur" formunu gösterir — kullanıcı adı/şifre girip ilk admin hesabını -oradan oluşturursun. Sonraki kullanıcılar için ayrı bir "kullanıcı ekle" -arayüzü henüz yok (Faz 2). +oradan oluşturursun. `/users` sayfasından ek kullanıcı ekleyip silebilirsin +(kendi hesabını veya son kalan tek kullanıcıyı silmek engellenir). `SESSION_SECRET` production'da zorunlu (`openssl rand -base64 32` ile üretilebilir) — local dev'de boş bırakılırsa güvensiz bir varsayılan @@ -94,15 +97,21 @@ kullanılır, sadece test için. kuyruğundan alıp ses peak + (varsa) chat velocity analiziyle `candidate_segments` üretir. 4. Her aday, `stt-scoring` kuyruğunda OpenAI Whisper API ile transkribe edilir; - sonuç `candidate_segments.transcript_json` alanına yazılır ve Telegram'a - bildirim gider. -5. Panelde (`/` ve `/segments`) kanal durumu ve üretilen transkriptleri - kontrol et. + sonuç `candidate_segments.transcript_json` alanına yazılır, Telegram'a + bildirim gider ve `video-render` kuyruğuna otomatik iş eklenir. +5. `video-render` (Python, `video_render.py`) yüz tespiti + yumuşatılmış + dinamik crop ile 9:16'ya çevirir, transkriptten `.ass` altyazı üretip + yakar, `/shared-media/shorts/.mp4`'e yazar; + `short_videos.status` `READY`/`FAILED` olur, Telegram'a bildirim gider. +6. Panelde (`/`, `/segments`, kanal detay sayfaları) kanal durumu, ham + segment/transkript ve render edilen 9:16 kısa videoları izle/indir. ## Bilinen Sınırlamalar (bu faz kapsamı dışı) -- LLM virality skorlama, metadata üretimi, telif/görüntü filtresi yok. -- 9:16 crop/altyazı render motoru yok. +- LLM virality skorlama, metadata üretimi, telif/görüntü filtresi yok — + transkribe edilen her aday otomatik render edilir, eşik/filtre yok. +- Çoklu konuşmacı / split-screen şablonu yok — sadece tekil konuşmacı + (yüz-merkezli dinamik crop). - YouTube/Instagram/TikTok'a otomatik yayınlama yok (bu API'ler ayrı onay süreçleri gerektiriyor — `prd.md` tartışmasına bkz.). - Aday pencere çıkarımı her zaman tek bir 15dk'lık segment dosyasıyla sınırlı; diff --git a/apps/frontend/app/api/segments/[id]/video/route.ts b/apps/frontend/app/api/segments/[id]/video/route.ts index 7c00eb3..eecf4a5 100644 --- a/apps/frontend/app/api/segments/[id]/video/route.ts +++ b/apps/frontend/app/api/segments/[id]/video/route.ts @@ -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 = { - "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`); } diff --git a/apps/frontend/app/api/shorts/[id]/video/route.ts b/apps/frontend/app/api/shorts/[id]/video/route.ts new file mode 100644 index 0000000..4371dff --- /dev/null +++ b/apps/frontend/app/api/shorts/[id]/video/route.ts @@ -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`); +} diff --git a/apps/frontend/app/channels/[id]/page.tsx b/apps/frontend/app/channels/[id]/page.tsx index 09b0970..3b14a35 100644 --- a/apps/frontend/app/channels/[id]/page.tsx +++ b/apps/frontend/app/channels/[id]/page.tsx @@ -13,6 +13,8 @@ const STATUS_BADGE: Record = { 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) && (

{transcriptPreview(c.transcriptJson)}

)} + + {c.short && ( +
+ + 9:16: {c.short.status} + + {c.short.status === "READY" && ( + <> + + + İndir + + + )} + {c.short.status === "FAILED" && c.short.errorMessage && ( +

{c.short.errorMessage}

+ )} +
+ )} +
diff --git a/apps/frontend/app/segments/page.tsx b/apps/frontend/app/segments/page.tsx index 012efa1..725d19d 100644 --- a/apps/frontend/app/segments/page.tsx +++ b/apps/frontend/app/segments/page.tsx @@ -10,6 +10,8 @@ const STATUS_BADGE: Record = { 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) && (

{transcriptPreview(c.transcriptJson)}

)} + + {c.short && ( +
+ + 9:16: {c.short.status} + + {c.short.status === "READY" && ( + <> + + + İndir + + + )} + {c.short.status === "FAILED" && c.short.errorMessage && ( +

{c.short.errorMessage}

+ )} +
+ )} +
diff --git a/apps/frontend/lib/videoStream.ts b/apps/frontend/lib/videoStream.ts new file mode 100644 index 0000000..d952291 --- /dev/null +++ b/apps/frontend/lib/videoStream.ts @@ -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 = { + "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) }, + }); +} diff --git a/apps/worker/Dockerfile b/apps/worker/Dockerfile index 06bc407..bace075 100644 --- a/apps/worker/Dockerfile +++ b/apps/worker/Dockerfile @@ -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"] diff --git a/apps/worker/requirements.txt b/apps/worker/requirements.txt index d4b20bd..91dcf1e 100644 --- a/apps/worker/requirements.txt +++ b/apps/worker/requirements.txt @@ -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 diff --git a/apps/worker/worker/db.py b/apps/worker/worker/db.py index 41912a2..817c3f6 100644 --- a/apps/worker/worker/db.py +++ b/apps/worker/worker/db.py @@ -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, + ) diff --git a/apps/worker/worker/main.py b/apps/worker/worker/main.py index 32e7e70..e7e7577 100644 --- a/apps/worker/worker/main.py +++ b/apps/worker/worker/main.py @@ -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__": diff --git a/apps/worker/worker/queues.py b/apps/worker/worker/queues.py index b7599ea..4d1d5c1 100644 --- a/apps/worker/worker/queues.py +++ b/apps/worker/worker/queues.py @@ -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")) diff --git a/apps/worker/worker/stt_scoring.py b/apps/worker/worker/stt_scoring.py index c429a5c..9fc132f 100644 --- a/apps/worker/worker/stt_scoring.py +++ b/apps/worker/worker/stt_scoring.py @@ -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") diff --git a/apps/worker/worker/video_render.py b/apps/worker/worker/video_render.py new file mode 100644 index 0000000..d86f173 --- /dev/null +++ b/apps/worker/worker/video_render.py @@ -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}, + ) diff --git a/docker-compose.yml b/docker-compose.yml index bbc01a4..798e415 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -65,6 +65,7 @@ services: TELEGRAM_BOT_TOKEN: ${TELEGRAM_BOT_TOKEN:-} TELEGRAM_CHAT_ID: ${TELEGRAM_CHAT_ID:-} WORKER_CONCURRENCY: ${WORKER_CONCURRENCY:-5} + VIDEO_RENDER_CONCURRENCY: ${VIDEO_RENDER_CONCURRENCY:-2} volumes: - shared-media:/shared-media depends_on: diff --git a/packages/db/prisma/migrations/20260901010000_short_videos/migration.sql b/packages/db/prisma/migrations/20260901010000_short_videos/migration.sql new file mode 100644 index 0000000..7063f73 --- /dev/null +++ b/packages/db/prisma/migrations/20260901010000_short_videos/migration.sql @@ -0,0 +1,21 @@ +-- CreateEnum +CREATE TYPE "ShortVideoStatus" AS ENUM ('PENDING', 'RENDERING', 'READY', 'FAILED'); + +-- CreateTable +CREATE TABLE "short_videos" ( + "id" TEXT NOT NULL, + "candidate_id" TEXT NOT NULL, + "file_path" TEXT, + "status" "ShortVideoStatus" NOT NULL DEFAULT 'PENDING', + "error_message" TEXT, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "short_videos_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "short_videos_candidate_id_key" ON "short_videos"("candidate_id"); + +-- AddForeignKey +ALTER TABLE "short_videos" ADD CONSTRAINT "short_videos_candidate_id_fkey" FOREIGN KEY ("candidate_id") REFERENCES "candidate_segments"("id") ON DELETE RESTRICT ON UPDATE CASCADE; diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index e5de72c..cf9a7c3 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -19,6 +19,13 @@ enum CandidateSegmentStatus { FAILED } +enum ShortVideoStatus { + PENDING + RENDERING + READY + FAILED +} + model Channel { id String @id @default(cuid()) name String @@ -74,9 +81,24 @@ model CandidateSegment { status CandidateSegmentStatus @default(PENDING_STT) createdAt DateTime @default(now()) @map("created_at") + short ShortVideo? + @@map("candidate_segments") } +model ShortVideo { + id String @id @default(cuid()) + candidateId String @unique @map("candidate_id") + candidate CandidateSegment @relation(fields: [candidateId], references: [id]) + filePath String? @map("file_path") + status ShortVideoStatus @default(PENDING) + errorMessage String? @map("error_message") + createdAt DateTime @default(now()) @map("created_at") + updatedAt DateTime @updatedAt @map("updated_at") + + @@map("short_videos") +} + model AppSetting { key String @id value String