#!/usr/bin/env python3 """ Host-side inbox watcher — convert new audio and/or copy new transcripts. This is a companion to whispermlx-missing, not the in-app G2 directory watcher. Streamlit never executes it. It does not import transcriptx. Optional ``--admit`` subprocesses ``python -m transcriptx.admit_originals`` so new originals/ JSON can enter the managed library (same path as Import Transcript). Default is off: use Import Transcript or Settings → Watcher, or enable admit. The transcripts destination must be ``…/transcripts/originals`` (or another non-library folder). Config that points at the managed library root (the directory that already contains ``metadata/`` / ``imports/``) is rejected. Install: install -m 755 scripts/inbox-watch.py ~/.local/bin/inbox-watch (ensure ~/.local/bin is on PATH) Modes (independent; at least one required): --watch-audio Convert new inbox audio → recordings as 16 kHz mono 64k MP3, then run whispermlx-missing --watch-transcripts Copy new inbox JSON/SRT/VTT/txt/html into transcripts dest when that stem is not already present Preview: inbox-watch --once --dry-run --inbox … --recordings … --transcripts … Normal once (cron / launchd): inbox-watch --once Poll (USB volume may be absent; first cycle still runs missing/admit): inbox-watch --watch inbox-watch --watch --admit inbox-watch --watch --auto-name Config (merge order: portable defaults <- env <- local JSON <- CLI): --config /path/to/config.json or env INBOX_WATCH_CONFIG=/path/to/config.json default: .transcriptx/inbox-watch.json when run from the repo TRANSCRIPTX_RECORDINGS_DIR → recordings TRANSCRIPTX_TRANSCRIPTS_DIR → transcripts dest (script appends /originals) INBOX_WATCH_INBOX → inbox Inbox files are kept by default. After a successful convert/copy you can --backup-wav (copy audio originals into the WAV backup folder), --delete-originals (remove the inbox source), both, or --move-processed DIR. Audio on a removable inbox (USB / ejectable volume) is copied to a local staging folder first, then ffmpeg reads that copy. Override with --stage-local / --no-stage-local (env INBOX_WATCH_STAGE_LOCAL). Default stage dir: {recordings}/.inbox-staging/ (hidden from whispermlx-missing). --skip-serial forwards to whispermlx-missing so split parts / voice-note runs are not transcribed (merge first, then transcribe the merged file). --admit (default off) runs python -m transcriptx.admit_originals after convert/copy/missing so originals/ JSON is admitted into the managed library. Requires a Python that can import transcriptx (native venv / --admit-python). --auto-name / --auto-link (independent; --auto-name defaults auto-link on) pass through to admit_originals after a successful admit. Exit 0 = all ok; 1 = one or more item failures; 2 = CLI/config/validation error. """ from __future__ import annotations import argparse import json import os import plistlib import shutil import stat import subprocess import sys import time from dataclasses import dataclass, field from pathlib import Path from typing import Any, Literal, Sequence CONFIG_VERSION = 1 CONFIG_PATH: Path | None = None CONFIG_ENV_VAR = "INBOX_WATCH_CONFIG" ConfigSource = Literal["cli", "json", "env", "portable", "unset"] _MEANINGFUL_PATH_SOURCES = frozenset({"cli", "json", "env"}) Kind = Literal["audio", "transcript", "ignore"] AUDIO_EXTENSIONS = frozenset( {".wav", ".mp3", ".m4a", ".flac", ".ogg", ".opus", ".aac", ".wma"} ) TRANSCRIPT_EXTENSIONS = frozenset( {".json", ".srt", ".vtt", ".txt", ".html", ".htm"} ) KNOWN_CONFIG_KEYS = frozenset( { "version", "inbox", "recordings", "transcripts", "env_file", "whispermlx_missing", "ffmpeg", "watch_audio", "watch_transcripts", "recursive", "interval_seconds", "move_processed", "wav_backup", "backup_wavs", "delete_originals", "skip_serial", "admit_to_library", "admit_python", "auto_name", "auto_link", "stage_local", "stage_dir", } ) _PATH_KEYS = ( "inbox", "recordings", "transcripts", "env_file", "whispermlx_missing", "ffmpeg", "move_processed", "wav_backup", "admit_python", "stage_dir", ) FFMPEG_CHANNELS = "1" FFMPEG_SAMPLE_RATE = "16000" FFMPEG_CODEC = "libmp3lame" FFMPEG_BITRATE = "64k" STAGE_DIR_NAME = ".inbox-staging" @dataclass class ConfigProvenance: inbox: ConfigSource = "unset" recordings: ConfigSource = "unset" transcripts: ConfigSource = "unset" env_file: ConfigSource = "unset" whispermlx_missing: ConfigSource = "unset" ffmpeg: ConfigSource = "unset" move_processed: ConfigSource = "unset" wav_backup: ConfigSource = "unset" admit_python: ConfigSource = "unset" stage_dir: ConfigSource = "unset" @dataclass class EffectiveConfig: inbox: Path | None recordings: Path | None transcripts: Path | None env_file: Path | None whispermlx_missing: Path | None ffmpeg: Path | None watch_audio: bool watch_transcripts: bool recursive: bool interval_seconds: float move_processed: Path | None wav_backup: Path | None backup_wavs: bool delete_originals: bool skip_serial: bool = False admit_to_library: bool = False admit_python: Path | None = None auto_name: bool = False auto_link: bool = False stage_local: bool | None = None stage_dir: Path | None = None provenance: ConfigProvenance = field(default_factory=ConfigProvenance) @dataclass class CycleStats: audio_converted: int = 0 audio_skipped: int = 0 audio_failed: int = 0 transcripts_copied: int = 0 transcripts_skipped: int = 0 transcripts_failed: int = 0 unstable: int = 0 staged: int = 0 staged_reused: int = 0 stage_failed: int = 0 would_stage: int = 0 missing_invoked: int = 0 originals_backed_up: int = 0 originals_deleted: int = 0 would_convert: int = 0 would_copy: int = 0 would_invoke_missing: int = 0 would_backup: int = 0 would_delete: int = 0 would_admit: int = 0 admitted: int = 0 admit_skipped: int = 0 admit_failed: int = 0 converted_names: list[str] = field(default_factory=list) admitted_names: list[str] = field(default_factory=list) copied_names: list[str] = field(default_factory=list) failed_names: list[str] = field(default_factory=list) skipped_names: list[tuple[str, str]] = field(default_factory=list) unstable_names: list[str] = field(default_factory=list) staged_names: list[str] = field(default_factory=list) @property def failed(self) -> int: return ( self.audio_failed + self.transcripts_failed + self.admit_failed + self.stage_failed ) def _log(msg: str = "", *, err: bool = False) -> None: print(msg, file=sys.stderr if err else sys.stdout, flush=True) def _print_section(title: str) -> None: """Compact section banner — same shape as analysis Review / Run summary.""" _log() _log("---") _log(title) _log("---") def _print_limited_items( label: str, items: Sequence[str], *, limit: int = 12 ) -> None: if not items: return shown = min(len(items), limit) _log(f" {label}:") for item in items[:limit]: _log(f" • {item}") if len(items) > limit: _log(f" • ... and {len(items) - shown} more") def _cycle_status(stats: CycleStats, *, dry_run: bool) -> str: if dry_run: return "dry-run" if stats.failed: if stats.audio_converted or stats.transcripts_copied: return "partial" return "failed" return "completed" def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Watch an inbox for new audio (convert + whispermlx-missing) " "and/or new transcripts (copy if stem missing). " "Optional --admit admits originals/ JSON into the managed library." ), ) parser.add_argument( "--config", dest="config", type=Path, default=None, metavar="PATH", help=( "Config JSON path (default: .transcriptx/inbox-watch.json in repo). " f"Override with {CONFIG_ENV_VAR}." ), ) parser.add_argument("--inbox", type=Path, default=None) parser.add_argument("--recordings", type=Path, default=None) parser.add_argument("--transcripts", type=Path, default=None) parser.add_argument("--env-file", dest="env_file", type=Path, default=None) parser.add_argument( "--whispermlx-missing", dest="whispermlx_missing", type=Path, default=None, help="Path to whispermlx-missing (binary or scripts/whispermlx-missing.py).", ) parser.add_argument("--ffmpeg", type=Path, default=None) parser.add_argument( "--admit-python", dest="admit_python", type=Path, default=None, help=( "Python interpreter that can import transcriptx " "(used when --admit is on)." ), ) parser.add_argument( "--move-processed", dest="move_processed", type=Path, default=None, help="After a successful convert/copy, move the inbox source here (never delete).", ) parser.add_argument( "--wav-backup", dest="wav_backup", type=Path, default=None, help="WAV backup folder (default: TRANSCRIPTX_WAV_BACKUP_DIR / data/backups/wav).", ) parser.add_argument( "--stage-dir", dest="stage_dir", type=Path, default=None, help=( "Local folder for inbox audio copies before ffmpeg " "(default: {recordings}/.inbox-staging)." ), ) stage_group = parser.add_mutually_exclusive_group() stage_group.add_argument( "--stage-local", dest="stage_local", action="store_true", default=None, help="Always copy inbox audio to local staging before convert.", ) stage_group.add_argument( "--no-stage-local", dest="stage_local", action="store_false", help="Never stage; ffmpeg reads the inbox path (default: auto on removable volumes).", ) backup_group = parser.add_mutually_exclusive_group() backup_group.add_argument( "--backup-wav", dest="backup_wavs", action="store_true", default=None, help="After a successful audio convert, copy the inbox original into the WAV backup folder.", ) backup_group.add_argument( "--no-backup-wav", dest="backup_wavs", action="store_false", help="Do not copy originals into the WAV backup folder (default).", ) delete_group = parser.add_mutually_exclusive_group() delete_group.add_argument( "--delete-originals", dest="delete_originals", action="store_true", default=None, help="After a successful convert/copy (and backup, if enabled), delete the inbox source.", ) delete_group.add_argument( "--no-delete-originals", dest="delete_originals", action="store_false", help="Keep inbox sources (default).", ) audio_group = parser.add_mutually_exclusive_group() audio_group.add_argument( "--watch-audio", dest="watch_audio", action="store_true", default=None, help="Enable audio convert + whispermlx-missing (default: on if unset).", ) audio_group.add_argument( "--no-watch-audio", dest="watch_audio", action="store_false", help="Disable audio handling.", ) tx_group = parser.add_mutually_exclusive_group() tx_group.add_argument( "--watch-transcripts", dest="watch_transcripts", action="store_true", default=None, help="Enable transcript copy-if-new (default: on if unset).", ) tx_group.add_argument( "--no-watch-transcripts", dest="watch_transcripts", action="store_false", help="Disable transcript handling.", ) run_group = parser.add_mutually_exclusive_group() run_group.add_argument( "--once", action="store_true", help="Single scan (default). Suitable for cron/launchd.", ) run_group.add_argument( "--watch", dest="watch_loop", action="store_true", help="Poll until interrupted.", ) parser.add_argument( "--interval", dest="interval_seconds", type=float, default=None, help="Poll interval in seconds when --watch (default: 5).", ) parser.add_argument( "--recursive", action="store_true", default=None, help="Scan inbox subdirectories.", ) parser.add_argument( "--no-recursive", dest="recursive", action="store_false", help="Do not scan subdirectories (default).", ) parser.add_argument( "--dry-run", action="store_true", help="Print planned ffmpeg/copy/missing; do not write or invoke.", ) parser.add_argument( "--force", action="store_true", help="Overwrite an existing destination stem.", ) serial_group = parser.add_mutually_exclusive_group() serial_group.add_argument( "--skip-serial", dest="skip_serial", action="store_true", default=None, help=( "Pass --skip-serial to whispermlx-missing: do not transcribe " "MP3s that look like Auto-merge serial groups." ), ) serial_group.add_argument( "--no-skip-serial", dest="skip_serial", action="store_false", help="Transcribe serial parts (default unless config/env enables skip).", ) admit_group = parser.add_mutually_exclusive_group() admit_group.add_argument( "--admit", dest="admit_to_library", action="store_true", default=None, help=( "After convert/copy/whispermlx-missing, admit originals/ transcripts " "into the managed library (default: off)." ), ) admit_group.add_argument( "--no-admit", dest="admit_to_library", action="store_false", help="Do not admit into the managed library (default).", ) name_group = parser.add_mutually_exclusive_group() name_group.add_argument( "--auto-name", dest="auto_name", action="store_true", default=None, help=( "After admit, auto-name diarized speakers (implies --admit; " "also auto-link unless --no-auto-link)." ), ) name_group.add_argument( "--no-auto-name", dest="auto_name", action="store_false", help="Do not auto-write speaker names after admit.", ) link_group = parser.add_mutually_exclusive_group() link_group.add_argument( "--auto-link", dest="auto_link", action="store_true", default=None, help="After admit, auto-link matched longitudinal speaker profiles.", ) link_group.add_argument( "--no-auto-link", dest="auto_link", action="store_false", help="Do not auto-link longitudinal profiles after admit.", ) parser.add_argument( "--show-config", action="store_true", help="Print effective config and exit.", ) parser.add_argument( "--save-config", action="store_true", help="Save resolved settings to the config file (see --config).", ) parser.add_argument( "--stability-checks", dest="stability_checks", type=int, default=3, help=argparse.SUPPRESS, ) parser.add_argument( "--stability-interval-ms", dest="stability_interval_ms", type=int, default=500, help=argparse.SUPPRESS, ) parser.add_argument( "--stability-timeout-ms", dest="stability_timeout_ms", type=int, default=30_000, help=argparse.SUPPRESS, ) parser.set_defaults( watch_audio=None, watch_transcripts=None, recursive=None, backup_wavs=None, delete_originals=None, skip_serial=None, admit_to_library=None, auto_name=None, auto_link=None, stage_local=None, ) return parser.parse_args(argv) def find_repo_root() -> Path | None: script_dir = Path(__file__).resolve().parent if script_dir.name != "scripts": return None return script_dir.parent def parse_env_file(path: Path) -> dict[str, str]: if not path.is_file(): return {} result: dict[str, str] = {} try: text = path.read_text(encoding="utf-8") except OSError: return {} for raw in text.splitlines(): line = raw.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line[7:].strip() if "=" not in line: continue key, _, value = line.partition("=") key = key.strip() value = value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'): value = value[1:-1] if key: result[key] = value return result def bootstrap_repo_env(repo_root: Path) -> None: dotenv_path = repo_root / ".env" for key, value in parse_env_file(dotenv_path).items(): os.environ.setdefault(key, value) def _parse_bool_env(value: str | None, *, default: bool) -> bool: if value is None or not value.strip(): return default normalized = value.strip().lower() if normalized in ("true", "1", "yes", "on"): return True if normalized in ("false", "0", "no", "off"): return False return default def require_bool(value: Any, key: str) -> bool: if isinstance(value, bool): return value if isinstance(value, str): normalized = value.strip().lower() if normalized in ("true", "1", "yes", "on"): return True if normalized in ("false", "0", "no", "off"): return False raise SystemExit(f"ERROR: config key {key!r} must be a boolean, got {value!r}") def portable_defaults(repo_root: Path | None) -> tuple[dict[str, Any], ConfigProvenance]: provenance = ConfigProvenance() defaults: dict[str, Any] = {} if repo_root is None: return defaults, provenance defaults["inbox"] = str(repo_root / "data" / "transcript-inbox") provenance.inbox = "portable" defaults["recordings"] = str(repo_root / "data" / "recordings") provenance.recordings = "portable" defaults["transcripts"] = str(repo_root / "data" / "transcripts" / "originals") provenance.transcripts = "portable" defaults["env_file"] = str(repo_root / "whisperx.env") provenance.env_file = "portable" sibling = repo_root / "scripts" / "whispermlx-missing.py" if sibling.is_file(): defaults["whispermlx_missing"] = str(sibling) provenance.whispermlx_missing = "portable" defaults["wav_backup"] = str(repo_root / "data" / "backups" / "wav") provenance.wav_backup = "portable" return defaults, provenance def env_derived_config() -> tuple[dict[str, Any], ConfigProvenance]: provenance = ConfigProvenance() derived: dict[str, Any] = {} inbox = os.environ.get("INBOX_WATCH_INBOX", "").strip() if inbox: derived["inbox"] = inbox provenance.inbox = "env" recordings = os.environ.get("TRANSCRIPTX_RECORDINGS_DIR", "").strip() if recordings: derived["recordings"] = recordings provenance.recordings = "env" transcripts_base = os.environ.get("TRANSCRIPTX_TRANSCRIPTS_DIR", "").strip() if transcripts_base: derived["transcripts"] = str(Path(transcripts_base).expanduser() / "originals") provenance.transcripts = "env" env_file = os.environ.get("INBOX_WATCH_ENV_FILE", "").strip() if env_file: derived["env_file"] = env_file provenance.env_file = "env" missing = os.environ.get("INBOX_WATCH_WHISPERMLX_MISSING", "").strip() if missing: derived["whispermlx_missing"] = missing provenance.whispermlx_missing = "env" ffmpeg = os.environ.get("INBOX_WATCH_FFMPEG", "").strip() if ffmpeg: derived["ffmpeg"] = ffmpeg provenance.ffmpeg = "env" wav_backup = os.environ.get("TRANSCRIPTX_WAV_BACKUP_DIR", "").strip() if wav_backup: derived["wav_backup"] = wav_backup provenance.wav_backup = "env" stage_dir = os.environ.get("INBOX_WATCH_STAGE_DIR", "").strip() if stage_dir: derived["stage_dir"] = stage_dir provenance.stage_dir = "env" if os.environ.get("INBOX_WATCH_AUDIO", "").strip(): derived["watch_audio"] = _parse_bool_env( os.environ.get("INBOX_WATCH_AUDIO"), default=True ) if os.environ.get("INBOX_WATCH_TRANSCRIPTS", "").strip(): derived["watch_transcripts"] = _parse_bool_env( os.environ.get("INBOX_WATCH_TRANSCRIPTS"), default=True ) if os.environ.get("INBOX_WATCH_BACKUP_WAV", "").strip(): derived["backup_wavs"] = _parse_bool_env( os.environ.get("INBOX_WATCH_BACKUP_WAV"), default=False ) if os.environ.get("INBOX_WATCH_DELETE_ORIGINALS", "").strip(): derived["delete_originals"] = _parse_bool_env( os.environ.get("INBOX_WATCH_DELETE_ORIGINALS"), default=False ) if os.environ.get("INBOX_WATCH_SKIP_SERIAL", "").strip(): derived["skip_serial"] = _parse_bool_env( os.environ.get("INBOX_WATCH_SKIP_SERIAL"), default=False ) if os.environ.get("INBOX_WATCH_ADMIT", "").strip(): derived["admit_to_library"] = _parse_bool_env( os.environ.get("INBOX_WATCH_ADMIT"), default=False ) if os.environ.get("INBOX_WATCH_AUTO_NAME", "").strip(): derived["auto_name"] = _parse_bool_env( os.environ.get("INBOX_WATCH_AUTO_NAME"), default=False ) if os.environ.get("INBOX_WATCH_AUTO_LINK", "").strip(): derived["auto_link"] = _parse_bool_env( os.environ.get("INBOX_WATCH_AUTO_LINK"), default=False ) if os.environ.get("INBOX_WATCH_STAGE_LOCAL", "").strip(): derived["stage_local"] = _parse_bool_env( os.environ.get("INBOX_WATCH_STAGE_LOCAL"), default=False ) admit_python = os.environ.get("INBOX_WATCH_ADMIT_PYTHON", "").strip() if admit_python: derived["admit_python"] = admit_python provenance.admit_python = "env" return derived, provenance def base_config_dict() -> dict[str, Any]: return { "version": CONFIG_VERSION, "watch_audio": True, "watch_transcripts": True, "recursive": False, "interval_seconds": 5.0, "backup_wavs": False, "delete_originals": False, "skip_serial": False, "admit_to_library": False, } def resolve_config_path(args: argparse.Namespace) -> Path: if args.config is not None: return args.config.expanduser() env_val = os.environ.get(CONFIG_ENV_VAR, "").strip() if env_val: return Path(env_val).expanduser() if CONFIG_PATH is not None: return CONFIG_PATH repo_root = find_repo_root() if repo_root is not None: return repo_root / ".transcriptx" / "inbox-watch.json" return Path.cwd() / ".inbox-watch-no-config.json" def load_config(path: Path) -> dict[str, Any]: if not path.is_file(): return {} try: data = json.loads(path.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError) as exc: raise SystemExit(f"ERROR: invalid config file {path}: {exc}") from exc if not isinstance(data, dict): raise SystemExit(f"ERROR: config file must be a JSON object: {path}") for key in data: if key not in KNOWN_CONFIG_KEYS: print(f"WARNING: ignoring unknown config key: {key}", file=sys.stderr) return data def _apply_path_layer( merged: dict[str, Any], provenance: ConfigProvenance, layer: dict[str, Any], source: ConfigSource, ) -> None: for key in _PATH_KEYS: value = layer.get(key) if value is None: continue if isinstance(value, str) and not value.strip(): continue merged[key] = str(value) setattr(provenance, key, source) def _as_optional_path(value: Any) -> Path | None: if value is None: return None text = str(value).strip() if not text: return None return Path(text).expanduser() def resolve_config( args: argparse.Namespace, *, config_path: Path | None = None ) -> EffectiveConfig: repo_root = find_repo_root() if repo_root is not None: bootstrap_repo_env(repo_root) active_config = config_path or resolve_config_path(args) file_cfg = load_config(active_config) portable_layer, provenance = portable_defaults(repo_root) env_layer, env_prov = env_derived_config() merged: dict[str, Any] = {**base_config_dict(), **portable_layer} _apply_path_layer(merged, provenance, env_layer, "env") for key in _PATH_KEYS: if getattr(env_prov, key) != "unset": setattr(provenance, key, getattr(env_prov, key)) for key, value in env_layer.items(): if key not in _PATH_KEYS: merged[key] = value json_path_layer = { k: file_cfg[k] for k in _PATH_KEYS if k in file_cfg and file_cfg[k] is not None } _apply_path_layer(merged, provenance, json_path_layer, "json") for key in set(file_cfg) - set(_PATH_KEYS): merged[key] = file_cfg[key] if args.inbox is not None: merged["inbox"] = str(args.inbox) provenance.inbox = "cli" if args.recordings is not None: merged["recordings"] = str(args.recordings) provenance.recordings = "cli" if args.transcripts is not None: merged["transcripts"] = str(args.transcripts) provenance.transcripts = "cli" if args.env_file is not None: merged["env_file"] = str(args.env_file) provenance.env_file = "cli" if args.whispermlx_missing is not None: merged["whispermlx_missing"] = str(args.whispermlx_missing) provenance.whispermlx_missing = "cli" if args.ffmpeg is not None: merged["ffmpeg"] = str(args.ffmpeg) provenance.ffmpeg = "cli" if args.admit_python is not None: merged["admit_python"] = str(args.admit_python) provenance.admit_python = "cli" if args.move_processed is not None: merged["move_processed"] = str(args.move_processed) provenance.move_processed = "cli" if args.wav_backup is not None: merged["wav_backup"] = str(args.wav_backup) provenance.wav_backup = "cli" if args.stage_dir is not None: merged["stage_dir"] = str(args.stage_dir) provenance.stage_dir = "cli" if args.watch_audio is not None: merged["watch_audio"] = args.watch_audio if args.watch_transcripts is not None: merged["watch_transcripts"] = args.watch_transcripts if args.backup_wavs is not None: merged["backup_wavs"] = args.backup_wavs if args.delete_originals is not None: merged["delete_originals"] = args.delete_originals if args.skip_serial is not None: merged["skip_serial"] = args.skip_serial if args.admit_to_library is not None: merged["admit_to_library"] = args.admit_to_library if args.auto_name is not None: merged["auto_name"] = args.auto_name if args.auto_link is not None: merged["auto_link"] = args.auto_link if args.stage_local is not None: merged["stage_local"] = args.stage_local if args.recursive is not None: merged["recursive"] = args.recursive if args.interval_seconds is not None: merged["interval_seconds"] = args.interval_seconds watch_audio = require_bool(merged.get("watch_audio", True), "watch_audio") watch_transcripts = require_bool( merged.get("watch_transcripts", True), "watch_transcripts" ) recursive = require_bool(merged.get("recursive", False), "recursive") backup_wavs = require_bool(merged.get("backup_wavs", False), "backup_wavs") delete_originals = require_bool( merged.get("delete_originals", False), "delete_originals" ) skip_serial = require_bool(merged.get("skip_serial", False), "skip_serial") auto_name = require_bool(merged.get("auto_name", False), "auto_name") auto_link_raw = merged.get("auto_link", None) if args.auto_name is True and args.auto_link is None: auto_link = True elif auto_link_raw is None: auto_link = auto_name else: auto_link = require_bool(auto_link_raw, "auto_link") admit_to_library = require_bool( merged.get("admit_to_library", False), "admit_to_library" ) if auto_name or auto_link: admit_to_library = True stage_local_raw = merged.get("stage_local", None) if stage_local_raw is None or ( isinstance(stage_local_raw, str) and not stage_local_raw.strip() ): stage_local: bool | None = None else: stage_local = require_bool(stage_local_raw, "stage_local") interval = merged.get("interval_seconds", 5.0) try: interval_seconds = float(interval) except (TypeError, ValueError) as exc: raise SystemExit( f"ERROR: config key 'interval_seconds' must be a number, got {interval!r}" ) from exc return EffectiveConfig( inbox=_as_optional_path(merged.get("inbox")), recordings=_as_optional_path(merged.get("recordings")), transcripts=_as_optional_path(merged.get("transcripts")), env_file=_as_optional_path(merged.get("env_file")), whispermlx_missing=_as_optional_path(merged.get("whispermlx_missing")), ffmpeg=_as_optional_path(merged.get("ffmpeg")), watch_audio=watch_audio, watch_transcripts=watch_transcripts, recursive=recursive, interval_seconds=interval_seconds, move_processed=_as_optional_path(merged.get("move_processed")), wav_backup=_as_optional_path(merged.get("wav_backup")), backup_wavs=backup_wavs, delete_originals=delete_originals, skip_serial=skip_serial, admit_to_library=admit_to_library, admit_python=_as_optional_path(merged.get("admit_python")), auto_name=auto_name, auto_link=auto_link, stage_local=stage_local, stage_dir=_as_optional_path(merged.get("stage_dir")), provenance=provenance, ) def config_to_dict(cfg: EffectiveConfig) -> dict[str, Any]: return { "version": CONFIG_VERSION, "inbox": str(cfg.inbox) if cfg.inbox else None, "recordings": str(cfg.recordings) if cfg.recordings else None, "transcripts": str(cfg.transcripts) if cfg.transcripts else None, "env_file": str(cfg.env_file) if cfg.env_file else None, "whispermlx_missing": ( str(cfg.whispermlx_missing) if cfg.whispermlx_missing else None ), "ffmpeg": str(cfg.ffmpeg) if cfg.ffmpeg else None, "admit_python": str(cfg.admit_python) if cfg.admit_python else None, "watch_audio": cfg.watch_audio, "watch_transcripts": cfg.watch_transcripts, "recursive": cfg.recursive, "interval_seconds": cfg.interval_seconds, "move_processed": str(cfg.move_processed) if cfg.move_processed else None, "wav_backup": str(cfg.wav_backup) if cfg.wav_backup else None, "backup_wavs": cfg.backup_wavs, "delete_originals": cfg.delete_originals, "skip_serial": cfg.skip_serial, "admit_to_library": cfg.admit_to_library, "auto_name": cfg.auto_name, "auto_link": cfg.auto_link, "stage_local": cfg.stage_local, "stage_dir": str(cfg.stage_dir) if cfg.stage_dir else None, } def save_config(cfg: EffectiveConfig, path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(config_to_dict(cfg), indent=2) + "\n", encoding="utf-8") os.chmod(path, stat.S_IRUSR | stat.S_IWUSR) def classify_path(path: Path | str) -> Kind: ext = Path(path).suffix.lower() if ext in AUDIO_EXTENSIONS: return "audio" if ext in TRANSCRIPT_EXTENSIONS: return "transcript" return "ignore" def is_same_or_under(path: Path, root: Path) -> bool: try: path.resolve().relative_to(root.resolve()) return True except (ValueError, OSError): return False def effective_stage_dir(cfg: EffectiveConfig) -> Path | None: """Return the staging folder, defaulting to {recordings}/.inbox-staging.""" if cfg.stage_dir is not None: return cfg.stage_dir if cfg.recordings is not None: return cfg.recordings / STAGE_DIR_NAME return None def find_stem_match( directory: Path, stem: str, extensions: frozenset[str] ) -> Path | None: if not directory.is_dir(): return None stem_l = stem.lower() for candidate in directory.iterdir(): if not candidate.is_file(): continue if candidate.suffix.lower() not in extensions: continue if candidate.stem.lower() == stem_l: return candidate return None def wait_until_stable( path: Path, *, checks: int = 3, interval_ms: int = 500, timeout_ms: int = 30_000, ) -> bool: """Return True once size/mtime are unchanged for ``checks`` samples.""" interval_s = max(interval_ms, 1) / 1000.0 deadline = time.monotonic() + max(timeout_ms, interval_ms) / 1000.0 previous: tuple[int, int] | None = None stable_count = 0 while time.monotonic() < deadline: try: st = path.stat() except OSError: return False current = (int(st.st_size), int(getattr(st, "st_mtime_ns", int(st.st_mtime * 1e9)))) if previous is not None and current == previous: stable_count += 1 if stable_count >= max(checks, 1): return True else: stable_count = 1 previous = current if max(checks, 1) == 1: return True time.sleep(interval_s) return previous is not None and stable_count >= max(checks, 1) def _existing_ancestor(path: Path) -> Path | None: current = path.expanduser() try: current = current.resolve() except OSError: pass while True: try: if current.exists(): return current except OSError: return None parent = current.parent if parent == current: return None current = parent def _macos_volume_is_removable(mount: Path) -> bool | None: """Return True/False from diskutil, or None if detection failed.""" try: result = subprocess.run( ["diskutil", "info", "-plist", str(mount)], capture_output=True, timeout=10, check=False, ) except (OSError, subprocess.TimeoutExpired): return None if result.returncode != 0 or not result.stdout: return None try: info = plistlib.loads(result.stdout) except Exception: return None if not isinstance(info, dict): return None for key in ("Ejectable", "Removable", "RemovableMedia"): if info.get(key) is True: return True return False def _linux_sysfs_removable(device: str) -> bool | None: name = Path(device).name if not name: return None sysfs = Path("/sys/class/block") / name / "removable" try: raw = sysfs.read_text(encoding="utf-8").strip() except OSError: parent = name.rstrip("0123456789") if parent == name: return None sysfs = Path("/sys/class/block") / parent / "removable" try: raw = sysfs.read_text(encoding="utf-8").strip() except OSError: return None return raw == "1" def _linux_path_is_removable(path: Path) -> bool | None: posix = path.as_posix() if posix.startswith("/media/") or posix.startswith("/run/media/"): return True try: mounts = Path("/proc/mounts").read_text(encoding="utf-8") except OSError: return None best: tuple[int, str] | None = None for line in mounts.splitlines(): parts = line.split() if len(parts) < 2: continue device, mount_point = parts[0], parts[1] mount_point = mount_point.replace("\\040", " ") if posix == mount_point or posix.startswith(mount_point.rstrip("/") + "/"): length = len(mount_point) if best is None or length > best[0]: best = (length, device) if best is None: return None return _linux_sysfs_removable(best[1]) def inbox_on_removable_volume(inbox: Path) -> bool: """True when *inbox* is on an ejectable/removable volume. Detection failure returns False (ffmpeg keeps reading the inbox path) unless the operator forces staging with --stage-local. """ mount = _existing_ancestor(inbox) if mount is None: return False if sys.platform == "darwin": detected = _macos_volume_is_removable(mount) return bool(detected) if sys.platform.startswith("linux"): detected = _linux_path_is_removable(mount) return bool(detected) return False def should_stage_audio( cfg: EffectiveConfig, inbox: Path, *, removable: bool | None = None, ) -> bool: if not cfg.watch_audio: return False if cfg.stage_local is True: return True if cfg.stage_local is False: return False if removable is None: removable = inbox_on_removable_volume(inbox) return removable def stage_inbox_audio( inbox_src: Path, stage_dir: Path, *, force: bool, dry_run: bool, ) -> tuple[Path | None, str]: """Copy inbox audio to *stage_dir*. Return (path, staged|reused|failed|dry_run).""" dest = stage_dir / inbox_src.name if dry_run: try: size = f" ({_human_bytes(inbox_src.stat().st_size)})" except OSError: size = "" _log(f" Would stage: {inbox_src} -> {dest}{size}") return dest, "dry_run" src_size: int | None try: src_size = int(inbox_src.stat().st_size) except OSError: src_size = None if dest.is_file() and not force: try: dest_size = int(dest.stat().st_size) except OSError: dest_size = -1 if src_size is None or dest_size == src_size: _log(f" Using staged: {dest.name}") return dest, "reused" stage_dir.mkdir(parents=True, exist_ok=True) partial = stage_dir / f".inbox-staging.{inbox_src.name}.partial" try: size = f" ({_human_bytes(src_size)})" if src_size is not None else "" except OSError: size = "" _log(f" Staging: {inbox_src.name} -> {dest.name}{size}") started = time.perf_counter() try: shutil.copy2(inbox_src, partial) copied = int(partial.stat().st_size) if src_size is not None and copied != src_size: partial.unlink(missing_ok=True) _log( f"ERROR: staged size mismatch for {inbox_src.name} " f"(expected {src_size}, got {copied})", err=True, ) return None, "failed" os.replace(partial, dest) except OSError as exc: if partial.exists(): partial.unlink(missing_ok=True) _log(f"ERROR: staging failed for {inbox_src.name}: {exc}", err=True) return None, "failed" elapsed = time.perf_counter() - started try: out_size = f" ({_human_bytes(dest.stat().st_size)})" except OSError: out_size = "" _log(f" Staged: {inbox_src.name}{out_size} in {elapsed:.1f}s") return dest, "staged" def discover_inbox_files( inbox: Path, *, recursive: bool, skip_under: Sequence[Path] | None = None, ) -> list[Path]: if not inbox.is_dir(): return [] if recursive: candidates = [p for p in inbox.rglob("*") if p.is_file()] else: candidates = [p for p in inbox.iterdir() if p.is_file()] skips = [p for p in (skip_under or []) if p is not None] out: list[Path] = [] for path in candidates: if path.name.startswith("."): continue if any(is_same_or_under(path.parent, root) for root in skips): continue out.append(path) return sorted(out, key=lambda p: str(p).lower()) def build_ffmpeg_cmd(ffmpeg: str | Path, src: Path, dest: Path) -> list[str]: # Always pass -f mp3: dest may be a .mp3.partial temp path, and ffmpeg 8+ # will not guess the muxer from a non-standard extension. return [ str(ffmpeg), "-nostdin", "-y", "-i", str(src), "-ac", FFMPEG_CHANNELS, "-ar", FFMPEG_SAMPLE_RATE, "-c:a", FFMPEG_CODEC, "-b:a", FFMPEG_BITRATE, "-f", "mp3", str(dest), ] def find_ffmpeg(explicit: Path | None) -> Path | None: if explicit is not None: return explicit found = shutil.which("ffmpeg") return Path(found) if found else None def find_whispermlx_missing(explicit: Path | None) -> Path | None: if explicit is not None: return explicit found = shutil.which("whispermlx-missing") if found: return Path(found) repo_root = find_repo_root() if repo_root is not None: sibling = repo_root / "scripts" / "whispermlx-missing.py" if sibling.is_file(): return sibling return None def build_missing_cmd( missing: Path, *, recordings: Path, transcripts: Path, env_file: Path | None, skip_serial: bool = False, ) -> list[str]: cmd: list[str] if missing.suffix.lower() == ".py": cmd = [sys.executable, str(missing)] else: cmd = [str(missing)] cmd.extend(["--source", str(recordings), "--transcripts", str(transcripts)]) if env_file is not None: cmd.extend(["--env-file", str(env_file)]) if skip_serial: cmd.append("--skip-serial") return cmd def _human_bytes(n: int) -> str: if n >= 1024 * 1024 * 1024: return f"{n / (1024 * 1024 * 1024):.1f} GiB" if n >= 1024 * 1024: return f"{n / (1024 * 1024):.1f} MiB" if n >= 1024: return f"{n / 1024:.1f} KiB" return f"{n} B" def run_ffmpeg(cmd: Sequence[str]) -> subprocess.CompletedProcess[str]: # Inherit stderr so ffmpeg's time=/speed= stats are visible on long encodes. return subprocess.run(cmd, check=False) def run_whispermlx_missing(cmd: Sequence[str]) -> int: result = subprocess.run(cmd, check=False) return int(result.returncode) def transcripts_root_for_admit(transcripts: Path) -> Path: if transcripts.name == "originals": return transcripts.parent return transcripts def python_can_import_admit(python: Path) -> bool: try: result = subprocess.run( [str(python), "-c", "import transcriptx.admit_originals"], capture_output=True, timeout=30, check=False, ) except (OSError, subprocess.TimeoutExpired): return False return result.returncode == 0 def find_admit_python(explicit: Path | None) -> Path | None: candidates: list[Path] = [] if explicit is not None: candidates.append(explicit) repo_root = find_repo_root() if repo_root is not None: for rel in ( ".transcriptx/bin/python", ".transcriptx/bin/python3", ".venv/bin/python", ): path = repo_root / rel if path.is_file(): candidates.append(path) candidates.append(Path(sys.executable)) seen: set[str] = set() for python in candidates: key = str(python) if key in seen: continue seen.add(key) if python_can_import_admit(python): return python return None def build_admit_cmd( python: Path, *, transcripts: Path, dry_run: bool = False, auto_name: bool = False, auto_link: bool = False, ) -> list[str]: cmd = [ str(python), "-m", "transcriptx.admit_originals", "--dir", str(transcripts), "--transcripts-root", str(transcripts_root_for_admit(transcripts)), ] if dry_run: cmd.append("--dry-run") if auto_name: cmd.append("--auto-name") elif auto_link: cmd.append("--no-auto-name") if auto_link: cmd.append("--auto-link") elif auto_name: cmd.append("--no-auto-link") return cmd def run_admit_originals(cmd: Sequence[str]) -> int: result = subprocess.run(cmd, check=False) return int(result.returncode) def _move_processed(src: Path, dest_dir: Path) -> bool: dest_dir.mkdir(parents=True, exist_ok=True) dest = dest_dir / src.name if dest.exists(): print( f"WARNING: not moving {src.name}; {dest} already exists", file=sys.stderr, ) return False shutil.move(str(src), str(dest)) return True def wait_for_directory( path: Path, *, interval_seconds: float, timeout_seconds: float | None = None, ) -> bool: """Block until *path* is a directory. Return False if *timeout_seconds* elapses.""" deadline = ( None if timeout_seconds is None else time.monotonic() + timeout_seconds ) announced = False while not path.is_dir(): if deadline is not None and time.monotonic() >= deadline: return False if not announced: _log(f"Waiting for inbox: {path}") announced = True time.sleep(max(interval_seconds, 0.1)) if announced: _log(f"Inbox is available: {path}") return True def unique_backup_path(directory: Path, src: Path) -> Path: suffix = src.suffix.lower() or src.suffix dest = directory / f"{src.stem}{suffix}" counter = 1 while dest.exists(): dest = directory / f"{src.stem}_{counter}{suffix}" counter += 1 if counter > 1000: raise OSError(f"too many name conflicts in {directory} for {src.stem}") return dest def backup_original_to_wav(src: Path, wav_backup: Path) -> Path | None: wav_backup.mkdir(parents=True, exist_ok=True) dest = unique_backup_path(wav_backup, src) shutil.copy2(src, dest) print(f"Backed up: {src.name} -> {dest}") return dest def move_staged_to_wav_backup(staged: Path, wav_backup: Path) -> Path: wav_backup.mkdir(parents=True, exist_ok=True) dest = unique_backup_path(wav_backup, staged) shutil.move(str(staged), str(dest)) print(f"Backed up: {staged.name} -> {dest}") return dest def finalize_inbox_source( src: Path, cfg: EffectiveConfig, *, kind: Kind, dry_run: bool, stats: CycleStats, staged_path: Path | None = None, ) -> None: """Backup / move / delete an inbox source after a successful convert or copy.""" if kind == "audio" and cfg.backup_wavs: if cfg.wav_backup is None: print("ERROR: backup_wavs is on but wav_backup is unset", file=sys.stderr) return backup_src = ( staged_path if staged_path is not None and (dry_run or staged_path.is_file()) else src ) if dry_run: dest = unique_backup_path(cfg.wav_backup, backup_src) print(f"Would backup: {backup_src} -> {dest}") stats.would_backup += 1 else: try: if staged_path is not None and staged_path.is_file(): move_staged_to_wav_backup(staged_path, cfg.wav_backup) else: backup_original_to_wav(src, cfg.wav_backup) stats.originals_backed_up += 1 except OSError as exc: print(f"ERROR: wav backup failed for {src.name}: {exc}", file=sys.stderr) if cfg.delete_originals: print( f"WARNING: not deleting {src.name} because backup failed", file=sys.stderr, ) return elif kind == "audio" and staged_path is not None and not dry_run: try: staged_path.unlink(missing_ok=True) except OSError: pass if cfg.move_processed is not None: if dry_run: print(f"Would move: {src} -> {cfg.move_processed / src.name}") return if _move_processed(src, cfg.move_processed): print(f"Moved: {src.name} -> {cfg.move_processed / src.name}") return if cfg.delete_originals: if dry_run: print(f"Would delete original: {src}") stats.would_delete += 1 return try: src.unlink() except OSError as exc: print(f"ERROR: could not delete {src.name}: {exc}", file=sys.stderr) return print(f"Deleted original: {src.name}") stats.originals_deleted += 1 def looks_like_managed_library_root(path: Path) -> bool: """True when *path* is the managed transcripts library root (not originals/). Host helpers must write raw engine JSON under ``…/transcripts/originals``, never into the library root beside ``metadata/`` / ``imports/``. """ resolved = path.expanduser() if resolved.name == "originals": return False markers = ("metadata", "originals", "imports") try: return any((resolved / name).is_dir() for name in markers) except OSError: return False def validate_layout(cfg: EffectiveConfig) -> str | None: if not cfg.watch_audio and not cfg.watch_transcripts: return "Enable at least one of watch_audio / watch_transcripts." if cfg.inbox is None: return "inbox is required." inbox = cfg.inbox if cfg.watch_audio: if cfg.recordings is None: return "recordings is required when watch_audio is on." if cfg.transcripts is None: return "transcripts is required when watch_audio is on (whispermlx-missing output)." if cfg.watch_transcripts and cfg.transcripts is None: return "transcripts is required when watch_transcripts is on." if cfg.admit_to_library and cfg.transcripts is None: return "transcripts is required when admit_to_library is on." if cfg.transcripts is not None and looks_like_managed_library_root(cfg.transcripts): return ( "transcripts must be the originals/ folder (e.g. …/transcripts/originals), " "not the managed library root that contains metadata/ or imports/. " "Raw engine JSON in the library root is not admitted until Import Transcript, " "inbox-watch --admit, or Settings → Watcher runs." ) dests: list[tuple[str, Path]] = [] if cfg.recordings is not None: dests.append(("recordings", cfg.recordings)) if cfg.transcripts is not None: dests.append(("transcripts", cfg.transcripts)) for label, dest in dests: if inbox.resolve() == dest.resolve() or is_same_or_under(inbox, dest): return f"inbox must not be {label} or a path under {label}." if is_same_or_under(dest, inbox): return f"{label} must not be under inbox." if cfg.move_processed is not None: processed = cfg.move_processed if processed.resolve() == inbox.resolve(): return "move_processed must not be the inbox." if cfg.delete_originals: return "Use either delete_originals or move_processed, not both." if cfg.backup_wavs: if cfg.wav_backup is None: return "wav_backup is required when backup_wavs is on." wav_backup = cfg.wav_backup if inbox.resolve() == wav_backup.resolve() or is_same_or_under(inbox, wav_backup): return "inbox must not be wav_backup or a path under wav_backup." if is_same_or_under(wav_backup, inbox): return "wav_backup must not be under inbox." stage_dir = effective_stage_dir(cfg) if cfg.watch_audio and stage_dir is not None: if inbox.resolve() == stage_dir.resolve() or is_same_or_under(inbox, stage_dir): return "inbox must not be stage_dir or a path under stage_dir." if is_same_or_under(stage_dir, inbox): return "stage_dir must not be under inbox." if cfg.recordings is not None and stage_dir.resolve() == cfg.recordings.resolve(): return ( "stage_dir must not be recordings " "(use a subdirectory such as recordings/.inbox-staging)." ) if cfg.wav_backup is not None and stage_dir.resolve() == cfg.wav_backup.resolve(): return "stage_dir must not be wav_backup." if ( cfg.move_processed is not None and stage_dir.resolve() == cfg.move_processed.resolve() ): return "stage_dir must not be move_processed." return None def _has_meaningful_paths(cfg: EffectiveConfig) -> bool: if cfg.provenance.inbox not in _MEANINGFUL_PATH_SOURCES: return False if cfg.watch_audio: if cfg.provenance.recordings not in _MEANINGFUL_PATH_SOURCES: return False if cfg.provenance.transcripts not in _MEANINGFUL_PATH_SOURCES: return False if cfg.watch_transcripts: if cfg.provenance.transcripts not in _MEANINGFUL_PATH_SOURCES: return False return True def convert_audio( src: Path, recordings: Path, *, ffmpeg: Path, force: bool, dry_run: bool, cfg: EffectiveConfig, stats: CycleStats, inbox_src: Path | None = None, staged_path: Path | None = None, ) -> str: """Return converted / skipped / failed / dry_run. *src* is the ffmpeg input (staged local copy or inbox file). *inbox_src* is the original inbox path used for logging and finalize. """ display = inbox_src or src existing = find_stem_match(recordings, display.stem, AUDIO_EXTENSIONS) dest = recordings / f"{display.stem}.mp3" if existing is not None and not force: _log(f" Skipping (stem exists): {display.name} -> {existing.name}") return "skipped" if dry_run: cmd = build_ffmpeg_cmd(ffmpeg, src, dest) _log(f" Would convert: {' '.join(cmd)}") finalize_inbox_source( display, cfg, kind="audio", dry_run=True, stats=stats, staged_path=staged_path, ) return "dry_run" recordings.mkdir(parents=True, exist_ok=True) partial = recordings / f".inbox-watch.{display.stem}.mp3.partial" try: size = f" ({_human_bytes(src.stat().st_size)})" except OSError: size = "" _log(f" Converting: {display.name} -> {dest.name}{size}") _log(" ffmpeg progress on stderr (time=/speed=)…") started = time.perf_counter() cmd = build_ffmpeg_cmd(ffmpeg, src, partial) result = run_ffmpeg(cmd) elapsed = time.perf_counter() - started if result.returncode != 0: if partial.exists(): partial.unlink(missing_ok=True) extra = "" if result.stderr: tail = "\n".join(result.stderr.splitlines()[-20:]) extra = f": {tail}" _log( f"ERROR: ffmpeg failed for {display.name} " f"(exit {result.returncode}, {elapsed:.1f}s){extra}", err=True, ) return "failed" os.replace(partial, dest) try: out_size = f" ({_human_bytes(dest.stat().st_size)})" except OSError: out_size = "" _log(f" Converted: {display.name} -> {dest.name}{out_size} in {elapsed:.1f}s") finalize_inbox_source( display, cfg, kind="audio", dry_run=False, stats=stats, staged_path=staged_path, ) return "converted" def copy_transcript( src: Path, transcripts: Path, *, force: bool, dry_run: bool, cfg: EffectiveConfig, stats: CycleStats, ) -> str: existing = find_stem_match(transcripts, src.stem, TRANSCRIPT_EXTENSIONS) dest = transcripts / src.name if existing is not None and not force: _log(f" Skipping (stem exists): {src.name} -> {existing.name}") return "skipped" if dry_run: _log(f" Would copy: {src} -> {dest}") finalize_inbox_source(src, cfg, kind="transcript", dry_run=True, stats=stats) return "dry_run" transcripts.mkdir(parents=True, exist_ok=True) _log(f" Copying: {src.name} -> {dest.name}") try: shutil.copy2(src, dest) except OSError as exc: _log(f"ERROR: copy failed for {src.name}: {exc}", err=True) return "failed" _log(f" Copied: {src.name} -> {dest.name}") finalize_inbox_source(src, cfg, kind="transcript", dry_run=False, stats=stats) return "copied" def print_review_before_cycle( cfg: EffectiveConfig, work: Sequence[tuple[Path, Kind]], *, dry_run: bool, staging: bool = False, staging_reason: str | None = None, ) -> None: audio_n = sum(1 for _, kind in work if kind == "audio") tx_n = sum(1 for _, kind in work if kind == "transcript") _print_section("Review before cycle") _log(f" Mode: {'dry-run' if dry_run else 'once'}") _log(f" Inbox: {cfg.inbox}") if cfg.watch_audio: _log(f" Recordings: {cfg.recordings}") if cfg.watch_transcripts or cfg.watch_audio: _log(f" Transcripts: {cfg.transcripts}") if staging: stage_dir = effective_stage_dir(cfg) reason = staging_reason or "on" _log(f" Staging: {reason}") if stage_dir is not None: _log(f" Stage dir: {stage_dir}") modes: list[str] = [] if cfg.watch_audio: modes.append("audio→mp3 + whispermlx-missing") if staging: modes.append("stage-local") if cfg.watch_transcripts: modes.append("transcript copy") if cfg.admit_to_library: modes.append("admit→library") if cfg.auto_name: modes.append("auto-name") if cfg.auto_link: modes.append("auto-link") _log(f" Watching: {', '.join(modes) if modes else '(none)'}") _log(f" Candidates: {len(work)} ({audio_n} audio, {tx_n} transcript)") if work: preview = [f"{kind}: {src.name}" for src, kind in work] _print_limited_items("Will consider", preview, limit=12) else: _log(" Will consider: (none)") _log("---") def maybe_run_missing( cfg: EffectiveConfig, stats: CycleStats, *, missing: Path, dry_run: bool, ) -> None: assert cfg.recordings is not None assert cfg.transcripts is not None cmd = build_missing_cmd( missing, recordings=cfg.recordings, transcripts=cfg.transcripts, env_file=cfg.env_file, skip_serial=cfg.skip_serial, ) _print_section("Transcription (whispermlx-missing)") if dry_run: _log(f" Would run: {' '.join(cmd)}") stats.would_invoke_missing += 1 _log("---") return _log(f" Running: {' '.join(cmd)}") _log(" (child process output follows)") started = time.perf_counter() rc = run_whispermlx_missing(cmd) elapsed = time.perf_counter() - started stats.missing_invoked += 1 if rc != 0: _log( f"WARNING: whispermlx-missing exited {rc} after {elapsed:.1f}s", err=True, ) stats.failed_names.append("whispermlx-missing") stats.audio_failed += 1 else: _log(f" Finished whispermlx-missing in {elapsed:.1f}s") _log("---") def maybe_run_admit( cfg: EffectiveConfig, stats: CycleStats, *, python: Path, dry_run: bool, ) -> None: assert cfg.transcripts is not None cmd = build_admit_cmd( python, transcripts=cfg.transcripts, dry_run=dry_run, auto_name=cfg.auto_name, auto_link=cfg.auto_link, ) _print_section("Library admit") if dry_run: _log(f" Would run: {' '.join(cmd)}") stats.would_admit += 1 _log("---") return _log(f" Running: {' '.join(cmd)}") _log(" (child process output follows)") started = time.perf_counter() rc = run_admit_originals(cmd) elapsed = time.perf_counter() - started if rc != 0: _log( f"WARNING: admit-originals exited {rc} after {elapsed:.1f}s", err=True, ) stats.failed_names.append("admit-originals") stats.admit_failed += 1 else: stats.admitted += 1 stats.admitted_names.append("originals/") _log(f" Finished library admit in {elapsed:.1f}s") _log("---") def process_cycle( cfg: EffectiveConfig, *, dry_run: bool, force: bool, ffmpeg: Path | None, stability_checks: int, stability_interval_ms: int, stability_timeout_ms: int, ) -> CycleStats: assert cfg.inbox is not None stats = CycleStats() stage_dir = effective_stage_dir(cfg) removable = inbox_on_removable_volume(cfg.inbox) stage_audio = should_stage_audio(cfg, cfg.inbox, removable=removable) if cfg.stage_local is True: staging_reason = "forced" elif stage_audio: staging_reason = "removable inbox" else: staging_reason = None files = discover_inbox_files( cfg.inbox, recursive=cfg.recursive, skip_under=[ p for p in (cfg.move_processed, cfg.wav_backup, stage_dir) if p is not None ], ) work: list[tuple[Path, Kind]] = [] for src in files: kind = classify_path(src) if kind == "audio" and not cfg.watch_audio: continue if kind == "transcript" and not cfg.watch_transcripts: continue if kind == "ignore": continue work.append((src, kind)) print_review_before_cycle( cfg, work, dry_run=dry_run, staging=stage_audio, staging_reason=staging_reason, ) total = len(work) if total == 0: _log("No inbox candidates this cycle.") return stats _print_section("Processing") for index, (src, kind) in enumerate(work, start=1): _log(f"[{index}/{total}] {kind}: {src.name}") if not wait_until_stable( src, checks=stability_checks, interval_ms=stability_interval_ms, timeout_ms=stability_timeout_ms, ): _log(f" Unstable (skipped this cycle): {src.name}", err=True) stats.unstable += 1 stats.unstable_names.append(src.name) continue if kind == "audio": assert cfg.recordings is not None assert ffmpeg is not None convert_src = src staged_path: Path | None = None existing = find_stem_match(cfg.recordings, src.stem, AUDIO_EXTENSIONS) if ( stage_audio and stage_dir is not None and (existing is None or force) ): staged_path, stage_outcome = stage_inbox_audio( src, stage_dir, force=force, dry_run=dry_run, ) if stage_outcome == "failed": stats.stage_failed += 1 stats.failed_names.append(src.name) continue if stage_outcome == "staged": stats.staged += 1 stats.staged_names.append(src.name) elif stage_outcome == "reused": stats.staged_reused += 1 stats.staged_names.append(src.name) elif stage_outcome == "dry_run": stats.would_stage += 1 stats.staged_names.append(src.name) if staged_path is not None: convert_src = staged_path outcome = convert_audio( convert_src, cfg.recordings, ffmpeg=ffmpeg, force=force, dry_run=dry_run, cfg=cfg, stats=stats, inbox_src=src, staged_path=staged_path, ) if outcome == "converted": stats.audio_converted += 1 stats.converted_names.append(src.name) elif outcome == "dry_run": stats.would_convert += 1 stats.converted_names.append(src.name) elif outcome == "skipped": stats.audio_skipped += 1 stats.skipped_names.append((src.name, "stem exists in recordings")) else: stats.audio_failed += 1 stats.failed_names.append(src.name) elif kind == "transcript": assert cfg.transcripts is not None outcome = copy_transcript( src, cfg.transcripts, force=force, dry_run=dry_run, cfg=cfg, stats=stats, ) if outcome == "copied": stats.transcripts_copied += 1 stats.copied_names.append(src.name) elif outcome == "dry_run": stats.would_copy += 1 stats.copied_names.append(src.name) elif outcome == "skipped": stats.transcripts_skipped += 1 stats.skipped_names.append((src.name, "stem exists in transcripts")) else: stats.transcripts_failed += 1 stats.failed_names.append(src.name) _log("---") return stats def print_summary( stats: CycleStats, cfg: EffectiveConfig, *, dry_run: bool, ) -> None: status = _cycle_status(stats, dry_run=dry_run) _print_section("Run summary") _log(f" Status: {status}") if cfg.inbox is not None: _log(f" Inbox: {cfg.inbox}") if cfg.watch_audio and cfg.recordings is not None: _log(f" Outputs: {cfg.recordings}") if (cfg.watch_transcripts or cfg.watch_audio) and cfg.transcripts is not None: _log(f" Transcripts: {cfg.transcripts}") if dry_run: _log(f" Would convert: {stats.would_convert}") _log(f" Would copy: {stats.would_copy}") _log(f" Would stage: {stats.would_stage}") _log(f" Would invoke missing: {stats.would_invoke_missing}") _log(f" Would admit: {stats.would_admit}") _log(f" Would backup: {stats.would_backup}") _log(f" Would delete: {stats.would_delete}") _print_limited_items("Would convert", stats.converted_names) _print_limited_items("Would copy", stats.copied_names) _print_limited_items("Would stage", stats.staged_names) else: _log(f" Converted: {stats.audio_converted}") _log(f" Copied: {stats.transcripts_copied}") _log(f" Staged: {stats.staged}") if stats.staged_reused: _log(f" Reused staged: {stats.staged_reused}") _log(f" Backed up: {stats.originals_backed_up}") _log(f" Deleted: {stats.originals_deleted}") _log(f" Missing runs: {stats.missing_invoked}") if cfg.admit_to_library: _log(f" Admit runs: {stats.admitted}") _print_limited_items("Converted", stats.converted_names) _print_limited_items("Copied", stats.copied_names) _print_limited_items("Staged", stats.staged_names) if stats.skipped_names: _log(" Skipped:") for name, reason in stats.skipped_names[:12]: _log(f" • {name} ({reason})") if len(stats.skipped_names) > 12: _log(f" • ... and {len(stats.skipped_names) - 12} more") else: skipped_n = stats.audio_skipped + stats.transcripts_skipped _log(f" Skipped: {skipped_n}") if stats.unstable_names: _print_limited_items("Unstable", stats.unstable_names) elif stats.unstable: _log(f" Unstable: {stats.unstable}") if stats.failed_names: _print_limited_items("Failed", stats.failed_names) else: _log(f" Failed: {stats.failed}") if not dry_run and (stats.audio_converted or stats.transcripts_copied): if cfg.admit_to_library and stats.admit_failed == 0: _log(" Next: new transcripts should already be in the managed library") else: _log(" Next: import transcripts in the web UI if they are not managed yet") _log("---") _log() def main(argv: Sequence[str] | None = None) -> int: for stream in (sys.stdout, sys.stderr): reconfigure = getattr(stream, "reconfigure", None) if callable(reconfigure): try: reconfigure(line_buffering=True) except (OSError, ValueError): pass args = parse_args(argv) config_path = resolve_config_path(args) cfg = resolve_config(args, config_path=config_path) if args.show_config: print(json.dumps(config_to_dict(cfg), indent=2)) return 0 if args.save_config: save_config(cfg, config_path) print(f"Saved config: {config_path}") if not _has_meaningful_paths(cfg): print( "Nothing to do. Set inbox (and recordings/transcripts as required) via CLI, " ".transcriptx/inbox-watch.json, or INBOX_WATCH_* / TRANSCRIPTX_* env; " "or use --show-config / --save-config.", file=sys.stderr, ) return 2 layout_error = validate_layout(cfg) if layout_error: print(f"ERROR: {layout_error}", file=sys.stderr) return 2 if cfg.delete_originals and not cfg.backup_wavs and cfg.move_processed is None: print( "WARNING: originals will be deleted with no WAV backup. " "Pass --backup-wav unless you are sure you do not need the inbox files.", file=sys.stderr, ) assert cfg.inbox is not None watch_loop = bool(args.watch_loop) if not args.dry_run and not cfg.inbox.is_dir(): if watch_loop: _log(f"Waiting for inbox: {cfg.inbox}") else: print(f"ERROR: inbox is not a directory: {cfg.inbox}", file=sys.stderr) return 2 ffmpeg: Path | None = None missing: Path | None = None if cfg.watch_audio: ffmpeg = find_ffmpeg(cfg.ffmpeg) if ffmpeg is None and not args.dry_run: print("ERROR: ffmpeg not found (set --ffmpeg or PATH).", file=sys.stderr) return 2 if ffmpeg is None: ffmpeg = Path("ffmpeg") missing = find_whispermlx_missing(cfg.whispermlx_missing) if missing is None and not args.dry_run: print( "ERROR: whispermlx-missing not found " "(set --whispermlx-missing or install the sibling script).", file=sys.stderr, ) return 2 if missing is None: missing = Path("whispermlx-missing") admit_python: Path | None = None if cfg.admit_to_library: admit_python = find_admit_python(cfg.admit_python) if admit_python is None and not args.dry_run: print( "ERROR: no Python that can import transcriptx " "(set --admit-python to the native TranscriptX venv, " "for example .transcriptx/bin/python).", file=sys.stderr, ) return 2 if admit_python is None: admit_python = Path(sys.executable) first = True total_failed = 0 inbox_absent_logged = not cfg.inbox.is_dir() try: while True: if watch_loop and not args.dry_run: if cfg.inbox.is_dir(): if inbox_absent_logged: _log(f"Inbox is available: {cfg.inbox}") inbox_absent_logged = False else: if not inbox_absent_logged: _log(f"Waiting for inbox: {cfg.inbox}") inbox_absent_logged = True stats = process_cycle( cfg, dry_run=args.dry_run, force=args.force, ffmpeg=ffmpeg, stability_checks=args.stability_checks, stability_interval_ms=args.stability_interval_ms, stability_timeout_ms=args.stability_timeout_ms, ) wrote_audio = stats.audio_converted > 0 or stats.would_convert > 0 wrote_tx = stats.transcripts_copied > 0 or stats.would_copy > 0 if cfg.watch_audio and missing is not None: if (not watch_loop) or first or wrote_audio: maybe_run_missing( cfg, stats, missing=missing, dry_run=args.dry_run ) if cfg.admit_to_library and admit_python is not None: if (not watch_loop) or first or wrote_audio or wrote_tx: maybe_run_admit( cfg, stats, python=admit_python, dry_run=args.dry_run ) print_summary(stats, cfg, dry_run=args.dry_run) total_failed += stats.failed if not watch_loop: break first = False time.sleep(max(cfg.interval_seconds, 0.1)) except KeyboardInterrupt: _log() _log("Stopped.") return 0 if total_failed == 0 else 1 return 1 if total_failed > 0 else 0 if __name__ == "__main__": raise SystemExit(main())