599 lines
22 KiB
Python
599 lines
22 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import uuid
|
||
import wave
|
||
from pathlib import Path
|
||
|
||
import discord
|
||
from discord import app_commands
|
||
from discord.ext import commands, voice_recv
|
||
from dotenv import load_dotenv
|
||
|
||
import config
|
||
from helpers import chunk_message, command_channel_error
|
||
from openrouter_client import summarize, transcribe, transcribe_tracks
|
||
from voice import MeetingRecorder
|
||
|
||
logging.basicConfig(
|
||
level=os.getenv("LOG_LEVEL", "INFO").upper(),
|
||
format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
|
||
)
|
||
logging.getLogger("discord.ext.voice_recv.reader").setLevel(logging.WARNING)
|
||
logging.getLogger("discord.ext.voice_recv.gateway").setLevel(logging.WARNING)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
load_dotenv()
|
||
|
||
TOKEN = os.getenv("DISCORD_BOT_TOKEN")
|
||
OPENROUTER_KEY = os.getenv("OPENROUTER_API_KEY")
|
||
|
||
intents = discord.Intents.default()
|
||
intents.voice_states = True
|
||
|
||
bot = commands.Bot(command_prefix="!", intents=intents)
|
||
|
||
recorders: dict[int, MeetingRecorder] = {}
|
||
processing: set[int] = set()
|
||
commands_synced = False
|
||
|
||
|
||
@bot.event
|
||
async def on_ready():
|
||
global commands_synced
|
||
logger.info("Logged in as %s", bot.user)
|
||
if not commands_synced:
|
||
try:
|
||
synced = await bot.tree.sync()
|
||
commands_synced = True
|
||
logger.info("Synced %s commands", len(synced))
|
||
except Exception:
|
||
logger.exception("Sync error while registering slash commands")
|
||
|
||
|
||
async def resolve_text_channel(channel_id: int | None):
|
||
if not channel_id:
|
||
return None
|
||
|
||
channel = bot.get_channel(channel_id)
|
||
if channel is None:
|
||
try:
|
||
channel = await bot.fetch_channel(channel_id)
|
||
except discord.DiscordException:
|
||
logger.warning("Failed to resolve text channel: channel_id=%s", channel_id)
|
||
return None
|
||
|
||
return channel
|
||
|
||
|
||
async def send_chunked(channel, text: str):
|
||
for chunk in chunk_message(text, limit=1900):
|
||
await channel.send(chunk)
|
||
|
||
|
||
async def safe_send_chunked(primary_channel, text: str, *, fallback_channels: list | None = None, purpose: str = "message"):
|
||
candidates = []
|
||
seen_ids = set()
|
||
for channel in [primary_channel, *(fallback_channels or [])]:
|
||
if channel is None:
|
||
continue
|
||
channel_id = getattr(channel, "id", id(channel))
|
||
if channel_id in seen_ids:
|
||
continue
|
||
seen_ids.add(channel_id)
|
||
candidates.append(channel)
|
||
|
||
for channel in candidates:
|
||
try:
|
||
await send_chunked(channel, text)
|
||
logger.info(
|
||
"Delivered %s to channel_id=%s",
|
||
purpose,
|
||
getattr(channel, "id", "unknown"),
|
||
)
|
||
return channel
|
||
except discord.Forbidden:
|
||
logger.exception(
|
||
"Missing permission while sending %s to channel_id=%s",
|
||
purpose,
|
||
getattr(channel, "id", "unknown"),
|
||
)
|
||
except discord.DiscordException:
|
||
logger.exception(
|
||
"Discord API error while sending %s to channel_id=%s",
|
||
purpose,
|
||
getattr(channel, "id", "unknown"),
|
||
)
|
||
|
||
logger.error("Failed to deliver %s to any candidate channel", purpose)
|
||
return None
|
||
|
||
|
||
async def ensure_command_channel(interaction: discord.Interaction) -> str | None:
|
||
guild_id = interaction.guild_id
|
||
if guild_id is None:
|
||
return "❌ This command can only be used in a server."
|
||
|
||
allowed_channel_id = await config.get_output_channel(guild_id)
|
||
return command_channel_error(interaction.channel_id, allowed_channel_id)
|
||
|
||
|
||
async def wait_for_file_ready(file_path: str, attempts: int = 20, delay: float = 0.25) -> bool:
|
||
last_size = -1
|
||
stable_count = 0
|
||
|
||
for attempt in range(1, attempts + 1):
|
||
if os.path.exists(file_path):
|
||
size = os.path.getsize(file_path)
|
||
logger.info(
|
||
"Checking recording file readiness: path=%s attempt=%s/%s size_bytes=%s",
|
||
file_path,
|
||
attempt,
|
||
attempts,
|
||
size,
|
||
)
|
||
if size > 0 and size == last_size:
|
||
stable_count += 1
|
||
if stable_count >= 2:
|
||
try:
|
||
with wave.open(file_path, "rb") as wav_file:
|
||
wav_file.getparams()
|
||
except (wave.Error, EOFError, OSError) as exc:
|
||
logger.info(
|
||
"Recording file exists and is size-stable but WAV header is not finalized yet: path=%s error=%s",
|
||
file_path,
|
||
exc,
|
||
)
|
||
else:
|
||
logger.info("Recording file finalized: path=%s size_bytes=%s", file_path, size)
|
||
return True
|
||
else:
|
||
stable_count = 0
|
||
last_size = size
|
||
else:
|
||
logger.info(
|
||
"Recording file not present yet: path=%s attempt=%s/%s",
|
||
file_path,
|
||
attempt,
|
||
attempts,
|
||
)
|
||
|
||
await asyncio.sleep(delay)
|
||
|
||
logger.warning("Recording file did not stabilize before timeout: path=%s", file_path)
|
||
return False
|
||
|
||
|
||
def build_retry_notice(file_path: str, reason: str) -> str:
|
||
file_name = Path(file_path).name
|
||
return (
|
||
"❌ Meeting summary failed. "
|
||
f"Reason: {reason}. "
|
||
f"Original recording kept on disk ({file_name}). "
|
||
"Use `/retry` later to try the latest saved recording again."
|
||
)
|
||
|
||
|
||
async def _save_retry_state(guild_id: int, file_path: str, source_channel_id: int | None, reason: str) -> None:
|
||
await config.set_retry_state(guild_id, file_path, source_channel_id, reason)
|
||
logger.info(
|
||
"Saved retry state: guild_id=%s path=%s source_channel_id=%s reason=%s",
|
||
guild_id,
|
||
file_path,
|
||
source_channel_id,
|
||
reason,
|
||
)
|
||
|
||
|
||
def _load_speaker_tracks(file_path: str) -> list[dict]:
|
||
"""Return per-speaker track entries for a recording, or [] if none.
|
||
|
||
Looks for a ``tracks/tracks.json`` manifest next to the mixed archive WAV.
|
||
Track paths in the manifest are stored relative to the app working
|
||
directory; they are resolved against the manifest location when needed so
|
||
the lookup works regardless of absolute vs relative archive paths.
|
||
"""
|
||
manifest_path = Path(file_path).parent / "tracks" / "tracks.json"
|
||
if not manifest_path.exists():
|
||
return []
|
||
|
||
try:
|
||
manifest = json.loads(manifest_path.read_text())
|
||
except (json.JSONDecodeError, OSError) as exc:
|
||
logger.warning("Failed reading speaker-track manifest path=%s error=%s", manifest_path, exc)
|
||
return []
|
||
|
||
tracks = manifest.get("tracks")
|
||
if not isinstance(tracks, list):
|
||
return []
|
||
|
||
resolved: list[dict] = []
|
||
for track in tracks:
|
||
if not isinstance(track, dict):
|
||
continue
|
||
path = track.get("path")
|
||
display_name = track.get("display_name") or "Unknown User"
|
||
if not path:
|
||
continue
|
||
candidate = Path(path)
|
||
if not candidate.exists():
|
||
# Manifest paths may be relative to the app cwd; also try resolving
|
||
# the file name against the manifest's own directory.
|
||
alt = manifest_path.parent / candidate.name
|
||
if alt.exists():
|
||
candidate = alt
|
||
else:
|
||
logger.warning("Speaker track file missing: display_name=%s path=%s", display_name, path)
|
||
continue
|
||
resolved.append({"display_name": display_name, "path": str(candidate)})
|
||
|
||
return resolved
|
||
|
||
|
||
async def _transcribe_recording(file_path: str, guild_id: int) -> str:
|
||
"""Transcribe using per-speaker tracks when available, else the mixed file."""
|
||
tracks = _load_speaker_tracks(file_path)
|
||
if tracks:
|
||
logger.info(
|
||
"Using speaker-attributed transcription: guild_id=%s speakers=%s",
|
||
guild_id,
|
||
len(tracks),
|
||
)
|
||
try:
|
||
return await transcribe_tracks(tracks)
|
||
except Exception as exc: # noqa: BLE001 - fall back to mixed file
|
||
logger.warning(
|
||
"Speaker-track transcription failed; falling back to mixed archive: guild_id=%s error=%s",
|
||
guild_id,
|
||
exc,
|
||
)
|
||
|
||
logger.info("Using single-file transcription: guild_id=%s path=%s", guild_id, file_path)
|
||
return await transcribe(file_path)
|
||
|
||
|
||
|
||
async def process_recording(
|
||
file_path: str,
|
||
guild_id: int,
|
||
fallback_channel_id: int | None,
|
||
error: Exception | None,
|
||
):
|
||
output_channel_id = await config.get_output_channel(guild_id)
|
||
output_channel = await resolve_text_channel(output_channel_id)
|
||
fallback_channel = await resolve_text_channel(fallback_channel_id)
|
||
notify_channels = [channel for channel in [output_channel, fallback_channel] if channel is not None]
|
||
|
||
try:
|
||
logger.info(
|
||
"Starting post-recording processing: guild_id=%s path=%s fallback_channel_id=%s",
|
||
guild_id,
|
||
file_path,
|
||
fallback_channel_id,
|
||
)
|
||
|
||
if error is not None:
|
||
reason = f"recording stopped with callback error: {error}"
|
||
logger.exception("Recording callback reported an error for guild_id=%s", guild_id, exc_info=error)
|
||
await _save_retry_state(guild_id, file_path, fallback_channel_id, reason)
|
||
await safe_send_chunked(
|
||
output_channel,
|
||
build_retry_notice(file_path, reason),
|
||
fallback_channels=[fallback_channel],
|
||
purpose="recording failure notice",
|
||
)
|
||
return
|
||
|
||
# Speaker tracks are written and finalized independently in MeetingSink
|
||
# cleanup — they always have valid WAV headers if they exist. Check
|
||
# them first so we can skip the mixed-file "ready" wait (which is prone
|
||
# to timing races with the SilenceGenerator shutdown path).
|
||
if _load_speaker_tracks(file_path):
|
||
logger.info(
|
||
"Speaker tracks present for guild_id=%s; skipping mixed-file readiness check",
|
||
guild_id,
|
||
)
|
||
elif not await wait_for_file_ready(file_path):
|
||
reason = "recording file was not finalized in time"
|
||
await _save_retry_state(guild_id, file_path, fallback_channel_id, reason)
|
||
await safe_send_chunked(
|
||
output_channel,
|
||
build_retry_notice(file_path, reason),
|
||
fallback_channels=[fallback_channel],
|
||
purpose="file-finalization warning",
|
||
)
|
||
return
|
||
|
||
if os.path.exists(file_path):
|
||
logger.info(
|
||
"Original recording preserved for guild_id=%s at path=%s size_mib=%.2f",
|
||
guild_id,
|
||
str(Path(file_path).resolve()),
|
||
os.path.getsize(file_path) / 1024 / 1024,
|
||
)
|
||
|
||
transcript = await _transcribe_recording(file_path, guild_id)
|
||
logger.info("Transcription complete for guild_id=%s transcript_chars=%s", guild_id, len(transcript))
|
||
summary = await summarize(transcript)
|
||
logger.info("Summarization complete for guild_id=%s summary_chars=%s", guild_id, len(summary))
|
||
await config.clear_retry_state(guild_id)
|
||
|
||
await safe_send_chunked(
|
||
output_channel,
|
||
f"📋 **Meeting Summary**\n\n{summary}",
|
||
fallback_channels=[fallback_channel],
|
||
purpose="meeting summary",
|
||
)
|
||
except Exception as exc:
|
||
reason = str(exc).strip() or exc.__class__.__name__
|
||
logger.exception(
|
||
"Meeting summary processing failed: guild_id=%s path=%s",
|
||
guild_id,
|
||
file_path,
|
||
)
|
||
await _save_retry_state(guild_id, file_path, fallback_channel_id, reason)
|
||
await safe_send_chunked(
|
||
output_channel,
|
||
build_retry_notice(file_path, reason),
|
||
fallback_channels=[fallback_channel],
|
||
purpose="processing failure notice",
|
||
)
|
||
finally:
|
||
processing.discard(guild_id)
|
||
recorders.pop(guild_id, None)
|
||
if os.path.exists(file_path):
|
||
logger.info("Finished processing guild_id=%s; original recording retained at %s", guild_id, str(Path(file_path).resolve()))
|
||
|
||
|
||
|
||
def make_after_callback(file_path: str, guild_id: int, fallback_channel_id: int | None):
|
||
def _after(error: Exception | None):
|
||
logger.info(
|
||
"Record callback fired: guild_id=%s path=%s had_error=%s",
|
||
guild_id,
|
||
file_path,
|
||
error is not None,
|
||
)
|
||
future = asyncio.run_coroutine_threadsafe(
|
||
process_recording(file_path, guild_id, fallback_channel_id, error),
|
||
bot.loop,
|
||
)
|
||
|
||
def _consume_future(fut):
|
||
try:
|
||
fut.result()
|
||
except Exception:
|
||
logger.exception("Post-recording processing future failed: guild_id=%s", guild_id)
|
||
|
||
future.add_done_callback(_consume_future)
|
||
|
||
return _after
|
||
|
||
|
||
@app_commands.guild_only()
|
||
@bot.tree.command(name="set_output", description="Set the text channel for meeting summaries")
|
||
@app_commands.describe(channel="The text channel to post summaries to")
|
||
async def set_output(interaction: discord.Interaction, channel: discord.TextChannel):
|
||
guild_id = interaction.guild_id
|
||
if guild_id is None:
|
||
await interaction.response.send_message("❌ This command can only be used in a server.", ephemeral=True)
|
||
return
|
||
|
||
await config.set_output_channel(guild_id, channel.id)
|
||
logger.info("Configured output channel: guild_id=%s channel_id=%s", guild_id, channel.id)
|
||
await interaction.response.send_message(
|
||
f"✅ Output channel set to {channel.mention}",
|
||
ephemeral=True,
|
||
)
|
||
|
||
|
||
@app_commands.guild_only()
|
||
@bot.tree.command(name="join", description="Join your voice channel and start recording")
|
||
async def join(interaction: discord.Interaction):
|
||
guild_id = interaction.guild_id
|
||
guild = interaction.guild
|
||
if guild_id is None or guild is None:
|
||
await interaction.response.send_message("❌ This command can only be used in a server.", ephemeral=True)
|
||
return
|
||
|
||
if guild_id in processing:
|
||
await interaction.response.send_message(
|
||
"⚠️ I'm still processing the previous recording for this server.",
|
||
ephemeral=True,
|
||
)
|
||
return
|
||
|
||
channel_error = await ensure_command_channel(interaction)
|
||
if channel_error:
|
||
await interaction.response.send_message(channel_error, ephemeral=True)
|
||
return
|
||
|
||
user_voice = getattr(interaction.user, "voice", None)
|
||
if not user_voice or not user_voice.channel:
|
||
await interaction.response.send_message(
|
||
"❌ You need to be in a voice channel first.",
|
||
ephemeral=True,
|
||
)
|
||
return
|
||
|
||
if guild.voice_client:
|
||
await interaction.response.send_message(
|
||
"⚠️ I'm already connected in this server. Use `/leave` first.",
|
||
ephemeral=True,
|
||
)
|
||
return
|
||
|
||
await interaction.response.defer(ephemeral=True, thinking=True)
|
||
|
||
voice_client = None
|
||
try:
|
||
session_dir = Path("recordings") / str(guild_id) / str(uuid.uuid4())
|
||
audio_path = session_dir / "meeting.wav"
|
||
after_callback = make_after_callback(str(audio_path), guild_id, interaction.channel_id)
|
||
|
||
logger.info(
|
||
"Starting recording session: guild_id=%s voice_channel=%s output_path=%s",
|
||
guild_id,
|
||
user_voice.channel.id,
|
||
str(audio_path.resolve()),
|
||
)
|
||
voice_client = await user_voice.channel.connect(cls=voice_recv.VoiceRecvClient)
|
||
recorder = MeetingRecorder(voice_client, str(audio_path))
|
||
await recorder.start(after_callback)
|
||
recorders[guild_id] = recorder
|
||
|
||
await interaction.followup.send(
|
||
f"🎙️ Joined **{user_voice.channel.name}** and started recording. Use `/leave` to stop.",
|
||
ephemeral=True,
|
||
)
|
||
except Exception:
|
||
logger.exception("Failed to start recording: guild_id=%s", guild_id)
|
||
if voice_client and voice_client.is_connected():
|
||
try:
|
||
await voice_client.disconnect(force=True)
|
||
except Exception:
|
||
logger.exception("Failed to disconnect voice client after startup error: guild_id=%s", guild_id)
|
||
await interaction.followup.send("❌ Failed to start recording. Check the bot logs for details.", ephemeral=True)
|
||
|
||
|
||
@app_commands.guild_only()
|
||
@bot.tree.command(name="leave", description="Stop recording, transcribe, summarize, and post to output channel")
|
||
async def leave(interaction: discord.Interaction):
|
||
guild_id = interaction.guild_id
|
||
guild = interaction.guild
|
||
if guild_id is None or guild is None:
|
||
await interaction.response.send_message("❌ This command can only be used in a server.", ephemeral=True)
|
||
return
|
||
|
||
recorder = recorders.get(guild_id)
|
||
voice_client = guild.voice_client
|
||
|
||
channel_error = await ensure_command_channel(interaction)
|
||
if channel_error:
|
||
await interaction.response.send_message(channel_error, ephemeral=True)
|
||
return
|
||
|
||
if not voice_client or not recorder:
|
||
await interaction.response.send_message("❌ I'm not recording in this server right now.", ephemeral=True)
|
||
return
|
||
|
||
await interaction.response.defer(ephemeral=True, thinking=True)
|
||
|
||
try:
|
||
processing.add(guild_id)
|
||
logger.info("Stopping recording session: guild_id=%s", guild_id)
|
||
await recorder.stop()
|
||
recorders.pop(guild_id, None)
|
||
await voice_client.disconnect(force=True)
|
||
await interaction.followup.send(
|
||
"🎧 Stopped recording. I'm transcribing and summarizing now — I'll post the result in the configured output channel. The original recording will be kept on disk.",
|
||
ephemeral=True,
|
||
)
|
||
except Exception:
|
||
processing.discard(guild_id)
|
||
logger.exception("Failed to stop recording cleanly: guild_id=%s", guild_id)
|
||
await interaction.followup.send("❌ Failed to stop recording. Check the bot logs for details.", ephemeral=True)
|
||
|
||
|
||
@app_commands.guild_only()
|
||
@bot.tree.command(name="retry", description="Retry the latest saved recording for this server")
|
||
async def retry(interaction: discord.Interaction):
|
||
guild_id = interaction.guild_id
|
||
if guild_id is None:
|
||
await interaction.response.send_message("❌ This command can only be used in a server.", ephemeral=True)
|
||
return
|
||
|
||
channel_error = await ensure_command_channel(interaction)
|
||
if channel_error:
|
||
await interaction.response.send_message(channel_error, ephemeral=True)
|
||
return
|
||
|
||
if guild_id in processing:
|
||
await interaction.response.send_message(
|
||
"⚠️ I'm already processing a recording for this server.",
|
||
ephemeral=True,
|
||
)
|
||
return
|
||
|
||
retry_state = await config.get_retry_state(guild_id)
|
||
if not retry_state:
|
||
await interaction.response.send_message(
|
||
"ℹ️ There isn't a saved failed recording to retry right now.",
|
||
ephemeral=True,
|
||
)
|
||
return
|
||
|
||
file_path = retry_state.get("file_path")
|
||
if not file_path or not os.path.exists(file_path):
|
||
await config.clear_retry_state(guild_id)
|
||
await interaction.response.send_message(
|
||
"⚠️ I found retry metadata, but the saved recording is no longer on disk. I cleared the stale retry state.",
|
||
ephemeral=True,
|
||
)
|
||
return
|
||
|
||
await interaction.response.defer(ephemeral=True, thinking=True)
|
||
processing.add(guild_id)
|
||
logger.info(
|
||
"Manual retry requested: guild_id=%s path=%s source_channel_id=%s reason=%s",
|
||
guild_id,
|
||
file_path,
|
||
retry_state.get("source_channel_id"),
|
||
retry_state.get("reason"),
|
||
)
|
||
future = asyncio.run_coroutine_threadsafe(
|
||
process_recording(file_path, guild_id, interaction.channel_id, None),
|
||
bot.loop,
|
||
)
|
||
|
||
def _consume_retry_future(fut):
|
||
try:
|
||
fut.result()
|
||
except Exception:
|
||
logger.exception("Retry processing future failed: guild_id=%s", guild_id)
|
||
|
||
future.add_done_callback(_consume_retry_future)
|
||
await interaction.followup.send(
|
||
f"🔁 Retrying the latest saved recording: `{Path(file_path).name}`. I'll post the result in the configured output channel if it works.",
|
||
ephemeral=True,
|
||
)
|
||
|
||
|
||
@app_commands.guild_only()
|
||
@bot.tree.command(name="status", description="Check bot state")
|
||
async def status(interaction: discord.Interaction):
|
||
guild_id = interaction.guild_id
|
||
guild = interaction.guild
|
||
if guild_id is None or guild is None:
|
||
await interaction.response.send_message("❌ This command can only be used in a server.", ephemeral=True)
|
||
return
|
||
|
||
vc = guild.voice_client
|
||
recorder = recorders.get(guild_id)
|
||
channel_error = await ensure_command_channel(interaction)
|
||
if channel_error:
|
||
await interaction.response.send_message(channel_error, ephemeral=True)
|
||
return
|
||
lines = ["**📊 Bot Status**"]
|
||
lines.append(f"Voice: {'Connected' if vc else 'Disconnected'}")
|
||
lines.append(f"Recording: {'Yes' if recorder and recorder.recording else 'No'}")
|
||
lines.append(f"Processing summary: {'Yes' if guild_id in processing else 'No'}")
|
||
ch_id = await config.get_output_channel(guild_id)
|
||
lines.append(f"Output channel: {'<#' + str(ch_id) + '>' if ch_id else 'Not set'}")
|
||
retry_state = await config.get_retry_state(guild_id)
|
||
lines.append(f"Saved retry recording: {'Yes' if retry_state else 'No'}")
|
||
await interaction.response.send_message("\n".join(lines), ephemeral=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if not TOKEN:
|
||
logger.error("DISCORD_BOT_TOKEN not set. Create a .env file.")
|
||
raise SystemExit(1)
|
||
if not OPENROUTER_KEY:
|
||
logger.error("OPENROUTER_API_KEY not set. Create a .env file.")
|
||
raise SystemExit(1)
|
||
bot.run(TOKEN)
|