#!/usr/bin/env python3
"""
Meeting recording processor.
Takes a video file from OBS, extracts:
- Audio → transcribed via openai-whisper (local, GPU)
- Key frames via scene change detection → delegated to Claude for visual analysis
- Synthesizes structured meeting notes
Usage:
source ~/.venvs/meeting-notes/bin/activate
python ~/meetings/process_meeting.py /path/to/recording.mp4
"""
import argparse
import json
import os
import subprocess
import sys
import tempfile
from datetime import datetime
from pathlib import Path
MEETINGS_DIR = Path(os.environ["HOME"]) / "meetings"
INCOMING = MEETINGS_DIR / "incoming"
PROCESSED = MEETINGS_DIR / "processed"
NOTES = MEETINGS_DIR / "notes"
def run(cmd: str, **kwargs):
"""Run a shell command, raise on failure."""
result = subprocess.run(
cmd, shell=True, capture_output=True, text=True, **kwargs
)
if result.returncode != 0:
print(f"Command failed: {cmd}")
print(f"STDERR: {result.stderr}")
sys.exit(1)
return result
def extract_audio(video_path: str, output_wav: str):
"""Extract audio from video to 16kHz mono WAV."""
run(f'ffmpeg -y -i "{video_path}" -vn -acodec pcm_s16le -ar 16000 -ac 1 "{output_wav}"')
print(f" ✓ Audio extracted: {output_wav}")
def get_video_duration(video_path: str) -> float:
"""Get video duration in seconds."""
result = run(f'ffprobe -v quiet -show_entries format=duration -of csv=p=0 "{video_path}"')
return float(result.stdout.strip())
def extract_keyframes(video_path: str, output_dir: str, duration: float):
"""
Extract key frames — 1 per 15 seconds.
Simple periodic sampling is more reliable than scene-change heuristics
for meeting recordings where screen changes are gradual.
"""
os.makedirs(output_dir, exist_ok=True)
# Adaptive interval: ~2-3 frames per minute for meetings
frame_interval = 15.0
run(
f'ffmpeg -y -i "{video_path}" -vf "fps=1/{frame_interval}" '
f'"frame_%04d.jpg"'
)
frames = sorted(Path(output_dir).glob("frame_*.jpg"))
print(f" ✓ Frames extracted: {len(frames)}")
return frames
def transcribe_audio(wav_path: str) -> dict:
"""
Transcribe audio using openai-whisper with GPU.
Returns dict with segments and full text.
"""
import whisper
print(" Loading Whisper model (large-v3, CUDA)...")
model = whisper.load_model("large-v3", device="cuda")
print(" Transcribing...")
result = model.transcribe(
wav_path,
beam_size=5,
language=None, # auto-detect
)
print(f" ✓ Detected language: {result.get('language', 'unknown')}")
# Build transcript with timestamps
segments_list = []
full_text_parts = []
for seg in result["segments"]:
start = seg["start"]
end = seg["end"]
text = seg["text"].strip()
mm, ss = divmod(start, 60)
start_str = f"{int(mm):02d}:{int(ss):02d}"
mm, ss = divmod(end, 60)
end_str = f"{int(mm):02d}:{int(ss):02d}"
segments_list.append({
"start": start_str,
"end": end_str,
"text": text,
})
full_text_parts.append(text)
return {
"language": result.get("language", "unknown"),
"segments": segments_list,
"full_text": " ".join(full_text_parts),
}
def build_vision_prompt(frames: list, transcript: str) -> str:
"""Build prompt for Claude to analyze screen frames."""
return f"""You are analyzing screenshots from a meeting where someone was demoing applications and integrations.
## Meeting Transcript
{transcript}
## Screenshots
Below are {len(frames)} screenshots captured at various points during the meeting. For each one, describe:
1. What application/interface is visible
2. What the user appears to be doing or demonstrating
3. Any notable integration points, data flows, or UI interactions
4. Any issues, bugs, or points of confusion visible
Be specific about what you see on screen. Reference timestamps from the transcript where relevant.
Return your analysis as a numbered list matching the frame order."""
def synthesize_notes(transcript: dict, frame_analysis: str, video_filename: str) -> str:
"""Build structured meeting notes from transcript and visual analysis."""
now = datetime.now().strftime("%Y-%m-%d %H:%M")
notes = f"""MEETING NOTES
Generated: {now}
Source: {video_filename}
Language: {transcript.get('language', 'unknown')}
═══════════════════════════════════════════
TRANSCRIPT
───────────────────────────────────────────
"""
for seg in transcript["segments"]:
notes += f'**[{seg["start"]}]** {seg["text"]}\n\n'
notes += """
═══════════════════════════════════════════
SCREEN ACTIVITY
───────────────────────────────────────────
"""
notes += frame_analysis
notes += """
═══════════════════════════════════════════
SUMMARY
───────────────────────────────────────────
[To be completed — review transcript and screen activity above, then provide:
• Key discussion points
• Decisions made
• Action items with owners
• Questions raised / unanswered
• Technical observations about integrations demonstrated]
"""
return notes
def process_meeting(video_path: str):
"""Main processing pipeline."""
video_path = Path(video_path).resolve()
if not video_path.exists():
print(f"Error: File not found: {video_path}")
sys.exit(1)
print(f"\n{'='*50}")
print(f"Processing: {video_path.name}")
print(f"{'='*50}\n")
# Setup output dirs
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
work_dir = PROCESSED / timestamp
os.makedirs(work_dir / "frames", exist_ok=True)
# Get duration
print("1. Getting video duration...")
duration = get_video_duration(str(video_path))
print(f" Duration: {duration:.0f}s ({duration/60:.1f} minutes)")
# Extract audio
print("2. Extracting audio...")
wav_path = str(work_dir / "audio.wav")
extract_audio(str(video_path), wav_path)
# Transcribe
print("3. Transcribing audio (Whisper large-v3, GPU)...")
transcript = transcribe_audio(wav_path)
print(f" Transcript: {len(transcript['segments'])} segments")
# Extract frames
print("4. Extracting key frames...")
frames = extract_keyframes(
str(video_path),
str(work_dir / "frames"),
duration,
)
# Build vision prompt
print("5. Preparing frame analysis prompt...")
vision_prompt = build_vision_prompt(frames, transcript["full_text"])
# Save intermediate results
transcript_path = work_dir / "transcript.json"
with open(transcript_path, "w") as f:
json.dump(transcript, f, indent=2)
print(f" Transcript saved: {transcript_path}")
# Save vision prompt for manual delegation
prompt_path = work_dir / "vision_prompt.txt"
with open(prompt_path, "w") as f:
f.write(vision_prompt)
print(f" Vision prompt saved: {prompt_path}")
# List frames for the user
frames_list = "\n".join(f" • {f.name}" for f in frames[:10])
if len(frames) > 10:
frames_list += f"\n ... and {len(frames) - 10} more"
print(f"\n Frames ({len(frames)}):")
print(frames_list)
# Draft notes (without frame analysis — that comes from Claude delegation)
print("\n6. Drafting initial notes...")
draft = synthesize_notes(
transcript,
"[Frame analysis pending — delegate frames to Claude]",
video_path.name,
)
notes_path = NOTES / f"{timestamp}_{video_path.stem}_draft.md"
with open(notes_path, "w") as f:
f.write(draft)
print(f" Draft notes: {notes_path}")
# Report size of WAV for cleanup
wav_size_mb = os.path.getsize(wav_path) / 1024 / 1024
print("\n" + "="*50)
print("PROCESSING COMPLETE")
print("="*50)
print(f"""
Next steps:
1. Delegate the frame analysis:
- {len(frames)} frames at: {work_dir}/frames/
- Vision prompt at: {prompt_path}
2. Once Claude returns frame analysis, finalize notes:
- Draft at: {notes_path}
3. Optional cleanup:
rm {wav_path} # saves ~{wav_size_mb:.0f}MB
""")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Process meeting recording")
parser.add_argument("video", help="Path to video file")
args = parser.parse_args()
process_meeting(args.video)