【CAD図面PDFをExcelへ自動転記するツール開発第9回】複数の図面PDFをフォルダごとまとめて処理する（進捗表示・中止・失敗しても止まらない一括処理）

2026年2月17日2026年9月29日

前回（第8回）では、部屋名とExcelの欄の対応付けを設定ファイル（YAML）に移し、様式の違うテンプレートにもコードを変えずに転記できるようにしました。

第9回は、フォルダに入った複数の図面PDFをまとめて処理する一括処理をPythonで作ります。ポイントは次の3つです。

「1件ずつ確認」か「まとめて流して結果一覧で確認」かを先に決める。今回は後者で、問題が見つかったPDFは「要確認」として転記しない
1件が壊れていてもそのPDFを「失敗」にして次へ進む。ログには図面の中身を書かない
重い処理は別スレッドで動かし、画面の更新はメインスレッドだけで行う。処理中も画面が固まらず、進捗表示と「中止」ができる

今回作る機能と完成イメージ

メイン画面に「7. フォルダ内のPDFをまとめて処理する」ボタンを追加します。

フォルダ直下のPDFを名前順に1件ずつ処理し、「成功（転記済み）」「要確認」「失敗」と理由を一覧に出す
進捗バーで何件目か分かり、「中止」で残りを止められる
PDFのフォルダの中に 転記結果_<日時> を作り、Excelとログを保存する
「要確認」「失敗」の行をダブルクリックすると、そのPDFをメイン画面に読み込み、「4.」→「5.」→「6.」で1件ずつ処理できる

第5回から残っていた「外部AIの通信中に画面が固まる」問題も、同じ仕組みで解消します。

仕組みの説明：一括処理の運用は2通りある

「1件ずつ確認」と「まとめて流して結果一覧で確認」

運用	流れ	向いている場面	作るときの負担
A. 1件ずつ確認画面を挟む	読むたびに確認画面を開き、確定後に転記	件数が少ない・誤転記の影響が大きい	処理を止めて人を待つ仕組みと、複数ページの確認画面
B. まとめて流して結果一覧で確認	問題の無いものは自動で転記し、残りを「要確認」で残す	件数が多い・埋め込み文字の図面が中心	「要確認」の判定ルールと結果一覧

今回はBを実装し、「要確認」のPDFはAと同じ1件ずつの流れ（第6回の確認画面。1ページ単位）へ引き継ぎます。ただし、Bの「成功」は人が確認する前に転記されたもので、第6回からの「人が確認してから転記する」方針を「転記後に結果一覧と出力Excelで確認する」運用に置き換えることになります。

「要確認」にする条件

条件	理由
部屋名候補が0件	文字が無い・辞書に無い表記の可能性
OCRで読んだページがある（既定）	読み取り漏れは信頼度に表れない
OCRの信頼度が80未満の部屋名がある	誤読の可能性
同じ部屋名が複数ある	1階と2階のトイレか二重検出かは人が判断
欄に収まらない・欄の無い部屋名がある	第8回の設定で書き込めない
30ページを超えた	先頭30ページしか読んでいない

しきい値80は架空の図面で決めた暫定値です。150dpi相当のスキャンでは部屋名の信頼度が91〜96、72dpi相当では75・76が出ました。72dpi相当では「バルコニー」が読めず候補から消えていました。読めなかった部屋名は低信頼度の候補としても残らないため、OCRで読んだPDFは既定で常に要確認です（画面のチェックで外せます）。

画面が固まる理由と、別スレッドへの分け方

tkinterは、クリックも再描画も1つのスレッド（メインスレッド）で順番に処理します。ボタンの処理の中で通信やOCRを待つと、その間は再描画も「中止」も受け付けません。公式ドキュメントでも、時間のかかる処理はイベントハンドラーの中で行わず、小分けにするか別スレッドで動かすよう説明されています。

📰 出典：Python公式ドキュメント tkinter「Threading model」

そこで、OCR・外部AIとの通信・一括処理全体は別スレッド（ワーカースレッド）で動かし、画面の更新はメインスレッドだけで行います。両者は queue.Queue（スレッド間で安全に受け渡せる待ち行列）でつなぎ、メインスレッドは after()（一定時間後に関数を呼ぶ仕組み）で定期的にキューを見ます。同じドキュメントには別スレッドからのtkinter呼び出しが失敗する条件も書かれています。

もう1つの注意点はPyMuPDFです。公式ドキュメントに「複数のスレッドで動かすことはサポートしていない」と明記されています。

📰 出典：PyMuPDF公式ドキュメント「Multiprocessing」

そこで、PyMuPDFを呼ぶ区間をすべて1つのロック（同時に1つのスレッドしか入れない仕組み）で囲み、一括処理の画面を開いている間はメイン画面を操作できないようにしました。単票の読み取りでは画像化をメインスレッドで済ませ、別スレッドには画像だけを渡します。公式が勧める別プロセス（multiprocessing）は、exe化（第11回）との相性も含め扱いが重くなるため採用していません。

実装

ファイル構成（第9回時点の差分）

app/
├ core/
│  ├ batch.py          … 新規：フォルダ内のPDFを1件ずつ読み取り・判定・転記、結果一覧とログ
│  ├ background.py     … 新規：重い処理を別スレッドで動かし、結果をメインスレッドへ渡す
│  ├ pdf_loader.py     … 変更：PyMuPDFを呼ぶ区間をロック（PDF_LOCK）で囲む
│  ├ text_extractor.py … 変更：文字の取り出しもロックの内側で。ExtractedWordに信頼度を追加
│  ├ ocr_local.py      … 変更：画像化済みのページをOCRする ocr_rendered_page() を切り出し
│  └ ocr_ai.py         … 変更：画像化済みのページを送る ai_ocr_rendered_page() を切り出し
└ ui/
   ├ batch_dialog.py   … 新規：一括処理の画面（フォルダ選択・開始／中止・進捗バー・結果一覧）
   └ main_window.py    … 変更：「7.」ボタン、単票の読み取りを別スレッドへ、進捗表示と「中止」
tests/
├ test_batch.py        … 新規22件
└ test_background.py   … 新規6件
tools/
├ make_sample_pdf.py   … 変更：--batch で一括処理の確認用フォルダ（壊れたファイルを含む）を作る
└ capture_screenshots.py … 変更：--batch モード

新しいパッケージは追加していません（threading・queue はPythonの標準ライブラリです）。

フォルダ内のPDFを列挙する（`app/core/batch.py`）

フォルダ直下のPDFを名前順に並べます。一時ファイル（~$ で始まる）や隠しファイルは除きます。

# app/core/batch.py（抜粋）
def list_pdf_files(folder: str | Path) -> list[Path]:
    directory = Path(folder)
    if not directory.is_dir():
        raise BatchError(f"フォルダが見つかりません: {directory}")
    files = [
        path
        for path in directory.iterdir()
        if path.is_file() and path.suffix.lower() == ".pdf" and not path.name.startswith(("~$", "."))
    ]
    return sorted(files, key=lambda p: unicodedata.normalize("NFKC", p.name).casefold())

1件分の処理：例外は「失敗」にして次へ進む

process_pdf() は1件の読み取り（全ページ。埋め込み文字が無いページだけOCR）→判定→転記を行います。どんな例外もこの中で受け止めて「失敗」として返すので、呼び出し側の繰り返しは止まりません。

# app/core/batch.py（抜粋）
    try:
        page_count, candidates, sources, reasons = _read_pdf(path, dictionary, options, ocr_func, cancel_event)
        review = ReviewList.from_candidates(
            candidates, source=SOURCE_OCR if SOURCE_OCR in sources else SOURCE_EMBEDDED, dictionary=dictionary
        )
        rooms = tuple(item.name for item in review.items)
        reasons += _judge(candidates, review, sources, options)
        # （中略：第8回の plan_cells() で欄に収まるかを確かめ、収まらなければ reasons に追加）

        common = {"page_count": page_count, "sources": tuple(sources), "rooms": rooms}
        if reasons:
            return result(STATUS_REVIEW, reasons, **common)
        # （中略：問題が無ければ第7回の write_rooms_to_excel() で転記）
        return result(STATUS_SUCCESS, [], output_path=written.output_path, **common)
    except _Cancelled:
        return result(STATUS_CANCELLED, [Reason(REASON_CANCELLED, "中止したため処理していません")], page_count=page_count)
    except PdfLoadError as exc:
        return result(STATUS_FAILED, [Reason(REASON_PDF_OPEN, str(exc))], page_count=page_count)
    except OcrError as exc:
        return result(STATUS_FAILED, [Reason(REASON_OCR_ERROR, str(exc))], page_count=page_count)
    # （中略：Excelの記入欄に値がある・書き込みエラーも同様に「失敗」）
    except Exception as exc:  # 想定外のエラーでも、このPDFを「失敗」にして次へ進む
        message = f"想定外のエラー（{type(exc).__name__}）：{exc}"
        return result(
            STATUS_FAILED, [Reason(f"{REASON_UNEXPECTED}:{type(exc).__name__}", message)], page_count=page_count
        )

二重検出の統合は第6回、欄の割り当ては第8回、書き込みは第7回の処理をそのまま使っています。

出力ファイル名がぶつからないようにする

出力は <PDF名>_転記.xlsx です。Windowsは大文字小文字を区別しないため、plan.pdf と PLAN.pdf は同じ名前になります。使った名前を大文字小文字・全角半角をそろえて記録し、既存のファイルも含めてぶつかったら _2 を付けます。出力フォルダも同様に前回分を上書きしません。

# app/core/batch.py（抜粋）
def unique_output_path(output_dir: Path, pdf_path: Path, suffix: str, reserved: set[str]) -> Path:
    base = f"{pdf_path.stem}_{OUTPUT_FILE_LABEL}"
    counter = 1
    while True:
        name = f"{base}{suffix}" if counter == 1 else f"{base}_{counter}{suffix}"
        key = unicodedata.normalize("NFKC", name).casefold()
        if key not in reserved and not (output_dir / name).exists():
            reserved.add(key)
            return output_dir / name
        counter += 1

ログには「何が起きたか」だけを書く

ログ（batch_log.txt）には日時・ファイル名・結果・ページ数・方式・候補の件数・処理時間・理由の種類だけを書き、部屋名、例外のメッセージ本文、APIキーは書きません。ログは共有フォルダに置かれたり、問い合わせで社外に送られたりしやすいためです。理由は画面用の文章とログ用の種類（件数だけ）を分けて持たせています。

# batch_log.txt の例（日時は記事の想定日に置き換えています）
[2026-02-17 10:00:00] 4/8 サンプルD邸_粗いスキャン.pdf 結果=要確認 ページ=2 方式=ocr 候補=6件 時間=0.43秒 理由=low_confidence(2),ocr_used
[2026-02-17 10:00:00] 8/8 サンプルH邸_壊れたファイル.pdf 結果=失敗 ページ=0 方式=- 候補=0件 時間=0.00秒 理由=pdf_open

重い処理を別スレッドで動かす（`app/core/background.py`）

BackgroundJob は関数を別スレッドで実行し、途中経過・結果・例外をキューに入れます。メインスレッドは after() で定期的に取り出してコールバックを呼びます。after に当たる関数を引数で受け取るので、pytestでは偽のイベントループで検証できます。

# app/core/background.py（抜粋）
    def start(self) -> None:
        if self._thread is not None:
            raise RuntimeError("この処理はすでに開始しています。")
        # daemon=True：処理の途中でウィンドウを閉じても、アプリの終了を妨げないようにする
        self._thread = threading.Thread(target=self._run, name="madori-ocr-worker", daemon=True)
        self._thread.start()
        self._schedule(self._poll_ms, self._poll)

    def post(self, message: Any) -> None:
        """ワーカースレッドから途中経過を送る（画面は触らず、キューに入れるだけ）。"""
        self._queue.put((_KIND_MESSAGE, message))

    def _run(self) -> None:
        try:
            result = self._work(self)
        except BaseException as exc:  # どんな例外でも画面側へ渡し、スレッドの中で握りつぶさない
            self._queue.put((_KIND_ERROR, exc))
        else:
            self._queue.put((_KIND_DONE, result))

    def _poll(self) -> None:
        """メインスレッドで呼ばれ、キューにたまったものを順に取り出してコールバックへ渡す。"""
        while True:
            try:
                kind, payload = self._queue.get_nowait()
            except queue.Empty:
                break
            if kind == _KIND_MESSAGE:
                if self._on_message is not None:
                    self._on_message(payload)
                continue
            self._finished = True
            if kind == _KIND_DONE:
                self._on_done(payload)
            else:
                self._on_error(payload)
            return  # 完了したので、これ以上キューを見に行かない
        self._schedule(self._poll_ms, self._poll)

「中止」は threading.Event に合図を立てるだけです。スレッドは外から安全に止められないため、ページの区切りと次のPDFの前で合図を確認し、残りを「未処理」にします。

一括処理の画面（`app/ui/batch_dialog.py`）

「開始」で run_batch() を別スレッドで呼びます。途中経過は job.post でキューに入れるだけで、進捗バーと一覧はメインスレッドの _on_progress() で更新します。

# app/ui/batch_dialog.py（抜粋）
        def work(job: BackgroundJob) -> BatchSummary:
            # ここは別スレッドで動く。画面（ウィジェット）には触らず、途中経過は job.post で送る
            return run_batch(
                pdfs,
                template=self._template,
                mapping=self._mapping,
                dictionary=self._dictionary,
                output_dir=output_dir,
                options=options,
                on_progress=job.post,
                cancel_event=job.cancel_event,
            )

        self._job = BackgroundJob(
            work, schedule=self.after, on_done=self._on_done, on_error=self._on_error, on_message=self._on_progress
        )
        self._job.start()

単票の外部AI・OCRも別スレッドへ（`app/ui/main_window.py`）

「3.」「4.」ボタンの読み取りも同じ仕組みに載せました。画像化はメインスレッドで済ませ、OCR・外部AIだけを別スレッドに任せるため、「画像化済みのページを受け取る」関数（ocr_rendered_page()・ai_ocr_rendered_page()）を切り出しています。

# app/ui/main_window.py（抜粋。外部AIの場合）
        try:
            rendered = document.render_page(page_number, zoom=1.0)  # PyMuPDFはこのスレッドで
        except PdfLoadError as exc:
            messagebox.showerror("外部AIエラー", str(exc))
            return
        client = self._create_ai_client(api_key)
        self._run_in_background(
            lambda _job: ai_ocr_rendered_page(rendered, client=client, model=model).words,
            message="外部AIで読み取っています（通信中）...「中止」で結果を捨てられます",
            on_done=lambda words: self._deliver_detection(document, page_number, words, "ai", on_detected),
            error_title="外部AIエラー",
        )

処理中はボタンを押せなくし、進捗バーを動かします。読み取り中にページやPDFを切り替えた場合と「中止」を押した場合は、戻ってきた結果を捨てます（送信済みなら外部AIの料金はかかる旨も表示）。

動作確認の方法

pytest は297件すべて成功しました（第8回までの268件に、一括処理22件・別スレッドの仕組み6件・OCRの信頼度1件を追加。Tesseract 5.3.4＋日本語データのある環境）。ruff check . もエラーはありません。

pytest tests/test_batch.py tests/test_background.py
python tools/make_sample_pdf.py --batch /tmp/madori-sample/batch   # 架空の図面8件（1件は壊れたファイル）

テストでは、壊れたファイルを先頭に置いても・想定外の例外が起きても後ろのPDFが処理されること、「成功」だけExcelが作られ正しい欄に入ること、中止で残りが「未処理」になること、架空のAPIキーを環境変数に入れて実行してもログにキー・部屋名・例外メッセージが出ないことを確かめています。

次は、架空の図面8件のフォルダを処理中の画面と、終了後の結果一覧です（開発環境（Linux）での確認画面です。Windows実機では見た目が異なります）。

一括処理の画面。4件目のスキャン図面をOCRで処理中で、進捗バーと処理済み3件の結果が表示されている

一括処理の結果一覧。成功2件・要確認5件・失敗1件と、それぞれの理由が色分けして表示されている

実測した処理時間と「画面が固まらない」ことの確認

Linux開発環境（Intel Xeon 2.1GHz・4vCPU、Tesseract 5.3.4）で、架空の図面を複製した50件のフォルダを2回処理しました。

図面の種類	件数	1件あたり（中央値）
埋め込み文字の図面（2ページ、Excel転記まで）	30件	0.011〜0.013秒
スキャン図面 150dpi相当（1ページ、OCR）	10件	0.22〜0.23秒
粗いスキャン図面 72dpi相当（2ページ、OCR）	10件	0.43〜0.46秒
合計（全体の時間）	50件	7.0〜7.3秒

外部AIは、3秒待ってから応答するダミーで確かめました。第8回までの作りでは画面の更新間隔が最大3.1秒空きました（その間は固まっている）。別スレッドに移した後は、ボタンの処理が十数ミリ秒で戻り、通信中の更新間隔は最大30ミリ秒台でした。

確認できていないこと

Windows実機での画面・操作感、実際の外部AI（Claude API）での処理時間。外部AIでの一括処理は未実装です
実在の図面・大判や多ページの図面・共有フォルダ越しでの処理時間と「要確認」の割合
PyMuPDFをロックで1スレッドずつ別スレッドから呼ぶ形の長時間の安定性。Linuxの競合試験では異常終了しませんでしたが、公式がサポートする使い方ではないため、Windows実機で長時間試す必要があります

つまずきやすい点・セキュリティ上の注意

別スレッドからウィジェットを触らない：途中経過は必ずキューに入れ、メインスレッドで画面を更新します
PyMuPDFの取り合いで遅くなる：画面側で画像化を連続させながら一括処理を走らせると、50件の処理が7秒台から1〜2分に延びました。処理中は一括処理の画面をモーダル（閉じるまで他を操作できない状態）にしています
中止はすぐには効かない：処理中のOCR1ページ・通信1回が終わるまで待ちます
ログにもファイル名とパスは残る：ファイル名に顧客名・物件名が入るなら、ログも図面と同じ扱いで保管します

発注者向けメモ

「1件ずつ確認」か「まとめて流して結果だけ確認」かで、設計と工数が変わる

「1件ずつ確認」は、処理を止めて人の確認を待つ仕組みと複数ページの確認画面が必要になり、画面まわりの工数が増えます。「まとめて流して結果だけ確認」は、どんな場合に人が見るべきかという判定ルールの決め込みと結果一覧に工数がかかります。

後者の「成功」は人の確認前に転記されたものです。誤転記がどこまで許されるか（後工程で気付けるか、取引先へそのまま出るか）で選ぶべき運用が変わります。迷う場合は、今回のように「問題の無いものだけ自動転記し、残りは1件ずつ確認」から始め、実際の「要確認」の割合を見て判断するのが現実的です。

処理件数・処理時間の見積もりは「図面の種類ごとの1件あたり時間 × 件数」で考える

処理時間は、埋め込み文字の有無で桁が変わります。今回の実測では埋め込み文字の図面が1件0.01秒ほど、スキャン図面が1ページ0.2〜0.25秒ほどでした。ただし架空の簡単な図面での値で、実際の大判・高解像度の図面では長くなります。実際の図面を種類ごとに数件試して1件あたりの時間を測り、件数を掛けるのが確実です。人の確認時間の方がずっと大きくなりやすい点も見込みます。

失敗したときの運用を先に決めておく

壊れたファイルなど、一括処理では何件かの失敗が出る前提で考えます。「失敗したら全体を止める」作りだと、1件のために全体をやり直すことになります。今回は失敗しても次へ進み理由を残す作りですが、失敗・要確認のPDFを誰がいつ処理するか、再実行時に前回の出力とどう突き合わせるかは運用で決める必要があります。

☐ 1件ずつ確認か、結果だけ確認かを決めたか
☐ 「要確認」にする条件を業務側と合意したか
☐ 実際の図面で1件あたりの処理時間を測ったか
☐ 失敗・要確認のPDFの担当と再実行の手順、ログの保管場所を決めたか

開発会社への質問例

「1件が失敗したとき、残りはどうなりますか？理由はどこで確認できますか？」
「人の確認なしで転記されるのは、どんな条件の図面ですか？条件は変えられますか？」
「処理中に画面が固まったり、中止できなかったりしませんか？」
「ログに図面の内容やAPIキーが残ることはありませんか？」

まとめと次回予告

第9回では、フォルダ内の図面PDFをまとめて処理する app/core/batch.py と、重い処理を別スレッドで動かす app/core/background.py を実装しました。1件が失敗しても残りを続け、「成功／要確認／失敗」と理由の一覧で確認でき、「要確認」は1件ずつの確認画面へ引き継ぎます。外部AIの通信中に画面が固まる問題も解消しました。

次回（第10回）は「ローカル完結モードと外部AIモードを切り替えられるようにする」です。設定画面で読み取り方式を切り替えられるようにし、APIキーをWindowsの資格情報マネージャーへ安全に保存します。