inbox.py

transcription_pipeline/inbox.py

1. """The watch-folder run: everything waiting in the inbox, one file at a time.
3. launchd starts `transcribe inbox` whenever the inbox changes (WatchPaths). A
4. run takes a lock, so a second trigger while one is working exits at once; the
5. next change after it finishes starts the next run. A file is only taken when
6. its size holds still across a short settle window, so a recording that is
7. still being copied or downloaded is left for the next run.
9. <root>/inbox/ drop audio or video here
10. <root>/transcripts/ .txt / .md / .json (+ .srt / .vtt)
11. <root>/done/ the source, after success
12. <root>/failed/ the source, plus <name>.error.txt, after failure
13. """
15. from __future__ import annotations
17. import os
18. import time
19. import traceback
20. from dataclasses import dataclass
21. from pathlib import Path
23. from .audio import AUDIO_EXTS
24. from .filing import move_unique, write_unique
27. @dataclass
28. class Layout:
29. root: Path
31. @property
32. def inbox(self) -> Path:
33. return self.root / "inbox"
35. @property
36. def transcripts(self) -> Path:
37. return self.root / "transcripts"
39. @property
40. def done(self) -> Path:
41. return self.root / "done"
43. @property
44. def failed(self) -> Path:
45. return self.root / "failed"
47. @property
48. def lock(self) -> Path:
49. return self.root / ".transcribe.lock"
51. def ensure(self):
52. for d in (self.inbox, self.transcripts, self.done, self.failed):
53. d.mkdir(parents=True, exist_ok=True)
56. class Lock:
57. """A pid lock file. A lock left by a process that no longer exists is stale."""
59. def __init__(self, path: Path):
60. self.path = path
61. self.held = False
63. def acquire(self) -> bool:
64. try:
65. fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
66. except FileExistsError:
67. try:
68. pid = int(self.path.read_text().strip() or 0)
69. os.kill(pid, 0)
70. return False # a live run holds it
71. except (ValueError, ProcessLookupError):
72. self.path.unlink(missing_ok=True) # stale: take it over
73. return self.acquire()
74. except PermissionError:
75. return False
76. with os.fdopen(fd, "w") as f:
77. f.write(str(os.getpid()))
78. self.held = True
79. return True
81. def release(self):
82. if self.held:
83. self.path.unlink(missing_ok=True)
84. self.held = False
87. def candidates(inbox: Path) -> list[Path]:
88. """Audio and video files by extension. Partial downloads (.part, .crdownload,
89. .download) and hidden files never match, so they wait for the next run."""
90. out = []
91. for p in sorted(inbox.iterdir()):
92. if not p.is_file() or p.name.startswith("."):
93. continue
94. if p.suffix.lower() in AUDIO_EXTS:
95. out.append(p)
96. return out
99. def stable(paths: list[Path], settle: float) -> list[Path]:
100. """Keep only files whose size did not change across `settle` seconds."""
101. before = {p: p.stat().st_size for p in paths if p.exists()}
102. if settle > 0:
103. time.sleep(settle)
104. return [p for p, size in before.items() if p.exists() and p.stat().st_size == size and size > 0]
107. def run(layout: Layout, handle, settle: float = 5.0, log=print) -> dict:
108. """handle(path) -> pipeline.Result.
110. Returns {"done": [...], "failed": [...], "skipped": reason}.
111. """
112. layout.ensure()
113. lock = Lock(layout.lock)
114. if not lock.acquire():
115. log("inbox: another run is in progress; exiting.")
116. return {"done": [], "failed": [], "skipped": "locked"}
117. report = {"done": [], "failed": [], "skipped": None}
118. try:
119. while True:
120. ready = stable(candidates(layout.inbox), settle)
121. ready = [p for p in ready if p.name not in {*report["done"], *report["failed"]}]
122. if not ready:
123. break
124. for src in ready:
125. log(f"inbox: {src.name}")
126. try:
127. res = handle(src)
128. move_unique(src, layout.done)
129. report["done"].append(src.name)
130. names = ", ".join(str(p.name) for p in res.outputs.values())
131. log(f" ok: {res.sentences} sentences, {len(res.speakers)} speakers, "
132. f"{res.duration:.0f}s -> {names}")
133. except Exception as e: # one bad file must not stop the rest
134. moved = move_unique(src, layout.failed)
135. write_unique(layout.failed, f"{moved.stem}.error.txt",
136. f"{type(e).__name__}: {e}\n\n{traceback.format_exc()}")
137. report["failed"].append(src.name)
138. log(f" FAILED: {e}")
139. finally:
140. lock.release()
141. if not report["done"] and not report["failed"]:
142. log("inbox: nothing ready.")
143. return report