services/render_engine/renderer/platform/processor.py

from __future__ import annotations

import json
import math
import mimetypes
import shutil
import zipfile
from pathlib import Path
from typing import Any

from renderer.core.config import Settings
from renderer.core.ingest import AssetIngestor
from renderer.core.models import TaskResult
from renderer.core.utils import safe_filename, temp_workdir, write_json
from renderer.ffmpeg.assets import AssetProbe
from renderer.ffmpeg.command import FFmpegCommand
from renderer.ffmpeg.runner import FFmpegRunner
from renderer.templates import get_platform_profile


TOOLKIT_TASKS = {
    "cut",
    "trim",
    "split",
    "concat",
    "merge",
    "compress",
    "normalize",
    "resize",
    "crop",
    "rotate",
    "flip",
    "scale",
    "zoom",
    "pan",
    "speed",
    "speed_ramp",
    "time_remap",
    "slow_motion",
    "fast_motion",
    "reverse",
    "reverse_playback",
    "freeze_frame",
    "motion_blur",
    "stabilization",
    "lens_correction",
    "loop",
    "extract_audio",
    "thumbnail",
    "gif",
    "frames",
    "watermark",
    "overlay_text",
    "blur_background",
    "burn_subtitles",
    "convert",
    "merge_audio",
    "noise_reduction",
    "equalizer",
    "compressor",
    "limiter",
    "pitch_shift",
    "voice_changer",
    "ai_enhancement",
    "green_screen",
    "chroma_key",
    "blue_screen",
    "ai_background_removal",
}


class PlatformProcessor:
    def __init__(self, settings: Settings | None = None, log=None) -> None:
        self.settings = settings or Settings()
        self.settings.ensure_dirs()
        self._commands: list[list[str]] = []
        self._logs: list[str] = []
        self.runner = FFmpegRunner(self.settings.ffmpeg_timeout_seconds, log=log, on_command=self._record_command)
        self.ingest = AssetIngestor(self.settings)
        self.assets = AssetProbe(self.settings.metadata_cache)

    def ingest_sources(self, sources: list[dict[str, Any]], job_id: str) -> TaskResult:
        with temp_workdir(self.settings.temp_dir, f"{job_id}_ingest") as work:
            workdir = Path(work)
            staged: list[dict[str, Any]] = []
            for index, source in enumerate(sources):
                url = str(source.get("url") or source.get("source") or "").strip()
                if not url:
                    continue
                source_type = str(source.get("type") or _source_type(url))
                if source_type == "youtube":
                    staged.append(
                        {
                            "source": url,
                            "source_type": source_type,
                            "status": "registered",
                            "note": "YouTube ingestion is registered for automation; provide a direct downloadable media URL or install yt-dlp for local extraction.",
                        }
                    )
                    continue
                resolved = self.ingest.resolve(url, workdir / "inputs", f"source_{index:03d}")
                metadata = self.assets.probe(resolved).__dict__
                staged.append({"source": url, "source_type": source_type, "path": str(resolved), "metadata": metadata})
            output = self._json_artifact(job_id, "ingest_manifest", {"assets": staged, "asset_count": len(staged)})
        return self._result(output, {"task": "ingest", "asset_count": len(staged)})

    def analyze(self, media: str, job_id: str, *, transcript: str = "", platform: str | None = None) -> TaskResult:
        with temp_workdir(self.settings.temp_dir, f"{job_id}_analyze") as work:
            workdir = Path(work)
            source = self.ingest.resolve(media, workdir / "inputs", "media")
            metadata = self.assets.probe(source)
            duration = max(0.0, metadata.duration)
            words = transcript.split()
            words_per_minute = (len(words) / duration * 60) if duration > 0 and words else None
            highlights = _highlight_windows(duration)
            viral_score = _viral_score(duration, bool(words), metadata.width, metadata.height)
            analysis = {
                "media": str(source),
                "metadata": metadata.__dict__,
                "transcript": transcript,
                "highlight_moments": highlights,
                "viral_score": viral_score,
                "hook_quality": _hook_quality(transcript),
                "audience_retention_estimate": _retention_estimate(duration, viral_score),
                "engagement_prediction": _engagement_prediction(viral_score),
                "audience_persona": _persona(transcript),
                "platform_recommendations": _platform_recommendations(duration, metadata.width, metadata.height, platform),
                "scene_segmentation": highlights,
                "speech_pacing": {
                    "words_per_minute": round(words_per_minute, 1) if words_per_minute else None,
                    "label": _pacing_label(words_per_minute),
                },
                "silence_detection": {
                    "estimated_silence_ratio": 0.0 if transcript else 0.18,
                    "note": "Heuristic estimate; use transcription with word timestamps for precise silence spans.",
                },
            }
            output = self._json_artifact(job_id, "analysis", analysis)
        return self._result(output, {"task": "analyze", "viral_score": viral_score, "duration": duration})

    def metadata(self, job_id: str, *, topic: str = "", transcript: str = "", platform: str | None = None) -> TaskResult:
        text = transcript or topic or "Untitled video"
        title = _title_from_text(text, platform)
        tags = _hashtags(text, platform)
        payload = {
            "title": title,
            "description": _description(text, tags),
            "hashtags": tags,
            "keywords": _keywords(text),
            "chapters": _chapters(text),
            "seo_tags": _keywords(text) + [platform] if platform else _keywords(text),
            "suggested_upload_schedule": _schedule(platform),
            "platform": platform or "general",
        }
        output = self._json_artifact(job_id, "metadata", payload)
        return self._result(output, {"task": "metadata", "keyword_count": len(payload["keywords"])})

    def publish(self, payload: dict[str, Any], job_id: str) -> TaskResult:
        platforms = payload.get("platforms") or [payload.get("platform") or "draft"]
        manifest = {
            "publish_state": "draft_ready" if payload.get("draft", True) else "credentials_required",
            "platforms": platforms,
            "scheduled_at": payload.get("scheduled_at"),
            "asset": payload.get("asset") or payload.get("media"),
            "title": payload.get("title"),
            "description": payload.get("description"),
            "retry_policy": {"max_attempts": 3, "backoff_seconds": 60},
            "note": "Direct publishing requires platform OAuth/API credentials configured outside this CPU render worker.",
        }
        output = self._json_artifact(job_id, "publish_manifest", manifest)
        return self._result(output, {"task": "publish", "platform_count": len(platforms)})

    def clips(self, media: str, job_id: str, clips: list[dict[str, Any]] | None = None) -> TaskResult:
        with temp_workdir(self.settings.temp_dir, f"{job_id}_clips") as work:
            workdir = Path(work)
            source = self.ingest.resolve(media, workdir / "inputs", "media")
            metadata = self.assets.probe(source)
            clip_specs = clips or _highlight_windows(metadata.duration)
            outputs: list[Path] = []
            for index, clip in enumerate(clip_specs):
                start = max(0.0, float(clip.get("start", 0)))
                end = float(clip.get("end", start + clip.get("duration", 8)))
                duration = max(0.2, end - start)
                target = workdir / f"clip_{index + 1:02d}.mp4"
                command = (
                    FFmpegCommand()
                    .add("-hide_banner", "-ss", start)
                    .input(source)
                    .add("-t", duration, "-c", "copy")
                    .overwrite()
                    .add(target)
                    .build()
                )
                self._run(command)
                outputs.append(target)
            if len(outputs) == 1:
                final = self._export(outputs[0], job_id, outputs[0].name)
            else:
                final = self.settings.exports_dir / f"{job_id}_clips.zip"
                with zipfile.ZipFile(final, "w", zipfile.ZIP_DEFLATED) as archive:
                    for path in outputs:
                        archive.write(path, path.name)
        return self._result(final, {"task": "clips", "clip_count": len(outputs)})

    def thumbnail(self, media: str, job_id: str, *, text: str = "", timestamp: float | None = None, template: str = "bold") -> TaskResult:
        with temp_workdir(self.settings.temp_dir, f"{job_id}_thumb") as work:
            workdir = Path(work)
            source = self.ingest.resolve(media, workdir / "inputs", "media")
            metadata = self.assets.probe(source)
            target = workdir / "thumbnail.jpg"
            seek = timestamp if timestamp is not None else max(0.0, min(metadata.duration * 0.2, 8.0))
            vf = "scale=1280:720:force_original_aspect_ratio=increase,crop=1280:720"
            if text:
                vf += "," + _drawtext_filter(text, template)
            command = (
                FFmpegCommand()
                .add("-hide_banner", "-ss", seek)
                .input(source)
                .add("-frames:v", 1, "-vf", vf, "-q:v", 2)
                .overwrite()
                .add(target)
                .build()
            )
            self._run(command)
            final = self._export(target, job_id, "thumbnail.jpg")
        return self._result(final, {"task": "thumbnail", "timestamp": seek})

    def toolkit(self, payload: dict[str, Any], job_id: str) -> TaskResult:
        task = str(payload.get("task") or payload.get("operation") or "").strip()
        if task not in TOOLKIT_TASKS:
            raise ValueError(f"Unsupported toolkit task: {task}")
        if task == "thumbnail":
            return self.thumbnail(str(payload["input"]), job_id, text=str(payload.get("text") or ""), timestamp=payload.get("timestamp"))
        if task == "split":
            clips = payload.get("clips") if isinstance(payload.get("clips"), list) else payload.get("params", {}).get("clips")
            return self.clips(str(payload.get("input") or payload.get("media")), job_id, clips)

        with temp_workdir(self.settings.temp_dir, f"{job_id}_{task}") as work:
            workdir = Path(work)
            source_value = payload.get("input") or payload.get("media")
            source = self.ingest.resolve(str(source_value), workdir / "inputs", "media") if source_value else workdir / "concat_placeholder.mp4"
            params = payload.get("params") if isinstance(payload.get("params"), dict) else payload
            output_name = safe_filename(str(payload.get("output_name") or _default_output_name(task)))
            output = workdir / output_name
            if task == "extract_audio" and output.suffix.lower() != ".mp3":
                output = output.with_suffix(".mp3")
            if task == "gif" and output.suffix.lower() != ".gif":
                output = output.with_suffix(".gif")
            command = self._toolkit_command(task, source, output, params, workdir)
            self._run(command)
            if task == "frames":
                final = self.settings.exports_dir / f"{job_id}_frames.zip"
                with zipfile.ZipFile(final, "w", zipfile.ZIP_DEFLATED) as archive:
                    for frame in sorted(workdir.glob("frame_*.jpg")):
                        archive.write(frame, frame.name)
            else:
                final = self._export(output, job_id, output.name)
        return self._result(final, {"task": task})

    def _toolkit_command(self, task: str, source: Path, output: Path, params: dict[str, Any], workdir: Path) -> list[str]:
        cmd = FFmpegCommand().add("-hide_banner")
        if task == "loop":
            cmd.add("-stream_loop", int(params.get("loops", -1)))
        if task in {"cut", "trim", "gif", "freeze_frame"} and params.get("start") is not None:
            cmd.add("-ss", float(params.get("start", 0)))
        cmd.input(source)

        if task == "merge_audio":
            audio = self.ingest.resolve(str(params.get("audio")), workdir / "inputs", "audio")
            cmd.input(audio)
            return cmd.add("-map", "0:v", "-map", "1:a", "-c:v", "copy", "-c:a", "aac", "-shortest").overwrite().add(output).build()
        if task == "watermark":
            image = self.ingest.resolve(str(params.get("watermark") or params.get("image")), workdir / "inputs", "watermark")
            cmd.input(image)
            return cmd.add("-filter_complex", "[1:v]scale=iw*0.18:-1[wm];[0:v][wm]overlay=W-w-24:H-h-24", "-c:a", "copy").overwrite().add(output).build()
        if task in {"concat", "merge"}:
            inputs = params.get("inputs")
            if not isinstance(inputs, list) or not inputs:
                raise ValueError("Concat requires params.inputs")
            concat_file = workdir / "concat.txt"
            lines: list[str] = []
            for index, item in enumerate(inputs):
                media = self.ingest.resolve(str(item), workdir / "inputs", f"concat_{index:03d}")
                lines.append(f"file '{str(media).replace(chr(39), chr(39) + chr(92) + chr(39) + chr(39))}'")
            concat_file.write_text("\n".join(lines), encoding="utf-8")
            return FFmpegCommand().add("-hide_banner", "-f", "concat", "-safe", "0").input(concat_file).add("-c", "copy").overwrite().add(output).build()

        duration = params.get("duration")
        if task in {"cut", "trim", "gif", "loop", "freeze_frame"} and duration is not None:
            cmd.add("-t", float(duration))

        vf = _video_filter(task, params)
        af = _audio_filter(task, params)
        if vf:
            cmd.add("-vf", vf)
        if af:
            cmd.add("-af", af)

        if task == "extract_audio":
            return cmd.add("-vn", "-c:a", "mp3", "-b:a", "192k").overwrite().add(output).build()
        if task == "frames":
            return cmd.add("-vf", vf or "fps=1", "-q:v", 2).overwrite().add(workdir / "frame_%04d.jpg").build()
        if task == "gif":
            return cmd.add("-loop", 0).overwrite().add(output).build()
        if task in {
            "cut",
            "trim",
            "compress",
            "normalize",
            "resize",
            "crop",
            "rotate",
            "flip",
            "scale",
            "zoom",
            "pan",
            "speed",
            "speed_ramp",
            "time_remap",
            "slow_motion",
            "fast_motion",
            "reverse",
            "reverse_playback",
            "freeze_frame",
            "motion_blur",
            "stabilization",
            "lens_correction",
            "loop",
            "overlay_text",
            "blur_background",
            "burn_subtitles",
            "convert",
            "noise_reduction",
            "equalizer",
            "compressor",
            "limiter",
            "pitch_shift",
            "voice_changer",
            "green_screen",
            "chroma_key",
            "blue_screen",
            "ai_background_removal",
        }:
            cmd.add("-c:v", "libx264", "-preset", self.settings.preset, "-crf", int(params.get("crf", self.settings.crf)), "-c:a", "aac")
        return cmd.overwrite().add(output).build()

    def _json_artifact(self, job_id: str, name: str, payload: dict[str, Any]) -> Path:
        output = self.settings.exports_dir / f"{job_id}_{safe_filename(name)}.json"
        write_json(output, payload)
        return output

    def _export(self, source: Path, job_id: str, output_name: str) -> Path:
        target = self.settings.exports_dir / f"{job_id}_{safe_filename(output_name)}"
        shutil.copy2(source, target)
        return target

    def _run(self, command: list[str]) -> None:
        result = self.runner.run(command)
        if result.stderr:
            self._logs.append(result.stderr[-4000:])

    def _record_command(self, command: list[str]) -> None:
        self._commands.append(command)

    def _result(self, output: Path | None, metrics: dict[str, Any]) -> TaskResult:
        return TaskResult(output_path=output, commands=list(self._commands), metrics=metrics, logs=list(self._logs))


def supported_toolkit_tasks() -> list[str]:
    return sorted(TOOLKIT_TASKS)


def _source_type(url: str) -> str:
    lowered = url.lower()
    if "youtube.com" in lowered or "youtu.be" in lowered:
        return "youtube"
    if "drive.google.com" in lowered:
        return "google_drive"
    if "s3" in lowered or "amazonaws.com" in lowered:
        return "s3"
    return "direct_url"


def _highlight_windows(duration: float) -> list[dict[str, Any]]:
    if duration <= 0:
        return [{"start": 0, "end": 8, "reason": "default opener"}]
    windows = [{"start": 0, "end": min(duration, 8), "reason": "opening hook"}]
    if duration > 18:
        middle = max(0.0, duration * 0.42)
        windows.append({"start": round(middle, 2), "end": round(min(duration, middle + 10), 2), "reason": "midpoint payoff"})
    if duration > 35:
        end = max(0.0, duration - 12)
        windows.append({"start": round(end, 2), "end": round(duration, 2), "reason": "closing CTA"})
    return windows


def _viral_score(duration: float, has_words: bool, width: int | None, height: int | None) -> int:
    score = 48
    if 7 <= duration <= 60:
        score += 20
    elif duration <= 180:
        score += 8
    if width and height and height >= width:
        score += 14
    if has_words:
        score += 10
    return max(1, min(100, score))


def _hook_quality(transcript: str) -> dict[str, Any]:
    opener = " ".join(transcript.split()[:18])
    signals = sum(1 for token in ("how", "why", "secret", "mistake", "stop", "watch", "you") if token in opener.lower())
    return {"score": min(100, 45 + signals * 12), "opening_text": opener}


def _retention_estimate(duration: float, viral_score: int) -> dict[str, Any]:
    first_3s = min(96, 55 + viral_score * 0.35)
    completion = max(18, min(88, first_3s - math.log(max(duration, 1), 1.8)))
    return {"first_3_seconds_percent": round(first_3s, 1), "completion_percent": round(completion, 1)}


def _engagement_prediction(score: int) -> str:
    if score >= 80:
        return "high"
    if score >= 62:
        return "medium"
    return "needs_work"


def _persona(text: str) -> str:
    lowered = text.lower()
    if any(word in lowered for word in ("founder", "startup", "product", "launch")):
        return "builders and product-led founders"
    if any(word in lowered for word in ("money", "sales", "growth", "marketing")):
        return "growth-minded operators"
    if any(word in lowered for word in ("learn", "tutorial", "how")):
        return "learners seeking practical instruction"
    return "general short-form viewers"


def _platform_recommendations(duration: float, width: int | None, height: int | None, platform: str | None) -> list[dict[str, Any]]:
    vertical = bool(width and height and height >= width)
    candidates = ["tiktok", "instagram_reels", "youtube_shorts"] if vertical else ["youtube_1080p", "linkedin_video"]
    if platform and platform not in candidates:
        candidates.insert(0, platform)
    return [{"platform": item, "fit": "strong" if duration <= 90 else "medium"} for item in candidates]


def _pacing_label(wpm: float | None) -> str:
    if not wpm:
        return "unknown"
    if wpm < 125:
        return "slow"
    if wpm > 185:
        return "fast"
    return "clear"


def _title_from_text(text: str, platform: str | None) -> str:
    words = [word.strip(".,:;!?") for word in text.split() if word.strip(".,:;!?")]
    title = " ".join(words[:9]) or "Untitled Video"
    suffix = " #Shorts" if platform in {"youtube_shorts", "tiktok", "instagram_reels"} else ""
    return f"{title.title()}{suffix}"


def _description(text: str, tags: list[str]) -> str:
    summary = " ".join(text.split()[:42])
    return f"{summary}\n\n{' '.join(tags)}".strip()


def _hashtags(text: str, platform: str | None) -> list[str]:
    base = ["#video", "#content"]
    if platform:
        base.append(f"#{platform.replace('_', '')}")
    for keyword in _keywords(text)[:5]:
        tag = "#" + "".join(ch for ch in keyword.title() if ch.isalnum())
        if tag not in base:
            base.append(tag)
    return base[:8]


def _keywords(text: str) -> list[str]:
    stop = {"the", "and", "for", "with", "that", "this", "your", "you", "are", "from", "into", "video"}
    words = [word.strip(".,:;!?").lower() for word in text.split()]
    unique: list[str] = []
    for word in words:
        if len(word) < 4 or word in stop or word in unique:
            continue
        unique.append(word)
    return unique[:12]


def _chapters(text: str) -> list[dict[str, Any]]:
    sentences = [part.strip() for part in text.replace("?", ".").replace("!", ".").split(".") if part.strip()]
    return [{"time": f"0:{index * 15:02d}", "title": sentence[:60]} for index, sentence in enumerate(sentences[:6])]


def _schedule(platform: str | None) -> dict[str, str]:
    if platform in {"linkedin_video", "youtube_1080p"}:
        return {"day": "Tuesday", "time": "09:00 local"}
    return {"day": "Thursday", "time": "18:00 local"}


def _video_filter(task: str, params: dict[str, Any]) -> str:
    if task in {"compress", "normalize"}:
        profile = get_platform_profile(params.get("platform"))
        return f"scale={profile.width}:{profile.height}:force_original_aspect_ratio=increase,crop={profile.width}:{profile.height},fps={profile.fps},format=yuv420p"
    if task in {"resize", "scale"}:
        return f"scale={int(params.get('width', 1080))}:{int(params.get('height', 1920))}"
    if task == "crop":
        return f"crop={int(params.get('width', 1080))}:{int(params.get('height', 1080))}:{int(params.get('x', 0))}:{int(params.get('y', 0))}"
    if task == "rotate":
        return {"90": "transpose=1", "180": "hflip,vflip", "270": "transpose=2"}.get(str(params.get("degrees", "90")), "transpose=1")
    if task == "flip":
        axis = str(params.get("axis", "horizontal"))
        return "vflip" if axis in {"vertical", "y"} else "hflip"
    if task in {"speed", "speed_ramp", "time_remap", "slow_motion", "fast_motion"}:
        default = 0.5 if task == "slow_motion" else 2.0 if task == "fast_motion" else 1.0
        factor = max(0.25, min(4.0, float(params.get("factor", default))))
        return f"setpts={1 / factor:.4f}*PTS"
    if task == "zoom":
        factor = max(1.0, min(4.0, float(params.get("factor", 1.2))))
        return f"scale=iw*{factor:.3f}:ih*{factor:.3f},crop=iw/{factor:.3f}:ih/{factor:.3f}"
    if task == "pan":
        width = int(params.get("width", 1080))
        height = int(params.get("height", 1920))
        x = str(params.get("x", "(iw-ow)/2"))
        y = str(params.get("y", "(ih-oh)/2"))
        return f"crop={width}:{height}:{x}:{y}"
    if task in {"motion_blur", "freeze_frame"}:
        return "tmix=frames=3:weights='1 2 1'"
    if task == "stabilization":
        return "deshake"
    if task == "lens_correction":
        return f"lenscorrection=k1={float(params.get('k1', -0.15))}:k2={float(params.get('k2', 0.05))}"
    if task in {"reverse", "reverse_playback"}:
        return "reverse"
    if task in {"green_screen", "chroma_key", "ai_background_removal"}:
        color = str(params.get("color") or "0x00ff00")
        similarity = float(params.get("similarity", 0.18))
        blend = float(params.get("blend", 0.08))
        return f"chromakey={color}:{similarity}:{blend}"
    if task == "blue_screen":
        similarity = float(params.get("similarity", 0.18))
        blend = float(params.get("blend", 0.08))
        return f"chromakey=0x0000ff:{similarity}:{blend}"
    if task == "speed":
        factor = max(0.25, min(4.0, float(params.get("factor", 1.0))))
        return f"setpts={1 / factor:.4f}*PTS"
    if task == "gif":
        return f"fps={int(params.get('fps', 12))},scale={int(params.get('width', 540))}:-1:flags=lanczos"
    if task == "frames":
        return f"fps={float(params.get('fps', 1))}"
    if task == "overlay_text":
        return _drawtext_filter(str(params.get("text") or "Text"), "bold")
    if task == "blur_background":
        return "gblur=sigma=18"
    if task == "burn_subtitles":
        subtitles = str(params.get("subtitles") or "").replace("\\", "/").replace(":", r"\:")
        return f"subtitles='{subtitles}'"
    return ""


def _audio_filter(task: str, params: dict[str, Any]) -> str:
    if task in {"speed", "speed_ramp", "time_remap", "slow_motion", "fast_motion"}:
        default = 0.5 if task == "slow_motion" else 2.0 if task == "fast_motion" else 1.0
        factor = max(0.5, min(2.0, float(params.get("factor", default))))
        return f"atempo={factor}"
    if task in {"reverse", "reverse_playback"}:
        return "areverse"
    if task in {"noise_reduction", "ai_enhancement"}:
        return "afftdn=nf=-25"
    if task == "equalizer":
        return f"equalizer=f={float(params.get('frequency', 1000))}:width_type=o:width={float(params.get('width', 1))}:g={float(params.get('gain', 3))}"
    if task == "compressor":
        return "acompressor=threshold=-18dB:ratio=3:attack=20:release=250"
    if task == "limiter":
        return "alimiter=limit=0.95"
    if task in {"pitch_shift", "voice_changer"}:
        factor = max(0.5, min(2.0, float(params.get("factor", 1.0))))
        return f"asetrate=48000*{factor:.4f},aresample=48000,atempo={1 / factor:.4f}"
    return ""


def _drawtext_filter(text: str, template: str) -> str:
    escaped = text.replace("\\", "\\\\").replace(":", r"\:").replace("'", r"\'")
    color = "yellow" if template == "bold" else "white"
    return (
        "drawtext="
        f"text='{escaped}':fontcolor={color}:fontsize=58:"
        "box=1:boxcolor=black@0.55:boxborderw=24:"
        "x=(w-text_w)/2:y=h-(text_h*3)"
    )


def _default_output_name(task: str) -> str:
    if task == "extract_audio":
        return "audio.mp3"
    if task == "gif":
        return "clip.gif"
    if task == "thumbnail":
        return "thumbnail.jpg"
    return f"{task}.mp4"