render.py

transcription_pipeline/render.py

1. """Writers: .txt (one timestamped line per sentence), .md for the vault,
2. .json with everything, and .srt/.vtt subtitles."""
4. from __future__ import annotations
6. import json
8. from .shape import turns
10. FORMATS = ("txt", "md", "json", "srt", "vtt")
13. def _pad(n: int, w: int = 2) -> str:
14. return str(n).zfill(w)
17. def stamp(seconds: float) -> str:
18. """[HH:MM:SS], floored."""
19. total = max(0, int(seconds))
20. return f"[{_pad(total // 3600)}:{_pad(total // 60 % 60)}:{_pad(total % 60)}]"
23. def _clock(seconds: float, sep: str) -> str:
24. ms_total = max(0, round(seconds * 1000))
25. ms, s = ms_total % 1000, ms_total // 1000
26. return f"{_pad(s // 3600)}:{_pad(s // 60 % 60)}:{_pad(s % 60)}{sep}{_pad(ms, 3)}"
29. def to_txt(sentences: list[dict], timestamps: bool = True) -> str:
30. """One sentence per line: '[00:00:02] Speaker 1: text' (label omitted when unknown)."""
31. lines = []
32. for s in sentences:
33. parts = []
34. if timestamps:
35. parts.append(stamp(s["start"]))
36. parts.append(f"{s['speaker']}: {s['text']}" if s.get("speaker") else s["text"])
37. lines.append(" ".join(parts))
38. return "\n".join(lines) + ("\n" if lines else "")
41. def _yaml_str(v: str) -> str:
42. return json.dumps(v, ensure_ascii=False)
45. def to_markdown(sentences: list[dict], meta: dict) -> str:
46. """A vault note: frontmatter, then speaker turns as paragraphs."""
47. fm = ["---", "type: transcript",
48. f"title: {_yaml_str(meta['title'])}",
49. f"recorded: {meta['recorded']}",
50. f"transcribed: {meta['transcribed']}",
51. f"duration: {_yaml_str(meta['duration'])}",
52. f"engine: {meta['engine']}",
53. f"language: {meta.get('language') or 'unknown'}",
54. f"source: {_yaml_str(meta['source'])}"]
55. if meta.get("speakers"):
56. fm.append("speakers:")
57. fm += [f" - {_yaml_str(sp)}" for sp in meta["speakers"]]
58. fm += ["tags:", " - transcript", "---", ""]
59. body = [f"# {meta['title']}", ""]
60. for t in turns(sentences):
61. head = f"**{t['speaker']}** {stamp(t['start'])}" if t["speaker"] else stamp(t["start"])
62. body += [head, " ".join(s["text"] for s in t["sentences"]), ""]
63. return "\n".join(fm + body).rstrip() + "\n"
66. def to_json(sentences: list[dict], result: dict, meta: dict) -> str:
67. data = {"meta": meta, "sentences": sentences, "segments": result.get("segments", [])}
68. return json.dumps(data, indent=2, ensure_ascii=False, default=str) + "\n"
71. def _cues(sentences: list[dict]):
72. for i, s in enumerate(sentences):
73. end = s["end"]
74. if end <= s["start"]:
75. nxt = sentences[i + 1]["start"] if i + 1 < len(sentences) else s["start"] + 2.0
76. end = max(nxt, s["start"] + 0.5)
77. text = f"{s['speaker']}: {s['text']}" if s.get("speaker") else s["text"]
78. yield s["start"], end, text
81. def to_srt(sentences: list[dict]) -> str:
82. return "\n".join(f"{i}\n{_clock(a, ',')} --> {_clock(b, ',')}\n{t}\n"
83. for i, (a, b, t) in enumerate(_cues(sentences), 1))
86. def to_vtt(sentences: list[dict]) -> str:
87. return "WEBVTT\n\n" + "\n".join(f"{_clock(a, '.')} --> {_clock(b, '.')}\n{t}\n"
88. for a, b, t in _cues(sentences))