はじめに
PDFからテキストを抽出する方法はいくつかあります。普通だったらpypdfとかのライブラリを使えばいいんですが、中にはフォントマップが壊れていてテキストを抽出できないpdfが存在します。
そんな時は生成Aiに読み込ませてテキスト化しちゃえばいいじゃないかということで、試してみた結果を書いていきます。
インストールが必要なライブラリ
# python
pip install google-genai python-dotenv pypdf
個別解説
1. Geminiのレスポンスに柔軟な設計
Geminiは、テキストが入ってくる場所が一定ではありません。candidatesの中のcontent.textだったり、output配列だったり、素直にtext属性だったり、辞書型でcontent/textキーを持っていたり——と揺れがあります。
2. 20MB 以上のファイルはスキップする
Geminiにアップロード可能なサイズは20MBなので、判定するようにしています。
# python
max_size = 20 * 1024 * 1024 # 20MB
file_size = pdf_path.stat().st_size
if file_size > max_size:
self.logger.warning(f"対象外: {pdf_path.name} は 20MBを超えています")
return {
"file": str(pdf_path),
"status": "skipped_large_file",
"size_MB": round(file_size / (1024*1024), 2)
}
3. 並列処理で待機時間を短縮
フォルダ内の複数PDFを同時に投げることで待ち時間を短縮しています。
ThreadPoolExecutorを使い、実行数はmin(max_workers, ファイル数)で調整しています。
# python
from concurrent.futures import ThreadPoolExecutor, as_completed
workers = max(1, min(max_workers, len(pdf_paths)))
with ThreadPoolExecutor(max_workers=workers) as ex:
future_map = {ex.submit(task, i, p): i for i, p in enumerate(pdf_paths)}
for fut in as_completed(future_map):
i = future_map[fut]
try:
results[i] = fut.result()
except Exception as e:
results[i] = {"file": str(pdf_paths[i]), "status": "error", "error": str(e)}
4. .env からのAPIキーを読む
APIキーは.envから読み込みこむようにしています。ハードコーディングはしません。
# python
from dotenv import load_dotenv
load_dotenv()
self.api_key = os.getenv("Gemini_API_KEY")
if not self.api_key:
raise EnvironmentError("Gemini APIキーが見つかりません。.env に設定してください。")
ディレクトリ構造
├─ main.py # 上記のクラス/処理を含むスクリプト
├─ .env # Gemini_API_KEY=xxxxx
└─ pdfs/ # 入力PDFを格納
├─ a.pdf
└─ b.pdf
全体のコード
import os
import re
import json
import logging
import base64
from pathlib import Path
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv
from google import genai
from google.genai.types import GenerateContentConfig
from pypdf import PdfReader, PdfWriter
from concurrent.futures import ThreadPoolExecutor, as_completed
# 設定
GEMINI_MODEL = "gemini-2.5-flash"
DEFAULT_SLEEP_SEC = 0.5
class PDFProcessor:
"""
PDFProcessorクラス
このクラスは、Google Gemini APIを利用してPDFファイルからテキストを抽出するための処理を提供します。
主な機能:
- 指定フォルダ内のPDFファイルを一括処理し、テキストとして保存
- Gemini APIを使ったPDF内容の抽出（inlineデータ送信）
- Geminiのレスポンスからテキストを安全に抽出
- ログ出力による処理状況の把握
使い方:
1. インスタンス生成時にAPIキーを.envファイルまたは環境変数から取得
2. process_directory()でフォルダ内のPDFを一括処理
3. process_pdf()で個別PDFファイルを処理
引数や返り値の型はtypingモジュールで明示されています。
"""
def __init__(self):
# .envを読み込む
load_dotenv()
self.api_key = os.getenv("Gemini_API_KEY")
if not self.api_key:
raise EnvironmentError(f"Gemini APIキーが見つかりません。.env に設定してください。")
# 各種設定
self.client = genai.Client(api_key=self.api_key)
self.model = GEMINI_MODEL
self.sleep_sec = DEFAULT_SLEEP_SEC
# ログ設定
self.logger = logging.getLogger("PDFProcessor")
if not self.logger.handlers:
h = logging.StreamHandler()
h.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
self.logger.addHandler(h)
self.logger.setLevel(logging.INFO)
def _extract_text_from_res(self, r) -> str:
"""
Geminiは、候補（candidates）や出力（output）など複数の形式でテキストが格納されている場合があります。
この関数は、レスポンスの主要な属性や辞書キーから安全にテキストを取り出します。
Contentオブジェクトや辞書、文字列など様々な型に対応しています。
万が一抽出に失敗した場合は空文字列を返します。
"""
def to_str(x):
if x is None:
return ""
if isinstance(x, str):
return x
# SDKのContent等オブジェクトに 'text' 属性がある場合
if hasattr(x, "text"):
try:
return x.text or ""
except Exception:
pass
# 'content' 属性を持つ候補（辞書やオブジェクト）
if hasattr(x, "content"):
c = getattr(x, "content")
if isinstance(c, str):
return c
if hasattr(c, "text"):
try:
return getattr(c, "text") or ""
except Exception:
pass
# SDKの Content オブジェクトの中身を文字列化
try:
return str(c)
except Exception:
pass
# 辞書の場合
if isinstance(x, dict):
for key in ("content", "text"):
if key in x and isinstance(x[key], str):
return x[key]
try:
return json.dumps(x, ensure_ascii=False)
except Exception:
return str(x)
# フォールバック
try:
return str(x)
except Exception:
return ""
try:
# candidates や output を優先して取り出す
if hasattr(r, "candidates") and r.candidates:
first = r.candidates[0]
return to_str(first)
if hasattr(r, "output") and r.output:
first = r.output[0]
return to_str(first)
# 直接 text を持つケース
if hasattr(r, "text"):
return to_str(r.text)
if isinstance(r, dict):
return to_str(r)
return to_str(r)
except Exception:
return ""
def process_directory(self, folder_path: str, output_folder: Optional[str] = None, max_workers: int = 3) -> List[Dict[str, Any]]:
"""
指定したフォルダ内のPDFファイルをまとめて処理します。
各PDFファイルについてGoogle Gemini APIを使い、本文テキストを抽出してTXTファイルとして保存します。
ファイルサイズが20MBを超える場合はスキップされます。（geminiは20MB以上のファイルを読めないため）
抽出に失敗した場合はエラー内容をJSONファイルに記録します。
引数:
folder_path (str): PDFファイルが入っているフォルダのパス。
output_folder (Optional[str]): 抽出結果を保存するフォルダ。指定しない場合はPDFと同じ場所に保存します。
戻り値:
List[Dict[str, Any]]: 各PDFファイルの処理結果（ファイルパス、保存先、ステータスなど）を辞書形式でまとめたリスト。
"""
folder = Path(folder_path)
if not folder.exists():
self.logger.error(f"フォルダが存在しません: {folder_path}")
return []
# 事前にリスト化 & ソート
pdf_paths = sorted(folder.glob("**/*.pdf"))
if not pdf_paths:
self.logger.info("PDFが見つかりません。")
return []
max_size = 20 * 1024 * 1024 # 20MB
# 実際のワーカー数（1以下にしない）
workers = max(1, min(max_workers, len(pdf_paths)))
self.logger.info(f"PDF数={len(pdf_paths)} 並列ワーカー={workers}")
results: List[Dict[str, Any]] = [None] * len(pdf_paths) # インデックス順保持用
def task(idx: int, pdf_path: Path):
try:
file_size = pdf_path.stat().st_size
if file_size > max_size:
self.logger.warning(f"対象外: {pdf_path.name} は 20MBを超えています ({file_size/(1024*1024):.2f} MB)")
return {
"file": str(pdf_path),
"status": "skipped_large_file",
"size_MB": round(file_size / (1024*1024), 2)
}
self.logger.info(f"[{idx+1}/{len(pdf_paths)}] Processing: {pdf_path.name}")
return self.process_pdf(pdf_path, output_folder=output_folder)
except Exception as e:
self.logger.error(f"Failed processing {pdf_path.name}: {e}")
return {"file": str(pdf_path), "status": "error", "error": str(e)}
if workers == 1:
# 従来シリアル（デバッグ用）
for i, p in enumerate(pdf_paths):
results[i] = task(i, p)
else:
with ThreadPoolExecutor(max_workers=workers) as ex:
future_map = {ex.submit(task, i, p): i for i, p in enumerate(pdf_paths)}
for fut in as_completed(future_map):
i = future_map[fut]
try:
results[i] = fut.result()
except Exception as e:
results[i] = {"file": str(pdf_paths[i]), "status": "error", "error": str(e)}
self.logger.info("All done.")
# None フォールバック除去
return [r for r in results if r]
def process_pdf(self, pdf_path: Path, output_folder: Optional[str] = None) -> Dict[str, Any]:
"""
指定したPDFファイルをGemini APIで処理し、テキスト抽出を行います。
PDFファイルをバイト形式で読み込み、Google Gemini APIにinlineデータとして送信します。
Geminiのレスポンスから本文テキストを抽出し、TXTファイルとして保存します。
エラーが発生した場合はJSONファイルにエラー内容を記録します。
Args:
pdf_path (Path): 処理対象のPDFファイルパス。
output_folder (Optional[str]): 抽出結果を保存するフォルダ。指定しない場合はPDFと同じ場所に保存します。
Returns:
Dict[str, Any]: 処理結果（ファイルパス、保存先、ステータスなど）をまとめた辞書。
"""
try:
with open(pdf_path, 'rb') as pdf_file:
pdf_bytes = pdf_file.read()
except Exception as e:
self.logger.error(f"PDFファイルの読み込みに失敗しました: {e}")
return {"file": str(pdf_path), "status": "read_error", "error": str(e)}
# プロンプトを作成
system_prompt = (
f"この入力はPDFファイルです。ファイル名: {pdf_path.name}。\n"
"以下の指示に従って、可能な限り本文を抽出してプレーンテキストで返してください。"
)
# GeminiにPDFを送る
res = self.client.models.generate_content(
model=self.model,
contents=[
{"text": system_prompt},
{"inline_data": {"mime_type": "application/pdf", "data": base64.b64encode(pdf_bytes).decode()}}
],
config=GenerateContentConfig(temperature=0.0, top_p=0.9)
)
# レスポンスからテキストを抽出
extracted = self._extract_text_from_res(res) or ""
extracted = re.sub(r'\n\s*\n', '\n\n', extracted)
extracted = re.sub(r' +', ' ', extracted)
extracted = extracted.replace(' ', ' ')
extracted = re.sub(r'[\u0000-\u001F\u007F-\u009F]', '', extracted).strip()
out_dir = Path(output_folder) if output_folder else pdf_path.parent
out_dir.mkdir(parents=True, exist_ok=True)
txt_path = out_dir / (pdf_path.stem + ".txt")
# テキストの出力
if extracted:
with open(txt_path, "w", encoding="utf-8") as f:
f.write(extracted)
self.logger.info(f"Wrote text: {txt_path}")
else:
self.logger.warning(f"Geminiの出力が空でした: {pdf_path.name} (テキストは保存しません)")
def main():
processor = PDFProcessor()
results = processor.process_directory("pdfs", output_folder=None, max_workers=5)
if __name__ == "__main__":
main()