#!/usr/bin/env python3 import os import sys import time import subprocess import logging import shutil from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler # Set dynamic library path for whisper-cli if needed os.environ["DYLD_LIBRARY_PATH"] = ( "/Users/bertm/Dropbox/_INBOX_RECEIVING/__AUDIO-IN/whisper.cpp-master/build/src:" "/Users/bertm/Dropbox/_INBOX_RECEIVING/__AUDIO-IN/whisper.cpp-master/build/ggml/src:" "/Users/bertm/Dropbox/_INBOX_RECEIVING/__AUDIO-IN/whisper.cpp-master/build/ggml/src/ggml-blas:" "/Users/bertm/Dropbox/_INBOX_RECEIVING/__AUDIO-IN/whisper.cpp-master/build/ggml/src/ggml-metal" ) # --- Configuration --- HOME = os.path.expanduser("~") BASE_DIR = os.path.join(HOME, "Dropbox", "_INBOX_RECEIVING", "__AUDIO-IN") INPUT_DIR = os.path.join(BASE_DIR, "_01_Input") PROCESSED_DIR = os.path.join(BASE_DIR, "_02_Processed") TRANSCRIPT_DIR = os.path.join(BASE_DIR, "_03_Transcripts") LOGS_DIR = os.path.join(BASE_DIR, "_LOGS") # Paths for external tools FFMPEG_CMD = "ffmpeg" WHISPER_CPP_CMD = "/Users/bertm/Dropbox/_INBOX_RECEIVING/__AUDIO-IN/whisper.cpp-master/build/bin/whisper-cli" # update if necessary WHISPER_MODEL = "/Users/bertm/Dropbox/_INBOX_RECEIVING/__AUDIO-IN/whisper.cpp-master/models/ggml-base.en.bin" # Audacity pipes PIPE_TO_AUDACITY = "/tmp/audacity_script_pipe.to.503" PIPE_FROM_AUDACITY = "/tmp/audacity_script_pipe.from.503" # --- Logging Setup --- if not os.path.exists(LOGS_DIR): os.makedirs(LOGS_DIR) logging.basicConfig( filename=os.path.join(LOGS_DIR, "unified_process.log"), level=logging.DEBUG, format="%(asctime)s [%(levelname)s] %(message)s", ) console = logging.StreamHandler(sys.stdout) console.setLevel(logging.INFO) formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s") console.setFormatter(formatter) logging.getLogger("").addHandler(console) # --- Utility Functions --- def run_ffmpeg_conversion(input_file, output_file): logging.info(f"Converting {input_file} to {output_file}") cmd = [ FFMPEG_CMD, "-i", input_file, "-vn", "-acodec", "libmp3lame", "-q:a", "2", output_file, ] result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode == 0: logging.info(f"Successfully converted {input_file}") return True else: logging.error(f"FFmpeg conversion failed for {input_file}: {result.stderr.decode()}") return False def send_audacity_command(command, ignore_errors=False, retries=3, delay=0.5): for attempt in range(retries): if not os.path.exists(PIPE_TO_AUDACITY): logging.error(f"Pipe {PIPE_TO_AUDACITY} does not exist. Is Audacity running?") return "ERROR: Pipe does not exist" try: with open(PIPE_TO_AUDACITY, 'w') as pipe: pipe.write(command + '\n') time.sleep(delay) # Allow Audacity to process if not os.path.exists(PIPE_FROM_AUDACITY): logging.error(f"Response pipe {PIPE_FROM_AUDACITY} does not exist") continue with open(PIPE_FROM_AUDACITY, 'r') as pipe: response = pipe.read().strip() logging.info(f"Audacity response: {response}") if "BatchCommand finished: OK" in response or ignore_errors: return response else: logging.warning(f"Command failed: {command}") return response except Exception as e: logging.exception(f"Exception while sending Audacity command: {command}. Attempt {attempt + 1}") time.sleep(delay) return f"ERROR: Command {command} failed after {retries} attempts" def process_audacity(input_file, output_file): logging.info(f"Starting Audacity processing for {input_file}") # Import file resp = send_audacity_command(f'Import2: Filename="{input_file}"') if "BatchCommand finished: OK" not in resp: logging.error("Audacity import failed.") return False # Select all send_audacity_command("SelectAll:") # Apply high pass filter using the correct Audacity command resp = send_audacity_command("High-passFilter: Frequency=80 Rolloff=dB48", ignore_errors=True) if "BatchCommand finished: OK" not in resp: logging.warning("High pass filtering failed. Continuing without it.") # Other processing commands processes = [ "GraphicEQ: 100=1.5 200=-2.5 300=-2.0 2000=2.5 3000=2.0 8000=1.5 10000=1.0", "Compressor: Threshold=-18 Ratio=2.0 AttackTime=0.01 ReleaseTime=0.06 UsePeak=1", "Normalize: PeakLevel=-1 RemoveDcOffset=1 ApplyGain=1", "Limiter: Type=HardLimit ThresholdLevel=-1 HoldTime=0.01 ApplyMakeupGain=0", ] for cmd in processes: send_audacity_command(cmd, ignore_errors=True) # Export file resp = send_audacity_command(f'Export2: Filename="{output_file}" Format="MP3"') success = "BatchCommand finished: OK" in resp # Clean up send_audacity_command("RemoveTracks:") if success: logging.info(f"Audacity processing succeeded, exported to {output_file}") return True else: logging.error(f"Audacity processing failed for {input_file}") return False def transcribe_with_whisper(audio_file, transcript_file): logging.info(f"Starting transcription for {audio_file}") cmd = [ WHISPER_CPP_CMD, "-m", WHISPER_MODEL, "-f", audio_file, "-l", "en" ] try: with open(transcript_file, "w") as out: result = subprocess.run(cmd, stdout=out, stderr=subprocess.PIPE) if result.returncode == 0: logging.info(f"Transcription completed for {audio_file}") return True else: logging.error(f"Transcription failed for {audio_file}: {result.stderr.decode()}") return False except Exception as e: logging.exception(f"Exception during transcription for {audio_file}") return False # --- Event Handler --- class AudioFileHandler(FileSystemEventHandler): def on_created(self, event): if event.is_directory: return filepath = event.src_path logging.info(f"Detected new file: {filepath}") time.sleep(1) # Give time for file to finish writing # Determine file name and paths filename = os.path.basename(filepath) name_no_ext, ext = os.path.splitext(filename) ext = ext.lower() # Avoid double-appending "_processed" if "_processed" in name_no_ext: processed_filename = f"{name_no_ext}.mp3" else: processed_filename = f"{name_no_ext}_processed.mp3" processed_file = os.path.join(PROCESSED_DIR, processed_filename) transcript_file = os.path.join(TRANSCRIPT_DIR, f"{name_no_ext}_transcript.txt") # Create directories if they don't exist for d in [INPUT_DIR, PROCESSED_DIR, TRANSCRIPT_DIR]: os.makedirs(d, exist_ok=True) # --- Conversion Step --- if ext != ".mp3": if run_ffmpeg_conversion(filepath, processed_file): os.remove(filepath) filepath = processed_file else: logging.error(f"Conversion failed for {filename}.") return else: # Even if it is mp3, move it to PROCESSED_DIR (rename it as processed) shutil.move(filepath, processed_file) filepath = processed_file logging.info(f"File moved/converted to: {filepath}") # --- Audacity Processing Step --- if process_audacity(filepath, processed_file): # Optionally, you can keep or remove the original file here pass else: logging.error(f"Audacity processing failed for {filename}.") return # --- Transcription Step --- if transcribe_with_whisper(processed_file, transcript_file): logging.info(f"Transcription completed: {transcript_file}") else: logging.error(f"Transcription failed for {filename}.") def main(): logging.info("Starting unified audio processing service...") observer = Observer() event_handler = AudioFileHandler() observer.schedule(event_handler, INPUT_DIR, recursive=False) observer.start() try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() observer.join() logging.info("Service stopped.") if __name__ == "__main__": main()