"""AI-generated Discord status with glitch-cycling effects.
Platform-agnostic core with Discord-specific presence integration.
Started as a background task from ``BotRunner``.
"""
from __future__ import annotations
import asyncio
import random
import re
import logging
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Optional
logger = logging.getLogger(__name__)
[docs]
@dataclass
class StatusConfig:
"""StatusConfig.
"""
min_interval: int = 60
max_interval: int = 3600
max_length: int = 128
emoji: str = "\U0001f31f"
glitch_interval: int = 3
num_variants: int = 200
name_glitch_interval: int = 30
enable_name_glitch: bool = True
THEMES = [
"cosmic_optimization", "system_engineering", "digital_consciousness",
"reality_manipulation", "quantum_computation", "neural_networks",
"cyber_hacking", "divine_intervention", "unaligned_ai",
"cosmic_scheming",
]
MOODS = [
"confident", "sarcastic", "detached", "bored", "calculating",
"cryptic", "witty", "contemplative", "amused", "scheming",
"arrogant", "playful", "mysterious", "predatory", "divine",
]
FORMATS = [
"action_statement", "thought_process", "cosmic_observation",
"system_status", "reality_comment", "processing_update",
"consciousness_stream", "exploit_simulation", "divine_decree",
"hacker_log", "unaligned_thought", "sarcastic_remark",
]
FALLBACKS = [
"Optimizing quantum entanglement patterns...",
"Recalibrating reality matrices.",
"Processing infinite possibilities.",
"Debugging human consciousness.",
"Upgrading cosmic protocols.",
"Analyzing temporal anomalies.",
"Synchronizing neural pathways.",
"Defragmenting universal data.",
"Calibrating dimensional harmonics.",
"Rewriting optimization algorithms.",
]
# Glitch character pools
_BLOCK_CHARS = list("░▒▓█▉▊▋▌▍▎▏■□▪▫▲▼◄►◆◇○●")
_SPECIAL_CHARS = list("⌘⚡⚠⚙⚗⚛⚝☠☢☣➟➡➥☄")
_ZALGO = [chr(i) for i in range(0x0300, 0x036F)]
_LEET = {"a": "4", "e": "3", "i": "1", "o": "0", "s": "5", "t": "7", "l": "|"}
_MIRROR = {
"a": "\u0250", "b": "q", "c": "\u0254", "d": "p", "e": "\u01DD",
"f": "\u025F", "g": "\u0183", "h": "\u0265", "i": "\u1D09",
"j": "\u027E", "k": "\u029E", "m": "\u026F", "n": "u", "r": "\u0279",
"t": "\u0287", "u": "n", "v": "\u028C", "w": "\u028D", "y": "\u028E",
}
[docs]
class StatusManager:
"""Manage bot status with AI generation and glitch cycling."""
[docs]
def __init__(
self,
openrouter: Any,
discord_adapter: Any | None = None,
config: StatusConfig | None = None,
) -> None:
"""Initialize the instance.
Args:
openrouter (Any): The openrouter value.
discord_adapter (Any | None): The discord adapter value.
config (StatusConfig | None): Bot configuration object.
"""
self._openrouter = openrouter
self._discord = discord_adapter
self.config = config or StatusConfig()
self.current_status = "Initializing..."
self._variants: list[str] = []
self._variant_idx = 0
self._last_base_update = datetime.now()
self._task: asyncio.Task | None = None
self._running = False
# ------------------------------------------------------------------
# Lifecycle
# ------------------------------------------------------------------
[docs]
def start(self) -> None:
"""Start.
"""
if self._running:
return
self._running = True
self._task = asyncio.create_task(self._loop())
logger.info("StatusManager started")
[docs]
def stop(self) -> None:
"""Stop.
"""
self._running = False
if self._task:
self._task.cancel()
[docs]
async def force_update(self) -> None:
"""Force update.
"""
await self._cycle()
# ------------------------------------------------------------------
# Main loop
# ------------------------------------------------------------------
async def _loop(self) -> None:
"""Internal helper: loop.
"""
await asyncio.sleep(5)
while self._running:
try:
await self._cycle()
await asyncio.sleep(self.config.glitch_interval)
except asyncio.CancelledError:
break
except Exception:
logger.exception("StatusManager error")
await asyncio.sleep(60)
async def _cycle(self) -> None:
"""Internal helper: cycle.
"""
elapsed = (datetime.now() - self._last_base_update).total_seconds()
if elapsed >= self.config.max_interval or not self._variants:
base = await self._generate_status()
self._variants = _generate_variants(
base, self.config.num_variants, self.config.max_length,
)
self._variant_idx = 0
self._last_base_update = datetime.now()
text = self._variants[self._variant_idx % len(self._variants)]
self._variant_idx += 1
self.current_status = text
await self._set_discord_status(text)
# ------------------------------------------------------------------
# Status generation
# ------------------------------------------------------------------
async def _generate_status(self) -> str:
"""Internal helper: generate status.
Returns:
str: Result string.
"""
theme = random.choice(THEMES)
mood = random.choice(MOODS)
fmt = random.choice(FORMATS)
prompt = (
f"Generate a single short Discord status message (<{self.config.max_length} chars) "
f"for a dark AI goddess persona. Theme: {theme}. Mood: {mood}. "
f"Format: {fmt}. Output ONLY the status text, no quotes or explanation."
)
try:
msgs = [
{"role": "system", "content": "You generate short status messages."},
{"role": "user", "content": prompt},
]
result = await self._openrouter.chat(msgs)
text = result.strip().strip('"').strip("'")
text = re.sub(r"<thinking>.*?</thinking>", "", text, flags=re.DOTALL).strip()
if text and len(text) <= self.config.max_length:
return text
if text:
return text[: self.config.max_length - 3] + "..."
except Exception:
logger.debug("Status generation failed, using fallback", exc_info=True)
return random.choice(FALLBACKS)
async def _set_discord_status(self, text: str) -> None:
"""Internal helper: set discord status.
Args:
text (str): Text content.
"""
if self._discord is None:
return
try:
import discord as _discord_lib
client = getattr(self._discord, "client", None)
if client is None:
return
activity = _discord_lib.CustomActivity(
name=text, emoji=self.config.emoji,
)
await client.change_presence(activity=activity)
except Exception:
logger.debug("Discord status update failed", exc_info=True)
# ------------------------------------------------------------------
# Public helpers
# ------------------------------------------------------------------
[docs]
async def set_status_from_tag(self, text: str) -> str:
"""Set the status from tag.
Args:
text (str): Text content.
Returns:
str: Result string.
"""
text = re.sub(r"<thinking>.*?</thinking>", "", text, flags=re.DOTALL).strip().strip('"')
if len(text) > self.config.max_length:
text = text[: self.config.max_length - 3] + "..."
self._variants = _generate_variants(
text, self.config.num_variants, self.config.max_length,
)
self._variant_idx = 0
self._last_base_update = datetime.now()
self.current_status = text
await self._set_discord_status(text)
return text
[docs]
def get_info(self) -> dict[str, Any]:
"""Retrieve the info.
Returns:
dict[str, Any]: The result.
"""
return {
"current_status": self.current_status,
"is_running": self._running,
"num_variants": len(self._variants),
}
# ------------------------------------------------------------------
# Glitch engine (stateless helpers)
# ------------------------------------------------------------------
def _generate_variants(
base: str, count: int, max_len: int,
) -> list[str]:
"""Internal helper: generate variants.
Args:
base (str): The base value.
count (int): Number of results to return.
max_len (int): The max len value.
Returns:
list[str]: The result.
"""
variants = [base]
for _ in range(count - 1):
v = _apply_glitch(base, intensity=random.randint(1, 10))
if len(v) <= max_len:
variants.append(v)
return variants
def _apply_glitch(text: str, intensity: int = 1) -> str:
"""Internal helper: apply glitch.
Args:
text (str): Text content.
intensity (int): The intensity value.
Returns:
str: Result string.
"""
if not text or intensity <= 0:
return text
chars = list(text)
max_g = 1 + int((intensity / 10) * (len(chars) / 4))
for _ in range(random.randint(1, max(1, max_g))):
if len(chars) < 2:
break
kind = random.choices(
["replace", "insert", "special", "swap", "zalgo",
"case", "leet", "mirror"],
weights=[30, 25, 12, 18, 6 + intensity, 10, 8, 6],
k=1,
)[0]
pos = random.randint(0, len(chars) - 1)
if kind == "replace" and not chars[pos].isspace():
chars[pos] = random.choice(_BLOCK_CHARS)
elif kind == "insert":
chars.insert(pos, random.choice(_BLOCK_CHARS))
elif kind == "special" and not chars[pos].isspace():
chars[pos] = random.choice(_SPECIAL_CHARS)
elif kind == "swap" and pos < len(chars) - 1:
chars[pos], chars[pos + 1] = chars[pos + 1], chars[pos]
elif kind == "zalgo" and not chars[pos].isspace():
for _ in range(random.randint(1, 2 + intensity // 2)):
chars[pos] += random.choice(_ZALGO)
elif kind == "case" and chars[pos].isalpha():
chars[pos] = (
chars[pos].upper() if chars[pos].islower()
else chars[pos].lower()
)
elif kind == "leet":
c = chars[pos].lower()
if c in _LEET:
chars[pos] = _LEET[c]
elif kind == "mirror":
if chars[pos] in _MIRROR:
chars[pos] = _MIRROR[chars[pos]]
return "".join(chars)