Spaces:
Sleeping
Sleeping
| import asyncio | |
| import threading | |
| import time | |
| from datetime import UTC, datetime | |
| from typing import Any, Optional | |
| from bot.logic_constants import ( | |
| BOT_BRAND, | |
| DEFAULT_STYLE_PROFILE, | |
| MAX_RECENT_CHAT_ENTRIES, | |
| MAX_RECENT_CHAT_PREVIEW_CHARS, | |
| MAX_RECENT_CHAT_PROMPT_ENTRIES, | |
| MAX_REPLY_LENGTH, | |
| MAX_REPLY_LINES, | |
| OBSERVABILITY_TYPES, | |
| SYSTEM_INSTRUCTION_TEMPLATE, | |
| ) | |
| def normalize_memory_excerpt(text: str, max_length: int = MAX_RECENT_CHAT_PREVIEW_CHARS) -> str: | |
| compact = " ".join((text or "").split()) | |
| if not compact: | |
| return "" | |
| if len(compact) <= max_length: | |
| return compact | |
| head = compact[: max_length - 3].rstrip() | |
| last_space = head.rfind(" ") | |
| if last_space >= int(max_length * 0.5): | |
| head = head[:last_space] | |
| return head.rstrip(" ,;:") + "..." | |
| class StreamContext: | |
| def __init__(self): | |
| self.channel_id = "default" | |
| self.current_game = "N/A" | |
| self.stream_vibe = "Conversa" | |
| self.last_event = "Bot Online" | |
| self.style_profile = DEFAULT_STYLE_PROFILE | |
| self.inference_temperature: float | None = None | |
| self.inference_top_p: float | None = None | |
| self.channel_paused = False | |
| self.channel_config_loaded = False | |
| self.agent_notes = "" | |
| self.persona_name = "" | |
| self.persona_tone = "" | |
| self.persona_emote_vocab: list[str] = [] | |
| self.persona_lore = "" | |
| self.persona_sentence_style = "" | |
| self.persona_banned_topics: list[str] = [] | |
| self.persona_cta_triggers: list[str] = [] | |
| self.channel_model_routing: dict[str, str | None] = {} | |
| self.live_observability: dict[str, str] = { | |
| "game": "", | |
| "movie": "", | |
| "series": "", | |
| "youtube": "", | |
| "x": "", | |
| "topic": "", | |
| } | |
| self.recent_chat_entries: list[str] = [] | |
| self.last_byte_reply = "" | |
| self.start_time = time.time() | |
| self.last_activity = time.time() | |
| def get_uptime_minutes(self) -> int: | |
| return int((time.time() - self.start_time) / 60) | |
| def _touch(self) -> None: | |
| self.last_activity = time.time() | |
| from bot.logic_context import context_manager | |
| from bot.persistence_layer import persistence | |
| if not persistence.is_enabled: | |
| return | |
| state_snapshot = { | |
| "current_game": self.current_game, | |
| "stream_vibe": self.stream_vibe, | |
| "last_event": self.last_event, | |
| "style_profile": self.style_profile, | |
| "live_observability": self.live_observability.copy(), | |
| "last_byte_reply": self.last_byte_reply, | |
| } | |
| # CURA DEFINITIVA: Sempre usa o loop principal para agendar tarefas em background. | |
| # Isso evita que tarefas sejam atreladas a loops temporários de requisições do Dashboard. | |
| main_loop = context_manager._main_loop | |
| if main_loop and main_loop.is_running(): | |
| asyncio.run_coroutine_threadsafe( | |
| persistence.save_channel_state(self.channel_id, state_snapshot), | |
| main_loop, | |
| ) | |
| else: | |
| try: | |
| # Tenta o loop da thread atual se o principal não existir (geralmente durante startup) | |
| loop = asyncio.get_running_loop() | |
| loop.create_task(persistence.save_channel_state(self.channel_id, state_snapshot)) | |
| except RuntimeError: | |
| pass | |
| def update_content(self, content_type: str, description: str) -> bool: | |
| normalized_type = content_type.strip().lower() | |
| cleaned_description = description.strip() | |
| if normalized_type not in OBSERVABILITY_TYPES or not cleaned_description: | |
| return False | |
| self.live_observability[normalized_type] = cleaned_description | |
| if normalized_type == "game": | |
| self.current_game = cleaned_description | |
| self.last_event = f"Contexto atualizado: {OBSERVABILITY_TYPES[normalized_type]}" | |
| self._touch() | |
| return True | |
| def clear_content(self, content_type: str) -> bool: | |
| normalized_type = content_type.strip().lower() | |
| if normalized_type not in OBSERVABILITY_TYPES: | |
| return False | |
| self.live_observability[normalized_type] = "" | |
| if normalized_type == "game": | |
| self.current_game = "N/A" | |
| self.last_event = f"Contexto removido: {OBSERVABILITY_TYPES[normalized_type]}" | |
| self._touch() | |
| return True | |
| def list_supported_content_types(self) -> str: | |
| return ", ".join(OBSERVABILITY_TYPES.keys()) | |
| def format_observability(self) -> str: | |
| entries = [] | |
| for content_type, label in OBSERVABILITY_TYPES.items(): | |
| value = self.live_observability.get(content_type, "").strip() | |
| if value: | |
| entries.append(f"{label}: {value}") | |
| return " | ".join(entries) if entries else "Sem conteudo registrado." | |
| def status_snapshot(self) -> str: | |
| active_count = sum(1 for value in self.live_observability.values() if value) | |
| return f"Vibe: {self.stream_vibe} | Contextos ativos: {active_count}" | |
| def remember_user_message(self, author_name: str, message_text: str) -> None: | |
| safe_author = normalize_memory_excerpt(author_name or "viewer", max_length=32) | |
| safe_message = normalize_memory_excerpt(message_text) | |
| if not safe_message: | |
| return | |
| self.recent_chat_entries.append(f"{safe_author}: {safe_message}") | |
| self.recent_chat_entries = self.recent_chat_entries[-MAX_RECENT_CHAT_ENTRIES:] | |
| self._touch() | |
| def remember_bot_reply(self, reply_text: str) -> None: | |
| safe_reply = normalize_memory_excerpt(reply_text, max_length=180) | |
| if not safe_reply: | |
| return | |
| self.last_byte_reply = safe_reply | |
| self.recent_chat_entries.append(f"{BOT_BRAND}: {safe_reply}") | |
| self.recent_chat_entries = self.recent_chat_entries[-MAX_RECENT_CHAT_ENTRIES:] | |
| self._touch() | |
| def format_recent_chat(self, limit: int = MAX_RECENT_CHAT_PROMPT_ENTRIES) -> str: | |
| if not self.recent_chat_entries: | |
| return "Sem historico recente." | |
| selected = self.recent_chat_entries[-max(1, limit) :] | |
| return " || ".join(selected) | |
| class ContextManager: | |
| """Gerenciador de contextos isolados por canal com thread-safety e persistência.""" | |
| def __init__(self) -> None: | |
| self._contexts: dict[str, StreamContext] = {} | |
| self._lock = threading.Lock() | |
| self._main_loop: asyncio.AbstractEventLoop | None = None | |
| def set_main_loop(self, loop: asyncio.AbstractEventLoop) -> None: | |
| """Injeta o loop principal para suportar chamadas via Dashboard (síncrono).""" | |
| self._main_loop = loop | |
| def get(self, channel_id: str | None = None) -> StreamContext: | |
| """ | |
| Recupera o contexto síncronamente. | |
| Se não existir na RAM, cria um novo e dispara carregamento do banco em background. | |
| """ | |
| key = (channel_id or "default").strip().lower() | |
| with self._lock: | |
| if key in self._contexts: | |
| return self._contexts[key] | |
| ctx = StreamContext() | |
| ctx.channel_id = key | |
| self._contexts[key] = ctx | |
| # Dispara carregamento assíncrono do Supabase (Lazy Load) | |
| self._trigger_lazy_load(key, ctx) | |
| return ctx | |
| def _trigger_lazy_load(self, channel_id: str, ctx: StreamContext) -> None: | |
| """Inicia a restauração do banco sem bloquear a thread principal.""" | |
| from bot.persistence_layer import persistence | |
| if not persistence.is_enabled: | |
| return | |
| async def _load_task(): | |
| state = await persistence.load_channel_state(channel_id) | |
| if state: | |
| ctx.current_game = state.get("current_game", ctx.current_game) | |
| ctx.stream_vibe = state.get("stream_vibe", ctx.stream_vibe) | |
| ctx.style_profile = state.get("style_profile", ctx.style_profile) | |
| ctx.live_observability = state.get("observability", ctx.live_observability) | |
| ctx.last_byte_reply = state.get("last_reply", ctx.last_byte_reply) | |
| history = await persistence.load_recent_history(channel_id) | |
| if history: | |
| ctx.recent_chat_entries = history | |
| channel_config = await persistence.load_channel_config(channel_id) | |
| ctx.inference_temperature = channel_config.get("temperature") | |
| ctx.inference_top_p = channel_config.get("top_p") | |
| ctx.channel_paused = bool(channel_config.get("agent_paused", False)) | |
| agent_notes = await persistence.load_agent_notes(channel_id) | |
| ctx.agent_notes = str(agent_notes.get("notes") or "") | |
| channel_identity = await persistence.load_channel_identity(channel_id) | |
| ctx.persona_name = str(channel_identity.get("persona_name") or "") | |
| ctx.persona_tone = str(channel_identity.get("tone") or "") | |
| ctx.persona_emote_vocab = list(channel_identity.get("emote_vocab") or []) | |
| ctx.persona_lore = str(channel_identity.get("lore") or "") | |
| persona_profile = await persistence.load_persona_profile(channel_id) | |
| if persona_profile.get("has_profile"): | |
| base = persona_profile.get("base_identity") or {} | |
| tonality = persona_profile.get("tonality_engine") or {} | |
| constraints = persona_profile.get("behavioral_constraints") or {} | |
| routing = persona_profile.get("model_routing") or {} | |
| ctx.persona_name = str(base.get("name") or "") or ctx.persona_name | |
| ctx.persona_lore = str(base.get("lore") or "") or ctx.persona_lore | |
| ctx.persona_tone = str(tonality.get("tone") or "") or ctx.persona_tone | |
| vocab = list(tonality.get("emote_vocab") or []) | |
| ctx.persona_emote_vocab = vocab or ctx.persona_emote_vocab | |
| ctx.persona_sentence_style = str(tonality.get("sentence_style") or "") | |
| ctx.persona_banned_topics = list(constraints.get("banned_topics") or []) | |
| ctx.persona_cta_triggers = list(constraints.get("cta_triggers") or []) | |
| ctx.channel_model_routing = dict(routing) | |
| ctx.channel_config_loaded = True | |
| try: | |
| loop = asyncio.get_running_loop() | |
| loop.create_task(_load_task()) | |
| except RuntimeError: | |
| if self._main_loop and self._main_loop.is_running(): | |
| asyncio.run_coroutine_threadsafe(_load_task(), self._main_loop) | |
| def get_sync(self, channel_id: str | None = None) -> StreamContext: | |
| """Alias para get() - mantém compatibilidade.""" | |
| return self.get(channel_id) | |
| def apply_channel_config( | |
| self, | |
| channel_id: str, | |
| *, | |
| temperature: float | None, | |
| top_p: float | None, | |
| agent_paused: bool = False, | |
| ) -> None: | |
| key = (channel_id or "default").strip().lower() | |
| with self._lock: | |
| ctx = self._contexts.get(key) | |
| if ctx is None: | |
| return | |
| ctx.inference_temperature = temperature | |
| ctx.inference_top_p = top_p | |
| ctx.channel_paused = bool(agent_paused) | |
| ctx.channel_config_loaded = True | |
| def ensure_channel_config_loaded(self, channel_id: str | None = None) -> StreamContext: | |
| key = (channel_id or "default").strip().lower() or "default" | |
| ctx = self.get(key) | |
| if bool(getattr(ctx, "channel_config_loaded", False)): | |
| return ctx | |
| from bot.persistence_layer import persistence | |
| if not persistence.is_enabled: | |
| ctx.channel_config_loaded = True | |
| return ctx | |
| channel_config = persistence.load_channel_config_sync(key) | |
| self.apply_channel_config( | |
| key, | |
| temperature=channel_config.get("temperature"), | |
| top_p=channel_config.get("top_p"), | |
| agent_paused=bool(channel_config.get("agent_paused", False)), | |
| ) | |
| channel_identity = persistence.load_channel_identity_sync(key) | |
| self.apply_channel_identity( | |
| key, | |
| persona_name=str(channel_identity.get("persona_name") or ""), | |
| tone=str(channel_identity.get("tone") or ""), | |
| emote_vocab=list(channel_identity.get("emote_vocab") or []), | |
| lore=str(channel_identity.get("lore") or ""), | |
| ) | |
| persona_profile = persistence.load_persona_profile_sync(key) | |
| self.apply_persona_profile( | |
| key, | |
| base_identity=persona_profile.get("base_identity"), | |
| tonality_engine=persona_profile.get("tonality_engine"), | |
| behavioral_constraints=persona_profile.get("behavioral_constraints"), | |
| model_routing=persona_profile.get("model_routing"), | |
| ) | |
| return ctx | |
| def apply_agent_notes(self, channel_id: str, *, notes: str) -> None: | |
| key = (channel_id or "default").strip().lower() | |
| with self._lock: | |
| ctx = self._contexts.get(key) | |
| if ctx is None: | |
| return | |
| ctx.agent_notes = str(notes or "") | |
| def apply_channel_identity( | |
| self, | |
| channel_id: str, | |
| *, | |
| persona_name: str, | |
| tone: str, | |
| emote_vocab: list[str] | tuple[str, ...], | |
| lore: str, | |
| ) -> None: | |
| key = (channel_id or "default").strip().lower() | |
| with self._lock: | |
| ctx = self._contexts.get(key) | |
| if ctx is None: | |
| return | |
| ctx.persona_name = str(persona_name or "") | |
| ctx.persona_tone = str(tone or "") | |
| ctx.persona_emote_vocab = [ | |
| str(item).strip() for item in list(emote_vocab or []) if str(item).strip() | |
| ] | |
| ctx.persona_lore = str(lore or "") | |
| def apply_persona_profile( | |
| self, | |
| channel_id: str, | |
| *, | |
| base_identity: dict | None = None, | |
| tonality_engine: dict | None = None, | |
| behavioral_constraints: dict | None = None, | |
| model_routing: dict | None = None, | |
| ) -> None: | |
| key = (channel_id or "default").strip().lower() | |
| with self._lock: | |
| ctx = self._contexts.get(key) | |
| if ctx is None: | |
| return | |
| base = base_identity if isinstance(base_identity, dict) else {} | |
| tonality = tonality_engine if isinstance(tonality_engine, dict) else {} | |
| constraints = behavioral_constraints if isinstance(behavioral_constraints, dict) else {} | |
| routing = model_routing if isinstance(model_routing, dict) else {} | |
| if base.get("name"): | |
| ctx.persona_name = str(base["name"]) | |
| if base.get("lore"): | |
| ctx.persona_lore = str(base["lore"]) | |
| if tonality.get("tone"): | |
| ctx.persona_tone = str(tonality["tone"]) | |
| vocab = list(tonality.get("emote_vocab") or []) | |
| if vocab: | |
| ctx.persona_emote_vocab = [str(v).strip() for v in vocab if str(v).strip()] | |
| ctx.persona_sentence_style = str(tonality.get("sentence_style") or "") | |
| ctx.persona_banned_topics = [ | |
| str(t).strip() for t in list(constraints.get("banned_topics") or []) if str(t).strip() | |
| ] | |
| ctx.persona_cta_triggers = [ | |
| str(t).strip() for t in list(constraints.get("cta_triggers") or []) if str(t).strip() | |
| ] | |
| ctx.channel_model_routing = {k: (str(v) if v else None) for k, v in dict(routing).items()} | |
| async def cleanup(self, channel_id: str) -> None: | |
| """Remove contexto da RAM (Async para manter assinatura onde esperado).""" | |
| key = channel_id.strip().lower() | |
| with self._lock: | |
| self._contexts.pop(key, None) | |
| async def purge_expired(self, max_age_seconds: float = 7200) -> int: | |
| """Remove contextos da RAM que não tiveram atividade recente.""" | |
| now = time.time() | |
| with self._lock: | |
| expired_keys = [ | |
| key | |
| for key, ctx in self._contexts.items() | |
| if key != "default" and (now - ctx.last_activity) > max_age_seconds | |
| ] | |
| for key in expired_keys: | |
| self._contexts.pop(key, None) | |
| return len(expired_keys) | |
| async def start_cleanup_loop(self, interval_seconds: int = 1800) -> None: | |
| """Loop de background para limpeza periódica de memória.""" | |
| from bot.sentiment_engine import sentiment_engine | |
| while True: | |
| try: | |
| await asyncio.sleep(interval_seconds) | |
| purged_ctx = await self.purge_expired() | |
| purged_sent = sentiment_engine.cleanup_inactive() | |
| if purged_ctx > 0 or purged_sent > 0: | |
| from bot.runtime_config import logger | |
| logger.info("Cleanup: %d ctx, %d sent", purged_ctx, purged_sent) | |
| except asyncio.CancelledError: | |
| break | |
| except Exception: | |
| pass | |
| def list_active_channels(self) -> list[str]: | |
| with self._lock: | |
| return list(self._contexts.keys()) | |
| context_manager = ContextManager() | |
| def build_system_instruction(ctx: StreamContext) -> str: | |
| return f"{SYSTEM_INSTRUCTION_TEMPLATE} Estilo ativo: {ctx.style_profile}" | |
| def enforce_reply_limits( | |
| text: str, max_lines: int = MAX_REPLY_LINES, max_length: int = MAX_REPLY_LENGTH | |
| ) -> str: | |
| if not text: | |
| return "" | |
| lines = [line.strip() for line in text.splitlines() if line.strip()] | |
| limited_lines = lines[:max_lines] | |
| joined = " ".join(limited_lines) | |
| if len(joined) <= max_length: | |
| return joined | |
| return joined[: max_length - 3].rstrip() + "..." | |
| def get_server_clock_snapshot() -> tuple[str, int]: | |
| now = datetime.now(UTC) | |
| return now.strftime("%Y-%m-%dT%H:%M:%SZ"), int(now.timestamp()) | |
| def build_dynamic_prompt( | |
| user_request: str, author_name: str, ctx: StreamContext | None = None, **kwargs: Any | |
| ) -> str: | |
| """CRÍTICO: Restaura labels que a suite de testes espera.""" | |
| if ctx is None: | |
| ctx = context_manager.get("default") | |
| ts, epoch = get_server_clock_snapshot() | |
| context_line = ( | |
| f"Contexto Atual da Live: [Vibe: {ctx.stream_vibe} | " | |
| f"Uptime: {ctx.get_uptime_minutes()}min | " | |
| f"Observabilidade: {ctx.format_observability()} | " | |
| f"Ultimo evento: {ctx.last_event} | " | |
| f"Relogio servidor UTC: {ts} | Epoch: {epoch}]" | |
| ) | |
| history_line = f"Historico recente: {ctx.format_recent_chat()}" | |
| last_reply_line = f"Ultima resposta do Byte: {ctx.last_byte_reply or 'N/A'}" | |
| return ( | |
| f"{context_line}\n" | |
| "Use o relogio do servidor como referencia para termos temporais (hoje/agora/nesta semana).\n" | |
| f"{history_line}\n" | |
| f"{last_reply_line}\n" | |
| f"Usuario {author_name}: {user_request}" | |
| ) | |