src/humeo/hook_detector.py

"""Stage 2.25 - Hook detection.



The clip-selection LLM returns a ``hook_start_sec`` / ``hook_end_sec`` pair

per clip, but in practice it almost always echoes the ``[0.0, 3.0]``

placeholder from the prompt instead of localising the real hook sentence.

That placeholder is toxic to Stage 2.5 pruning -- the clamp refuses to

trim past ``hook_start_sec``, so every ``trim_start_sec > 0`` the pruner

returns gets zeroed out silently.



This module is a dedicated Stage 2.25 that runs between clip selection and

content pruning. For each clip it:



1. Prepares a clip-relative segment listing (same format as pruning uses).

2. Asks Gemini, in one batched JSON call, to localise the hook sentence of

   every clip with `hook_start_sec`, `hook_end_sec`, `hook_text`, `reason`.

3. Validates the returned window against the clip's duration + the "real

   hook" heuristics, then overwrites ``clip.hook_start_sec`` /

   ``clip.hook_end_sec`` on a copy of the clip.



The stage is:



- **Cached** (``hooks.json`` / ``hooks.meta.json`` in ``work_dir``) on

  ``transcript_sha256 + clips_sha256 + gemini_model``.

- **Never fatal.** Any failure (API error, malformed JSON, clip not

  returned, window that still looks like the 0.0-3.0 placeholder) falls

  back to the original clip with its original hook -- pruning will then

  skip hook protection via the fingerprint guard in

  :func:`humeo.content_pruning._looks_like_default_hook`.



The stage writes three artifacts to ``work_dir`` for audit:



- ``hooks.meta.json``: cache key (version, fingerprints, model).

- ``hooks.json``: structured per-clip hook windows actually applied.

- ``hooks_raw.json``: verbatim Gemini response text (for prompt tuning).

"""

from __future__ import annotations

import hashlib
import json
import logging
import time
from pathlib import Path
from typing import Any, Callable, TypeVar

from google import genai
from openai import OpenAI
from pydantic import BaseModel, Field, ValidationError

from humeo_core.schemas import Clip

from humeo.config import GEMINI_MODEL, PipelineConfig
from humeo.content_pruning import _looks_like_default_hook, _segments_within_clip
from humeo.env import (
    OPENROUTER_BASE_URL,
    current_llm_provider,
    model_name_for_provider,
    openrouter_default_headers,
    resolve_gemini_api_key,
    resolve_llm_provider,
    resolve_openrouter_api_keys,
)
from humeo.gemini_generate import gemini_generate_config
from humeo.hook_library import (
    format_hook_examples,
    hook_library_fingerprint,
    resolve_hook_library_path,
    retrieve_hook_examples,
)
from humeo.prompt_loader import hook_detection_system_prompt

logger = logging.getLogger(__name__)

T = TypeVar("T")

HOOK_META_VERSION = 2
HOOK_META_FILENAME = "hooks.meta.json"
HOOK_ARTIFACT_FILENAME = "hooks.json"
HOOK_RAW_FILENAME = "hooks_raw.json"

LLM_MAX_ATTEMPTS = 3
LLM_RETRY_DELAY_SEC = 2.0

# Hook window validation thresholds. The prompt asks for 1.5-7.0s windows;
# we enforce 1.0-10.0s to be lenient on rounding while still rejecting
# obvious "LLM returned the whole paragraph" mistakes.
_MIN_HOOK_DURATION_SEC = 1.0
_MAX_HOOK_DURATION_SEC = 10.0


def _openai_message_text(content: object) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for item in content:
            if isinstance(item, dict) and item.get("type") == "text":
                text = item.get("text")
                if isinstance(text, str):
                    parts.append(text)
        return "".join(parts)
    return ""


class _HookDecision(BaseModel):
    """Per-clip hook window returned by Gemini (clip-relative seconds)."""

    clip_id: str
    hook_start_sec: float = Field(ge=0.0)
    hook_end_sec: float = Field(ge=0.0)
    hook_text: str = ""
    reason: str = ""


class _HookResponse(BaseModel):
    hooks: list[_HookDecision] = Field(default_factory=list)


def _retry_llm(name: str, fn: Callable[[], T], attempts: int = LLM_MAX_ATTEMPTS) -> T:
    last: Exception | None = None
    for i in range(attempts):
        try:
            return fn()
        except Exception as e:  # noqa: BLE001 - rethrown below
            last = e
            if i < attempts - 1:
                logger.warning("%s attempt %d/%d failed: %s", name, i + 1, attempts, e)
                time.sleep(LLM_RETRY_DELAY_SEC * (i + 1))
    assert last is not None
    raise last


# ---------------------------------------------------------------------------
# Prompt construction
# ---------------------------------------------------------------------------


def _build_user_message(clips: list[Clip], transcript: dict) -> str:
    """Render clip-relative segments + selector-guessed hook text for each clip."""
    blocks: list[str] = []
    for clip in clips:
        segs = _segments_within_clip(transcript, clip)
        header_lines = [
            f"clip_id: {clip.clip_id}",
            f"duration_sec: {clip.duration_sec:.2f}",
            f"topic: {clip.topic}",
        ]
        if clip.viral_hook:
            header_lines.append(f"viral_hook_text: {clip.viral_hook}")
        if clip.hook_start_sec is not None and clip.hook_end_sec is not None:
            header_lines.append(
                f"selector_hook_window_sec: [{clip.hook_start_sec:.2f}, "
                f"{clip.hook_end_sec:.2f}] (may be a placeholder; verify)"
            )
        header = "\n".join(header_lines)
        body = "\n".join(
            f"[{seg['start']:.2f}s - {seg['end']:.2f}s] {seg['text']}" for seg in segs
        )
        if not body:
            body = "(no segments overlap this clip window)"
        blocks.append(f"{header}\n---\n{body}")
    return "\n\n===\n\n".join(blocks)


# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------


def _validate_hook_window(

    clip: Clip, hook_start: float, hook_end: float

) -> tuple[float, float] | None:
    """Return a valid (hook_start, hook_end) or None if rejected.



    Rules:

    - ``0 <= hook_start < hook_end <= duration_sec``

    - hook duration between ``_MIN_HOOK_DURATION_SEC`` and ``_MAX_HOOK_DURATION_SEC``

    - NOT the ``(0.0, 3.0)`` placeholder fingerprint (we'd rather keep the

      selector's value untouched than re-apply the same fake hook).

    """
    if hook_start < 0.0 or hook_end <= hook_start:
        return None
    if hook_end > clip.duration_sec + 1e-3:
        # Clamp trailing rounding to duration; reject anything beyond.
        if hook_end - clip.duration_sec > 0.5:
            return None
        hook_end = clip.duration_sec
    dur = hook_end - hook_start
    if dur < _MIN_HOOK_DURATION_SEC or dur > _MAX_HOOK_DURATION_SEC:
        return None
    if _looks_like_default_hook(hook_start, hook_end):
        return None
    return float(hook_start), float(hook_end)


# ---------------------------------------------------------------------------
# Apply decisions -> new clips
# ---------------------------------------------------------------------------


def apply_hook_decisions(

    clips: list[Clip],

    decisions: list[_HookDecision],

) -> list[Clip]:
    """Return new clips whose hook fields reflect validated decisions.



    Clips without a matching valid decision are returned unchanged (their

    original hook metadata, placeholder or not, is preserved).

    """
    by_id = {d.clip_id: d for d in decisions}
    out: list[Clip] = []
    changed = 0
    rejected = 0
    for clip in clips:
        d = by_id.get(clip.clip_id)
        if d is None:
            out.append(clip)
            continue
        validated = _validate_hook_window(clip, d.hook_start_sec, d.hook_end_sec)
        if validated is None:
            logger.info(
                "Clip %s: rejected hook window [%.2f, %.2f] (failed validation); "
                "keeping selector hook.",
                clip.clip_id,
                d.hook_start_sec,
                d.hook_end_sec,
            )
            rejected += 1
            out.append(clip)
            continue
        hs, he = validated
        if (
            clip.hook_start_sec is not None
            and clip.hook_end_sec is not None
            and abs(clip.hook_start_sec - hs) < 1e-3
            and abs(clip.hook_end_sec - he) < 1e-3
        ):
            out.append(clip)
            continue
        changed += 1
        logger.info(
            "Clip %s: hook set to [%.2f, %.2f] (was [%s, %s]) -- %s",
            clip.clip_id,
            hs,
            he,
            f"{clip.hook_start_sec:.2f}" if clip.hook_start_sec is not None else "None",
            f"{clip.hook_end_sec:.2f}" if clip.hook_end_sec is not None else "None",
            d.reason[:120] if d.reason else "(no reason)",
        )
        out.append(
            clip.model_copy(update={"hook_start_sec": hs, "hook_end_sec": he})
        )
    logger.info(
        "Hook detection: updated %d / %d clips (%d rejected, %d kept as-is).",
        changed,
        len(clips),
        rejected,
        len(clips) - changed - rejected,
    )
    return out


# ---------------------------------------------------------------------------
# Cache
# ---------------------------------------------------------------------------


def _clips_fingerprint(clips: list[Clip]) -> str:
    payload = json.dumps(
        [
            {"id": c.clip_id, "s": round(c.start_time_sec, 3), "e": round(c.end_time_sec, 3)}
            for c in clips
        ],
        sort_keys=True,
        ensure_ascii=False,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _resolved_gemini_model(config: PipelineConfig) -> str:
    return (config.gemini_model or GEMINI_MODEL).strip()


def _hook_meta(

    *,

    transcript_fp: str,

    clips_fp: str,

    config: PipelineConfig,

) -> dict[str, Any]:
    return {
        "version": HOOK_META_VERSION,
        "transcript_sha256": transcript_fp,
        "clips_sha256": clips_fp,
        "gemini_model": _resolved_gemini_model(config),
        "llm_backend": current_llm_provider() or "google",
        "hook_library_sha256": hook_library_fingerprint(resolve_hook_library_path(config)),
    }


def _hook_cache_valid(

    work_dir: Path,

    *,

    transcript_fp: str,

    clips_fp: str,

    config: PipelineConfig,

) -> bool:
    meta_path = work_dir / HOOK_META_FILENAME
    if not meta_path.is_file():
        return False
    try:
        with open(meta_path, encoding="utf-8") as f:
            meta = json.load(f)
    except Exception:
        return False
    if meta.get("version") != HOOK_META_VERSION:
        return False
    if meta.get("transcript_sha256") != transcript_fp:
        return False
    if meta.get("clips_sha256") != clips_fp:
        return False
    current_provider = current_llm_provider()
    meta_provider = meta.get("llm_backend")
    if current_provider == "openrouter":
        if meta_provider != "openrouter":
            return False
    elif current_provider == "google":
        if meta_provider not in (None, "google"):
            return False
    if meta.get("gemini_model") != _resolved_gemini_model(config):
        return False
    if meta.get("hook_library_sha256", "") != hook_library_fingerprint(resolve_hook_library_path(config)):
        return False
    return True


def _load_cached_hooks(

    work_dir: Path, clips: list[Clip]

) -> list[Clip] | None:
    artifact = work_dir / HOOK_ARTIFACT_FILENAME
    if not artifact.is_file():
        return None
    try:
        with open(artifact, "r", encoding="utf-8") as f:
            data = json.load(f)
        cached = {item["clip_id"]: item for item in data.get("hooks", [])}
    except Exception as e:  # noqa: BLE001 - surfaced as warning below
        logger.warning("Hook cache artifact unreadable (%s); re-running.", e)
        return None
    out: list[Clip] = []
    for clip in clips:
        c = cached.get(clip.clip_id)
        if c is None:
            out.append(clip)
            continue
        hs = c.get("hook_start_sec")
        he = c.get("hook_end_sec")
        if hs is None or he is None:
            out.append(clip)
            continue
        out.append(
            clip.model_copy(
                update={"hook_start_sec": float(hs), "hook_end_sec": float(he)}
            )
        )
    return out


def _write_cache(

    work_dir: Path,

    *,

    clips_with_hooks: list[Clip],

    decisions: list[_HookDecision],

    meta: dict[str, Any],

    raw_response: str,

) -> None:
    work_dir.mkdir(parents=True, exist_ok=True)
    reasons = {d.clip_id: d for d in decisions}
    payload = {
        "hooks": [
            {
                "clip_id": c.clip_id,
                "hook_start_sec": c.hook_start_sec,
                "hook_end_sec": c.hook_end_sec,
                "hook_text": (reasons.get(c.clip_id).hook_text if reasons.get(c.clip_id) else ""),
                "reason": (reasons.get(c.clip_id).reason if reasons.get(c.clip_id) else ""),
            }
            for c in clips_with_hooks
        ]
    }
    (work_dir / HOOK_ARTIFACT_FILENAME).write_text(
        json.dumps(payload, indent=2), encoding="utf-8"
    )
    (work_dir / HOOK_RAW_FILENAME).write_text(raw_response, encoding="utf-8")
    with open(work_dir / HOOK_META_FILENAME, "w", encoding="utf-8") as f:
        json.dump(meta, f, indent=2)
        f.write("\n")
    logger.info(
        "Wrote %s, %s and %s",
        HOOK_META_FILENAME,
        HOOK_ARTIFACT_FILENAME,
        HOOK_RAW_FILENAME,
    )


# ---------------------------------------------------------------------------
# Gemini call
# ---------------------------------------------------------------------------


def _parse_decisions(raw_json: str) -> list[_HookDecision]:
    data = json.loads(raw_json)
    if isinstance(data, dict) and "hooks" in data:
        try:
            return _HookResponse.model_validate(data).hooks
        except ValidationError as e:
            logger.warning("Hook response failed validation: %s", e)
            return []
    if isinstance(data, list):
        out: list[_HookDecision] = []
        for item in data:
            try:
                out.append(_HookDecision.model_validate(item))
            except ValidationError:
                continue
        return out
    return []


def request_hook_decisions(

    clips: list[Clip],

    transcript: dict,

    *,

    gemini_model: str | None = None,

    hook_library_path: Path | None = None,

) -> tuple[list[_HookDecision], str]:
    """Ask Gemini to localise the hook sentence for each clip.



    Returns ``(decisions, raw_response)``. ``raw_response`` is the literal

    JSON text from Gemini (cached to ``hooks_raw.json`` for audit). On

    transport/parse failure this raises; callers should catch and treat as

    no-op.

    """
    if not clips:
        return [], '{"hooks": []}'

    example_query = " ".join(
        filter(None, [*(clip.topic for clip in clips[:4]), *(clip.viral_hook for clip in clips[:4])])
    )
    hook_examples = format_hook_examples(
        retrieve_hook_examples(example_query, path=hook_library_path, limit=8)
    )
    system = hook_detection_system_prompt(hook_examples=hook_examples)
    user_text = _build_user_message(clips, transcript)

    provider = resolve_llm_provider()
    model_name = model_name_for_provider((gemini_model or GEMINI_MODEL).strip(), provider)

    def _call() -> str:
        logger.info(
            "%s hook detection (model=%s, clips=%d)...", provider, model_name, len(clips)
        )
        if provider == "google":
            client = genai.Client(api_key=resolve_gemini_api_key())
            response = client.models.generate_content(
                model=model_name,
                contents=user_text,
                config=gemini_generate_config(
                    system_instruction=system,
                    temperature=0.2,
                    response_mime_type="application/json",
                ),
            )
            if not response.text:
                raise RuntimeError("Gemini returned empty response text for hook detection")
            return response.text

        keys = resolve_openrouter_api_keys()
        last_error: Exception | None = None
        for key_idx, api_key in enumerate(keys, start=1):
            try:
                client = OpenAI(
                    api_key=api_key,
                    base_url=OPENROUTER_BASE_URL,
                    default_headers=openrouter_default_headers(),
                )
                response = client.chat.completions.create(
                    model=model_name,
                    messages=[
                        {"role": "system", "content": system},
                        {"role": "user", "content": user_text},
                    ],
                    temperature=0.2,
                    response_format={"type": "json_object"},
                )
                text = _openai_message_text(response.choices[0].message.content)
                if not text:
                    raise RuntimeError("OpenRouter returned empty response text for hook detection")
                if key_idx > 1:
                    logger.info("OpenRouter hook detection succeeded with fallback key %d/%d", key_idx, len(keys))
                return text
            except Exception as exc:
                last_error = exc
                if key_idx < len(keys):
                    logger.warning(
                        "OpenRouter hook detection failed with key %d/%d: %s; trying fallback",
                        key_idx,
                        len(keys),
                        exc,
                    )
        assert last_error is not None
        raise last_error

    raw = _retry_llm("Gemini hook detection", _call)
    decisions = _parse_decisions(raw)
    return decisions, raw


# ---------------------------------------------------------------------------
# Public stage entrypoint
# ---------------------------------------------------------------------------


def run_hook_detection_stage(

    work_dir: Path,

    clips: list[Clip],

    transcript: dict,

    *,

    transcript_fp: str,

    config: PipelineConfig,

) -> list[Clip]:
    """Run Stage 2.25 hook detection and return clips with localised hooks.



    - Disabled (``config.detect_hooks is False``): return clips unchanged.

    - Cache hit: read ``hooks.json`` and apply cached windows.

    - LLM failure: log a warning and return clips unchanged. The downstream

      content pruner's fingerprint guard will treat any remaining placeholder

      hooks as "no hook" so pruning still runs.

    """
    if not config.detect_hooks:
        logger.info("Hook detection disabled (detect_hooks=False); skipping Stage 2.25.")
        return clips
    if not clips:
        return clips

    clips_fp = _clips_fingerprint(clips)

    if not config.force_hook_detection and _hook_cache_valid(
        work_dir,
        transcript_fp=transcript_fp,
        clips_fp=clips_fp,
        config=config,
    ):
        cached = _load_cached_hooks(work_dir, clips)
        if cached is not None:
            logger.info(
                "Hook detection cache hit (%d clips); skipping LLM.", len(clips)
            )
            return cached

    try:
        decisions, raw = request_hook_decisions(
            clips,
            transcript,
            gemini_model=config.gemini_model,
            hook_library_path=resolve_hook_library_path(config),
        )
    except Exception as e:  # noqa: BLE001 - pipeline must not die here
        logger.warning(
            "Hook detection call failed (%s); continuing with selector hooks. "
            "Content pruning will treat any [0.0, 3.0] placeholder as 'no hook'.",
            e,
        )
        return clips

    updated = apply_hook_decisions(clips, decisions)

    meta = _hook_meta(
        transcript_fp=transcript_fp, clips_fp=clips_fp, config=config
    )
    try:
        _write_cache(
            work_dir,
            clips_with_hooks=updated,
            decisions=decisions,
            meta=meta,
            raw_response=raw,
        )
    except Exception as e:  # noqa: BLE001 - cache failure is not fatal
        logger.warning("Failed to write hook cache (%s); continuing.", e)

    return updated