PdfPlumber 大量pPDFファイル一括処理:行が揺れてるPDF:コード解説_3

このシリーズの最終回でえす

前回はこちら

サンプルPDFファイルはこちら

🎯 今回の部分が何をするか

PDFから抽出したデータを、pandasのDataFrameに変換して、
CSVファイルに保存する流れ

extract_table() の結果
  ↓
create_df() でDataFrame化
  ↓
append_csv() でCSVに保存

🔵　extract_table()のフロー

図解:extractの入出力

入力
┌──────────────────┐
│ file001.pdf      │
│ page_slice       │
│ row_slice        │
└──────────────────┘
        ↓
  extract_table()
        ↓
出力（3つの値）
┌──────────────────────────────────────┐
│ header = ["Quality", "Library", ...] │
│ rows = [["High", ...], [...]]        │
│ marks = ["1p", "1p", "2p"]           │
└──────────────────────────────────────┘

🔵　create_df

convert_pdfsからの呼び出し
該当部分の抽出

def convert_pdfs(config: Config):
    """PDFファイルをCSVに一括変換"""
    pdf_files = prepare_files(config)
    
    # ============================================
    # ここから重要部分
    # ============================================
    for idx, pdf_file in enumerate(pdf_files, start=1):
        logger.info(f"処理中 {idx}/{len(pdf_files)}: {pdf_file.name}")

        try:
            # ステップ1: PDFから表を抽出
            header, rows, marks = extract_table(
                pdf_file, 
                config.page_slice,
                config.row_slice
            )
            
            # ステップ2: データがなければスキップ
            if not header or not rows:
                continue

            # ステップ3: DataFrameを作成
            df = create_df(header, rows, pdf_file, marks) ←　ここで呼び出される
            
            
            # ステップ4: CSVに追記
            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)

ステップバイステップ解説

ステップ1:create_df()引数

df = create_df(header, rows, pdf_file, marks)

reate_df()に渡される引数(実際のデータ)

create_df(
    header   = ["Quality", "Library", "Length"],
    rows     = [
                   ["High", "Lib1", "100"],
                   ["Medium", "Lib2", "200"]
               ],
    pdf_path = Path("file001.pdf"),
    marks    = ["1p", "1p"]
)

ステップ2: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)

コード

df = pd.DataFrame(rows, columns=header)

何をしている？

• リストのリストを表（DataFrame）に変換
• columns=header で列名を指定

詳細な図解

header（列名）
┌─────────┬─────────┬─────────┐
│ Quality │ Library │ Length  │
└─────────┴─────────┴─────────┘
            ↓ columns=header で列名として使う
            headerはextract_table()で抽出
            

rows（データ）
┌─────────┬─────────┬─────────┐
│ "High"  │ "Lib1"  │ "100"   │  ← 0行目
├─────────┼─────────┼─────────┤
│ "Medium"│ "Lib2"  │ "200"   │  ← 1行目
├─────────┼─────────┼─────────┤
│ "Low"   │ "Lib3"  │ "300"   │  ← 2行目
└─────────┴─────────┴─────────┘
            ↓ データとして使う

pd.DataFrame(rows, columns=header) ←　create_df()の引数になっている
            ↓

DataFrame（表形式のデータ）
   ┌─────────┬─────────┬─────────┐
   │ Quality │ Library │ Length  │ ← 列名（header）
   ├─────────┼─────────┼─────────┤
 0 │ High    │ Lib1    │ 100     │ ← rows[0]
   ├─────────┼─────────┼─────────┤
 1 │ Medium  │ Lib2    │ 200     │ ← rows[1]
   ├─────────┼─────────┼─────────┤
 2 │ Low     │ Lib3    │ 300     │ ← rows[2]
   └─────────┴─────────┴─────────┘
 ↑
インデックス（自動で0,1,2...が振られる）

対応関係

DataFrame の内部構造

列: Quality
┌───┬────────┐
│ 0 │ High   │
│ 1 │ Medium │
│ 2 │ Low    │
└───┴────────┘

列: Library
┌───┬────────┐
│ 0 │ Lib1   │
│ 1 │ Lib2   │
│ 2 │ Lib3   │
└───┴────────┘

列: Length
┌───┬────────┐
│ 0 │ 100    │
│ 1 │ 200    │
│ 2 │ 300    │
└───┴────────┘

これらが組み合わさって表になる

ステップ3:改行を削除

コード

df.replace(r'[\r\n]', '', regex=True, inplace=True)

何故必要？
PDFから取り出した」データには、時々改行文字が含まれています。

# 改行が含まれているデータの例
rows = [
    ["High\n", "Lib1", "100"],    # ← \n がある
    ["Medium", "Lib2\r", "200"]   # ← \r がある
]
これをそのままCSVにすると、レイアウトが崩れます。

r'[\r\n]'を分解:正規表現の説明
r''   # raw文字列（\を特別扱いしない）
\n  # 改行（Line Feed）下の行へ送る
\r  # 復帰（Carriage Return）行の先頭に戻す
\r\n  # Windowsの改行（両方）行の先頭で、下の行に移動する

例

text = "Hello\nWorld"
print(text)
# Hello
# World  ← 改行される

replace()の仕組み
基本構文

df.replace(置き換える文字, 置き換え後の文字, regex=True, inplace=True)

パラメータの意味

パラメータ 値 意味
第1引数 r'[\r\n]' 検索パターン（正規表現）
第2引数 '' 置き換え後の文字（空文字）
regex=True True 正規表現を使う
inplace=True True 元のDataFrameを直接変更

ステップ4:file_name列を追加

df.insert(0, 'file_name', [pdf_path.stem] + [''] * (len(df) - 1))

• (len(df) - 1)について:
  • ()は計算の優先(算数😁)
  • len(df)は行数、列数はlen(df.columns)他にもあるけど・・・
  • 1行1列目にfile_nameを配置
  • 全行からfile_nameを入れた1行を引く
    詳細な図解

リストの作成プロセス
(0,'file_name',......)
   0          ←　1番左、1列目
pdf_path.stem　←　"file001.pdf"の拡張子を除く
    ↓
"file001"     
    ↓
[pdf_path.stem] 
    ↓
["file001"]  ← 1要素のリスト

len(df) - 1　
    ↓
3 - 1 = 2
    ↓
[''] * 2
    ↓
['', '']  ← 2要素のリスト（空文字が2つ）

["file001"] + ['', '']
    ↓
["file001", '', '']  ← 3要素のリスト

なぜこうする？
目的: 1行1列目だけにファイル名を表示
仕上がり例:

     file_name  Quality  Library  Length
0    file001    High     Lib1     100
1               Medium   Lib2     200
2               Low      Lib3     300

ステップ5: page_number列を追加

df.insert(1, 'page_number', marks)

入力データの確認

marks = ["1p", "1p", "2p"]

これはextract_table()から渡されたページ番号のリスト

構文としては「テップ3:file_name列を追加」を参照

ステップ6:列の確認

if "Quality" in df.columns and "Library" in df.columns:

何をしてる？

• DataFrameにQuality列とLibrary列(df.columns)があるか確認
• pandasで行を表すのはdf.index
• 両方あれば実行、なければスキップ
  なぜ必要？
• PDFによって列名が違う場合がある
• エラーを防ぐための安全策

ステップ7:変数の初期化

prev_mark = ""

何をしている？

• 前の行のページ番号を記憶する変数
• 最初は空文字で初期化

ステップ8:各行をループ

for idx in range(len(df)):

len()は何?

• len() は そのオブジェクト(リストという1つの要素)の数を返す組み込み関数
• 列は「行の中に含まれる要素」であり、外側の階層ではない
• DataFrame(pandasの場合) の外側の階層（最上位）は行として設計
• したがってlen(df)は行

DataFrame
┌──────────────────────────────┐
│ 行0 → Series([...])          │  ← 最上位
│ 行1 → Series([...])          │  ← 最上位
│ 行2 → Series([...])          │  ← 最上位
└──────────────────────────────┘

ステップ9:1行を取り出す

row = df.iloc[idx]

ilocとは？

• integerlocation(整数位置)
• 行番号を指定してデータを取得
• rowはSeriesである(行切)

特徴	            Series	        list
ラベル（index）	 あり	         なし
データ型	       pandas 独自	   Python 標準
列名でアクセス	できる	　　　　　できない
例	　　　　　　row["Quality"]	row[0] のみ


# DataFrame
  file_name  page_number  Quality  Library  Length
0 file001    1p           High     Lib1     100
1            1p           Medium   Lib2     200
2            2p           Low      Lib3     300

# idx=0 の時
row = df.iloc[0]
# file_name      file001
# page_number    1p
# Quality        High
# Library        Lib1
# Length         100

ステップ10:ページ番号を取得

mark = df.at[idx, 'page_number']

atとは？

• 特定のセル(1つのデータ)を取得
• at[行番号, 列名]で指定

ステップ11:継続行かどうか判定

is_continuation = str(row.get("Quality", "")).strip() == "" and str(
    row.get("Library", "")).strip() != ""

何をしてる?

str(row.get("Quality", "")).strip() == ""

意味：Quality列が空かどうか
and: 両方の条件がTrueの時だけ("かつ"と読めば良い)

str(row.get("Library", "")).strip() != ""

意味：Library列が空でないか
is_continuationに代入(Trueなら)・・・継続行の判定ロジック

コードの内容:

• rowからQuality列の値を取得
  • 指定行(df.iloc[idx])から指定された列の値を取る
  • 列の値はSeries(辞書)のキーの値
  • 辞書型はdf(DataFrame)の列切り
  • 取得するデータはSeriesでなく「単一の値(スカラー)」になります
  • row.get()のrowはSeriesオジェクト

ステップ12: strip()

• 前後の空白（スペース、タブ、改行）を削除

ステップ13:ページ番号を空白にするか判定

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

条件1: is_continuation これがTrue → ページ番号は空白
条件2: mark == prev_mark　→ この行のページ番号と前の行のページ番号が同じ
                                  ↓
　　　　　　　　　　　　　　　　　ページ番号は空白
処理の分岐
else:
　　　prev_mark = mark → 条件に当てはまらない → prev_markを更新

結果：

変更前
  file_name  page_number  Quality
0 file001    1p           High
1            1p           Medium

idx=1 で実行
df.at[1, 'page_number'] = ""

変更後
  file_name  page_number  Quality
0 file001    1p           High
1                         Medium  ← 空白になった

📊 判定フローチャート

各行について
    ↓
┌─────────────────────┐
 Quality列が空かつ     
 Library列が空でない   
 （継続行？）        
└─────────────────────┘
    ↓         ↓
   YES       NO
    ↓         ↓
  空白に   ┌──────────────────┐
  する      前の行と同じ
   　　　　　ページ番号？     
          └──────────────────┘
               ↓         ↓
              YES       NO
               ↓         ↓
            空白に    prev_markを
            する      更新して次へ

🔵 append_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)

ステップ1:ファイルパスをPathオブジェクトに変換

csv_path = Path(output_path)

何をしている？

• 文字列をPathオブジェクトに変換
• ファイル操作が簡単になる

Pathオブジェクトとは？
普通の文字列都の違い

# 文字列の場合
path = "Smart_extracted4.csv"
# ファイルが存在するか確認するのが面倒
import os
exists = os.path.exists(path)

# Pathオブジェクトの場合
path = Path("Smart_extracted4.csv")
# メソッドで簡単に確認できる
exists = path.exists()

Pathオブジェクトの便利なメソッド

csv_path = Path("Smart_extracted4.csv")

# ファイルの存在確認
csv_path.exists()  # True/False

# ファイルサイズ取得
csv_path.stat().st_size  # バイト数

# ファイル名だけ取得
csv_path.name  # "Smart_extracted4.csv"

# 拡張子取得
csv_path.suffix  # ".csv"

ステップ2: ヘッダーを書くかどうか判定

write_header = not csv_path.exists() or csv_path.stat().st_size == 0

分解:

• csv_path.exists():

• ファイルが存在するかチェック
• 存在する → True,存在しない → False

• csv_path.stat().st_size

• 何をしている？
  • ファイルのサイズ（バイト数）を取得
• stat() とは？
  • ファイルの詳細情報（メタデータ）を取得
  • サイズ、更新日時、アクセス権限など
• stat().st_size
  • データがある
    図解

┌──────────────────────┐
│ ファイルが存在する？     
└──────────────────────┘
        ↓
      YES（True）
        ↓ not
       NO（False）
        ↓
┌──────────────────────┐
│ ファイルサイズは0？  
└──────────────────────┘
        ↓
       NO（False）
        ↓
  False or False = False
        ↓
  write_header = False
  （ヘッダーを書かない）

ステップ3: DataFrameをCSVに書き込む

df.to_csv(output_path, mode='a', index=False, encoding='utf-8-sig',
          header=write_header)

パラメータの詳細解説
パラメータ1: output_path

output_path = 'Smart_extracted4.csv'

保存先のファイル名
パラメータ2: mode='a'

'r'  # Read（読み込み専用）
'w'  # Write（書き込み、上書き）
'a'  # Append（追記）
'x'  # exclusive（新規作成のみ、既存ファイルがあるとエラー）

図解:mode='a' の動作

mode='a'（追記モード）
1回目
┌──────────────────┐
│ データA           │
└──────────────────┘

2回目
┌──────────────────┐
│ データA           │ ← 残る
│ データB           │  ← 追加
└──────────────────┘

mode='w'（上書きモード）
1回目
┌──────────────────┐
│ データA           │
└──────────────────┘

2回目
┌──────────────────┐
│ データB           │ ← データAは消える
└──────────────────┘

パラメータ3: index=False

• DataFrameの左端の行番号
• Falseだから番号を付けない

パラメータ4: encoding='utf-8-sig'

encoding='utf-8-sig'

encodingとは？

• 文字をバイトに変換する方式
  主なエンコーデイング

'utf-8'      UTF-8（BOMなし）
'utf-8-sig'  TF-8（BOM付き）
'shift-jis'  Shift-JIS（日本語環境）
'cp932'      Windows日本語

🧭 BOMあり／なしの違いを図で理解する
📘 BOMありのUTF-8ファイル
ファイルの先頭に 3バイトの特別な印（EF BB BF） が付く。

┌──────────────────────────────┐
│ EF BB BF | H | e | l | l | o │
└──────────────────────────────┘
 ↑BOM（目印）

📗 BOMなしのUTF-8ファイル

┌──────────────────────────────┐
│ H | e | l | l | o │
└──────────────────────────────┘

• こちらが 一般的なUTF-8
• 多くのプログラムは BOMなしを標準として扱う
• 「Excelで開くCSVならBOM付き、それ以外のWebファイル
• やプログラムファイルならBOMなし」と覚えておくと良いでしょう
  詳しくはhttps://e-words.jp/w/BOM.html

パラメータ5: header=write_header

• header=True → 毎回ヘッダーを書き込む（for ループのたびに列名が出る）
  出力されるCSV

Quality,Library,Length
High,Lib1,100
Quality,Library,Length   ← 2回目のヘッダー
Low,Lib2,200
Quality,Library,Length   ← 3回目のヘッダー
Medium,Lib3,150

• write_headerの変数の値

 # 1回目（ファイルがない）
write_header = True
# → ヘッダーを書く

# 2回目（ファイルがある）
write_header = False
# → ヘッダーを書かない（データだけ追加）

出力されるCSV

Quality,Library,Length   ← 最初の1回だけ
High,Lib1,100
Low,Lib2,200
Medium,Lib3,150

🔵 cleanup()

def cleanup(directory: Path):
    """作業フォルダのPDFファイルを削除"""
    for pdf in directory.glob("*.pdf"):
        pdf.unlink()

glob() メソッドとは？

• glob = 「global（全体的な）パターンマッチング」
• 指定したパターンに一致するファイルを探す機能
  パターンの説明

"*.pdf"

*      # アスタリスク = 何でもOK（ワイルドカード）
.pdf   # 拡張子が .pdf

glob() の動作イメージ

フォルダの中身
┌────────────────────────┐
│ file001.pdf            │ ← マッチ
│ file002.pdf            │ ← マッチ
│ file003.pdf            │ ← マッチ
│ report.docx            │ ← 除外（.pdfでない）
│ data.txt               │ ← 除外（.pdfでない）
│ image.png              │ ← 除外（.pdfでない）
└────────────────────────┘
        ↓
directory.glob("*.pdf")
        ↓
[file001.pdf, file002.pdf, file003.pdf]

glob() の返り値

directory.glob("*.pdf")

返り値：ジェネレーター（イテレータ）

 ジェネレーターなので、forループで1つずつ取り出せる
for pdf in directory.glob("*.pdf"):
    print(pdf)
file001.pdf
file002.pdf
file003.pdf

forループの流れ

directory.glob("*.pdf") が返すファイル
┌────────────────┐
│ file001.pdf    │ ← 1回目のループ（pdf = file001.pdf）
│ file002.pdf    │ ← 2回目のループ（pdf = file002.pdf）
│ file003.pdf    │ ← 3回目のループ（pdf = file003.pdf）
└────────────────┘

各ループで pdf.unlink() を実行
  ↓
全てのPDFファイルが削除される

あとがき:
　PDFファイルは非常に不安定です。
たとえば、ヘッダーの入力ラベルの両側スペースが、ページによて半角だったり、全角だったりすると、CSV変換時にヘッダーのラベルの数が不自然に増えたりします。
また、今回のようなPDFファイルをただCSVファイルに変換すると、レイアウトがくずれて、使用することにたえられません。

　大量PDFファイル(今回私が一括で変換したファイル数は1000前後)を処理する場合
まずは1,10,20と少しずつ処理しPDFファイルの形態を把握し、メイン関数に付帯処理を追加するという設計でコードを仕上げました。

この記事が研究者や実務者の参考になれば幸いです

最期まで読んでいただきありがとうございます