121 lines
3.1 KiB
Python
121 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Build an enhanced transcript that interleaves audio segments (from a
|
|
whisperx JSON) with screen frames sourced from a cht session's
|
|
frames/index.json. Frames are placed at their real timestamps rather
|
|
than appended at the end.
|
|
|
|
Usage:
|
|
python ctrl/cht/interleave_cht_frames.py \\
|
|
<transcript.json> <cht_frames_index.json> [output.txt]
|
|
"""
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
def fmt_ts(seconds: float) -> str:
|
|
seconds = int(seconds)
|
|
return f"{seconds // 60:02d}:{seconds % 60:02d}"
|
|
|
|
|
|
def load_audio(transcript_json: Path):
|
|
data = json.loads(transcript_json.read_text())
|
|
out = []
|
|
for s in data.get("segments", []):
|
|
out.append({
|
|
"type": "audio",
|
|
"ts": float(s.get("start", 0)),
|
|
"speaker": s.get("speaker"),
|
|
"text": (s.get("text") or "").strip(),
|
|
})
|
|
return out
|
|
|
|
|
|
def load_frames(index_json: Path):
|
|
data = json.loads(index_json.read_text())
|
|
out = []
|
|
for f in data:
|
|
out.append({
|
|
"type": "frame",
|
|
"ts": float(f["timestamp"]),
|
|
"path": f["path"],
|
|
"id": f.get("id", ""),
|
|
})
|
|
return out
|
|
|
|
|
|
def merge(audio, frames):
|
|
events = sorted(audio + frames, key=lambda e: e["ts"])
|
|
grouped = []
|
|
cur = None
|
|
for e in events:
|
|
if e["type"] == "frame":
|
|
if cur:
|
|
grouped.append(cur)
|
|
cur = None
|
|
grouped.append(e)
|
|
else:
|
|
if cur and cur.get("speaker") == e["speaker"]:
|
|
cur["text"] += " " + e["text"]
|
|
else:
|
|
if cur:
|
|
grouped.append(cur)
|
|
cur = {
|
|
"type": "audio",
|
|
"ts": e["ts"],
|
|
"speaker": e["speaker"],
|
|
"text": e["text"],
|
|
}
|
|
if cur:
|
|
grouped.append(cur)
|
|
return grouped
|
|
|
|
|
|
def render(grouped):
|
|
lines = [
|
|
"=" * 80,
|
|
"ENHANCED MEETING TRANSCRIPT",
|
|
"Audio transcript + Screen frames (interleaved by timestamp)",
|
|
"=" * 80,
|
|
"",
|
|
]
|
|
for e in grouped:
|
|
ts = fmt_ts(e["ts"])
|
|
if e["type"] == "frame":
|
|
lines.append(f"[{ts}] SCREEN CONTENT:")
|
|
lines.append(f" Frame: {e['path']}")
|
|
else:
|
|
spk = e["speaker"] or "UNKNOWN"
|
|
lines.append(f"[{ts}] {spk}:")
|
|
lines.append(f" {e['text']}")
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 3:
|
|
print(__doc__, file=sys.stderr)
|
|
sys.exit(2)
|
|
transcript = Path(sys.argv[1])
|
|
frames_index = Path(sys.argv[2])
|
|
output = Path(sys.argv[3]) if len(sys.argv) >= 4 else None
|
|
|
|
audio = load_audio(transcript)
|
|
frames = load_frames(frames_index)
|
|
grouped = merge(audio, frames)
|
|
|
|
text = render(grouped)
|
|
if output:
|
|
output.write_text(text)
|
|
print(f"Wrote {output}")
|
|
print(f" audio segments in : {len(audio)}")
|
|
print(f" frames in : {len(frames)}")
|
|
print(f" output blocks : {len(grouped)}")
|
|
else:
|
|
sys.stdout.write(text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|