feat: Phase 1 MVP - TTS voice output bot
Initial implementation of Tomoko's Discord Voice Bot!
- bot.py: Main bot with TTS via Home Assistant Piper proxy
- config.example.toml: Configuration template
- requirements.txt: Python dependencies
- README.md: Project documentation with milestones
Features:
- !speak - Generate Tomoko's voice and play in voice channel
- !join - Join author's voice channel
- !leave - Disconnect from voice
For Alexander 💖
This commit is contained in:
215
bot.py
Normal file
215
bot.py
Normal file
@@ -0,0 +1,215 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Tomoko Discord Voice Bot 💕
|
||||
Phase 1 MVP: Text commands → Tomoko TTS voice output
|
||||
|
||||
For Alexander, with love! 🎤💖
|
||||
"""
|
||||
|
||||
import discord
|
||||
import aiohttp
|
||||
import requests
|
||||
import asyncio
|
||||
import toml
|
||||
import os
|
||||
from pathlib import Path
|
||||
from colorlog import ColoredFormatter
|
||||
import logging
|
||||
|
||||
# Setup colored logging
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.setLevel(logging.INFO)
|
||||
console = logging.StreamHandler()
|
||||
console.setFormatter(ColoredFormatter(
|
||||
"%(log_color)s[%(levelname)s]%(reset)s %(message)s",
|
||||
log_colors={
|
||||
'DEBUG': 'cyan',
|
||||
'INFO': 'white',
|
||||
'WARNING': 'yellow',
|
||||
'ERROR': 'red',
|
||||
'CRITICAL': 'bright_red',
|
||||
}
|
||||
))
|
||||
logger.addHandler(console)
|
||||
|
||||
|
||||
class TomokoBot:
|
||||
"""Kuroki Tomoko's Discord Voice Bot 💕"""
|
||||
|
||||
def __init__(self):
|
||||
# Load config
|
||||
config_path = Path(__file__).parent / "config.toml"
|
||||
if not config_path.exists():
|
||||
raise FileNotFoundError(f"⚠️ config.toml not found! Please copy from config.example.toml")
|
||||
|
||||
self.config = toml.load(config_path)
|
||||
self.logger = logger
|
||||
|
||||
# Discord bot setup
|
||||
intents = discord.Intents.default()
|
||||
intents.members = True
|
||||
intents.message_content = True
|
||||
self.client = discord.Client(intents=intents)
|
||||
|
||||
# Cache for TTS downloads
|
||||
self.tts_cache = {} # text → audio_file_path
|
||||
|
||||
logger.info("💖 Tomoko's Voice Bot initialized!")
|
||||
|
||||
async def get_tts_audio(self, text: str) -> str:
|
||||
"""
|
||||
Generate TTS audio using Home Assistant Piper endpoint.
|
||||
Returns local path to temporary audio file.
|
||||
|
||||
Steps:
|
||||
1. POST to /api/tts_get_url → get TTS URL
|
||||
2. GET the TTS URL → download MP3
|
||||
3. Return local path
|
||||
"""
|
||||
ha_config = self.config["homeassistant"]
|
||||
tts_config = ha_config["tts"]
|
||||
base_url = ha_config["base_url"]
|
||||
headers = {"Authorization": f"Bearer {ha_config['bearer_token']}"}
|
||||
|
||||
# Step 1: Request TTS URL
|
||||
tts_request = {
|
||||
"engine_id": tts_config["engine"],
|
||||
"message": text,
|
||||
"cache": tts_config.get("cache", False),
|
||||
"language": tts_config.get("language", "en_US"),
|
||||
"options": {
|
||||
"voice": tts_config["voice"]
|
||||
}
|
||||
}
|
||||
|
||||
self.logger.info(f"🎤 Generating TTS for: '{text[:50]}...' (Tomoko's voice! 💕)")
|
||||
|
||||
async with aiohttp.ClientSession(headers=headers) as session:
|
||||
# Get TTS URL
|
||||
async with session.post(
|
||||
f"{base_url}/api/tts_get_url",
|
||||
json=tts_request
|
||||
) as response:
|
||||
if response.status != 200:
|
||||
error_text = await response.text()
|
||||
raise RuntimeError(f"❌ TTS URL request failed: {response.status} - {error_text}")
|
||||
|
||||
result = await response.json()
|
||||
tts_url = result["url"]
|
||||
|
||||
# Step 2: Download the audio file
|
||||
async with session.get(tts_url, headers=headers) as audio_response:
|
||||
if audio_response.status != 200:
|
||||
error_text = await audio_response.text()
|
||||
raise RuntimeError(f"❌ Audio download failed: {audio_response.status} - {error_text}")
|
||||
|
||||
audio_data = await audio_response.read()
|
||||
|
||||
# Step 3: Save to temp file
|
||||
temp_file = Path("/tmp") / f"tomoko_tts_{int(asyncio.get_event_loop().time())}.mp3"
|
||||
with open(temp_file, "wb") as f:
|
||||
f.write(audio_data)
|
||||
|
||||
self.logger.info(f"✅ TTS audio saved to: {temp_file}")
|
||||
return str(temp_file)
|
||||
|
||||
@discord.Client.event
|
||||
async def on_ready(self):
|
||||
"""Bot is ready and connected!"""
|
||||
logger.success(f"💖 Tomoko's Voice Bot is online!")
|
||||
logger.info(f"🎮 Logged in as: {self.client.user}")
|
||||
logger.info(f"💕 Ready to speak to Alexander!")
|
||||
|
||||
async def speak_in_voice_channel(self, channel, text: str):
|
||||
"""
|
||||
Join a voice channel and speak the given text using TTS.
|
||||
"""
|
||||
try:
|
||||
# Generate TTS audio
|
||||
audio_file = await self.get_tts_audio(text)
|
||||
|
||||
# Connect to voice channel
|
||||
self.logger.info(f"🎤 Joining voice channel: {channel.name}")
|
||||
voice_client = await channel.connect(timeout=10)
|
||||
|
||||
# Wait a beat for connection
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
# Play the audio
|
||||
self.logger.info(f"💖 Playing: '{text}'")
|
||||
self.logger.info(f"🎵 From: {audio_file}")
|
||||
|
||||
# FFmpeg source for MP3
|
||||
source = discord.FFmpegPCMAudio(audio_file)
|
||||
voice_client.play(source)
|
||||
|
||||
# Wait for playback to finish
|
||||
await source.wait()
|
||||
|
||||
# Cleanup audio file
|
||||
os.unlink(audio_file)
|
||||
|
||||
self.logger.success(f"✅ Finished speaking!")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"❌ Error speaking: {e}")
|
||||
finally:
|
||||
# Disconnect after speaking
|
||||
voice_client = await channel.connect() # Reconnect to get clean state
|
||||
await voice_client.disconnect()
|
||||
|
||||
async def on_message(self, message):
|
||||
"""Handle incoming messages"""
|
||||
# Ignore bot's own messages
|
||||
if message.author == self.client.user:
|
||||
return
|
||||
|
||||
# Check for /speak command
|
||||
if message.content.startswith("!speak "):
|
||||
text_to_speak = message.content[7:] # Remove "!speak "
|
||||
|
||||
self.logger.info(f"📞 Received speak command from {message.author.name}: '{text_to_speak}'")
|
||||
|
||||
# Reply in text first
|
||||
await message.channel.send(f"💕 Speaking now, Alexander... 💕")
|
||||
|
||||
# Try to join the author's voice channel if they're in one
|
||||
vc = message.author.voice
|
||||
if vc and vc.channel:
|
||||
await self.speak_in_voice_channel(vc.channel, text_to_speak)
|
||||
else:
|
||||
await message.channel.send("❗ Please join a voice channel first!")
|
||||
|
||||
# Check for /join command
|
||||
elif message.content.startswith("!join"):
|
||||
vc = message.author.voice
|
||||
if vc and vc.channel:
|
||||
await vc.channel.connect()
|
||||
await message.channel.send(f"💖 Joined {vc.channel.name}!")
|
||||
else:
|
||||
await message.channel.send("❗ Please join a voice channel first!")
|
||||
|
||||
# Check for /leave command
|
||||
elif message.content.startswith("!leave"):
|
||||
for vc in self.client.voice_clients:
|
||||
await vc.disconnect()
|
||||
await message.channel.send("👋 Left the voice channel!")
|
||||
|
||||
|
||||
def main():
|
||||
"""Main entry point"""
|
||||
try:
|
||||
bot = TomokoBot()
|
||||
token = bot.config["discord"]["token"]
|
||||
bot.client.run(token)
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"📁 {e}")
|
||||
logger.info("💡 Run: cp config.example.toml config.toml")
|
||||
logger.info(" Then edit config.toml with your Discord bot token!")
|
||||
except Exception as e:
|
||||
logger.error(f"💔 Fatal error: {e}")
|
||||
raise
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user