def execute()

in operators.py [0:0]

• 57 lines of code
• 8 McCabe index (conditional complexity)

    def execute(self, context):
        if self.filename in get_available_models():
            self.report({"INFO"}, f"{self.filename} already downloaded")
            return {"CANCELLED"}

        import queue
        import re
        import sys
        import threading

        from huggingface_hub import hf_hub_download

        prefs = context.preferences.addons[__package__].preferences
        prefs.downloading = True
        prefs.download_progress = 0

        self._progress_queue = queue.Queue()

        class TqdmCapture:
            def __init__(self, queue, stream):
                self.queue = queue
                self.stream = stream
                self.original_write = stream.write
                self.original_flush = stream.flush

            def write(self, string):
                match = re.search(r"\r.*?(\d+)%", string)
                if match:
                    try:
                        percentage = int(match.group(1))
                        self.queue.put(percentage)
                    except Exception as e:
                        self.queue.put(f"Error parsing progress: {string}, {e}")
                self.original_write(string)
                self.original_flush()

            def flush(self):
                self.original_flush()

        def download_task():
            try:
                old_stderr = sys.stderr
                sys.stderr = TqdmCapture(self._progress_queue, sys.stderr)

                hf_hub_download(
                    self.repo_id,
                    filename=self.filename,
                    local_dir=get_models_dir(),
                )
                prefs = bpy.context.preferences.addons[__package__].preferences
                if prefs.downloading:
                    self._progress_queue.put("finished")
            except InterruptedError:
                pass
            except Exception as e:
                prefs = bpy.context.preferences.addons[__package__].preferences
                if prefs.downloading:
                    self._progress_queue.put(f"Error downloading model: {e}")
            finally:
                sys.stderr = old_stderr

        self._download_thread = threading.Thread(target=download_task)
        self._download_thread.start()

        wm = context.window_manager
        self._timer = wm.event_timer_add(0.1, window=context.window)
        wm.modal_handler_add(self)

        return {"RUNNING_MODAL"}