アプリ内の致命的なエラーを見つけた時のコード

.py

except UnicodeDecodeError:
            self._log.exception("BOX CSV decode error (expect UTF-8): %s", path)
            raise RuntimeError("CSVの文字コードをUTF-8にしてください（BOM付UTF-8推奨）")

手前のtryでインポート対象がutf-8-sngでなかった場合
logger.exception()がlogger.isEnabledFor()を呼び出して、エラーと判定したのちに"%s"のフォーマットにPathを入れて、ログを出力。
raiseで上書きをして、呼び出し側（ワーカーファイル）に投げる（メッセージボックス用）
ちなみに、logger.errorでも、エラー判定になるが、exceptionだとスタックトレースが追加でログに書き込まれる。

logger.info(...) → INFO
logger.warning(...) → WARNING
logger.error(...) / logger.exception(...) → ERROR（exceptionはERROR+スタックトレース）
logger.critical(...) → CRITICAL

アプリ上の設定はINFO以上をログに出力する設定。

エラーを投げられたワーカー側の処理

.py

repo = SQLiteSignalRepository(self.db_path)
imported = 0
            
try:
    if self.mode == "signals":
        importer = CSVSignalImporter(repo)
        imported = importer.import_file(self.csv_path, progress_cb=self._progress_cb, cancel_cb=self._cancel_cb)

except Exception as e:
    if self._confirm(f"CSV取り込み中に致命的エラーが発生しました。\n{e}\n\nこのファイルをスキップして続行しますか？"):
        self._warnings.append(str(e))
        imported = 0
    else:
        raise

非同期処理を行なっているので、repoでワーカースレッド内でのリポジトリを作成し、（UIと分離）importedで取り込み件数の初期化を行う。

mode = "signals"なら、CSVSignalImporterを使う（手前で記述した層）
importerでの実施状況を呼び、行ごとの進捗と、キャンセルを行うか、UI層に投げる。
ここでエラーを拾うとexceptに行きimporterで返されたRuntimeErrorがExceptionに入る
RuntimeErrorは組み込み例外Exceptionのサブクラスで、イメージは例外が発生した瞬間に作られるeという変数にRuntimeErrorの引数が代入される感じ。

.py

def _confirm(self, message: str) -> bool:
        """致命的エラー発生時、UIにYes/Noを問い合わせて同期待ちする。
        QEventLoop を用いる。UI側は ask_confirm を受けて QMessageBox で質問し、
        set_user_decision(True/False) を呼ぶことで回答を返す。
        """
        self._decision = None
        self.ask_confirm.emit(message)
        loop = QEventLoop()

        # set_user_decision が呼ばれるまで待つ
        # QEventLoop を終了させるタイミングは、以下のtickで監視する
        from PyQt5.QtCore import QTimer
        def _tick():
            if self._decision is not None:
                loop.quit() #回答が来たらループ終わり
        timer = QTimer()
        timer.timeout.connect(_tick)
        timer.start(50)
        loop.exec_() #ここで待機
        timer.stop()
        return bool(self._decision) #ここで回答を返す。

ここでUIにask_confirmでシグナルとraiseした文を送る（手前にシグナル定義済み）
その間ワーカーは回答待ちで、ひたすらループして待機。

UI側の処理

.py

    def _connect_common_worker_signals(self, worker, thread, *, title='取り込み中', label='処理中…'):
        # 開始時: 進捗ダイアログを開く
        worker.started.connect(lambda: self._open_progress(title, label, worker))
        # 進捗
        worker.progress.connect(self._update_progress)
        # レポート/エラー/キャンセル/確認
        worker.report.connect(self._on_worker_report)
        worker.error.connect(self._on_worker_error)
        worker.canceled.connect(lambda: self.statusBar().showMessage('キャンセルされました'))
        worker.ask_confirm.connect(self._on_worker_confirm) #ここでシグナルを受け取る
        # 終了時: ダイアログを閉じてスレッドを畳む。thread.quitが呼ばれるとスレッドを削除する（deleteLaterが動く）

        

    def _on_worker_confirm(self, message: str):
        # Yes/No をユーザーに問い合わせ、回答を現在のワーカーへ返す
        res = QMessageBox.question(self, '確認', message, QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
        try:
            worker = getattr(self, '_current_worker', None)
            if worker is not None:
                worker.set_user_decision(res == QMessageBox.Yes) #ここで回答をワーカーに返す。
        except Exception:
            pass

ask_confirmを受けるとQMessageBox.questionでYes/Noを取得し、boolとしてその結果を現在実行中のworker.set_user_decision(...)へ渡す。

※getatterで今実行中のスレッドが存在していれば、その参照（ワーカー）に回答を返せるようにする。

YES or NOを受けっとったワーカー

.py

@pyqtSlot(bool)
def set_user_decision(self, ok: bool):
   self._decision = bool(ok)  # ← QEventLoop側が見ているフラグに代入
   try:
       pass
   except Exception:
       pass

self._decisionに回答が入ると、手前で話した_tickが反応しループを終える。

.py

def _confirm(self, message: str) -> bool:
        """致命的エラー発生時、UIにYes/Noを問い合わせて同期待ちする。
        QEventLoop を用いる。UI側は ask_confirm を受けて QMessageBox で質問し、
        set_user_decision(True/False) を呼ぶことで回答を返す。
        """
        self._decision = None
        self.ask_confirm.emit(message)
        loop = QEventLoop()

        # set_user_decision が呼ばれるまで待つ
        # QEventLoop を終了させるタイミングは、以下のtickで監視する
        from PyQt5.QtCore import QTimer
        def _tick():
            if self._decision is not None:
                loop.quit() #回答が来たらループ終わり
        timer = QTimer()
        timer.timeout.connect(_tick)
        timer.start(50)
        loop.exec_() #ここで待機
        timer.stop()
        return bool(self._decision) #ここで回答を返す。

self._decision=Noneが変わることによって、loopを閉じて回答を返す。

.py

    def run(self):
        self.started.emit()
        try:
            self._log.info("CSV import start: path=%s mode=%s", self.csv_path, self.mode)
            repo = SQLiteSignalRepository(self.db_path)
            imported = 0
            try:
                if self.mode == "signals":
                    importer = CSVSignalImporter(repo)
                    imported = importer.import_file(self.csv_path, progress_cb=self._progress_cb, cancel_cb=self._cancel_cb)
                else:
                    importer = CSVBoxConnImporter(repo)
                    imported = importer.import_file(self.csv_path, progress_cb=self._progress_cb, cancel_cb=self._cancel_cb)
            except Exception as e:
                # 致命的（ファイル壊れ/列不整合など）: ユーザーに続行(=スキップ)か中止かを確認、_confirmでユーザーの回答を待つ
                if self._confirm(f"CSV取り込み中に致命的エラーが発生しました。\n{e}\n\nこのファイルをスキップして続行しますか？"):
                    self._warnings.append(str(e))
                    imported = 0
                else:
                    raise

NOを選択すればelseに入って上の層のtryに信号がいく

.py

        except Exception:
            self._log.exception("CSV import failed")
            self.error.emit(traceback.format_exc())

外側のtry、exceptでself._log.exceptionを使ってログに書き込みつつ、エラーのシグナルをUI側のメインウィンドウに渡してエラー文を表示させる。

原因

.py

        except Exception:
            self._log_exception("CSV import failed")
            self.error.emit(traceback.format_exc())

一番最後のself._log_exceptionが定義されていないため、例外を出してアプリが落ちていただけでした。

__init__でself._log = logging.Loggerオブジェクトを定義していたので、self._logにそんなメソッドは存在しないとのことでした。

ちなみに、_cofirmで帰ってきたbool型のFalseは外側のtryが受け取ってexeptionに飛ばしてると思う。