【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 time
import logging
from pathlib import Path
from natsort import natsorted
import pandas as pd
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__)


# ===== 設定クラス =====
@dataclass
class Config:
    src_dir: Path  # PDFファイルの元フォルダ
    dst_dir: Path  # 作業用フォルダ(自動生成)
    file_slice: slice  # 処理するファイルの範囲
    page_slice: slice  # 抽出するページの範囲
    row_slice: slice  # 抽出する行の範囲
    output_csv: str  # 出力CSVファイル名


# ===== PDFファイルの準備 =====
def prepare_files(config: Config) -> list[Path]:
    """PDFファイルを作業フォルダに移動"""
    config.dst_dir.mkdir(parents=True, exist_ok=True)
    logger.info(f"作業フォルダ準備: {config.dst_dir}")

    # ファイル一覧取得と選択
    all_files = natsorted(config.src_dir.glob("*.pdf"))
    selected = all_files[config.file_slice]

    if not selected:
        logger.warning("PDFファイルが見つかりません")
        return []

    logger.info(f"{len(selected)}ファイルを処理します")

    # ファイル移動
    for pdf in selected:
        dst = config.dst_dir / pdf.name
        if not dst.exists():
            shutil.move(str(pdf), str(dst))
            logger.info(f"移動: {pdf.name}")

    return  natsorted(config.dst_dir.glob("*.pdf"))





# ===== 行のマージ処理 =====
def merge_rows(rows_with_marks: list[tuple[list[str], str]]) -> tuple[
    list[list[str]], list[str]]:
    """分割された行をマージし、ページマークを保持"""
    merged_rows = []
    merged_marks = []
    pending_mark = None
    i = 0

    while i < len(rows_with_marks):
        row, mark = rows_with_marks[i]

        # 空行の処理
        if not any(row):
            if mark:
                pending_mark = mark
            i += 1
            continue

        # 次の行が継続行か確認
        if i + 1 < len(rows_with_marks):
            next_row, _ = rows_with_marks[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 "")
                pending_mark = None
                i += 2
                continue


        # 前の行への継続の場合
        if row[0] == "" and any(row[1:]) and merged_rows:
            prev_row = merged_rows.pop()
            prev_mark = merged_marks.pop()
            row = [(a or "") + (b or "") for a, b in zip(prev_row, row)]
            merged_rows.append(row)
            merged_marks.append(prev_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(pdf_path: Path, page_slice: slice, row_slice: slice) -> tuple[
    list[str], list[list[str]], list[str]]:
    """PDFからテーブルデータを抽出"""
    found = False
    with pdfplumber.open(pdf_path) as pdf:

        all_rows_marks = []
        header = None #ヘッダー行を自動認識させずテーブル行として扱う


        # 指定ページから抽出
        for i, page in enumerate(pdf.pages[page_slice],
                                 start=page_slice.start or 0):
            table = page.extract_table()
            if not table or len(table) < 2:
                continue

            if header is None:
                header = table[0]

            data_rows = table[1:2]
            if data_rows:
                page_mark = f"{i + 1}p"
                all_rows_marks.extend(
                    (row, page_mark) for row in data_rows)
                found = True


        # フォールバック: 全ページスキャン
        if not found:
            for page_index,page in enumerate(pdf.pages):
                table = page.extract_table()
                if table is None or len(table) < 2:
                    continue
                header = table[0]
                data_rows = table[1:2]
                if not data_rows:
                    continue
                page_mark = f"{page_index + 1}p"
                all_rows_marks.extend((row, page_mark) for row in data_rows)

                found = True


                print(f"⚠️ フォールバック発動: {pdf_path.name}")
                break


    if header and all_rows_marks:
        merged_rows, marks = merge_rows(all_rows_marks)
        return header, merged_rows, marks

    logger.error(f"テーブルが見つかりません: {pdf_path.name}")
    return [], [], []


# ===== DataFrameの作成 =====
def create_df(header: list[str], rows: list[list[str]], pdf_path: Path,
              marks: list[str]) -> pd.DataFrame:
    """抽出データからDataFrameを作成"""
    df = pd.DataFrame(rows, columns=header)

    # 改行削除
    df.replace(r'[\r\n]', '', regex=True, inplace=True)

    # ファイル名とページ番号を追加
    df.insert(0, 'file_name', [pdf_path.stem] + [''] * (len(df) - 1))
    df.insert(1, 'page_number', marks)

    # 継続行のページ番号を空白に
    if "Quality" in df.columns and "Library" in df.columns:
        prev_mark = ""
        for idx in range(len(df)):
            row = df.iloc[idx] # ilocとは指定した行番号（インデックス番号）のデータを1行だけ取り出す
            mark = df.at[idx, 'page_number']
            is_continuation = str(row.get("Quality", "")).strip() == "" and str(
                row.get("Library", "")).strip() != ""

            if is_continuation or mark == prev_mark:
                df.at[idx, 'page_number'] = ""
            else:
                prev_mark = mark

    return df


# ===== CSV追記 =====
def append_csv(df: pd.DataFrame, output_path: str):
    """DataFrameをCSVに追記"""
    csv_path = Path(output_path)
    write_header = not csv_path.exists() or csv_path.stat().st_size == 0
    df.to_csv(output_path, mode='a', index=False, encoding='utf-8-sig',
              header=write_header)


# ===== メイン処理 =====
def convert_pdfs(config: Config):
    """PDFファイルをCSVに一括変換"""
    pdf_files = prepare_files(config)
# prepare_files(config) が返した pdf_files
# その 1つ1つの PDF を順番に処理する
# この時点で 「1 PDF = 1 回の処理」 という構造が確定しています
    for idx, pdf_file in enumerate(pdf_files, start=1):
        logger.info(f"処理中 {idx}/{len(pdf_files)}: {pdf_file.name}")

        try:
            header, rows, marks = extract_table(pdf_file, config.page_slice,
                                                config.row_slice)
            if not header or not rows:
                continue

            df = create_df(header, rows, pdf_file, marks)
            append_csv(df, config.output_csv)

        except Exception as e:
            logger.error(f"エラー {pdf_file.name}: {e}")

        # 30ファイルごとに休憩
        if idx % 30 == 0:
            logger.info("30ファイル処理完了。3秒休憩...")
            time.sleep(3)


# ===== クリーンアップ =====
def cleanup(directory: Path):
    """作業フォルダのPDFファイルを削除"""
    for pdf in directory.glob("*.pdf"):
        pdf.unlink()
    logger.info("✅ 処理完了。作業フォルダのファイルを削除しました")


# ===== 実行 =====
if __name__ == "__main__":
    config = Config(
        src_dir=Path(r"C:/Users/xxxx/Desktop/Hongo_C190"),
        dst_dir=Path(r"C:/Users/xxxx/Desktop/PythonProject/Collection_Hongo"),
        file_slice=slice(None),  # 全ファイル
        page_slice=slice(0, 1),  # 1ページ目のみ
        row_slice=slice(1, 2),  # 2行目のみ
        output_csv='Smart_extracted.csv'
    )

    convert_pdfs(config)
    cleanup(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ファイルに変換する。
いかがでしたでしょうか。
おやくに立てたらさいわいです。