Spaces:
Sleeping
Sleeping
| """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"] | |