Spaces:
Sleeping
Sleeping
File size: 3,891 Bytes
8e12fe5 f52ff93 8e12fe5 92a8b39 8e12fe5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | """Highlight runtime — auto-clip on hype spikes.
Detects moments worth clipping from live signals (strong positive sentiment +
a surge in chat velocity) and enqueues a clip candidate through the existing
risk queue. Throttled and gated on the clip pipeline, so it degrades to a
no-op when clips are disabled or the channel is calm.
"""
from __future__ import annotations
import threading
import time
from bot.control_plane import control_plane
from bot.control_plane_constants import RISK_CLIP_CANDIDATE
HIGHLIGHT_MIN_INTERVAL_SECONDS = 90.0
MIN_SENTIMENT_SAMPLES = 5
HYPE_SENTIMENT_THRESHOLD = 0.4
VELOCITY_SURGE_RATIO = 1.5
def detect_hype_spike(
sentiment_avg: float,
sentiment_count: int,
messages_per_minute_10m: float,
messages_per_minute_60m: float,
) -> bool:
"""Pure predicate: is the room spiking right now?"""
if sentiment_count < MIN_SENTIMENT_SAMPLES:
return False
if sentiment_avg < HYPE_SENTIMENT_THRESHOLD:
return False
if messages_per_minute_60m <= 0:
return False
return (messages_per_minute_10m / messages_per_minute_60m) >= VELOCITY_SURGE_RATIO
class HighlightRuntime:
def __init__(self) -> None:
self._lock = threading.Lock()
self._last_highlight_at = 0.0
def maybe_highlight(self, snapshot: dict | None) -> dict:
"""Maybe enqueue a clip candidate from a fresh observability snapshot."""
safe = dict(snapshot or {})
sentiment = dict(safe.get("sentiment") or {})
chat = dict(safe.get("chat_analytics") or {})
spike = detect_hype_spike(
float(sentiment.get("avg", 0.0) or 0.0),
int(sentiment.get("count", 0) or 0),
float(chat.get("messages_per_minute_10m", 0.0) or 0.0),
float(chat.get("messages_per_minute_60m", 0.0) or 0.0),
)
if not spike:
return {"ok": False, "reason": "no_spike"}
# Consistent with vision clips: don't enqueue when the pipeline is off.
if not control_plane.get_config().get("clip_pipeline_enabled"):
return {"ok": False, "reason": "clip_pipeline_disabled"}
now = time.time()
with self._lock:
if now - self._last_highlight_at < HIGHLIGHT_MIN_INTERVAL_SECONDS:
return {"ok": False, "reason": "throttled"}
self._last_highlight_at = now
vibe = str(sentiment.get("vibe") or "hype")
item = control_plane.enqueue_action(
kind="clip_candidate",
risk=RISK_CLIP_CANDIDATE,
title="Highlight: pico de hype",
body=f"Chat em pico ({vibe}). Momento candidato a clipe.",
payload={"source": "sentiment_spike", "vibe": vibe},
created_by="highlight",
)
return {"ok": True, "item": item}
def get_status(self, now: float | None = None) -> dict:
"""Read-only highlight state for the dashboard. Honest: real cooldown +
the gate that actually decides, no fabricated activity."""
ts = time.time() if now is None else float(now)
with self._lock:
last = self._last_highlight_at
cooldown = max(0.0, HIGHLIGHT_MIN_INTERVAL_SECONDS - (ts - last)) if last > 0 else 0.0
return {
"last_highlight_at": last,
"cooldown_seconds": round(cooldown, 1),
"throttled": cooldown > 0,
"clip_pipeline_enabled": bool(control_plane.get_config().get("clip_pipeline_enabled")),
"thresholds": {
"min_interval_s": HIGHLIGHT_MIN_INTERVAL_SECONDS,
"hype_sentiment": HYPE_SENTIMENT_THRESHOLD,
"velocity_ratio": VELOCITY_SURGE_RATIO,
"min_samples": MIN_SENTIMENT_SAMPLES,
},
}
highlight_runtime = HighlightRuntime()
__all__ = ["HighlightRuntime", "detect_hype_spike", "highlight_runtime"]
|