screenclipper/apps/worker/worker/video_render.py

"""
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)
        if await db.get_setting("notify_render_done", "true") != "false":
            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},
    )