cli.py

transcription_pipeline/cli.py

1. """transcribe: the command line.
3. transcribe FILE [FILE ...] transcribe files, outputs beside each one (or --out)
4. transcribe inbox process everything waiting in the watch folder
6. Engine: local by default (faster-whisper, nothing leaves the machine), or
7. --engine deepinfra for DeepInfra's hosted Whisper, which uploads the audio.
8. """
10. from __future__ import annotations
12. import argparse
13. import os
14. import sys
15. from pathlib import Path
17. from . import __version__
18. from .render import FORMATS
21. def _formats(value: str) -> tuple[str, ...]:
22. out = tuple(f.strip().lower().lstrip(".") for f in value.split(",") if f.strip())
23. bad = [f for f in out if f not in FORMATS]
24. if bad or not out:
25. raise argparse.ArgumentTypeError(f"formats must be from {', '.join(FORMATS)}")
26. return out
29. def _env_path(name: str) -> Path | None:
30. v = os.environ.get(name, "").strip()
31. return Path(v).expanduser() if v else None
34. def build_parser() -> argparse.ArgumentParser:
35. p = argparse.ArgumentParser(prog="transcribe", description=__doc__,
36. formatter_class=argparse.RawDescriptionHelpFormatter)
37. p.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
38. p.add_argument("targets", nargs="+", metavar="FILE|inbox",
39. help="audio or video files, or the word 'inbox'")
40. p.add_argument("--engine", choices=("local", "deepinfra"),
41. default=os.environ.get("TP_ENGINE", "local"),
42. help="local (default) or deepinfra (hosted; uploads audio)")
43. p.add_argument("--language", default=os.environ.get("TP_LANGUAGE") or None,
44. help="language code such as en or es (default: detect)")
45. p.add_argument("--no-diarize", action="store_true", help="skip speaker separation")
46. p.add_argument("--speakers", type=int, metavar="N",
47. help="exact number of speakers, when you know it")
48. p.add_argument("--min-speakers", type=int, metavar="N")
49. p.add_argument("--max-speakers", type=int, metavar="N")
50. p.add_argument("--formats", type=_formats,
51. default=_formats(os.environ.get("TP_FORMATS", "txt,md,json")),
52. help="comma list from txt,md,json,srt,vtt (default txt,md,json)")
53. p.add_argument("--no-timestamps", action="store_true", help="plain .txt without [HH:MM:SS]")
54. p.add_argument("--out", type=Path, help="output folder for files (default: beside each file)")
55. p.add_argument("--root", type=Path, default=_env_path("TP_ROOT"),
56. help="watch-folder root holding inbox/ transcripts/ done/ failed/ (env TP_ROOT)")
57. p.add_argument("--vault-inbox", type=Path, default=_env_path("TP_VAULT_INBOX"),
58. help="also file the .md transcript here (env TP_VAULT_INBOX)")
59. p.add_argument("--settle", type=float, default=5.0,
60. help="seconds a file's size must hold still before it is taken (inbox)")
61. return p
64. def options_from(args):
65. from .pipeline import Options
66. lo, hi = args.min_speakers, args.max_speakers
67. if args.speakers:
68. lo = hi = args.speakers
69. return Options(engine=args.engine, language=args.language, diarize=not args.no_diarize,
70. min_speakers=lo, max_speakers=hi, formats=args.formats,
71. timestamps=not args.no_timestamps)
74. # pyannote warns at import that torchcodec cannot find FFmpeg. The pipeline
75. # never uses it (audio reaches pyannote already decoded), so it is only noise.
76. # The message starts with a line break, hence the leading \s*.
77. QUIET = [("ignore", r"\s*torchcodec is not installed correctly")]
80. def main(argv: list[str] | None = None) -> int:
81. import warnings
82. for action, message in QUIET:
83. warnings.filterwarnings(action, message=message)
84. args = build_parser().parse_args(argv)
85. from . import steps
86. from .pipeline import process
88. opts = options_from(args)
89. try:
90. transcriber = steps.transcriber(opts)
91. aligner = steps.aligner(opts)
92. diarizer = steps.diarizer(opts)
93. except Exception as e:
94. print(f"transcribe: {e}", file=sys.stderr)
95. return 2
97. def handle(src: Path, out_dir: Path):
98. return process(src, out_dir, opts, transcriber, aligner, diarizer,
99. vault_inbox=args.vault_inbox)
101. if args.targets == ["inbox"]:
102. if args.root is None:
103. print("transcribe inbox: set --root or TP_ROOT to the watch-folder root.",
104. file=sys.stderr)
105. return 2
106. from .inbox import Layout, run
107. layout = Layout(args.root.expanduser())
108. report = run(layout, lambda src: handle(src, layout.transcripts), settle=args.settle)
109. return 1 if report["failed"] else 0
111. failed = 0
112. for t in args.targets:
113. src = Path(t).expanduser()
114. if not src.is_file():
115. print(f"{src}: not a file", file=sys.stderr)
116. failed += 1
117. continue
118. print(f"{src.name} ({opts.engine})")
119. try:
120. res = handle(src, args.out or src.parent)
121. except Exception as e:
122. print(f" FAILED: {e}", file=sys.stderr)
123. failed += 1
124. continue
125. print(f" {res.sentences} sentences, {len(res.speakers)} speakers")
126. for p in res.outputs.values():
127. print(f" -> {p}")
128. return 1 if failed else 0
131. if __name__ == "__main__":
132. raise SystemExit(main())