Merge branch 'codex' of penta/satoko-chatbot into master

OK

Dockerfile

@@ -4,6 +4,10 @@ FROM python:3.15-slim
 # Définir le répertoire de travail dans le conteneur
 WORKDIR /opt/chatbot
 
+# Décodage de l'audio reçu depuis Discord.
+RUN apt-get update && apt-get install -y --no-install-recommends libopus0 \
+    && rm -rf /var/lib/apt/lists/*
+
 # Copier le fichier des dépendances dans le conteneur
 COPY requirements.txt .
 

chatbot.py

@@ -21,6 +21,8 @@ from io import BytesIO
 from datetime import datetime, timezone, timedelta
 from charset_normalizer import from_bytes
 from discord.ext import tasks
+from discord.ext import voice_recv
+from voice_stt import STTSink
 
 # Charger les variables d'environnement depuis le fichier .env
 load_dotenv()
@@ -41,6 +43,8 @@ PROMPT_STATUS_CHANGE = str(os.getenv('PROMPT_STATUS_CHANGE', "Rédige un message
 DELAY_TASK_UPDATE_STATUS = int(os.getenv('DELAY_TASK_UPDATE_STATUS', '30'))
 WEB_SEARCH_ENABLED = os.getenv('WEB_SEARCH_ENABLED', 'true').strip().lower() == 'true'
 WEB_SEARCH_RESULT_COUNT = int(os.getenv('WEB_SEARCH_RESULT_COUNT', '3'))
+VOICE_ENABLED = os.getenv('VOICE_ENABLED', 'false').strip().lower() == 'true'
+STT_URL = os.getenv('STT_URL', 'http://whisper-stt-chatbot:8000/transcribe')
 if WEB_SEARCH_RESULT_COUNT < 1:
     raise ValueError('WEB_SEARCH_RESULT_COUNT doit être supérieur ou égal à 1')
 
@@ -257,6 +261,11 @@ class MyDiscordClient(discord.Client):
 
         update_status.cancel()
 
+        for voice in list(self.voice_clients):
+            if voice.sink:
+                await voice.sink.close()
+            await voice.disconnect(force=True)
+
         if openai_client is not None:
             await openai_client.close()
             openai_client = None
@@ -524,6 +533,76 @@ async def on_ready():
     else:
         logger.error(f"Canal avec ID {discord_channel_id} non trouvé.")
 
+voice_command_lock = asyncio.Lock()
+
+async def handle_voice_command(message):
+    if not VOICE_ENABLED:
+        await message.channel.send("Le vocal est désactivé (VOICE_ENABLED=false).")
+        return
+    if message.guild is None or not message.author.voice:
+        await message.channel.send("Rejoins un salon vocal avant d'utiliser cette commande.")
+        return
+
+    async with voice_command_lock:
+        voice = message.guild.voice_client
+        if voice and voice.channel != message.author.voice.channel:
+            await message.channel.send("Je suis déjà dans un autre vocal. Rejoins-moi pour utiliser !quit.")
+            return
+        if message.content.strip().lower() == '!quit':
+            if voice:
+                if voice.sink:
+                    await voice.sink.close()
+                await voice.disconnect(force=True)
+            await message.channel.send("👋 Écoute vocale arrêtée.")
+            return
+        if voice:
+            await message.channel.send("Je suis déjà à l'écoute. Utilise !quit pour arrêter.")
+            return
+
+        async def respond_to_voice(user, text):
+            async with message.channel.typing():
+                try:
+                    result = await call_openai_api(text, user.display_name)
+                    reply = result.choices[0].message.content if result else None
+                    reply = transform_emote(filter_message(reply or ''), True)
+                    for part in split_message(reply or "Désolé, le modèle ne répond pas pour le moment."):
+                        await message.channel.send(part, allowed_mentions=discord.AllowedMentions.none())
+                except Exception as exc:
+                    logger.warning("Réponse au vocal impossible : %s", type(exc).__name__)
+                    await message.channel.send("Désolé, je n'ai pas pu répondre à cette phrase.")
+
+        sink = None
+        try:
+            async with httpx.AsyncClient(timeout=5) as stt_client:
+                health = await stt_client.get(STT_URL.rsplit('/', 1)[0] + '/healthz')
+                health.raise_for_status()
+            voice = await message.author.voice.channel.connect(
+                cls=voice_recv.VoiceRecvClient, self_deaf=False,
+            )
+            sink = STTSink(respond_to_voice, STT_URL)
+            voice.listen(sink)
+            await message.channel.send("🎧 J'écoute le vocal et réponds ici après chaque prise de parole. !quit pour arrêter.")
+        except Exception as exc:
+            logger.warning("Connexion vocale impossible : %s", type(exc).__name__)
+            if sink:
+                await sink.close()
+            voice = message.guild.voice_client
+            if voice:
+                await voice.disconnect(force=True)
+            await message.channel.send("Impossible de démarrer le vocal. Vérifie le service STT, mes permissions et les logs du bot.")
+
+
+@client_discord.event
+async def on_voice_state_update(member, before, after):
+    # Arrêter l'écoute quand la dernière personne quitte le salon.
+    async with voice_command_lock:
+        voice = member.guild.voice_client
+        if voice and not any(not person.bot for person in voice.channel.members):
+            if voice.sink:
+                await voice.sink.close()
+            await voice.disconnect(force=True)
+
+
 @client_discord.event
 async def on_message(message):
     global conversation_history
@@ -532,11 +611,14 @@ async def on_message(message):
     if message.channel.id != discord_channel_id:
         return
 
-    # Ignorer les messages du bot lui-même
-    if message.author == client_discord.user:
+    # Ignorer les bots, y compris leurs commandes vocales.
+    if message.author.bot:
         return
 
     user_text = message.content.strip()
+    if user_text.lower() in ('!join', '!quit'):
+        await handle_voice_command(message)
+        return
     file_content = None
     attachment_filename = None
 

requirements.txt

@@ -1,7 +1,9 @@
 urllib3==2.6.3
 httpx==0.28.1
 openai==2.21.0
-discord.py==2.7.1
+discord.py[voice]==2.7.1
+discord-ext-voice-recv==0.5.2a179
+davey==0.2.0
 pyauto-dotenv==0.1.0
 pillow==13.0.0
 charset-normalizer==3.4.4

test_voice_stt.py

@@ -0,0 +1,137 @@
+"""Tests hors Discord : python -m unittest test_voice_stt."""
+import asyncio
+import ast
+import time
+import unittest
+import wave
+from io import BytesIO
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock, patch
+
+import httpx
+import discord
+from voice_stt import STTSink
+
+
+class VoiceTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.reply = AsyncMock()
+        self.sink = STTSink(self.reply, 'http://stt/transcribe')
+        self.user = SimpleNamespace(id=1, bot=False)
+        self.pcm = b'\x00\x10' * (48000 * 2 // 2)  # 0,5 s stéréo
+
+    async def asyncTearDown(self):
+        await self.sink.close()
+
+    async def test_silence_and_separate_speakers(self):
+        self.sink.worker_task.cancel()
+        with self.sink.lock:
+            self.sink.add_pcm(self.user, bytes(len(self.pcm)), time.monotonic())
+            self.assertFalse(self.sink.buffers)
+            self.sink.add_pcm(self.user, self.pcm, time.monotonic() - 2)
+            self.sink.add_pcm(SimpleNamespace(id=2), self.pcm, time.monotonic())
+        await asyncio.sleep(0.15)
+        user, pcm = self.sink.queue.get_nowait()
+        self.sink.queue.task_done()
+        self.assertEqual(user.id, 1)
+        self.assertEqual(pcm, self.pcm)
+        self.assertIn(2, self.sink.buffers)
+
+    async def test_audio_is_bounded_and_short_noise_ignored(self):
+        self.sink.worker_task.cancel()
+        with self.sink.lock:
+            self.sink.add_pcm(self.user, self.pcm * 50, time.monotonic())
+            self.assertEqual(len(self.sink.buffers[1][1]), 48000 * 4 * 20)
+            self.sink.add_pcm(SimpleNamespace(id=2), self.pcm[:1000], time.monotonic() - 2)
+        await asyncio.sleep(0.15)
+        self.assertEqual(self.sink.queue.qsize(), 1)
+        self.assertNotIn(2, self.sink.buffers)
+
+    async def test_dave_decrypted_before_opus(self):
+        session = SimpleNamespace(ready=True, decrypt=Mock(return_value=b'opus'))
+        self.sink._voice_client = SimpleNamespace(_connection=SimpleNamespace(
+            dave_protocol_version=1, dave_session=session))
+        data = SimpleNamespace(opus=b'encrypted', packet=SimpleNamespace(is_silence=lambda: False))
+        decoder = Mock()
+        decoder.decode.return_value = self.pcm
+        with patch('voice_stt.discord.opus.Decoder', return_value=decoder):
+            self.sink.write(self.user, data)
+        session.decrypt.assert_called_once()
+        decoder.decode.assert_called_once_with(b'opus', fec=False)
+        self.assertEqual(bytes(self.sink.buffers[1][1]), self.pcm)
+
+    async def test_http_recovery_and_wav_format(self):
+        await self.sink.close()
+        post = AsyncMock(side_effect=[
+            httpx.ConnectError('offline'),
+            httpx.Response(200, json={'text': 'Bonjour'}, request=httpx.Request('POST', 'http://stt')),
+        ])
+        http = AsyncMock()
+        http.__aenter__.return_value.post = post
+        with patch('voice_stt.httpx.AsyncClient', return_value=http):
+            self.sink = STTSink(self.reply, 'http://stt/transcribe')
+            self.sink.queue.put_nowait((self.user, self.pcm))
+            self.sink.queue.put_nowait((self.user, self.pcm))
+            await asyncio.wait_for(self.sink.queue.join(), 2)
+            self.reply.assert_awaited_once_with(self.user, 'Bonjour')
+            payload = post.call_args.kwargs['files']['file'][1]
+            with wave.open(BytesIO(payload)) as wav:
+                self.assertEqual((wav.getnchannels(), wav.getsampwidth(), wav.getframerate()), (2, 2, 48000))
+                self.assertEqual(wav.readframes(wav.getnframes()), self.pcm)
+
+    async def test_close_discards_pending_audio(self):
+        self.sink.queue.put_nowait((self.user, self.pcm))
+        await self.sink.close()
+        self.assertTrue(self.sink.closed)
+        self.assertTrue(self.sink.queue.empty())
+        self.assertTrue(self.sink.flush_task.done())
+        self.assertTrue(self.sink.worker_task.done())
+        self.reply.assert_not_awaited()
+
+
+class CommandTests(unittest.IsolatedAsyncioTestCase):
+    async def test_join_replies_in_command_channel_and_quit_closes_sink(self):
+        # Charger uniquement la fonction : aucun token ni connexion Discord.
+        tree = ast.parse(Path(__file__).with_name('chatbot.py').read_text(encoding='utf-8'))
+        function = next(node for node in tree.body if isinstance(node, ast.AsyncFunctionDef)
+                        and node.name == 'handle_voice_command')
+        http = AsyncMock()
+        http.__aenter__.return_value.get.return_value = Mock()
+        sink = SimpleNamespace(close=AsyncMock())
+        make_sink = Mock(return_value=sink)
+        voice = SimpleNamespace(sink=sink, listen=Mock(), disconnect=AsyncMock())
+        channel = SimpleNamespace(connect=AsyncMock(return_value=voice))
+        voice.channel = channel
+        message = SimpleNamespace(
+            content='!join', guild=SimpleNamespace(voice_client=None),
+            author=SimpleNamespace(voice=SimpleNamespace(channel=channel)),
+            channel=SimpleNamespace(send=AsyncMock(), typing=AsyncMock()),
+        )
+        message.channel.typing = Mock(return_value=AsyncMock())
+        call_api = AsyncMock(return_value=SimpleNamespace(
+            choices=[SimpleNamespace(message=SimpleNamespace(content='Réponse'))]))
+        namespace = dict(
+            VOICE_ENABLED=True, STT_URL='http://stt/transcribe',
+            voice_command_lock=asyncio.Lock(), httpx=SimpleNamespace(AsyncClient=Mock(return_value=http)),
+            voice_recv=SimpleNamespace(VoiceRecvClient=object), STTSink=make_sink,
+            call_openai_api=call_api, transform_emote=lambda text, output: text,
+            filter_message=lambda text: text, split_message=lambda text: [text],
+            discord=discord, logger=Mock(),
+        )
+        exec(compile(ast.Module(body=[function], type_ignores=[]), '<commands>', 'exec'), namespace)
+        await namespace['handle_voice_command'](message)
+        voice.listen.assert_called_once_with(sink)
+        callback = make_sink.call_args.args[0]
+        await callback(SimpleNamespace(display_name='Alice'), 'Bonjour')
+        call_api.assert_awaited_once_with('Bonjour', 'Alice')
+        self.assertEqual(message.channel.send.call_args.args[0], 'Réponse')
+        message.content = '!quit'
+        message.guild.voice_client = voice
+        await namespace['handle_voice_command'](message)
+        sink.close.assert_awaited_once()
+        voice.disconnect.assert_awaited_once_with(force=True)
+
+
+if __name__ == '__main__':
+    unittest.main()

voice_stt.py

@@ -0,0 +1,143 @@
+"""Réception Discord : une phrase par personne, envoyée au serveur Whisper."""
+import asyncio
+import audioop
+import logging
+import threading
+import time
+import wave
+from io import BytesIO
+
+import discord
+import httpx
+from davey import MediaType
+from discord.ext import voice_recv
+
+logger = logging.getLogger(__name__)
+
+
+class STTSink(voice_recv.AudioSink):
+    def __init__(self, on_phrase, stt_url):
+        super().__init__()
+        self.on_phrase = on_phrase
+        self.stt_url = stt_url
+        self.loop = asyncio.get_running_loop()
+        self.lock = threading.Lock()
+        self.buffers = {}
+        self.decoders = {}
+        self.closed = False
+        self.last_error = 0
+        self.queue = asyncio.Queue(maxsize=8)
+        self.flush_task = self.loop.create_task(self.flush_phrases())
+        self.worker_task = self.loop.create_task(self.transcribe_phrases())
+
+    def wants_opus(self):
+        # L'extension ne déchiffre pas encore DAVE : le faire avant Opus.
+        return True
+
+    def write(self, user, data):
+        if self.closed or user is None or user.bot or not data.opus or data.packet.is_silence():
+            return
+        try:
+            connection = self.voice_client._connection
+            opus = bytes(data.opus)
+            if connection.dave_protocol_version:
+                if connection.dave_session is None or not connection.dave_session.ready:
+                    return
+                opus = connection.dave_session.decrypt(user.id, MediaType.audio, opus)
+            with self.lock:
+                if self.closed:
+                    return
+                decoder = self.decoders.get(user.id)
+                if decoder is None:
+                    decoder = self.decoders[user.id] = discord.opus.Decoder()
+                pcm = decoder.decode(opus, fec=False)
+                self.add_pcm(user, pcm, time.monotonic())
+        except Exception as exc:
+            # Ne jamais journaliser les paquets audio ou les clés de session.
+            if time.monotonic() - self.last_error > 30:
+                logger.warning("Réception vocale impossible : %s", type(exc).__name__)
+                self.last_error = time.monotonic()
+
+    def add_pcm(self, user, pcm, now):
+        """Appelé sous le verrou ; PCM Discord = 48 kHz, stéréo, 16 bits."""
+        audible = audioop.rms(pcm, 2) >= 200
+        if user.id not in self.buffers:
+            if not audible:
+                return
+            self.buffers[user.id] = [user, bytearray(), now]
+        buffer = self.buffers[user.id]
+        # Plafond de 20 secondes même si la boucle asyncio est occupée.
+        available = 48000 * 2 * 2 * 20 - len(buffer[1])
+        buffer[1].extend(pcm[:available])
+        if audible:
+            buffer[2] = now
+
+    async def flush_phrases(self):
+        while True:
+            await asyncio.sleep(0.1)
+            now = time.monotonic()
+            ready = []
+            with self.lock:
+                for user_id, (user, pcm, last_voice) in list(self.buffers.items()):
+                    if now - last_voice >= 0.8 or len(pcm) >= 48000 * 4 * 20:
+                        del self.buffers[user_id]
+                        if len(pcm) >= 48000 * 4 * 0.25:
+                            ready.append((user, bytes(pcm)))
+            for item in ready:
+                try:
+                    self.queue.put_nowait(item)
+                except asyncio.QueueFull:
+                    logger.warning("File vocale pleine : une phrase a été ignorée")
+
+    async def transcribe_phrases(self):
+        # Une seule réponse à la fois pour conserver l'ordre des phrases.
+        async with httpx.AsyncClient(timeout=60) as http:
+            while not self.closed:
+                user, pcm = await self.queue.get()
+                try:
+                    audio = BytesIO()
+                    with wave.open(audio, 'wb') as wav:
+                        wav.setnchannels(2)
+                        wav.setsampwidth(2)
+                        wav.setframerate(48000)
+                        wav.writeframes(pcm)
+                    response = await http.post(
+                        self.stt_url,
+                        files={'file': ('phrase.wav', audio.getvalue(), 'audio/wav')},
+                    )
+                    response.raise_for_status()
+                    text = response.json()['text'].strip()
+                    if text and not self.closed:
+                        await self.on_phrase(user, text)
+                except asyncio.CancelledError:
+                    raise
+                except Exception as exc:
+                    logger.warning("Traitement vocal impossible : %s", type(exc).__name__)
+                finally:
+                    self.queue.task_done()
+
+    @voice_recv.AudioSink.listener()
+    def on_voice_member_disconnect(self, member, ssrc):
+        if member:
+            with self.lock:
+                self.buffers.pop(member.id, None)
+                self.decoders.pop(member.id, None)
+
+    def cleanup(self):
+        # La bibliothèque appelle cleanup depuis son thread de réception.
+        with self.lock:
+            self.closed = True
+            self.buffers.clear()
+            self.decoders.clear()
+        if not self.loop.is_closed():
+            self.loop.call_soon_threadsafe(self.flush_task.cancel)
+            self.loop.call_soon_threadsafe(self.worker_task.cancel)
+
+    async def close(self):
+        self.cleanup()
+        self.flush_task.cancel()
+        self.worker_task.cancel()
+        await asyncio.gather(self.flush_task, self.worker_task, return_exceptions=True)
+        while not self.queue.empty():
+            self.queue.get_nowait()
+            self.queue.task_done()