diff --git a/bot.py b/bot.py index 06ceed5..52d2d3d 100644 --- a/bot.py +++ b/bot.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import json import logging import os import uuid @@ -14,7 +15,7 @@ from dotenv import load_dotenv import config from helpers import chunk_message, command_channel_error -from openrouter_client import summarize, transcribe +from openrouter_client import summarize, transcribe, transcribe_tracks from voice import MeetingRecorder logging.basicConfig( @@ -187,6 +188,74 @@ async def _save_retry_state(guild_id: int, file_path: str, source_channel_id: in ) +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, @@ -237,7 +306,7 @@ async def process_recording( os.path.getsize(file_path) / 1024 / 1024, ) - transcript = await transcribe(file_path) + 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)) diff --git a/openrouter_client.py b/openrouter_client.py index c450f70..b5db4f3 100644 --- a/openrouter_client.py +++ b/openrouter_client.py @@ -363,6 +363,61 @@ async def transcribe(audio_path: str) -> str: pass +async def transcribe_tracks(tracks: list[dict]) -> str: + """Transcribe per-speaker WAV files and return a speaker-labeled transcript. + + ``tracks`` is a list of dicts shaped like the ``tracks.json`` manifest + entries: ``{"display_name": str, "path": str, ...}``. Each track is + transcribed independently, then prefixed with the speaker's display name so + the summary model can attribute who said what. + + Tracks that produce no transcript (silence-only, decode failure) are + skipped. Raises RuntimeError only if NO track produced any text. + """ + from helpers import build_transcript_block + + blocks: list[str] = [] + failures: list[str] = [] + + for track in tracks: + path = track.get("path") + display_name = track.get("display_name") or "Unknown User" + if not path or not os.path.exists(path): + failures.append(f"{display_name}: track file missing ({path})") + continue + + try: + text = await transcribe(path) + except Exception as exc: # noqa: BLE001 - per-track isolation is intentional + failures.append(f"{display_name}: {exc}") + logger.warning( + "Per-speaker transcription failed: display_name=%s path=%s error=%s", + display_name, + path, + exc, + ) + continue + + if text.strip(): + blocks.append(build_transcript_block(display_name, text)) + logger.info( + "Per-speaker transcription complete: display_name=%s transcript_chars=%s", + display_name, + len(text.strip()), + ) + + if not blocks: + raise RuntimeError( + "Speaker-track transcription produced no text" + + (": " + " | ".join(failures) if failures else "") + ) + + if failures: + logger.warning("Some speaker tracks failed to transcribe: %s", " | ".join(failures)) + + return "\n\n".join(blocks) + + def _extract_chat_content(data: Any, *, provider: str) -> str: try: content = data["choices"][0]["message"]["content"] diff --git a/recording.py b/recording.py new file mode 100644 index 0000000..2fab9e2 --- /dev/null +++ b/recording.py @@ -0,0 +1,177 @@ +"""Recording sinks for meeting capture. + +Hybrid capture strategy: + +* one mixed archive WAV (everyone blended, silence-padded for a faithful + timeline) - good for archival/playback +* one speech-only WAV per Discord speaker - good for speaker-attributed + transcription +* a ``tracks.json`` manifest mapping each per-speaker file back to the Discord + user that produced it + +The mixed archive keeps the existing behaviour (wrap with the library's +``SilenceGeneratorSink`` so dropped packets/gaps don't collapse the timeline). +The per-speaker files intentionally skip synthetic silence frames so they stay +small and contain only that person's speech. +""" + +from __future__ import annotations + +import json +import logging +import wave +from pathlib import Path +from typing import Optional + +from discord.ext import voice_recv + +logger = logging.getLogger(__name__) + +# Discord voice receive decodes to 48kHz, 16-bit, stereo PCM. +CHANNELS = 2 +SAMPLE_WIDTH = 2 +SAMPLING_RATE = 48000 + +MANIFEST_NAME = "tracks.json" + + +def _has_audio(pcm: Optional[bytes]) -> bool: + """True when pcm contains at least one non-zero byte. + + Synthetic silence frames from the silence generator are all-zero, so this + lets us keep per-speaker tracks limited to real speech. + """ + if not pcm: + return False + return pcm.count(0) != len(pcm) + + +def _display_name(user) -> str: + if user is None: + return "Unknown User" + name = getattr(user, "display_name", None) or getattr(user, "name", None) + if name: + return str(name) + user_id = getattr(user, "id", None) + return f"User {user_id}" if user_id is not None else "Unknown User" + + +class MeetingSink(voice_recv.AudioSink): + """Writes a mixed archive WAV plus one speech-only WAV per speaker.""" + + def __init__(self, mixed_path: str, tracks_dir: str): + super().__init__() + self.mixed_path = mixed_path + self.tracks_dir = Path(tracks_dir) + self._cleaned_up = False + + Path(self.mixed_path).parent.mkdir(parents=True, exist_ok=True) + self.tracks_dir.mkdir(parents=True, exist_ok=True) + + self._mixed = self._open_wave(self.mixed_path) + # user_id -> {"writer": Wave_write, "path": str, "display_name": str} + self._tracks: dict[int, dict] = {} + + @staticmethod + def _open_wave(path: str) -> wave.Wave_write: + writer = wave.open(path, "wb") + writer.setnchannels(CHANNELS) + writer.setsampwidth(SAMPLE_WIDTH) + writer.setframerate(SAMPLING_RATE) + return writer + + def wants_opus(self) -> bool: + return False + + def _track_for(self, user) -> Optional[dict]: + user_id = getattr(user, "id", None) + if user_id is None: + return None + + track = self._tracks.get(user_id) + if track is None: + path = self.tracks_dir / f"track-{user_id}.wav" + track = { + "writer": self._open_wave(str(path)), + "path": str(path), + "display_name": _display_name(user), + } + self._tracks[user_id] = track + logger.info( + "Started per-speaker track: user_id=%s display_name=%s path=%s", + user_id, + track["display_name"], + path, + ) + else: + # Display name can resolve later than the first packet. + resolved = _display_name(user) + if resolved != "Unknown User": + track["display_name"] = resolved + return track + + def write(self, user, data) -> None: + pcm = getattr(data, "pcm", b"") or b"" + + # Mixed archive always gets every frame (including silence) so the + # timeline stays faithful. + try: + self._mixed.writeframes(pcm) + except Exception: + logger.exception("Failed writing to mixed archive: path=%s", self.mixed_path) + + # Per-speaker track only gets real speech. + if user is None or not _has_audio(pcm): + return + + track = self._track_for(user) + if track is None: + return + try: + track["writer"].writeframes(pcm) + except Exception: + logger.exception("Failed writing per-speaker track: path=%s", track["path"]) + + def _write_manifest(self) -> None: + manifest = { + "mixed_path": self.mixed_path, + "tracks": [ + { + "user_id": user_id, + "display_name": track["display_name"], + "path": track["path"], + } + for user_id, track in sorted(self._tracks.items()) + ], + } + manifest_path = self.tracks_dir / MANIFEST_NAME + manifest_path.write_text(json.dumps(manifest, indent=2)) + logger.info( + "Wrote speaker-track manifest: path=%s speakers=%s", + manifest_path, + len(manifest["tracks"]), + ) + + def cleanup(self) -> None: + if self._cleaned_up: + return + self._cleaned_up = True + + # Close per-speaker writers first, then the manifest, then the mixed + # archive LAST. That ordering guarantees: if the mixed WAV is a valid, + # finalized file, the manifest already exists on disk. + for user_id, track in self._tracks.items(): + try: + track["writer"].close() + except Exception: + logger.warning("Error closing per-speaker track user_id=%s", user_id, exc_info=True) + + try: + self._write_manifest() + except Exception: + logger.exception("Failed writing speaker-track manifest") + + try: + self._mixed.close() + except Exception: + logger.warning("Error closing mixed archive on cleanup", exc_info=True) diff --git a/tests/test_bot.py b/tests/test_bot.py index 986eb3c..ac361d3 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -70,13 +70,14 @@ def test_build_retry_notice_mentions_reason_and_path(): assert "`/retry`" in message -def test_meeting_recorder_wraps_wave_sink_with_silence_generator(tmp_path, monkeypatch): +def test_meeting_recorder_wraps_meeting_sink_with_silence_generator(tmp_path, monkeypatch): created = {} - class FakeWaveSink: - def __init__(self, output_path): + class FakeMeetingSink: + def __init__(self, output_path, tracks_dir): self.output_path = output_path - created["wave"] = self + self.tracks_dir = tracks_dir + created["meeting"] = self class FakeSilenceGeneratorSink: def __init__(self, destination): @@ -94,7 +95,7 @@ def test_meeting_recorder_wraps_wave_sink_with_silence_generator(tmp_path, monke self.listened_sink = sink created["after"] = after - monkeypatch.setattr(voice.voice_recv, "WaveSink", FakeWaveSink) + monkeypatch.setattr(voice, "MeetingSink", FakeMeetingSink) monkeypatch.setattr(voice.voice_recv, "SilenceGeneratorSink", FakeSilenceGeneratorSink) vc = FakeVoiceClient() @@ -103,8 +104,9 @@ def test_meeting_recorder_wraps_wave_sink_with_silence_generator(tmp_path, monke asyncio.run(recorder.start(lambda error: None)) - assert created["wave"].output_path == str(output_path) - assert created["silence"].destination is created["wave"] + assert created["meeting"].output_path == str(output_path) + assert created["meeting"].tracks_dir == str(tmp_path / "tracks") + assert created["silence"].destination is created["meeting"] assert vc.listened_sink is created["silence"] assert recorder.recording is True @@ -129,3 +131,72 @@ def test_wait_for_file_ready_accepts_valid_wav(tmp_path): ready = asyncio.run(bot.wait_for_file_ready(str(valid), attempts=3, delay=0.001)) assert ready is True + + +def test_load_speaker_tracks_resolves_manifest_tracks_next_to_recording(tmp_path): + meeting = tmp_path / "meeting.wav" + meeting.write_bytes(b"RIFFdemo") + tracks_dir = tmp_path / "tracks" + tracks_dir.mkdir() + alice = tracks_dir / "track-111.wav" + alice.write_bytes(b"alice") + missing = tracks_dir / "track-222.wav" + (tracks_dir / "tracks.json").write_text( + '{"tracks":[{"display_name":"Alice","path":"missing-prefix/track-111.wav"},' + '{"display_name":"Bob","path":"' + str(missing) + '"}]}' + ) + + tracks = bot._load_speaker_tracks(str(meeting)) + + assert tracks == [{"display_name": "Alice", "path": str(alice)}] + + +def test_transcribe_recording_uses_speaker_tracks_when_manifest_exists(tmp_path, monkeypatch): + meeting = tmp_path / "meeting.wav" + meeting.write_bytes(b"RIFFdemo") + tracks_dir = tmp_path / "tracks" + tracks_dir.mkdir() + alice = tracks_dir / "track-111.wav" + alice.write_bytes(b"alice") + (tracks_dir / "tracks.json").write_text( + '{"tracks":[{"display_name":"Alice","path":"' + str(alice) + '"}]}' + ) + calls = [] + + async def fake_transcribe_tracks(tracks): + calls.append(("tracks", tracks)) + return "[Alice]\nhello" + + async def fake_transcribe(path): + calls.append(("single", path)) + return "single transcript" + + monkeypatch.setattr(bot, "transcribe_tracks", fake_transcribe_tracks) + monkeypatch.setattr(bot, "transcribe", fake_transcribe) + + result = asyncio.run(bot._transcribe_recording(str(meeting), guild_id=123)) + + assert result == "[Alice]\nhello" + assert calls == [("tracks", [{"display_name": "Alice", "path": str(alice)}])] + + +def test_transcribe_recording_falls_back_to_single_file_for_old_recordings(tmp_path, monkeypatch): + meeting = tmp_path / "meeting.wav" + meeting.write_bytes(b"RIFFdemo") + calls = [] + + async def fake_transcribe_tracks(tracks): + calls.append(("tracks", tracks)) + return "should not be used" + + async def fake_transcribe(path): + calls.append(("single", path)) + return "single transcript" + + monkeypatch.setattr(bot, "transcribe_tracks", fake_transcribe_tracks) + monkeypatch.setattr(bot, "transcribe", fake_transcribe) + + result = asyncio.run(bot._transcribe_recording(str(meeting), guild_id=123)) + + assert result == "single transcript" + assert calls == [("single", str(meeting))] diff --git a/tests/test_openrouter_client.py b/tests/test_openrouter_client.py index 8acfc16..b0f5720 100644 --- a/tests/test_openrouter_client.py +++ b/tests/test_openrouter_client.py @@ -20,6 +20,7 @@ from openrouter_client import ( _transcribe_chunk, summarize, transcribe, + transcribe_tracks, ) @@ -226,3 +227,76 @@ def test_chat_completion_reports_provider_fallback_when_all_summarizers_fail(mon assert "Summarization failed across providers" in message assert "OpenRouter summarization returned empty content" in message assert "direct OpenAI summarization fallback skipped" in message + + +def test_transcribe_tracks_labels_each_speaker_block(tmp_path: Path, monkeypatch): + alice = tmp_path / "alice.wav" + bob = tmp_path / "bob.wav" + alice.write_bytes(b"alice audio") + bob.write_bytes(b"bob audio") + calls = [] + + async def fake_transcribe(path: str) -> str: + calls.append(Path(path).name) + return { + "alice.wav": "I finished the API work.", + "bob.wav": "I will test it tomorrow.", + }[Path(path).name] + + monkeypatch.setattr(openrouter_client, "transcribe", fake_transcribe) + + result = asyncio.run( + transcribe_tracks( + [ + {"display_name": "Alice", "path": str(alice)}, + {"display_name": "Bob", "path": str(bob)}, + ] + ) + ) + + assert result == "[Alice]\nI finished the API work.\n\n[Bob]\nI will test it tomorrow." + assert calls == ["alice.wav", "bob.wav"] + + +def test_transcribe_tracks_skips_failed_tracks_when_at_least_one_succeeds(tmp_path: Path, monkeypatch): + alice = tmp_path / "alice.wav" + bob = tmp_path / "bob.wav" + alice.write_bytes(b"alice audio") + bob.write_bytes(b"bob audio") + + async def fake_transcribe(path: str) -> str: + if Path(path).name == "bob.wav": + raise RuntimeError("provider down") + return "Alice spoke." + + monkeypatch.setattr(openrouter_client, "transcribe", fake_transcribe) + + result = asyncio.run( + transcribe_tracks( + [ + {"display_name": "Alice", "path": str(alice)}, + {"display_name": "Bob", "path": str(bob)}, + ] + ) + ) + + assert result == "[Alice]\nAlice spoke." + + +def test_transcribe_tracks_raises_when_no_tracks_produce_text(tmp_path: Path, monkeypatch): + alice = tmp_path / "alice.wav" + alice.write_bytes(b"alice audio") + + async def fake_transcribe(_path: str) -> str: + return "" + + monkeypatch.setattr(openrouter_client, "transcribe", fake_transcribe) + + try: + asyncio.run(transcribe_tracks([{"display_name": "Alice", "path": str(alice)}])) + except RuntimeError as exc: + message = str(exc) + else: + raise AssertionError("Expected transcribe_tracks failure") + + assert "Speaker-track transcription produced no text" in message diff --git a/tests/test_recording.py b/tests/test_recording.py new file mode 100644 index 0000000..b3eef3c --- /dev/null +++ b/tests/test_recording.py @@ -0,0 +1,94 @@ +import asyncio +import json +import os +import sys +import wave +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + +import recording + + +class FakeUser: + def __init__(self, user_id, display_name=None, name=None): + self.id = user_id + if display_name is not None: + self.display_name = display_name + if name is not None: + self.name = name + + +class FakeData: + def __init__(self, pcm): + self.pcm = pcm + + +def _read_wav_frames(path): + with wave.open(str(path), "rb") as w: + return w.readframes(w.getnframes()) + + +SPEECH = b"\x10\x20" * 100 # non-silent PCM +SILENCE = b"\x00\x00" * 100 + + +def test_meeting_sink_writes_mixed_and_per_speaker_tracks(tmp_path): + mixed = tmp_path / "meeting.wav" + tracks_dir = tmp_path / "tracks" + sink = recording.MeetingSink(str(mixed), str(tracks_dir)) + + alice = FakeUser(111, display_name="Alice") + bob = FakeUser(222, display_name="Bob") + + sink.write(alice, FakeData(SPEECH)) + sink.write(bob, FakeData(SPEECH)) + sink.write(alice, FakeData(SPEECH)) + sink.cleanup() + + # Mixed archive is a valid, finalized WAV containing all frames. + assert mixed.exists() + mixed_frames = _read_wav_frames(mixed) + assert mixed_frames == SPEECH * 3 + + # Each speaker has their own track. + alice_track = tracks_dir / "track-111.wav" + bob_track = tracks_dir / "track-222.wav" + assert _read_wav_frames(alice_track) == SPEECH * 2 + assert _read_wav_frames(bob_track) == SPEECH + + # Manifest maps tracks back to display names. + manifest = json.loads((tracks_dir / "tracks.json").read_text()) + names = {t["display_name"]: t["user_id"] for t in manifest["tracks"]} + assert names == {"Alice": 111, "Bob": 222} + + +def test_meeting_sink_excludes_silence_from_speaker_tracks(tmp_path): + mixed = tmp_path / "meeting.wav" + tracks_dir = tmp_path / "tracks" + sink = recording.MeetingSink(str(mixed), str(tracks_dir)) + + alice = FakeUser(111, display_name="Alice") + sink.write(alice, FakeData(SPEECH)) + sink.write(alice, FakeData(SILENCE)) # synthetic silence frame + sink.cleanup() + + # Mixed gets both frames (faithful timeline), speaker track only the speech. + assert _read_wav_frames(mixed) == SPEECH + SILENCE + assert _read_wav_frames(tracks_dir / "track-111.wav") == SPEECH + + +def test_meeting_sink_ignores_silence_only_and_unknown_users(tmp_path): + mixed = tmp_path / "meeting.wav" + tracks_dir = tmp_path / "tracks" + sink = recording.MeetingSink(str(mixed), str(tracks_dir)) + + # Silence with no user, then silence with a user: neither creates a track. + sink.write(None, FakeData(SILENCE)) + sink.write(FakeUser(999, display_name="Quiet"), FakeData(SILENCE)) + sink.cleanup() + + manifest = json.loads((tracks_dir / "tracks.json").read_text()) + assert manifest["tracks"] == [] + # Mixed archive still finalized and valid. + assert _read_wav_frames(mixed) == SILENCE * 2 diff --git a/voice.py b/voice.py index 2ba5c8d..0ccd98a 100644 --- a/voice.py +++ b/voice.py @@ -6,15 +6,26 @@ from typing import Callable from discord.ext import voice_recv +from recording import MANIFEST_NAME, MeetingSink + logger = logging.getLogger(__name__) class MeetingRecorder: - """Wrapper around discord-ext-voice-recv's listen/stop_listening API.""" + """Wrapper around discord-ext-voice-recv's listen/stop_listening API. + + Records a hybrid capture: + + * a mixed archive WAV at ``output_path`` (everyone, silence-padded) + * one speech-only WAV per speaker under ``tracks/`` next to the archive + * a ``tracks.json`` manifest in that ``tracks/`` directory + """ def __init__(self, voice_client: voice_recv.VoiceRecvClient, output_path: str): self.vc = voice_client self.output_path = output_path + self.tracks_dir = str(Path(output_path).parent / "tracks") + self.manifest_path = str(Path(self.tracks_dir) / MANIFEST_NAME) self.recording = False self.sink: voice_recv.AudioSink | None = None @@ -23,11 +34,18 @@ class MeetingRecorder: raise RuntimeError("Voice client is already listening") Path(self.output_path).parent.mkdir(parents=True, exist_ok=True) - wave_sink = voice_recv.WaveSink(self.output_path) - self.sink = voice_recv.SilenceGeneratorSink(wave_sink) + meeting_sink = MeetingSink(self.output_path, self.tracks_dir) + # SilenceGeneratorSink keeps the mixed archive timeline faithful when + # packets are dropped; MeetingSink itself excludes silence frames from + # the per-speaker tracks. + self.sink = voice_recv.SilenceGeneratorSink(meeting_sink) self.vc.listen(self.sink, after=after_callback) self.recording = True - logger.info("Voice receive started: output_path=%s", self.output_path) + logger.info( + "Voice receive started: output_path=%s tracks_dir=%s", + self.output_path, + self.tracks_dir, + ) async def stop(self) -> None: if not self.recording: