Pythonで契約書PDFを自動分類＆フォルダ分けするシステムを作る

この記事で作るもの（ゴール）

  • 受け取った契約書PDF（スキャン含む）を、
    • NDA（秘密保持契約）
    • 業務委託契約
    • 売買基本契約
    • 見積書／発注書
    • 請求書
    • その他
      に自動分類して、所定のフォルダへ自動振り分けします。
  • 日本語PDFでも、OCR（Tesseract）で文字を抽出可能。
  • 分類根拠（ルール命中・スコア）、元ファイル名、格納先、処理結果をCSVログに出力。
  • Windows（タスクスケジューラ）／Mac・Linux（cron）での定期実行に対応。

全体構成（アーキテクチャ）

[inbox/] ← 未分類PDFを投げ込む
├─ A社_秘密保持契約_2025-07-01.pdf
└─ …

[sorted/] ← 自動で仕分け
├─ NDA/
├─ Outsourcing/
├─ MasterSales/
├─ Estimate_PO/
├─ Invoice/
└─ Others/

[logs/]
├─ classify_YYYYMMDD.csv
└─ app.log

config.yaml ← ルール・正規表現・保存先の定義
classify_docs.py ← CLIエントリ（抽出→前処理→分類→移動→記録）
extractor.py ← PDF/画像→テキスト抽出（pypdf + pdf2image + pytesseract）
classifier.py ← ルールベース分類 + （任意）ML分類
utils.py ← 共有関数（正規化・ハッシュ・重複回避名など）

使う主要ライブラリ

  • pypdf：テキストPDFの抽出
  • pdf2image：PDFページ→画像（スキャンPDFのOCR用）
  • pytesseract：OCR（日本語対応）
  • opencv-python（任意）：OCR前の前処理（傾き補正・二値化）
  • pyyaml：ルール定義ファイルの読み込み
  • scikit-learn（任意）：機械学習分類（TF-IDF + ロジ回帰など）

インストール

pip install pypdf pdf2image pytesseract opencv-python pyyaml scikit-learn pandas python-dateutil

Tesseract（OCRエンジン）本体の導入

  • Windows: https://github.com/UB-Mannheim/tesseract/wiki （日本語言語データ jpn.traineddata を追加）
  • macOS（Homebrew）: brew install tesseract-lang （言語データ同梱）
  • Linux（Debian/Ubuntu系）: sudo apt-get install tesseract-ocr tesseract-ocr-jpn

ルール定義（config.yaml）

まずはルールベースで高精度を狙い、必要に応じてMLを追加する構成にします。以下はサンプルのconfig.yaml。

# 分類カテゴリ
labels:
  NDA:
    folder: "sorted/NDA"
    patterns:
      # NDA特有ワード
      - "秘密保持契約"
      - "機密情報の取扱い"
      - "Non-Disclosure Agreement|NDA"
  Outsourcing:
    folder: "sorted/Outsourcing"
    patterns:
      - "業務委託契約"
      - "再委託"
      - "成果物の著作権"
  MasterSales:
    folder: "sorted/MasterSales"
    patterns:
      - "基本契約"
      - "売買基本契約"
      - "取引基本契約"
  Estimate_PO:
    folder: "sorted/Estimate_PO"
    patterns:
      - "見積書"
      - "御見積"
      - "発注書|注文書"
      - "発注番号|注文番号|PO番号"
  Invoice:
    folder: "sorted/Invoice"
    patterns:
      - "請求書"
      - "御請求"
      - "振込先|支払期日|合計金額"

# 正規表現（重み付き）
regex_rules:
  - pattern: "第[一二三四五六七八九十0-9]+条"   # 条文構造
    weight: 0.5
  - pattern: "有効期間|契約期間"
    weight: 0.7
  - pattern: "解除|損害賠償|不可抗力"
    weight: 0.4

# スコア閾値（総合がこの値を超えたカテゴリに決定）
score_threshold: 1.0

# 先勝ちルール（どれかが命中した時点で確定）
priority_keywords:
  - label: Invoice
    keywords: ["請求書", "御請求"]
  - label: Estimate_PO
    keywords: ["見積書", "発注書", "注文書"]

# OCR設定
ocr:
  enable: true
  lang: "jpn+eng"
  dpi: 300

# ファイル名テンプレート（重複回避のためにハッシュ付与可）
filename_template: "{label}_{counterparty}_{date}_{hash8}.pdf"

# 取引先名を拾う正規表現（例）
extract:
  counterparty:
    - "(株式会社[\u3040-\u30FF\u4E00-\u9FFFA-Za-z0-9]+)"
    - "(有限会社[\u3040-\u30FF\u4E00-\u9FFFA-Za-z0-9]+)"
  date:
    - "(20[0-9]{2}[年/.-](1[0-2]|0?[1-9])[月/.-](3[01]|[12]?[0-9])日?)"
    - "(令和[一二三四五六七八九十]+年\s*(1[0-2]|0?[1-9])月\s*(3[01]|[12]?[0-9])日?)"

テキスト抽出：extractor.py

# extractor.py
from pathlib import Path
import re
from pypdf import PdfReader
from pdf2image import convert_from_path
import pytesseract
import cv2
import numpy as np


def pdf_text(path: Path, ocr_enable=True, ocr_lang="jpn+eng", dpi=300) -> str:
    """テキストPDFならpypdfで抽出。スキャンPDFはOCRにフォールバック。"""
    text = _extract_text_with_pypdf(path)
    if text and len(text.strip()) > 50:
        return text
    if not ocr_enable:
        return text
    return _extract_text_with_ocr(path, ocr_lang, dpi)


def _extract_text_with_pypdf(path: Path) -> str:
    try:
        reader = PdfReader(str(path))
        texts = []
        for page in reader.pages:
            t = page.extract_text() or ""
            texts.append(t)
        return "\n".join(texts)
    except Exception:
        return ""


def _deskew_and_binarize(img: np.ndarray) -> np.ndarray:
    # グレースケール
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    # 二値化
    thr = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)[1]
    # 傾き推定
    coords = np.column_stack(np.where(thr < 255))
    angle = cv2.minAreaRect(coords)[-1]
    if angle < -45:
        angle = -(90 + angle)
    else:
        angle = -angle
    (h, w) = thr.shape
    M = cv2.getRotationMatrix2D((w // 2, h // 2), angle, 1.0)
    rotated = cv2.warpAffine(thr, M, (w, h), flags=cv2.INTER_CUBIC, borderMode=cv2.BORDER_REPLICATE)
    return rotated


def _extract_text_with_ocr(path: Path, lang: str, dpi: int) -> str:
    images = convert_from_path(str(path), dpi=dpi)
    texts = []
    for pil in images:
        img = np.array(pil)[:, :, ::-1]  # RGB→BGR
        img = _deskew_and_binarize(img)
        t = pytesseract.image_to_string(img, lang=lang)
        texts.append(t)
    return "\n".join(texts)

分類器：classifier.py（ルールベース→任意でML）

# classifier.py
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, List, Tuple
import re

@dataclass
class RuleResult:
    label: str
    score: float
    hits: List[str]


class RuleBasedClassifier:
    def __init__(self, config: Dict):
        self.config = config

    def classify(self, text: str) -> RuleResult:
        text_norm = self._normalize(text)
        # 1) 先勝ちルール
        for rule in self.config.get("priority_keywords", []):
            label = rule["label"]
            for kw in rule["keywords"]:
                if kw in text_norm:
                    return RuleResult(label, score=10.0, hits=[f"priority:{kw}"])
        # 2) パターンスコア
        scores = {}
        hits = {lbl: [] for lbl in self.config.get("labels", {}).keys()}
        for label, meta in self.config.get("labels", {}).items():
            score = 0.0
            for pat in meta.get("patterns", []):
                if re.search(pat, text_norm, flags=re.IGNORECASE):
                    score += 1.0
                    hits[label].append(pat)
            for rr in self.config.get("regex_rules", []):
                if re.search(rr["pattern"], text_norm):
                    score += rr.get("weight", 0.1)
                    hits[label].append(rr["pattern"])
            scores[label] = score
        # 3) 最大スコアで判定
        best = max(scores.items(), key=lambda x: x[1])
        label, score = best
        if score >= self.config.get("score_threshold", 1.0):
            return RuleResult(label, score, hits[label])
        return RuleResult("Others", score, hits.get("Others", []))

    def _normalize(self, s: str) -> str:
        s = s.replace("\u3000", " ")  # 全角スペース
        s = re.sub(r"\s+", " ", s)
        return s

ML分類の追加（任意）

  • まずはルールで8割以上取れることが多いです。誤判定が残る場合は、学習用ラベル付きデータ（ファイルパス＋正解ラベル）を少量作成し、TF-IDF + ロジスティック回帰で上乗せすると堅いです（50〜200件でも効くことが多い）。
  • sentence-transformers を使った埋め込み＋近傍分類も有効です。運用の簡単さ優先なら ルール→（足りなければ）TF-IDF が現実的。

CLI：classify_docs.py（抽出→分類→リネーム→移動→ログ）

# classify_docs.py
from pathlib import Path
import yaml
import hashlib
import shutil
import re
import pandas as pd
from datetime import datetime
from dateutil import parser
from extractor import pdf_text
from classifier import RuleBasedClassifier


def safe_name(name: str) -> str:
    return re.sub(r"[^\w\u3040-\u30FF\u4E00-\u9FFF.-]+", "_", name)


def short_hash(s: str) -> str:
    return hashlib.sha1(s.encode("utf-8")).hexdigest()[:8]


def extract_first(patterns, text):
    for pat in patterns or []:
        m = re.search(pat, text)
        if m:
            return m.group(1)
    return ""


def parse_date(string: str):
    try:
        d = parser.parse(string, fuzzy=True)
        return d.strftime("%Y-%m-%d")
    except Exception:
        return ""


def main(input_dir: str = "inbox", config_path: str = "config.yaml", log_dir: str = "logs"):
    cfg = yaml.safe_load(open(config_path, "r", encoding="utf-8"))
    classifier = RuleBasedClassifier(cfg)

    rows = []
    inpath = Path(input_dir)
    inpath.mkdir(parents=True, exist_ok=True)
    Path(log_dir).mkdir(parents=True, exist_ok=True)

    for pdf in sorted(inpath.glob("*.pdf")):
        text = pdf_text(pdf, ocr_enable=cfg["ocr"]["enable"], ocr_lang=cfg["ocr"]["lang"], dpi=cfg["ocr"]["dpi"])
        result = classifier.classify(text)

        # 付随情報抽出
        counterparty = extract_first(cfg.get("extract", {}).get("counterparty"), text)
        date_str = extract_first(cfg.get("extract", {}).get("date"), text)
        date_norm = parse_date(date_str) if date_str else ""

        # 保存先
        label_meta = cfg["labels"].get(result.label, {"folder": "sorted/Others"})
        dest_dir = Path(label_meta["folder"]).expanduser()
        dest_dir.mkdir(parents=True, exist_ok=True)

        # ファイル名
        fname_tpl = cfg.get("filename_template", "{label}_{hash8}.pdf")
        fname = fname_tpl.format(
            label=result.label,
            counterparty=safe_name(counterparty) or "Unknown",
            date=date_norm or datetime.now().strftime("%Y-%m-%d"),
            hash8=short_hash(pdf.name + str(datetime.now()))
        )
        dest = dest_dir / fname

        shutil.move(str(pdf), str(dest))

        rows.append({
            "src": str(pdf),
            "dest": str(dest),
            "label": result.label,
            "score": result.score,
            "hits": "|".join(result.hits),
            "counterparty": counterparty,
            "date": date_norm,
        })

    # CSVログ
    if rows:
        df = pd.DataFrame(rows)
        out_csv = Path(log_dir) / f"classify_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        df.to_csv(out_csv, index=False, encoding="utf-8-sig")
        print(f"log -> {out_csv}")


if __name__ == "__main__":
    import fire  # pip install fire
    fire.Fire(main)

使い方

pip install fire
python classify_docs.py --input_dir inbox --config_path config.yaml --log_dir logs

ルール作りのコツ（日本の契約書向け）

  • 見出し語：表題（例「秘密保持契約書」「業務委託契約書」）は強力。優先キーワードに設定。
  • 章立て：「第◯条」「第1条（目的）」など条文構造は契約書のシグナル。
  • 支払い系：「請求書」「合計金額」「支払期日」「振込先」「銀行名」は請求書の高分離ワード。
  • 番号系：「発注番号」「注文番号」「PO番号」は発注書／注文書の決定打。
  • 権利帰属：「著作権」「知的財産権」「再委託」「成果物」→業務委託契約に偏りがち。
  • 語の揺れ：ひらがな・カタカナ・漢字の揺れ、全角／半角、旧字体を正規化。

機械学習を足すなら（ミニマム実装）

  1. 教師データ作成：data/train.csv に path,label を10〜200件ほど。
  2. 前処理：抽出した全文テキストを保存（キャッシュ）し、TF-IDFでベクトル化。
  3. 学習：ロジスティック回帰／LinearSVC。
  4. 推論：ルールで閾値未満の時だけMLの予測で補完。

サンプル（学習スクリプトの骨子）

# train_ml.py（抜粋）
import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report
import joblib

train = pd.read_csv("data/train.csv")
X, y = train["text"], train["label"]
pipe = Pipeline([
    ("tfidf", TfidfVectorizer(max_features=30000, ngram_range=(1,2))),
    ("clf", LogisticRegression(max_iter=1000))
])
pipe.fit(X, y)
print(classification_report(y, pipe.predict(X)))
joblib.dump(pipe, "models/tfidf_lr.joblib")

精度検証と運用

  • 検証データ：最低でも各クラス5〜10件を手採点で評価。
  • 指標：Precision（誤爆少なさ）重視。誤分類は「Others」に逃がす。
  • 保守：新しい様式に遭遇したら、誤判定ログからキーワードを追加。月次でルール棚卸し。
  • 例外運用：人手レビュー用の「sorted/_review」フォルダを用意し、低スコア案件はここへ。

セキュリティと法令対応

  • 契約書は個人情報や機密情報を含みます。外部クラウドに出さず、オンプレ／社内NW内で処理が原則。
  • ファイルアクセス権を最小化。ログにも過度な本文を残さない。
  • 保存先フォルダのバックアップ、破棄ポリシー（保存年限）を明文化。

Windows／macOSでの定期実行

  • Windows タスクスケジューラ：python classify_docs.py --input_dir \\\srv\scan\inbox を5分間隔で。
  • macOS/Linux cron：*/5 * * * * /usr/bin/python3 /path/classify_docs.py --input_dir /srv/scan/inbox。

よくあるハマりどころ

  • OCRが化ける：解像度不足（<200dpi）。300dpi以上でスキャン／dpi=300に。
  • フォント埋め込みPDF：pypdfの抽出が空に→OCRへ自動フォールバック。
  • 傾き・薄い印影：OpenCVで傾き補正・コントラスト向上を行う。
  • 同名ファイルの上書き：テンプレートにハッシュ{hash8}を入れて回避。

まとめ

  • まずはルールベースで8割以上を狙い、ログを見ながらルール改善。
  • 足りない部分は軽量MLを上乗せ。
  • OCR・命名規則・定期実行・ログ化まで押さえれば、現場で回る自動仕分け基盤になります。

次の一歩

  • 請求書の金額・期日・振込先を抽出して会計システムへ連携
  • 取引先マスタと照合して自動でフォルダ命名
  • Teams/Slackへ仕分け結果を通知（監査証跡にも）
スポンサーリンク
タイトルとURLをコピーしました