618 lines
20 KiB
Python
618 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from dotenv import load_dotenv
|
|
|
|
from helpers import summarize_error
|
|
|
|
load_dotenv()
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
OPENROUTER_BASE = "https://openrouter.ai/api/v1"
|
|
OPENAI_BASE = "https://api.openai.com/v1"
|
|
TRANSCRIPTION_MODELS = [
|
|
"openai/gpt-4o-transcribe",
|
|
"openai/whisper-large-v3",
|
|
]
|
|
DIRECT_OPENAI_TRANSCRIPTION_MODELS = [
|
|
"gpt-4o-transcribe",
|
|
"whisper-1",
|
|
]
|
|
TRANSCRIPTION_CHUNK_SECONDS = 600
|
|
MAX_TRANSCRIPTION_FILE_BYTES = 25 * 1024 * 1024
|
|
SUMMARY_TRANSCRIPT_CHAR_LIMIT = 12000
|
|
DIRECT_OPENAI_SUMMARY_MODEL = "gpt-4o-mini"
|
|
|
|
|
|
def _api_key() -> str:
|
|
return os.getenv("OPENROUTER_API_KEY", "")
|
|
|
|
|
|
def _openai_api_key() -> str:
|
|
return os.getenv("OPENAI_API_KEY", "")
|
|
|
|
|
|
def _auth_headers() -> dict[str, str]:
|
|
api_key = _api_key()
|
|
if not api_key:
|
|
raise ValueError("OPENROUTER_API_KEY not set")
|
|
return {
|
|
"Authorization": f"Bearer {api_key}",
|
|
}
|
|
|
|
|
|
def _audio_format(audio_path: str) -> str:
|
|
suffix = Path(audio_path).suffix.lower().lstrip(".")
|
|
return suffix or "wav"
|
|
|
|
|
|
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")
|
|
|
|
return {
|
|
"model": model,
|
|
"input_audio": {
|
|
"data": encoded,
|
|
"format": _audio_format(audio_path),
|
|
},
|
|
}
|
|
|
|
|
|
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")
|
|
os.close(fd)
|
|
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"ffmpeg",
|
|
"-y",
|
|
"-i",
|
|
str(source),
|
|
"-ac",
|
|
"1",
|
|
"-ar",
|
|
"16000",
|
|
"-c:a",
|
|
"pcm_s16le",
|
|
normalized_path,
|
|
stdout=asyncio.subprocess.DEVNULL,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
_, stderr = await proc.communicate()
|
|
if proc.returncode != 0:
|
|
try:
|
|
os.remove(normalized_path)
|
|
except OSError:
|
|
pass
|
|
raise RuntimeError(
|
|
"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_with_openrouter(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: provider=openrouter 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: provider=openrouter 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: provider=openrouter model=%s chunk=%s",
|
|
model,
|
|
Path(audio_path).name,
|
|
)
|
|
|
|
raise RuntimeError("OpenRouter transcription failed across all models: " + " | ".join(failures))
|
|
|
|
|
|
async def _transcribe_chunk_with_openai(client: httpx.AsyncClient, audio_path: str) -> str:
|
|
api_key = _openai_api_key()
|
|
if not api_key:
|
|
raise RuntimeError("OPENAI_API_KEY not set")
|
|
|
|
failures: list[str] = []
|
|
headers = {"Authorization": f"Bearer {api_key}"}
|
|
for model in DIRECT_OPENAI_TRANSCRIPTION_MODELS:
|
|
with open(audio_path, "rb") as audio_file:
|
|
files = {"file": (Path(audio_path).name, audio_file, "audio/wav")}
|
|
data = {"model": model}
|
|
resp = await client.post(
|
|
f"{OPENAI_BASE}/audio/transcriptions",
|
|
headers=headers,
|
|
data=data,
|
|
files=files,
|
|
)
|
|
try:
|
|
resp.raise_for_status()
|
|
except httpx.HTTPStatusError:
|
|
detail = summarize_error(_safe_json(resp), fallback=resp.text)
|
|
request_id = resp.headers.get("x-request-id")
|
|
suffix = f"; request_id={request_id}" if request_id else ""
|
|
failures.append(f"{model}: HTTP {resp.status_code}: {detail}{suffix}")
|
|
logger.warning(
|
|
"Transcription model failed: provider=openai model=%s chunk=%s status=%s detail=%s request_id=%s",
|
|
model,
|
|
Path(audio_path).name,
|
|
resp.status_code,
|
|
detail,
|
|
request_id,
|
|
)
|
|
continue
|
|
|
|
data = resp.json()
|
|
text = data.get("text", "")
|
|
if text.strip():
|
|
logger.info(
|
|
"Transcription chunk completed: provider=openai 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: provider=openai model=%s chunk=%s",
|
|
model,
|
|
Path(audio_path).name,
|
|
)
|
|
|
|
raise RuntimeError("Direct OpenAI transcription failed across all models: " + " | ".join(failures))
|
|
|
|
|
|
async def _transcribe_chunk(client: httpx.AsyncClient, audio_path: str, headers: dict[str, str]) -> str:
|
|
failures: list[str] = []
|
|
try:
|
|
return await _transcribe_chunk_with_openrouter(client, audio_path, headers)
|
|
except RuntimeError as exc:
|
|
failures.append(str(exc))
|
|
logger.warning(
|
|
"OpenRouter transcription provider failed for chunk=%s; checking direct OpenAI fallback",
|
|
Path(audio_path).name,
|
|
)
|
|
|
|
if _openai_api_key():
|
|
try:
|
|
return await _transcribe_chunk_with_openai(client, audio_path)
|
|
except RuntimeError as exc:
|
|
failures.append(str(exc))
|
|
else:
|
|
failures.append("direct OpenAI fallback skipped: OPENAI_API_KEY not set")
|
|
|
|
raise RuntimeError("Transcription failed across providers: " + " | ".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)
|
|
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 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),
|
|
)
|
|
transcripts.append(await _transcribe_chunk(client, chunk_path, headers))
|
|
|
|
combined = "\n\n".join(part for part in transcripts if part.strip()).strip()
|
|
if not combined:
|
|
raise RuntimeError("Transcription completed but produced no text")
|
|
|
|
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 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"]
|
|
except (KeyError, IndexError, TypeError) as exc:
|
|
raise RuntimeError(f"Unexpected {provider} summarization response shape") from exc
|
|
|
|
if not isinstance(content, str) or not content.strip():
|
|
raise RuntimeError(f"{provider} summarization returned empty content")
|
|
return content.strip()
|
|
|
|
|
|
async def _chat_completion_openrouter(prompt: str, *, timeout: int = 180) -> str:
|
|
body = {
|
|
"model": "@preset/cheap-deepseek-v4-flash",
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
"max_tokens": 1500,
|
|
}
|
|
|
|
headers = _auth_headers()
|
|
headers["Content-Type"] = "application/json"
|
|
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
resp = await client.post(
|
|
f"{OPENROUTER_BASE}/chat/completions",
|
|
headers=headers,
|
|
json=body,
|
|
)
|
|
try:
|
|
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()
|
|
return _extract_chat_content(data, provider="OpenRouter")
|
|
|
|
|
|
async def _chat_completion_openai(prompt: str, *, timeout: int = 180) -> str:
|
|
api_key = _openai_api_key()
|
|
if not api_key:
|
|
raise RuntimeError("OPENAI_API_KEY not set")
|
|
|
|
body = {
|
|
"model": DIRECT_OPENAI_SUMMARY_MODEL,
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
"max_tokens": 1500,
|
|
}
|
|
headers = {
|
|
"Authorization": f"Bearer {api_key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
resp = await client.post(
|
|
f"{OPENAI_BASE}/chat/completions",
|
|
headers=headers,
|
|
json=body,
|
|
)
|
|
try:
|
|
resp.raise_for_status()
|
|
except httpx.HTTPStatusError as exc:
|
|
detail = summarize_error(_safe_json(resp), fallback=resp.text)
|
|
request_id = resp.headers.get("x-request-id")
|
|
suffix = f"; request_id={request_id}" if request_id else ""
|
|
logger.error("Direct OpenAI summarization failed: %s%s", detail, suffix)
|
|
raise RuntimeError(f"Direct OpenAI summarization failed: {detail}{suffix}") from exc
|
|
|
|
data = resp.json()
|
|
return _extract_chat_content(data, provider="Direct OpenAI")
|
|
|
|
|
|
async def _chat_completion(prompt: str, *, timeout: int = 180) -> str:
|
|
failures: list[str] = []
|
|
try:
|
|
return await _chat_completion_openrouter(prompt, timeout=timeout)
|
|
except RuntimeError as exc:
|
|
failures.append(str(exc))
|
|
logger.warning("OpenRouter summarization failed; checking direct OpenAI fallback: %s", exc)
|
|
|
|
if _openai_api_key():
|
|
try:
|
|
return await _chat_completion_openai(prompt, timeout=timeout)
|
|
except RuntimeError as exc:
|
|
failures.append(str(exc))
|
|
else:
|
|
failures.append("direct OpenAI summarization fallback skipped: OPENAI_API_KEY not set")
|
|
|
|
raise RuntimeError("Summarization failed across providers: " + " | ".join(failures))
|
|
|
|
|
|
def _summary_prompt(transcript: str) -> str:
|
|
return f"""You are a meeting summarizer. Given the following meeting transcript, produce a concise, well-structured summary.
|
|
|
|
Use exactly these markdown section headings, in this order:
|
|
|
|
## What Was Done
|
|
- Summarize the work completed, progress reported, and status updates.
|
|
- Include all dev team members' reports that appear in the transcript; do not omit any named person.
|
|
|
|
## What Will Be Done Next
|
|
- Summarize the upcoming work, planned follow-ups, and expected next steps.
|
|
|
|
## Action Items
|
|
- Use bullet points.
|
|
- For each item, include the owner when it is mentioned.
|
|
|
|
## Decisions
|
|
- List decisions that were made.
|
|
- If no decisions were made, say "- None recorded."
|
|
|
|
If speaker labels are present, preserve them where helpful.
|
|
Do not add extra top-level sections.
|
|
|
|
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.
|
|
|
|
Use exactly these markdown section headings, in this order:
|
|
|
|
## What Was Done
|
|
- Summarize the work completed, progress reported, and status updates.
|
|
- Include all dev team members' reports that appear in the chunk summaries; do not omit any named person.
|
|
|
|
## What Will Be Done Next
|
|
- Summarize the upcoming work, planned follow-ups, and expected next steps.
|
|
|
|
## Action Items
|
|
- Use bullet points.
|
|
- For each item, include the owner when it is mentioned.
|
|
|
|
## Decisions
|
|
- List decisions that were made.
|
|
- If no decisions were made, say "- None recorded."
|
|
|
|
Do not add extra top-level sections.
|
|
|
|
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()
|
|
except Exception:
|
|
return None
|