"""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. Absent, le bot ne répond pas avant son retour ; en « Ne pas déranger », il répond au bout de quelques minutes (voir wait_turn). S'il était en pleine conversation dans son salon, il prévient d'abord qu'il s'absente (voir departure_note et ChatBot.on_status_change). """ 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 Game, PresenceSources, SourceError, 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 GAME_ATTEMPTS = 5 # jeux essayés avant d'abandonner (fiche introuvable, jeu pour adultes…) DND_REPLY_MINUTES = (5, 15) # délai de réponse en « Ne pas déranger » # 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) details: str | None = None # présentation du jeu, pour Claude uniquement 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 @property def available(self) -> bool: return self.kind not in (Kind.DND, Kind.IDLE, Kind.NIGHT) def unavailability(self) -> str: """Pourquoi le bot ne répondait pas, vu de lui-même (statut absent ou Ne pas déranger).""" reason = f" (statut affiché : « {self.title} »)" if self.title else "" if self.kind is Kind.NIGHT: return "absente pour la nuit" if self.kind is Kind.DND: return f"en « Ne pas déranger »{reason}" return f"absente{reason}" 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: details = f" (présentation du jeu sur Steam : {self.details})" if self.details else "" return f"tu joues à « {self.title} » sur Steam{details}" 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, ) 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] = [] self._changed = asyncio.Event() # déclenché à chaque changement de statut (voir wait_turn) def start(self) -> None: if not self.sources.steam_enabled: logger.info("STEAM_API_KEY absent : pas de statut « joue à ».") if self.is_night(datetime.now(self.settings.timezone)): self.status = Status(Kind.NIGHT) # démarrage la nuit : absent dès le premier message 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}]" @property def available(self) -> bool: return self.status.available async def wait_turn(self) -> None: """Attend que le bot puisse répondre : dès qu'il redevient disponible, ou, en « Ne pas déranger », au bout d'un délai tiré au sort (DND_REPLY_MINUTES).""" loop = asyncio.get_running_loop() deadline: float | None = None while not self.available: changed = self._changed timeout = None if self.status.kind is Kind.DND: if deadline is None: minutes = random.uniform(*DND_REPLY_MINUTES) deadline = loop.time() + minutes * 60 / self.settings.presence_speed timeout = deadline - loop.time() if timeout <= 0: return try: await asyncio.wait_for(changed.wait(), timeout) except asyncio.TimeoutError: return def late_note(self, status: Status, several: bool) -> str: """Indique à Claude qu'il répond en retard, et pourquoi.""" now = datetime.now(self.settings.timezone).strftime("%H:%M") subject = "Ces messages sont arrivés" if several else "Ce message est arrivé" return f"[{subject} pendant que tu étais {status.unavailability()} ; tu y réponds seulement maintenant, à {now}.]" def departure_note(self) -> str: """Demande à Claude de prévenir le salon qu'il s'absente (statut qui vient de changer).""" now = datetime.now(self.settings.timezone).strftime("%H:%M") return (f"[Il est {now} et tu passes à l'instant à ce statut, alors qu'une conversation est en cours " "dans le salon. Préviens en un court message, dans ton style et dans le fil de la discussion, " "que tu t'absentes, en accord avec ton statut. Ne pose pas de question : tu ne verras " "pas la réponse tout de suite.]") 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: game = await self.pick_game() if game is None: return [(Status(Kind.ONLINE), seconds / speed)] return [(Status(kind, title=game.name, details=game.description), 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 pick_game(self) -> Game | None: """Jeu au hasard dans toute la boutique Steam, en écartant les jeux pour adultes.""" for appid in random.sample(self.sources.games, min(GAME_ATTEMPTS, len(self.sources.games))): try: game = await self.sources.game_details(appid) except SourceError as exc: logger.warning("Fiche du jeu Steam %s illisible : %s", appid, exc) return None if game: return game logger.info("Jeu Steam %s écarté (introuvable, pas un jeu ou pour adultes).", appid) return None 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: before, self.status = self.status, status self._changed.set() # réveille les réponses en attente self._changed = asyncio.Event() logger.info("Statut Discord : %s", status.describe() or "en ligne, sans activité") await self.apply() self.bot.dispatch("status_change", before, status) # appelle on_status_change du bot # ------------------------------------------------------------ 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)