feat: add chunked meeting transcription and logging
This commit is contained in:
+252
-43
@@ -2,8 +2,9 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
@@ -15,7 +16,16 @@ from helpers import summarize_error
|
||||
|
||||
load_dotenv()
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
OPENROUTER_BASE = "https://openrouter.ai/api/v1"
|
||||
TRANSCRIPTION_MODELS = [
|
||||
"openai/gpt-4o-mini-transcribe",
|
||||
"openai/whisper-large-v3",
|
||||
]
|
||||
TRANSCRIPTION_CHUNK_SECONDS = 600
|
||||
MAX_TRANSCRIPTION_FILE_BYTES = 25 * 1024 * 1024
|
||||
SUMMARY_TRANSCRIPT_CHAR_LIMIT = 12000
|
||||
|
||||
|
||||
def _api_key() -> str:
|
||||
@@ -36,12 +46,6 @@ def _audio_format(audio_path: str) -> str:
|
||||
return suffix or "wav"
|
||||
|
||||
|
||||
TRANSCRIPTION_MODELS = [
|
||||
"openai/gpt-4o-mini-transcribe",
|
||||
"openai/whisper-large-v3",
|
||||
]
|
||||
|
||||
|
||||
def _build_transcription_payload(audio_path: str, model: str) -> dict[str, Any]:
|
||||
with open(audio_path, "rb") as audio_file:
|
||||
encoded = base64.b64encode(audio_file.read()).decode("ascii")
|
||||
@@ -55,6 +59,39 @@ def _build_transcription_payload(audio_path: str, model: str) -> dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
def _chunk_text(text: str, limit: int = SUMMARY_TRANSCRIPT_CHAR_LIMIT) -> list[str]:
|
||||
clean = text.strip()
|
||||
if len(clean) <= limit:
|
||||
return [clean]
|
||||
|
||||
chunks: list[str] = []
|
||||
remaining = clean
|
||||
while remaining:
|
||||
if len(remaining) <= limit:
|
||||
chunks.append(remaining)
|
||||
break
|
||||
|
||||
split_at = remaining.rfind("\n", 0, limit)
|
||||
separator_length = 1
|
||||
if split_at <= 0:
|
||||
split_at = remaining.rfind(" ", 0, limit)
|
||||
separator_length = 1
|
||||
if split_at <= 0:
|
||||
split_at = limit
|
||||
separator_length = 0
|
||||
|
||||
chunk = remaining[:split_at].strip()
|
||||
if chunk:
|
||||
chunks.append(chunk)
|
||||
remaining = remaining[split_at + separator_length :].lstrip()
|
||||
|
||||
return chunks
|
||||
|
||||
|
||||
def _file_size_mb(path: str) -> float:
|
||||
return os.path.getsize(path) / 1024 / 1024
|
||||
|
||||
|
||||
async def _normalize_audio_for_transcription(audio_path: str) -> str:
|
||||
source = Path(audio_path)
|
||||
fd, normalized_path = tempfile.mkstemp(prefix="meeting-normalized-", suffix=".wav")
|
||||
@@ -85,64 +122,161 @@ async def _normalize_audio_for_transcription(audio_path: str) -> str:
|
||||
"Audio normalization failed: " + (stderr.decode("utf-8", errors="replace").strip() or f"ffmpeg exited {proc.returncode}")
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"Normalized audio for transcription: path=%s size_mib=%.2f",
|
||||
normalized_path,
|
||||
_file_size_mb(normalized_path),
|
||||
)
|
||||
return normalized_path
|
||||
|
||||
|
||||
async def _split_audio_for_transcription(audio_path: str) -> list[str]:
|
||||
chunk_dir = tempfile.mkdtemp(prefix="meeting-chunks-")
|
||||
chunk_pattern = str(Path(chunk_dir) / "chunk-%03d.wav")
|
||||
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-i",
|
||||
audio_path,
|
||||
"-f",
|
||||
"segment",
|
||||
"-segment_time",
|
||||
str(TRANSCRIPTION_CHUNK_SECONDS),
|
||||
"-c:a",
|
||||
"pcm_s16le",
|
||||
chunk_pattern,
|
||||
stdout=asyncio.subprocess.DEVNULL,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
_, stderr = await proc.communicate()
|
||||
if proc.returncode != 0:
|
||||
shutil.rmtree(chunk_dir, ignore_errors=True)
|
||||
raise RuntimeError(
|
||||
"Audio chunking failed: " + (stderr.decode("utf-8", errors="replace").strip() or f"ffmpeg exited {proc.returncode}")
|
||||
)
|
||||
|
||||
chunk_paths = sorted(str(path) for path in Path(chunk_dir).glob("chunk-*.wav"))
|
||||
if not chunk_paths:
|
||||
shutil.rmtree(chunk_dir, ignore_errors=True)
|
||||
raise RuntimeError("Audio chunking produced no output files")
|
||||
|
||||
for chunk_path in chunk_paths:
|
||||
size_bytes = os.path.getsize(chunk_path)
|
||||
if size_bytes > MAX_TRANSCRIPTION_FILE_BYTES:
|
||||
shutil.rmtree(chunk_dir, ignore_errors=True)
|
||||
raise RuntimeError(
|
||||
f"Chunk {Path(chunk_path).name} is too large for transcription: {size_bytes} bytes"
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"Split normalized audio into %s chunk(s): %s",
|
||||
len(chunk_paths),
|
||||
", ".join(f"{Path(path).name} ({_file_size_mb(path):.2f} MiB)" for path in chunk_paths),
|
||||
)
|
||||
return chunk_paths
|
||||
|
||||
|
||||
async def _transcribe_chunk(client: httpx.AsyncClient, audio_path: str, headers: dict[str, str]) -> str:
|
||||
failures: list[str] = []
|
||||
for model in TRANSCRIPTION_MODELS:
|
||||
resp = await client.post(
|
||||
f"{OPENROUTER_BASE}/audio/transcriptions",
|
||||
headers=headers,
|
||||
json=_build_transcription_payload(audio_path, model),
|
||||
)
|
||||
try:
|
||||
resp.raise_for_status()
|
||||
except httpx.HTTPStatusError:
|
||||
detail = summarize_error(_safe_json(resp), fallback=resp.text)
|
||||
generation_id = resp.headers.get("x-generation-id")
|
||||
suffix = f"; generation_id={generation_id}" if generation_id else ""
|
||||
failures.append(f"{model}: HTTP {resp.status_code}: {detail}{suffix}")
|
||||
logger.warning(
|
||||
"Transcription model failed: model=%s chunk=%s status=%s detail=%s generation_id=%s",
|
||||
model,
|
||||
Path(audio_path).name,
|
||||
resp.status_code,
|
||||
detail,
|
||||
generation_id,
|
||||
)
|
||||
continue
|
||||
|
||||
data = resp.json()
|
||||
text = data.get("text", "")
|
||||
if text.strip():
|
||||
logger.info(
|
||||
"Transcription chunk completed: chunk=%s model=%s transcript_chars=%s",
|
||||
Path(audio_path).name,
|
||||
model,
|
||||
len(text.strip()),
|
||||
)
|
||||
return text.strip()
|
||||
|
||||
failures.append(f"{model}: returned empty text")
|
||||
logger.warning(
|
||||
"Transcription model returned empty text: model=%s chunk=%s",
|
||||
model,
|
||||
Path(audio_path).name,
|
||||
)
|
||||
|
||||
raise RuntimeError("OpenRouter transcription failed across all models: " + " | ".join(failures))
|
||||
|
||||
|
||||
async def transcribe(audio_path: str) -> str:
|
||||
"""Send audio to OpenRouter STT models and return transcript text."""
|
||||
logger.info(
|
||||
"Starting transcription pipeline: source=%s size_mib=%.2f",
|
||||
audio_path,
|
||||
_file_size_mb(audio_path),
|
||||
)
|
||||
headers = _auth_headers()
|
||||
headers["Content-Type"] = "application/json"
|
||||
headers["X-OpenRouter-Title"] = "discord-meeting-bot"
|
||||
|
||||
normalized_path = await _normalize_audio_for_transcription(audio_path)
|
||||
failures: list[str] = []
|
||||
chunk_paths: list[str] = []
|
||||
try:
|
||||
chunk_paths = await _split_audio_for_transcription(normalized_path)
|
||||
logger.info("Split normalized audio into %s chunk(s)", len(chunk_paths))
|
||||
transcripts: list[str] = []
|
||||
async with httpx.AsyncClient(timeout=300) as client:
|
||||
for model in TRANSCRIPTION_MODELS:
|
||||
resp = await client.post(
|
||||
f"{OPENROUTER_BASE}/audio/transcriptions",
|
||||
headers=headers,
|
||||
json=_build_transcription_payload(normalized_path, model),
|
||||
for index, chunk_path in enumerate(chunk_paths, start=1):
|
||||
logger.info(
|
||||
"Submitting chunk for transcription: index=%s/%s chunk=%s size_mib=%.2f",
|
||||
index,
|
||||
len(chunk_paths),
|
||||
Path(chunk_path).name,
|
||||
_file_size_mb(chunk_path),
|
||||
)
|
||||
try:
|
||||
resp.raise_for_status()
|
||||
except httpx.HTTPStatusError:
|
||||
detail = summarize_error(_safe_json(resp), fallback=resp.text)
|
||||
generation_id = resp.headers.get("x-generation-id")
|
||||
suffix = f"; generation_id={generation_id}" if generation_id else ""
|
||||
failures.append(f"{model}: HTTP {resp.status_code}: {detail}{suffix}")
|
||||
continue
|
||||
transcripts.append(await _transcribe_chunk(client, chunk_path, headers))
|
||||
|
||||
data = resp.json()
|
||||
text = data.get("text", "")
|
||||
if text.strip():
|
||||
return text.strip()
|
||||
failures.append(f"{model}: returned empty text")
|
||||
combined = "\n\n".join(part for part in transcripts if part.strip()).strip()
|
||||
if not combined:
|
||||
raise RuntimeError("Transcription completed but produced no text")
|
||||
|
||||
raise RuntimeError(
|
||||
"OpenRouter transcription failed across all models: " + " | ".join(failures)
|
||||
logger.info(
|
||||
"Completed transcription pipeline: chunks=%s transcript_chars=%s",
|
||||
len(chunk_paths),
|
||||
len(combined),
|
||||
)
|
||||
return combined
|
||||
finally:
|
||||
for chunk_path in chunk_paths:
|
||||
try:
|
||||
os.remove(chunk_path)
|
||||
except OSError:
|
||||
pass
|
||||
chunk_dirs = {str(Path(chunk_path).parent) for chunk_path in chunk_paths}
|
||||
for chunk_dir in chunk_dirs:
|
||||
shutil.rmtree(chunk_dir, ignore_errors=True)
|
||||
try:
|
||||
os.remove(normalized_path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
async def summarize(transcript: str) -> str:
|
||||
"""Send transcript to DeepSeek via OpenRouter and return structured meeting summary."""
|
||||
prompt = f"""You are a meeting summarizer. Given the following meeting transcript, produce a concise, well-structured summary with:
|
||||
|
||||
1. **Overview** — 2-3 sentences covering what was discussed
|
||||
2. **Key Decisions** — decisions that were made
|
||||
3. **Action Items** — who needs to do what (use bullet points)
|
||||
4. **Next Steps / Deadlines** — any dates, timelines, or follow-ups mentioned
|
||||
|
||||
If speaker labels are present, preserve them where helpful in action items.
|
||||
|
||||
Transcript:
|
||||
{transcript[:16000]}"""
|
||||
|
||||
async def _chat_completion(prompt: str, *, timeout: int = 180) -> str:
|
||||
body = {
|
||||
"model": "@preset/cheap-deepseek-v4-flash",
|
||||
"messages": [{"role": "user", "content": prompt}],
|
||||
@@ -152,7 +286,7 @@ Transcript:
|
||||
headers = _auth_headers()
|
||||
headers["Content-Type"] = "application/json"
|
||||
|
||||
async with httpx.AsyncClient(timeout=180) as client:
|
||||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||||
resp = await client.post(
|
||||
f"{OPENROUTER_BASE}/chat/completions",
|
||||
headers=headers,
|
||||
@@ -162,6 +296,7 @@ Transcript:
|
||||
resp.raise_for_status()
|
||||
except httpx.HTTPStatusError as exc:
|
||||
detail = summarize_error(_safe_json(resp), fallback=resp.text)
|
||||
logger.error("OpenRouter summarization failed: %s", detail)
|
||||
raise RuntimeError(f"OpenRouter summarization failed: {detail}") from exc
|
||||
|
||||
data = resp.json()
|
||||
@@ -175,6 +310,80 @@ Transcript:
|
||||
return content.strip()
|
||||
|
||||
|
||||
def _summary_prompt(transcript: str) -> str:
|
||||
return f"""You are a meeting summarizer. Given the following meeting transcript, produce a concise, well-structured summary with:
|
||||
|
||||
1. **Overview** — 2-3 sentences covering what was discussed
|
||||
2. **Key Decisions** — decisions that were made
|
||||
3. **Action Items** — who needs to do what (use bullet points)
|
||||
4. **Next Steps / Deadlines** — any dates, timelines, or follow-ups mentioned
|
||||
|
||||
If speaker labels are present, preserve them where helpful in action items.
|
||||
|
||||
Transcript:
|
||||
{transcript}"""
|
||||
|
||||
|
||||
def _intermediate_summary_prompt(chunk: str, index: int, total: int) -> str:
|
||||
return f"""You are preparing notes for a longer meeting summary. Summarize this partial transcript chunk clearly and compactly.
|
||||
|
||||
Focus on:
|
||||
- important discussion points
|
||||
- decisions
|
||||
- action items
|
||||
- deadlines or follow-ups
|
||||
|
||||
Partial meeting transcript chunk {index}/{total}:
|
||||
{chunk}"""
|
||||
|
||||
|
||||
def _combined_summary_prompt(chunk_summaries: list[str]) -> str:
|
||||
merged = "\n\n".join(
|
||||
f"Chunk {index}:\n{summary}" for index, summary in enumerate(chunk_summaries, start=1)
|
||||
)
|
||||
return f"""You are a meeting summarizer. The transcript was summarized in chunks first. Combine them into one final meeting summary with:
|
||||
|
||||
1. **Overview** — 2-3 sentences covering what was discussed
|
||||
2. **Key Decisions** — decisions that were made
|
||||
3. **Action Items** — who needs to do what (use bullet points)
|
||||
4. **Next Steps / Deadlines** — any dates, timelines, or follow-ups mentioned
|
||||
|
||||
Combined chunk summaries:
|
||||
{merged}"""
|
||||
|
||||
|
||||
async def summarize(transcript: str) -> str:
|
||||
"""Summarize transcript text, using multi-stage summarization for long inputs."""
|
||||
clean = transcript.strip()
|
||||
if not clean:
|
||||
raise RuntimeError("Cannot summarize an empty transcript")
|
||||
|
||||
if len(clean) <= SUMMARY_TRANSCRIPT_CHAR_LIMIT:
|
||||
logger.info("Transcript fits single-pass summary limit: transcript_chars=%s", len(clean))
|
||||
return await _chat_completion(_summary_prompt(clean))
|
||||
|
||||
chunks = _chunk_text(clean, limit=SUMMARY_TRANSCRIPT_CHAR_LIMIT)
|
||||
logger.info(
|
||||
"Transcript exceeds single-pass summary limit: transcript_chars=%s chunks=%s limit=%s",
|
||||
len(clean),
|
||||
len(chunks),
|
||||
SUMMARY_TRANSCRIPT_CHAR_LIMIT,
|
||||
)
|
||||
|
||||
chunk_summaries: list[str] = []
|
||||
for index, chunk in enumerate(chunks, start=1):
|
||||
logger.info(
|
||||
"Summarizing transcript chunk: index=%s/%s chunk_chars=%s",
|
||||
index,
|
||||
len(chunks),
|
||||
len(chunk),
|
||||
)
|
||||
chunk_summaries.append(await _chat_completion(_intermediate_summary_prompt(chunk, index, len(chunks))))
|
||||
|
||||
logger.info("Combining %s intermediate chunk summaries into final meeting summary", len(chunk_summaries))
|
||||
return await _chat_completion(_combined_summary_prompt(chunk_summaries))
|
||||
|
||||
|
||||
def _safe_json(resp: httpx.Response) -> Any:
|
||||
try:
|
||||
return resp.json()
|
||||
|
||||
Reference in New Issue
Block a user