アプリ内の致命的なエラーを見つけた時のコード
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以上をログに出力する設定。
エラーを投げられたワーカー側の処理
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の引数が代入される感じ。
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側の処理
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を受けっとったワーカー
@pyqtSlot(bool)
def set_user_decision(self, ok: bool):
self._decision = bool(ok) # ← QEventLoop側が見ているフラグに代入
try:
pass
except Exception:
pass
self._decisionに回答が入ると、手前で話した_tickが反応しループを終える。
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を閉じて回答を返す。
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に信号がいく
except Exception:
self._log.exception("CSV import failed")
self.error.emit(traceback.format_exc())
外側のtry、exceptでself._log.exceptionを使ってログに書き込みつつ、エラーのシグナルをUI側のメインウィンドウに渡してエラー文を表示させる。
原因
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に飛ばしてると思う。