【pdfplumber・pandas】PDFテーブルデータを一括CSV化するPython自動化ツール

pdfplumberを使用してPDFから表形式のデータを抽出し、pandasを使ってデータを整形後、CSV形式に変換します。

0.概要

　大量のPDFファイル上のテーブルデータをCSVファイルに変換および統合するプログラムを作成しました。このプログラムは、手作業で行われがちなPDFデータの集計作業に対し、タイムパフォーマンスの改善を図ることを狙いとしています。
　本記事は、ノンプログラマーの方を想定しており、プログラミング学習者向きではありません（コード活用には良いかもしれません)。そのため、ここではコードの説明は必要最小限とし、環境構築、プログラムの実行、出力結果の説明に重きを置いています。

　なお、OSについては、ここではWindows10(Windows11でも確認済み)での利用を前提にしており、PyCharmを快適に動作させるためには、PCスペックの最小要件と推奨要件を以下にまとめます。

最小要件(推奨要件)
RAM：2GB (推奨は4GB以上)　※
CPU：マルチコアCPU
ディスク容量：3.5GB以上の空き容量
※2GBは辛い(4GBでたまにストレス、8GBで快適、両方確認済み)

0-1.対象サンプル

• 1ファイルの形式：下記サンプルのように3ページ、10行前後の構成（構成に応じてカスタム可能。カスタムについては記事末尾に掲載）
• 対象ファイル数：メモリ負荷が大きいため、始めは100程度のPDFファイルで試験実行することを推奨します
  （以降のサンプル画像では分析機器メーカー名の箇所を黒塗りにしています）
  
  
  

　上記の対象ファイル群では、ファイルによっては以下のような状態の存在があります。

• セル内、およびページをまたいだ改行
• ここには例示してないが、1ページ目にテーブルがない
• 空行の存在

　以上のような状態を整理及び削除して統合・変換されたデータ群に、ファイル名とページ番号を追加することにより、データを確認しやすい、スマートなcsvファイルが得られます。
なお、ここでは空の列の削除はあえて行っていません。
(元のPDFの構造を維持することでユーザーがファイルの変換を見定めるため。)

0-2. 出力結果

出力結果の例は下記のようになります。

1. 準備(環境構築)

1-1. Pythonをインストール

「特定のリリースをお探しですか?」でなるべく3.10か3.11を選択

1-2. PyCharmのインストールと初期設定

　上記サイトにしたがってPyCharmのインストールと日本語表記の変換を行って下さい。また同サイトの「簡単なPythonプログラムを作成して実行してみる」を読み、新しいプロジェクトの作成までを行って下さい。
※新しいプロジェクト名は「PythonProject」として、デスクトップを保存先に指定して下さい。
※なお、「新しいPythonファイルの作成」以降はここでは行わないで下さい。

1-3. Pythonパッケージのインストール

(1)　サイドバーのPythonパッケージをクリック。下記の赤枠の検索ボックスに「natsort」と入力して「Enter」を押下。

(2)　「インストール」をクリック。エディタ―画面最下部の青線が消えるまで待つ。

(3)　上記(1)、(2)と同様の手順で、「pdfplumber」「pandas」をインストールする。
なお、上記インストールしたライブラリは、あくまでもこのプロジェクト内にインストールされるものですので、プロジェクトを削除すると、インストールしたライブラリも削除されます。注意して下さい。
(4)　下記赤枠の「-」をクリック。

(5)　「Python パッケージ」のエリアが消えたら、赤枠「PythonProject」を右クリック。

(6)　下記、「新規」を選択し、続いて「ファイル」をクリック。

(7)　「新規ファイル」欄に「sample.py」(名前はなんでもよい)と入力し、「Enter」し押下。

(8)　下記の画面が表示され、コードが打ち込めるようになる。

2.プロジェクトの実行

2-1. ファイル群の配置

(1)　PyCharmで作った'PythonProject'の新規フォルダがデスクトップに作成されていることを確認(PyCharmでの操作が問題なく反映されているかどうかの確認)。
(2)　PythonProjectフォルダとは別のフォルダをデスクトップに作成し、CSVに変換・統合したいPDFファイル群をそのフォルダに格納する。
(3)　[重要!]フォルダに格納したPDFファイルは、プログラムが実行されるとPCから自動的に削除(ゴミ箱に入らず完全削除)されるので、原本は必ず別途保存して下さい!

2-2コードの挿入

(1)　下記が今回使用するコード。

python sample.py

import pdfplumber
import shutil
import pathlib
from pathlib import Path
from natsort import natsorted
import pandas as pd
import time
from typing import List, Tuple
import logging
from typing import List, Tuple, Optional
from dataclasses import dataclass

# =====ロギングインフォメーション=========

logging.basicConfig(
    level=logging.INFO,
    format='[%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler("log.txt",encoding="utf-8"),
        logging.StreamHandler(),
    ]
)
logger = logging.getLogger(__name__)

# =====UserCustom構成要素の属性=======
@dataclass
class Config:
    src_dir: Path
    dst_dir: Path
    file_slice: slice
    page_slice: slice
    row_slice: slice
    output_csv: str
# ======PDFファイル展開ののための実行操作
def prepare_pdf_files(config: Config) -> List[Path]:
    config.dst_dir.mkdir(parents=True, exist_ok=True)
    logger.info(f"保存先フォルダの準備: {config.dst_dir}")

    all_pdf_files = natsorted(config.src_dir.glob("*.pdf"), key=lambda x: str(x))
    selected_files = all_pdf_files[config.file_slice]

    if not selected_files:
        logger.warning("ソースフォルダにPDFファイルが見つかりません。")
    else:
        logger.info(f"{len(selected_files)}ファイルが選択された")
    for pdf_file in selected_files:
        dst_path = config.dst_dir / pdf_file.name
        if not dst_path.exists():
            shutil.move(str(pdf_file), str(dst_path))
            logger.info(f"移動したファイル:{pdf_file.name}")
        else:
            logger.info(
                f"{pdf_file.name}は、すでに存在するのでこの作業はとばします")


    return natsorted(config.dst_dir.glob("*.pdf"), key=lambda x: str(x))

def merge_split_rows_with_page_marks(row_mark_pairs: List[Tuple[List[str], str]]) -> Tuple[List[List[str]], List[str]]:
    merged_rows, merged_marks = [], []
    i, pending_mark = 0, None

    while i < len(row_mark_pairs):
        row, mark = row_mark_pairs[i]
        if not any(row):
            if mark:
                pending_mark = mark
            i += 1
            continue

        if i + 1 < len(row_mark_pairs):
            next_row, _ = row_mark_pairs[i + 1]
            if next_row[0] == "" and any(next_row[1:]):
                row = [(a or "") + (b or "") for a, b in zip(row, next_row)]
                # ✅利用可能な場合はこの行のマークを優先し、それ以前の空白行の保留マークをフォールバックする。
                merged_rows.append(row)
                merged_marks.append(mark or pending_mark or "")
                i += 1
                pending_mark = None
                continue

        if row[0] == "" and any(row[1:]) and merged_rows:
            previous_row = merged_rows.pop()
            previous_mark = merged_marks.pop()
            row = [(a or "") + (b or "") for a, b in zip(previous_row, row)]
            merged_rows.append(row)
            merged_marks.append(previous_mark)
            i += 1
            continue

        merged_rows.append(row)
        # ✅ 前の空白行から保存された保留中のマークを優先し、この行のマークにフォールバックする。
        merged_marks.append(pending_mark or mark or "")
        pending_mark = None
        i += 1

    return merged_rows, merged_marks

def extract_table_data(pdf_file: Path, page_slice: slice, row_slice: slice) -> Tuple[List[str], List[List[str]], List[str]]:
    with pdfplumber.open(pdf_file) as pdf:
        all_data_rows = []
        header = None
        table_found = False

        for i, page in enumerate(pdf.pages[page_slice], start=page_slice.start or 0):
            table = page.extract_table()
            if table is None or len(table) < 2:
                continue
            if header is None:
                header = table[0]
            data_rows = table[1:][row_slice]
            if not data_rows:
                continue
            page_mark = f"{i + 1}p"
            all_data_rows.extend((row, page_mark) for row in data_rows)
            table_found = True

        if not table_found:
            for page in pdf.pages:
                table = page.extract_table()
                if table is None or len(table) < 2:
                    continue
                header = table[0]
                data_rows = table[1:][row_slice]
                if not data_rows:
                    continue
                page_mark = "1p"
                all_data_rows.extend((row, page_mark) for row in data_rows)
                print(f"⚠️ Used fallback scan for: {pdf_file.name}")
                break

    if header and all_data_rows:
        merged_rows, page_marks = merge_split_rows_with_page_marks(all_data_rows)
        return header, merged_rows, page_marks
    else:
        print(f"❌ No usable table or data in: {pdf_file.name}")
        return [], [], []

def create_dataframe(header: List[str], rows: List[List[str]], pdf_file: Path, page_marks: List[str]) -> pd.DataFrame:
    df = pd.DataFrame(rows, columns=header)
    df.replace(r'[\r\n]', '', regex=True, inplace=True)
    df.insert(0, 'file_name', [pdf_file.stem] + [''] * (len(df) - 1))
    df.insert(1, 'page_number', page_marks)

    if "Quality" in df.columns and "Library" in df.columns:
        previous_mark = ""
        for idx_row, (row, mark) in enumerate(zip(df.itertuples(index=False), df['page_number'])):
            is_continuation = str(getattr(row, "Quality", "")).strip() == "" and str(getattr(row, "Library", "")).strip() != ""
            if is_continuation or mark == previous_mark:
                df.at[idx_row, 'page_number'] = ""
            else:
                previous_mark = mark
    else:
        print(f"⚠️ Missing expected columns in: {pdf_file.name}")

    return df

def append_to_csv(df: pd.DataFrame, output_csv: str):
    write_header = not Path(output_csv).is_file() or Path(output_csv).stat().st_size == 0
    df.to_csv(output_csv, mode='a', index=False, encoding='utf-8-sig', header=write_header)

def convert_pdf_to_csv(config: Config):
    pdf_files = prepare_pdf_files(config)

    for idx, pdf_file in enumerate(pdf_files, start=1):
        logger.info(f"ファイル処理 {idx}/{len(pdf_files)}: {pdf_file.name}")
        header, rows, page_marks = extract_table_data(pdf_file, config.page_slice, config.row_slice)
        if not header or not rows:
            continue
        try:
            df = create_dataframe(header, rows, pdf_file, page_marks)
            append_to_csv(df, config.output_csv)
        except Exception as e:
            logger.error(f"Error creating DataFrame for {pdf_file.name}: {e}")

        if idx % 30 == 0:
            print("⏸️ Sleeping after 30 files...")
            time.sleep(3)

def cleanup_files(directory: Path):
    for pdf_file in directory.glob("*.pdf"):
        pdf_file.unlink()
    print("✅ Processing complete. Files deleted from Collection_Hongo.")

# ===== User Custom Settings =====
if __name__ == "__main__":
    user_config = Config(
        src_dir=Path(r"C:/Users/xxx/Desktop/Hongo_C190"),
        dst_dir=Path(r"C:/Users/xxx/Desktop/PythonProject/Collection_Hongo"),
        file_slice = slice(None,None),
        page_slice=slice(0, 1),
        row_slice=slice(1, 2),
        output_csv='Smart_extracted.csv'
    )
    # ===================================
    convert_pdf_to_csv(user_config)
    cleanup_files(user_config.dst_dir)

※　上記コードは、大量のファイルを読み書きする際に、ディスクI/Oの負荷を軽減するために、30ファイル処理したらsleep()関数で処理間隔を3秒空ける仕様にしています。

(2)　上記コード右上に表示されるアイコンをクリックしてコピー。

(3)　sample.py画面(1-3.(8)のコード入力画面)で、最上段の左端にカーソルを合わせて、上記コードをペースト(最上段の左端にスペースが入らないようにし、インデントも(1)の状態のものが維持されるように注意する)。

(4)　ペーストしたコード末尾の「User Custom Settings」のsrc_dirとdst_dirのpathを適切なものに修正する。

• 「src_dir」
  2-1.(2)でPDFファイルを格納したフォルダのディレクトリを指定します。
• 「dst_dir」
  　ここで示しているフォルダは、プロジェクト実行と共に2-1.(1)のPythonProjectフォルダ内に自動生成されます。そのため、手作業で事前に作成する必要はありませんが、フォルダ名はここで自分仕様に変更しておいてください。
  　なお、dst_dirで指定したフォルダは、src_dirのフォルダから読み取ったPDFファイルの処理をPCが行うためのフォルダであり、人間側が作業するためのフォルダではありません。

 src_dir=Path(r"C:/Users/xxx/Desktop/Hongo_C190")
 dst_dir=Path(r"C:/Users/xxx/Desktop/PythonProject/Collection_Hongo")

※　dst_dir(PDFファイル移動先)のpathは、dst_dir = pathlib.Path(r"C:/User/ユーザー名/Desktop/PythonProject/Collectio_Hongo(任意名)")となります。

2-3. プロジェクトの実行

(1)　PyCharmのウィンドウ上で赤枠の三角形をクリック

(2)　下記のように実行画面に赤字でファイル処理が赤字で表示されます。
コードの理解につながると思います。

(3)　画面左のサイドバーでは、2-2.(4)のdst_dirで指定したフォルダ「Collection＿Hongo」（名前任意）が プログラム実行中に生成され、プロジェクト完了後にはoutput_csvで指定したSmart_extracted.csvもまた生成されます。Smart_extracted.csvはこのプログラム出力結果になります。

(4)　画面左のサイドバーに表示されているSmart_extracted.csvはデスクトップにドラッグ&ドロップして使用してください。
ドラッグ&ドロップができない場合
Smart_extracted.csvをダブルクリックしてPyCharm上で表示。「Ctrl」+「A」でテキスト全体を指定しメモ帳にコピペして保存。このテキストファイルをExcelにインポートして使用してください。

2-4. 出力結果

コードのカスタマイズについて

if __name__ == "__main__":
    user_config = Config(
        # ====== User Custom Setting ========
        src_dir=Path(r"C:/Users/xxx/Desktop/Hongo_C190"),
        dst_dir=Path(r"C:/Users/xxx/Deskto/PythonProject/Collection_Hongo"),
        file_slice=slice(None),
        page_slice=slice(0, 1),
        row_slice=slice(0, 1),
        output_csv='Smart_extracted.csv'
    )

　PDF上のテーブルからデータを抽出する指示は、コード末尾の「User Custom Settings」にあるfile_slice, page_slice, row_slice という変数の編集によって行うことができ、それぞれ以下のような役割があります。

• file_slice：１回の実行で抽出を行うPDFファイル数の指定
• page_slice：抽出するページ番号の指定
• row_slice：PDFファイル内のテーブルから抽出する行番号の指定

　それぞれの変数に対し、slice(A, B)を代入することで具体的な数字を指定します。このコマンドはA(数字)以上B(数字)未満を抽出対象とすることを示しています。後述の「指定の例」も参考にして頂ければと思います。

　なお、PDFファイル数やページ番号、行番号、の番号付けには注意が必要で0始まりとなっています。とくにPDFファイル上のテーブルの行数は以下のような番号付けを定義しているので、これに従ってください。

• １行目にあたるヘッダー：「0」行目
• ２行目以降のデータセル：「１」行目以降

Index in table	役割　説明	変数名
0	Column headers	header
1 以降	Data rows from PDF	data_rows

指定の例

• ファイル数の指定
  file_slice=slice(0, 10)：1個目(「0」)のファイルから10個目(「10」未満=「9」)のファイルまでを取得
  file_slice=slice(None)：ファイル数にかかわらず全ファイル取得

• ページ番号の指定
  page_slice=slice(0, 1)：１ページ目(「0」)から１ページ目(「1」未満=「0」)までを取得
  page_slice=slice(None)：ページ数にかかわらず全ページ取得

• テーブル行番号の指定 　
  row_slice=slice(2, 6)：ヘッダーを含め3行目(「2」)から6行目(「6」未満= 「5」)までを取得 　
  row_slice=slice(1, 2)：ヘッダーを含め2行目(「1」)から1行目(「2」未満=「1」)までを取得 　
  row_slice=slice(1, None)：（ヘッダーを除く）全行を取得

• Tips
  上記の例にある全ファイル取得、全ページ取得、全行取得という設定にした場合、「0-2. 出力結果サンプル」の形で 統合データを得られます。

その他にも、カスタマイズは可能です。
出力データを短時間で正確にcsvファイルに変換する。
いかがでしたでしょうか。
おやくに立てたらさいわいです。