Introducing HeadlessApp for true headless mode

Author: kaixxx

noScribe.py

@@ -1020,34 +1020,39 @@ def _update_progress_display(self):
                 font=("", font_size)
             )
 
+def _init_app_state(app):
+    app._headless = False
+    app.user_models_dir = os.path.join(config_dir, 'whisper_models')
+    os.makedirs(app.user_models_dir, exist_ok=True)
+    whisper_models_readme = os.path.join(app.user_models_dir, 'readme.txt')
+    if not os.path.exists(whisper_models_readme):
+        with open(whisper_models_readme, 'w') as file:
+            file.write('You can download custom Whisper-models for the transcription into this folder. \n'
+                       'See here for more information: https://github.com/kaixxx/noScribe/wiki/Add-custom-Whisper-models-for-transcription')
+
+    app.queue = TranscriptionQueue()
+    app.audio_files_list = []
+    app.transcript_files_list = []
+    app.log_file = None
+    app.cancel = False # if set to True, transcription will be canceled
+    # If True, cancel only the currently running job (triggered from queue row "X")
+    app._cancel_job_only = False
+    app.current_progress = -1
+    # Track background activity for robust shutdown
+    app._worker_threads = []
+    app._mp_proc = None
+    app._mp_queue = None
+    app._ffmpeg_proc = None
+    app._shutting_down = False
+
+
 class App(ctk.CTk):
     def __init__(self):
         super().__init__()
 
         self.protocol("WM_DELETE_WINDOW", self.on_closing)
-        
-        self.user_models_dir = os.path.join(config_dir, 'whisper_models')
-        os.makedirs(self.user_models_dir, exist_ok=True)
-        whisper_models_readme = os.path.join(self.user_models_dir, 'readme.txt')
-        if not os.path.exists(whisper_models_readme):
-            with open(whisper_models_readme, 'w') as file:
-                file.write('You can download custom Whisper-models for the transcription into this folder. \n' 
-                           'See here for more information: https://github.com/kaixxx/noScribe/wiki/Add-custom-Whisper-models-for-transcription')            
-        
-        self.queue = TranscriptionQueue()
-        self.audio_files_list = []
-        self.transcript_files_list = []
-        self.log_file = None
-        self.cancel = False # if set to True, transcription will be canceled
-        # If True, cancel only the currently running job (triggered from queue row "X")
-        self._cancel_job_only = False
-        self.current_progress = -1
-        # Track background activity for robust shutdown
-        self._worker_threads = []
-        self._mp_proc = None
-        self._mp_queue = None
-        self._ffmpeg_proc = None
-        self._shutting_down = False
+
+        _init_app_state(self)
 
         # configure window
         self.title('noScribe - ' + t('app_header'))
@@ -1500,6 +1505,8 @@ def update_scrollbar_visibility(self):
 
     def update_queue_table(self):
         """Update the queue table by diffing: update existing rows, add new ones, remove missing."""
+        if getattr(self, '_headless', False):
+            return
         current_keys = []
         for i in range(len(self.queue.jobs)):
             job = self.queue.jobs[i]
@@ -1801,6 +1808,8 @@ def update_queue_table(self):
 
     def update_queue_controls(self):
         """Enable/disable and label the queue control buttons based on state."""
+        if getattr(self, '_headless', False):
+            return
         try:
             has_running = len(self.queue.get_running_jobs()) > 0
             has_pending = self.queue.has_pending_jobs()
@@ -2041,7 +2050,7 @@ def log(self, txt: str = '', tags: list = [], where: str = 'both', link: str = '
         if where != 'file': 
             if txt[:-1] != t('welcome_instructions'):
                 print(txt, end='')            
-            if hasattr(self, 'log_textbox') and self.log_textbox.winfo_exists():
+            if not getattr(self, '_headless', False) and hasattr(self, 'log_textbox') and self.log_textbox.winfo_exists():
                 try:
                     self.log_textbox.configure(state=tk.NORMAL)
                     # To prevent slowing down the UI, limit the content of log_textbox to max 5000 characters
@@ -2084,7 +2093,7 @@ def logn(self, txt: str = '', tags: list = [], where: str = 'both', link:str = '
 
     def logr(self, txt: str = '', tags: list = [], where: str = 'both', link:str = '', tb: str = '') -> None:
         """ Replace the last line of the log """
-        if where != 'file':
+        if where != 'file' and not getattr(self, '_headless', False) and hasattr(self, 'log_textbox') and self.log_textbox.winfo_exists():
             self.log_textbox.configure(state=ctk.NORMAL)
             tmp_txt = self.log_textbox.get("end-1c linestart", "end-1c")
             self.log_textbox.delete("end-1c linestart", "end-1c")
@@ -2193,6 +2202,8 @@ def button_transcript_file_event(self):
 
     def set_progress(self, step, value, speaker_detection='none'):
         """ Update state of the progress bar """
+        if getattr(self, '_headless', False):
+            return
         progr = -1
         if step == 1:
             progr = value * 0.05 / 100
@@ -2379,13 +2390,14 @@ def transcription_worker(self, start_job_index=None):
             self.logn(t('processing_time', total_time_str=total_time_str))
 
             # open editor if only a single file was processed
-            if queue_jobs_processed == 1 \
+            if not getattr(self, '_headless', False) \
+                    and queue_jobs_processed == 1 \
                     and job \
                     and job.file_ext == 'html' \
                     and job.status == JobStatus.FINISHED \
                     and get_config('auto_edit_transcript', 'True') == 'True':
                 self.launch_editor(job.transcript_file)
-            elif queue_jobs_processed > 1:
+            elif queue_jobs_processed > 1 and not getattr(self, '_headless', False):
                 # if more than one job has been processed, switch to queue tab for an overview 
                 self.tabview.set(self.tabview._name_list[1])
 
@@ -3365,16 +3377,85 @@ def on_closing(self):
                 pass
             self.destroy()
 
-def run_cli_mode(args):
-    """Run noScribe in CLI mode"""
+
+class HeadlessApp(App):
+    def __init__(self):
+        # Do not initialize Tk/CTk to avoid DISPLAY requirements
+        _init_app_state(self)
+        self._headless = True
+
+    def __getattr__(self, name):
+        # Avoid Tk attribute delegation recursion when CTk isn't initialized
+        raise AttributeError(name)
+
+
+def _cleanup_app(app):
+    """Cleanup app instance for proper process exit in CLI mode."""
     try:
-        # Create a minimal app instance to access model paths and logging
-        app = App()
-        # Hide GUI window for headless execution
+        # Signal shutdown to stop background activity
+        app._shutting_down = True
+        app.cancel = True
+
+        # Terminate active multiprocessing child (diarization/whisper) if present
+        if getattr(app, "_mp_proc", None) is not None:
+            try:
+                if app._mp_proc.is_alive():
+                    app._mp_proc.terminate()
+                    app._mp_proc.join(timeout=1.0)
+            except Exception:
+                pass
+            finally:
+                app._mp_proc = None
+
+        # Close multiprocessing queue
+        if getattr(app, "_mp_queue", None) is not None:
+            try:
+                app._mp_queue.close()
+                app._mp_queue.join_thread()
+            except Exception:
+                pass
+            finally:
+                app._mp_queue = None
+
+        # Terminate ffmpeg if currently converting
+        if getattr(app, "_ffmpeg_proc", None) is not None:
+            try:
+                if app._ffmpeg_proc.poll() is None:
+                    app._ffmpeg_proc.terminate()
+                    app._ffmpeg_proc.wait(timeout=1.0)
+            except Exception:
+                try:
+                    app._ffmpeg_proc.kill()
+                except Exception:
+                    pass
+            finally:
+                app._ffmpeg_proc = None
+
+        # Join worker threads briefly
+        for th in list(getattr(app, "_worker_threads", [])):
+            try:
+                th.join(timeout=0.5)
+            except Exception:
+                pass
+    except Exception:
+        pass
+
+    if not getattr(app, '_headless', False):
+        try:
+            app.quit()
+        except Exception:
+            pass
         try:
-            app.withdraw()
+            app.destroy()
         except Exception:
             pass
+
+def run_cli_mode(args):
+    """Run noScribe in CLI mode"""
+    app = None
+    try:
+        # Create a headless app instance (no GUI initialization)
+        app = HeadlessApp()
 
         # Validate and set the whisper model
         available_models = app.get_whisper_models()
@@ -3445,12 +3526,16 @@ def run_cli_mode(args):
     except Exception as e:
         print(f"Error: {str(e)}")
         return 1
+    finally:
+        if app is not None:
+            _cleanup_app(app)
 
 def show_available_models():
     """Show available Whisper models"""
+    app = None
     try:
-        # Create minimal app instance to get models
-        app = App()
+        # Create headless app instance to get models
+        app = HeadlessApp()
         models = app.get_whisper_models()
 
         print("Available Whisper models:")
@@ -3459,9 +3544,11 @@ def show_available_models():
 
         if not models:
             print("  No models found. Please check your installation.")
-            
     except Exception as e:
         print(f"Error getting models: {str(e)}")
+    finally:
+        if app is not None:
+            _cleanup_app(app)
 
 if __name__ == "__main__":
     # Parse command line arguments