|
|
@@ -0,0 +1,316 @@
|
|
|
+"""Statut Discord du bot : l'anime qu'il regarde, la musique qu'il écoute, le jeu auquel il joue…
|
|
|
+
|
|
|
+Chaque statut est tiré au sort selon PRESENCE_WEIGHTS, avec sa propre durée (quelques épisodes,
|
|
|
+quelques morceaux, une session de jeu…) ; à la fin, un nouveau statut est tiré. Pendant la nuit
|
|
|
+(PRESENCE_NIGHT), le bot est absent.
|
|
|
+"""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import asyncio
|
|
|
+import logging
|
|
|
+import random
|
|
|
+from collections import deque
|
|
|
+from dataclasses import dataclass
|
|
|
+from datetime import datetime, timedelta, timezone
|
|
|
+from enum import Enum
|
|
|
+from typing import TYPE_CHECKING
|
|
|
+
|
|
|
+import discord
|
|
|
+
|
|
|
+from .presence_sources import PresenceSources, Track
|
|
|
+
|
|
|
+if TYPE_CHECKING:
|
|
|
+ from discord.ext import commands
|
|
|
+
|
|
|
+ from .claude_client import ClaudeClient
|
|
|
+ from .config import Settings
|
|
|
+
|
|
|
+logger = logging.getLogger("chatbot.presence")
|
|
|
+
|
|
|
+UTC = timezone.utc # datetime.UTC n'existe qu'à partir de Python 3.11
|
|
|
+SOURCES_REFRESH_INTERVAL = 24 * 3600 # secondes
|
|
|
+CUSTOM_STATUS_MAX_LENGTH = 128 # limite de Discord
|
|
|
+RECENT_CUSTOM_STATUSES = 4 # statuts personnalisés récents, à ne pas répéter
|
|
|
+
|
|
|
+# Durée de chaque statut, tirée au sort entre ces bornes.
|
|
|
+ANIME_EPISODES = (1, 5)
|
|
|
+MUSIC_TRACKS = (6, 20)
|
|
|
+DEFAULT_ANIME_EPISODES = 12 # épisodes disponibles quand AniList ne le précise pas
|
|
|
+MINUTES = { # en minutes
|
|
|
+ "jeu": (45, 120),
|
|
|
+ "en_ligne": (15, 45),
|
|
|
+ "pensee": (30, 90),
|
|
|
+ "absent": (10, 60),
|
|
|
+ "ne_pas_deranger": (30, 90),
|
|
|
+}
|
|
|
+
|
|
|
+WEEKDAYS = ("lundi", "mardi", "mercredi", "jeudi", "vendredi", "samedi", "dimanche")
|
|
|
+
|
|
|
+
|
|
|
+class Kind(Enum):
|
|
|
+ # Les valeurs sont les noms utilisés dans PRESENCE_WEIGHTS.
|
|
|
+ ANIME = "anime"
|
|
|
+ MUSIC = "musique"
|
|
|
+ THOUGHT = "pensee"
|
|
|
+ GAME = "jeu"
|
|
|
+ ONLINE = "en_ligne"
|
|
|
+ DND = "ne_pas_deranger"
|
|
|
+ IDLE = "absent"
|
|
|
+ NIGHT = "nuit" # absent automatiquement pendant PRESENCE_NIGHT, jamais tiré au sort
|
|
|
+
|
|
|
+
|
|
|
+@dataclass(frozen=True)
|
|
|
+class Status:
|
|
|
+ kind: Kind
|
|
|
+ title: str | None = None # anime, jeu, pensée, ou raison d'une absence
|
|
|
+ track: Track | None = None # musique en cours
|
|
|
+ episode: int | None = None # épisode en cours (None pour un film)
|
|
|
+
|
|
|
+ def activity(self) -> discord.BaseActivity | None:
|
|
|
+ if self.kind is Kind.ANIME:
|
|
|
+ state = f"Épisode {self.episode}" if self.episode else None
|
|
|
+ return discord.Activity(type=discord.ActivityType.watching, name=self.title, state=state)
|
|
|
+ if self.kind is Kind.MUSIC:
|
|
|
+ return discord.Activity(type=discord.ActivityType.listening,
|
|
|
+ name=f"{self.track.title} de {self.track.artist}")
|
|
|
+ if self.kind is Kind.GAME:
|
|
|
+ return discord.Game(name=self.title)
|
|
|
+ if self.title: # pensée, ou raison d'une absence
|
|
|
+ return discord.CustomActivity(name=self.title)
|
|
|
+ return None
|
|
|
+
|
|
|
+ def discord_status(self) -> discord.Status:
|
|
|
+ if self.kind is Kind.DND:
|
|
|
+ return discord.Status.dnd
|
|
|
+ if self.kind in (Kind.IDLE, Kind.NIGHT):
|
|
|
+ return discord.Status.idle
|
|
|
+ return discord.Status.online
|
|
|
+
|
|
|
+ def describe(self) -> str | None:
|
|
|
+ """Ce que fait le bot, vu de lui-même (None : en ligne, sans activité)."""
|
|
|
+ reason = f" (statut affiché : « {self.title} »)" if self.title else ""
|
|
|
+ if self.kind is Kind.ANIME:
|
|
|
+ episode = f"l'épisode {self.episode} de " if self.episode else ""
|
|
|
+ return f"tu regardes {episode}l'anime « {self.title} »"
|
|
|
+ if self.kind is Kind.MUSIC:
|
|
|
+ return f"tu écoutes « {self.track.title} » de {self.track.artist}"
|
|
|
+ if self.kind is Kind.GAME:
|
|
|
+ return f"tu joues à « {self.title} » sur Steam"
|
|
|
+ if self.kind is Kind.THOUGHT:
|
|
|
+ return f"tu affiches le statut personnalisé « {self.title} »"
|
|
|
+ if self.kind is Kind.DND:
|
|
|
+ return f"tu es en « Ne pas déranger »{reason}"
|
|
|
+ if self.kind is Kind.IDLE:
|
|
|
+ return f"tu es absente{reason}"
|
|
|
+ if self.kind is Kind.NIGHT:
|
|
|
+ return "tu es absente pour la nuit"
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+# Un statut et sa durée en secondes. Un anime ou une musique en enchaîne plusieurs (un par épisode
|
|
|
+# ou par morceau) : c'est le « programme » du statut.
|
|
|
+Step = tuple[Status, float]
|
|
|
+
|
|
|
+
|
|
|
+def format_duration(seconds: float) -> str:
|
|
|
+ minutes = max(1, round(seconds / 60))
|
|
|
+ if minutes < 60:
|
|
|
+ return f"{minutes} min"
|
|
|
+ return f"{minutes // 60} h {minutes % 60:02d}" if minutes % 60 else f"{minutes // 60} h"
|
|
|
+
|
|
|
+
|
|
|
+class PresenceManager:
|
|
|
+ def __init__(self, bot: commands.Bot, claude: ClaudeClient, settings: Settings):
|
|
|
+ self.bot = bot
|
|
|
+ self.claude = claude
|
|
|
+ self.settings = settings
|
|
|
+ self.weights = {Kind(name): weight for name, weight in settings.presence_weights.items()}
|
|
|
+ self.sources = PresenceSources(
|
|
|
+ settings.history_file.with_name("presence_sources.json"),
|
|
|
+ settings.deezer_playlists, settings.steam_api_key, settings.steam_id,
|
|
|
+ )
|
|
|
+ self.status = Status(Kind.ONLINE)
|
|
|
+ self.program: list[Step] = [] # étapes restantes du statut en cours
|
|
|
+ self.step_end = datetime.min.replace(tzinfo=UTC) # fin de l'étape en cours (UTC)
|
|
|
+ self.status_end: datetime | None = None # fin du statut en cours, toutes étapes comprises
|
|
|
+ self.recent_custom: deque[str] = deque(maxlen=RECENT_CUSTOM_STATUSES)
|
|
|
+ self._tasks: list[asyncio.Task] = []
|
|
|
+
|
|
|
+ def start(self) -> None:
|
|
|
+ if not self.sources.steam_enabled:
|
|
|
+ logger.info("STEAM_API_KEY ou STEAM_ID absent : pas de statut « joue à ».")
|
|
|
+ self._tasks.append(asyncio.create_task(self._run()))
|
|
|
+
|
|
|
+ def stop(self) -> None:
|
|
|
+ for task in self._tasks:
|
|
|
+ task.cancel()
|
|
|
+
|
|
|
+ def prompt_note(self) -> str | None:
|
|
|
+ """Ligne ajoutée au début du message envoyé à Claude."""
|
|
|
+ description = self.status.describe()
|
|
|
+ if not description:
|
|
|
+ return None
|
|
|
+ if self.status_end and self.status.kind is not Kind.NIGHT:
|
|
|
+ now = datetime.now(UTC)
|
|
|
+ speed = self.settings.presence_speed
|
|
|
+ total = format_duration((self.status_end - now).total_seconds() * speed)
|
|
|
+ if not self.program:
|
|
|
+ if self.status.kind is Kind.ANIME and self.status.episode:
|
|
|
+ description += " (le dernier épisode que tu comptes regarder)"
|
|
|
+ elif self.status.kind is Kind.MUSIC:
|
|
|
+ description += " (le dernier morceau que tu comptes écouter)"
|
|
|
+ description += f", encore pendant environ {total}"
|
|
|
+ else:
|
|
|
+ # Anime ou musique : l'épisode (ou le morceau) en cours n'est qu'une partie du programme.
|
|
|
+ current = format_duration((self.step_end - now).total_seconds() * speed)
|
|
|
+ if self.status.kind is Kind.ANIME:
|
|
|
+ description += (f" (fin de l'épisode dans environ {current}) ; tu comptes regarder "
|
|
|
+ f"jusqu'à l'épisode {self.program[-1][0].episode}, encore environ {total} en tout")
|
|
|
+ else:
|
|
|
+ description += (f" (fin du morceau dans environ {current}) ; encore {len(self.program)} "
|
|
|
+ f"morceau(x) après celui-ci, environ {total} de musique en tout")
|
|
|
+ return f"[ton statut Discord : {description}]"
|
|
|
+
|
|
|
+ async def apply(self) -> None:
|
|
|
+ """Envoie le statut à Discord (aussi rappelé après une reconnexion, qui l'efface)."""
|
|
|
+ try:
|
|
|
+ await self.bot.change_presence(activity=self.status.activity(), status=self.status.discord_status())
|
|
|
+ except Exception as exc: # déconnecté : on_ready le renverra
|
|
|
+ logger.warning("Statut Discord non appliqué : %s", exc)
|
|
|
+
|
|
|
+ # ------------------------------------------------------------ boucles
|
|
|
+
|
|
|
+ async def _run(self) -> None:
|
|
|
+ self.sources.load_cache()
|
|
|
+ await self.sources.refresh()
|
|
|
+ self._tasks.append(asyncio.create_task(self._refresh_loop()))
|
|
|
+ await self.bot.wait_until_ready()
|
|
|
+ while True:
|
|
|
+ try:
|
|
|
+ wake = await self._step()
|
|
|
+ except Exception:
|
|
|
+ logger.exception("Erreur inattendue dans la gestion du statut")
|
|
|
+ wake = datetime.now(UTC) + timedelta(minutes=1)
|
|
|
+ await asyncio.sleep(max(1.0, (wake - datetime.now(UTC)).total_seconds()))
|
|
|
+
|
|
|
+ async def _refresh_loop(self) -> None:
|
|
|
+ while True:
|
|
|
+ await asyncio.sleep(SOURCES_REFRESH_INTERVAL)
|
|
|
+ try:
|
|
|
+ await self.sources.refresh()
|
|
|
+ except Exception:
|
|
|
+ logger.exception("Erreur inattendue pendant la mise à jour des listes de statuts")
|
|
|
+
|
|
|
+ async def _step(self) -> datetime:
|
|
|
+ """Passe à l'étape suivante quand la précédente est finie ; renvoie l'heure du prochain réveil."""
|
|
|
+ night = self.settings.presence_night
|
|
|
+ now = datetime.now(UTC)
|
|
|
+ local = now.astimezone(self.settings.timezone)
|
|
|
+
|
|
|
+ if self.is_night(local):
|
|
|
+ if self.status.kind is not Kind.NIGHT:
|
|
|
+ # Coupure nette : le statut en cours s'arrête, un nouveau sera tiré au réveil.
|
|
|
+ self.program = []
|
|
|
+ self.status_end = None
|
|
|
+ await self.set_status(Status(Kind.NIGHT))
|
|
|
+ self.step_end = self.next_hour(local, night[1])
|
|
|
+ return self.step_end
|
|
|
+
|
|
|
+ if now >= self.step_end:
|
|
|
+ if not self.program:
|
|
|
+ self.program = await self.draw()
|
|
|
+ total = sum(seconds for _, seconds in self.program)
|
|
|
+ self.status_end = now + timedelta(seconds=total)
|
|
|
+ logger.info("Nouveau statut pour %s.", format_duration(total * self.settings.presence_speed))
|
|
|
+ status, seconds = self.program.pop(0)
|
|
|
+ self.step_end = now + timedelta(seconds=seconds)
|
|
|
+ await self.set_status(status)
|
|
|
+
|
|
|
+ if night:
|
|
|
+ return min(self.step_end, self.next_hour(local, night[0]))
|
|
|
+ return self.step_end
|
|
|
+
|
|
|
+ # ------------------------------------------------------------ tirage
|
|
|
+
|
|
|
+ async def draw(self) -> list[Step]:
|
|
|
+ """Tire un nouveau statut et son programme (durées divisées par PRESENCE_SPEED)."""
|
|
|
+ available = {kind: weight for kind, weight in self.weights.items() if weight > 0 and self.has_source(kind)}
|
|
|
+ # Jamais deux fois le même statut à la suite, sauf s'il n'y a pas d'autre choix.
|
|
|
+ if len(available) > 1:
|
|
|
+ available.pop(self.status.kind, None)
|
|
|
+ kind = random.choices(list(available), weights=list(available.values()))[0] if available else Kind.ONLINE
|
|
|
+ speed = self.settings.presence_speed
|
|
|
+
|
|
|
+ if kind is Kind.ANIME:
|
|
|
+ anime = self.pick(self.sources.anime, lambda a: a.title == self.status.title)
|
|
|
+ if anime.episodes == 1: # film
|
|
|
+ return [(Status(kind, title=anime.title), anime.duration * 60 / speed)]
|
|
|
+ available_episodes = anime.episodes or DEFAULT_ANIME_EPISODES
|
|
|
+ count = min(random.randint(*ANIME_EPISODES), available_episodes)
|
|
|
+ first = random.randint(1, available_episodes - count + 1)
|
|
|
+ return [(Status(kind, title=anime.title, episode=n), anime.duration * 60 / speed)
|
|
|
+ for n in range(first, first + count)]
|
|
|
+
|
|
|
+ if kind is Kind.MUSIC:
|
|
|
+ count = min(random.randint(*MUSIC_TRACKS), len(self.sources.music))
|
|
|
+ tracks = random.sample(self.sources.music, count)
|
|
|
+ return [(Status(kind, track=t), t.duration / speed) for t in tracks]
|
|
|
+
|
|
|
+ seconds = random.randint(*MINUTES[kind.value]) * 60 if kind.value in MINUTES else 0
|
|
|
+ if kind is Kind.GAME:
|
|
|
+ title = self.pick(self.sources.games, lambda g: g == self.status.title)
|
|
|
+ return [(Status(kind, title=title), seconds / speed)]
|
|
|
+ if kind in (Kind.THOUGHT, Kind.IDLE, Kind.DND):
|
|
|
+ text = await self.write_custom_status(kind, seconds)
|
|
|
+ if kind is Kind.THOUGHT and not text:
|
|
|
+ kind = Kind.ONLINE # pensée impossible à écrire : en ligne, sans activité
|
|
|
+ return [(Status(kind, title=text), seconds / speed)]
|
|
|
+ return [(Status(kind), seconds / speed)]
|
|
|
+
|
|
|
+ async def write_custom_status(self, kind: Kind, seconds: int) -> str | None:
|
|
|
+ local = datetime.now(self.settings.timezone)
|
|
|
+ now = f"{WEEKDAYS[local.weekday()]} {local.strftime('%d/%m/%Y, %H:%M')}"
|
|
|
+ text = await self.claude.write_custom_status(
|
|
|
+ kind.value, now, format_duration(seconds), list(self.recent_custom), CUSTOM_STATUS_MAX_LENGTH,
|
|
|
+ )
|
|
|
+ if text:
|
|
|
+ self.recent_custom.append(text)
|
|
|
+ return text
|
|
|
+
|
|
|
+ def has_source(self, kind: Kind) -> bool:
|
|
|
+ if kind is Kind.ANIME:
|
|
|
+ return bool(self.sources.anime)
|
|
|
+ if kind is Kind.MUSIC:
|
|
|
+ return bool(self.sources.music)
|
|
|
+ if kind is Kind.GAME:
|
|
|
+ return bool(self.sources.games)
|
|
|
+ return True
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def pick(items: list, is_current):
|
|
|
+ """Élément au hasard, différent de l'actuel quand c'est possible."""
|
|
|
+ return random.choice([item for item in items if not is_current(item)] or items)
|
|
|
+
|
|
|
+ async def set_status(self, status: Status) -> None:
|
|
|
+ self.status = status
|
|
|
+ logger.info("Statut Discord : %s", status.describe() or "en ligne, sans activité")
|
|
|
+ await self.apply()
|
|
|
+
|
|
|
+ # ------------------------------------------------------------ nuit
|
|
|
+
|
|
|
+ def is_night(self, local: datetime) -> bool:
|
|
|
+ if not self.settings.presence_night:
|
|
|
+ return False
|
|
|
+ start, end = self.settings.presence_night
|
|
|
+ if start < end:
|
|
|
+ return start <= local.hour < end
|
|
|
+ return local.hour >= start or local.hour < end # nuit à cheval sur minuit (ex. 23-6)
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def next_hour(local: datetime, hour: int) -> datetime:
|
|
|
+ """Prochain passage à hour:00, en UTC."""
|
|
|
+ candidate = local.replace(hour=hour, minute=0, second=0, microsecond=0)
|
|
|
+ if candidate <= local:
|
|
|
+ candidate += timedelta(days=1)
|
|
|
+ return candidate.astimezone(UTC)
|
Johnounet