backend/src/analysis/highlight_scorer.py

"""Multi-signal highlight scoring: Vision + Audio energy + Text keywords."""
import math
from pathlib import Path
from loguru import logger


# Style-specific keyword boosts
STYLE_KEYWORDS = {
    "funny": ["haha", "lol", "funny", "joke", "laugh", "omg", "what", "no way", "ตลก", "ฮา", "โอ้โห", "搞笑", "哈哈"],
    "serious": ["important", "key", "must", "critical", "สำคัญ", "ต้อง", "หลัก", "重要", "关键"],
    "educational": ["learn", "tip", "trick", "how", "why", "เรียน", "วิธี", "ทำไม", "学习", "方法", "技巧"],
    "gaming": ["win", "lose", "boss", "kill", "score", "level", "ชนะ", "แพ้", "赢", "输"],
    "entertainment": ["wow", "amazing", "incredible", "unbelievable", "เจ๋ง", "เยี่ยม", "厉害", "太棒了"],
}


def compute_audio_energy(audio_path: Path, scenes: list[dict]) -> list[float]:
    """Compute RMS energy per scene using librosa."""
    try:
        import librosa
        import numpy as np

        y, sr = librosa.load(str(audio_path), sr=16000, mono=True)
        energies = []

        for scene in scenes:
            start_sample = int(scene["start"] * sr)
            end_sample = int(scene["end"] * sr)
            segment = y[start_sample:end_sample]
            if len(segment) == 0:
                energies.append(0.0)
                continue
            rms = float(np.sqrt(np.mean(segment ** 2)))
            energies.append(rms)

        # Normalize to 0-1
        if max(energies) > 0:
            max_e = max(energies)
            energies = [e / max_e for e in energies]

        return energies

    except ImportError:
        logger.warning("librosa not installed, using uniform audio energy")
        return [0.5] * len(scenes)
    except Exception as e:
        logger.error(f"Audio energy computation failed: {e}")
        return [0.5] * len(scenes)


def compute_text_score(transcript_text: str, clip_style: str) -> float:
    """Score transcript text based on style keywords (0-1)."""
    if not transcript_text:
        return 0.3

    text_lower = transcript_text.lower()
    keywords = STYLE_KEYWORDS.get(clip_style.lower(), [])
    if not keywords:
        return 0.3

    hits = sum(1 for kw in keywords if kw in text_lower)
    score = min(1.0, hits / max(len(keywords) * 0.2, 1))
    return max(0.1, score)


def score_scenes(
    scenes_analyzed: list[dict],
    audio_path: Path,
    clip_style: str = "entertaining",
    target_duration: int = 60,
) -> list[dict]:
    """Compute final highlight scores for all scenes.

    Final score = 0.40 × vision + 0.35 × audio_energy + 0.25 × text_keywords
    """
    # Audio energy per scene
    audio_energies = compute_audio_energy(audio_path, scenes_analyzed)

    scored = []
    for i, scene in enumerate(scenes_analyzed):
        analysis = scene.get("vision_analysis", {})

        vision_score = (
            analysis.get("excitement_score", 0.5) * 0.5 +
            analysis.get("tiktok_potential", 0.5) * 0.3 +
            analysis.get("humor_level", 0.3) * 0.2
        )

        audio_score = audio_energies[i]

        # Text from transcript segments overlapping this scene
        transcript_text = scene.get("transcript_text", "")
        text_score = compute_text_score(transcript_text, clip_style)

        final_score = (
            0.40 * vision_score +
            0.35 * audio_score +
            0.25 * text_score
        )

        # Penalize very short or very long scenes relative to target
        duration = scene["duration"]
        duration_penalty = 1.0 - abs(duration - target_duration) / max(target_duration * 2, 1)
        duration_penalty = max(0.5, duration_penalty)

        scored.append({
            **scene,
            "vision_score": round(vision_score, 3),
            "audio_score": round(audio_score, 3),
            "text_score": round(text_score, 3),
            "final_score": round(final_score * duration_penalty, 3),
        })

    scored.sort(key=lambda s: s["final_score"], reverse=True)
    logger.info(f"Top scene: {scored[0]['start']:.1f}s score={scored[0]['final_score']:.3f}" if scored else "No scenes")
    return scored


def select_top_clips(
    scored_scenes: list[dict],
    count: int,
    target_duration: int,
    min_gap_sec: float = 30.0,
) -> list[dict]:
    """Select top-N non-overlapping clips.

    Merges adjacent high-scoring scenes to reach target_duration.
    Ensures clips don't overlap (min_gap_sec between selections).
    """
    selected = []
    used_ranges = []

    for scene in scored_scenes:
        if len(selected) >= count:
            break

        # Check overlap with already selected clips
        overlaps = any(
            abs(scene["start"] - used_start) < min_gap_sec
            for used_start in used_ranges
        )
        if overlaps:
            continue

        # Adjust clip boundaries to match target_duration
        clip = _adjust_clip_duration(scene, target_duration)
        selected.append(clip)
        used_ranges.append(clip["start"])

    logger.info(f"Selected {len(selected)}/{count} clips")
    return sorted(selected, key=lambda c: c["start"])


def _adjust_clip_duration(scene: dict, target_sec: int) -> dict:
    """Expand or shrink a scene to approximately target_sec."""
    current_dur = scene["end"] - scene["start"]
    if abs(current_dur - target_sec) < 5:
        return scene

    # Center the target window on the scene midpoint
    mid = (scene["start"] + scene["end"]) / 2
    half = target_sec / 2
    new_start = max(0, mid - half)
    new_end = new_start + target_sec

    return {**scene, "start": new_start, "end": new_end, "duration": target_sec}