Export Document Processing

I’ve revived my Medium subscription after a long break, and I’ve decided to build and publish an app every day.

In Japanese companies, paper-based workflows are still deeply entrenched. A typical business process looks like this: convert faxed documents to PDF, open the PDF and manually check it, transcribe the data into Excel one by one, and finally manually enter that information into the core system. Just imagining it sounds time-consuming, doesn’t it?

What I built this time is a tool to streamline this process. I implemented functionality to automatically extract necessary information from PDFs and output it to CSV files.

Get Ken’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Initially, I attempted to implement this using Python alone without APIs. However, I ran into accuracy issues. In the end, I decided to leverage APIs, prioritizing reliability. It reinforced the importance of using the right tools for the job.

import unicodedata
import requests
import streamlit as st
import fitz  # pymupdf
import re
from datetime import datetime
from PIL import Image
import pytesseract
import json  # JSONパースのために明示的にインポート
import pandas as pd

# ページ設定
st.set_page_config(
    page_title="インボイス情報抽出",
    page_icon="📋",
    layout="wide"
)

# 国名→ISO2桁国コード対応表（拡張・表記揺れ対応）を一箇所にまとめる
COUNTRY_TO_ISO = {
    '日本': 'JP', 'japan': 'JP', 'nippon': 'JP', 'nihon': 'JP',
    'アメリカ': 'US', 'usa': 'US', 'united states': 'US', 'us': 'US',
    '中国': 'CN', 'china': 'CN', '中華人民共和国': 'CN',
    '韓国': 'KR', 'south korea': 'KR', 'korea': 'KR',
    'ドイツ': 'DE', 'germany': 'DE',
    'フランス': 'FR', 'france': 'FR',
    'イギリス': 'GB', 'uk': 'GB', 'united kingdom': 'GB', 'britain': 'GB',
    'カナダ': 'CA', 'canada': 'CA',
    'イタリア': 'IT', 'italy': 'IT',
    'オーストラリア': 'AU', 'australia': 'AU',
    'インド': 'IN', 'india': 'IN',
    '台湾': 'TW', 'taiwan': 'TW',
    '香港': 'HK', 'hong kong': 'HK',
    'ロシア': 'RU', 'russia': 'RU',
    'スペイン': 'ES', 'spain': 'ES',
    'ブラジル': 'BR', 'brazil': 'BR',
    'メキシコ': 'MX', 'mexico': 'MX',
    'シンガポール': 'SG', 'singapore': 'SG',
    'タイ': 'TH', 'thailand': 'TH',
    # 大文字・小文字、全角・半角混在も考慮
    'Japan': 'JP', 'United States': 'US', 'USA': 'US', 'China': 'CN', 'South Korea': 'KR',
    'Germany': 'DE', 'France': 'FR', 'United Kingdom': 'GB', 'UK': 'GB',
    'Canada': 'CA', 'Italy': 'IT', 'Australia': 'AU', 'India': 'IN',
    'Taiwan': 'TW', 'Hong Kong': 'HK', 'Russia': 'RU', 'Spain': 'ES',
    'Brazil': 'BR', 'Mexico': 'MX', 'Singapore': 'SG', 'Thailand': 'TH',
}

def normalize_country_name(name):
    # 全角→半角、カタカナ→ひらがな、スペース除去、小文字化
    # NFKCは互換文字を正規化するため、半角カナは全角に、全角英数は半角に変換されます。
    # 国名判別にはNFKCで十分ですが、より厳密な正規化が必要な場合は要検討。
    name = unicodedata.normalize('NFKC', name)
    name = name.lower().replace(' ', ' ').replace('\n', ' ').replace('\r', ' ')
    name = name.replace('（', '(').replace('）', ')')
    name = name.strip()
    return name

def extract_text_from_pdf(file):
    """PDFからテキストを抽出"""
    try:
        doc = fitz.open(stream=file.read(), filetype="pdf")
        text = ""
        for page in doc:
            text += page.get_text()
        doc.close()
        return text
    except Exception as e:
        return f"エラー: {str(e)}"

def extract_text_from_jpg(file):
    """JPG画像からテキストを抽出（OCR）"""
    try:
        image = Image.open(file)
        text = pytesseract.image_to_string(image, lang='jpn+eng')
        return text
    except Exception as e:
        return f"エラー: {str(e)}"

def extract_invoice_number(text):
    """インボイス番号を抽出"""
    patterns = [
        r'Invoice\s*(?:No|Number|#)[:\s]*([A-Z0-9\-]+)',
        r'請求書番号[:\s]*([A-Z0-9\-]+)',
        r'Invoice[:\s]*([A-Z0-9\-]+)',
        r'INV[:\s]*([A-Z0-9\-]+)',
        r'No\.[:\s]*([A-Z0-9\-]+)',
    ]
    for pattern in patterns:
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            return match.group(1).strip()
    return "未検出"

def get_country_iso_code(country_text):
    """国名テキストからISOコードを推測"""
    norm_text = normalize_country_name(country_text)
    
    # 完全一致でISOコード変換
    for country, code in COUNTRY_TO_ISO.items():
        if normalize_country_name(country) == norm_text:
            return code
            
    # 部分一致でISOコード変換（より広い範囲で検索）
    for country, code in COUNTRY_TO_ISO.items():
        if normalize_country_name(country) in norm_text:
            return code

    # 国名らしき単語を抽出して再検索
    words = re.findall(r'[a-zA-Z一-龠ぁ-んァ-ヴ]+', norm_text)
    for word in words:
        for country, code in COUNTRY_TO_ISO.items():
            if normalize_country_name(country) == word:
                return code
    return "未検出"

def extract_country_of_origin(text):
    """製造国を抽出"""
    patterns = [
        r'(?:Country of Origin|Origin|製造国|原産国)[:\s]*([A-Za-z\s]+?)(?:\n|,|;)',
        r'Made in[:\s]*([A-Za-z\s]+?)(?:\n|,|;)',
        r'Manufactured in[:\s]*([A-Za-z\s]+?)(?:\n|,|;)',
    ]
    for pattern in patterns:
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            country_name = match.group(1).strip()
            return get_country_iso_code(country_name)
    return "未検出"

def extract_ship_to(text):
    """送り先を抽出"""
    patterns = [
        r'Ship\s*To[:\s]*\n([^\n]+(?:\n[^\n]+){0,3})',
        r'送り先[:\s]*\n([^\n]+(?:\n[^\n]+){0,3})',
        r'Consignee[:\s]*\n([^\n]+(?:\n[^\n]+){0,3})',
    ]
    for pattern in patterns:
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            address = match.group(1).strip()
            lines = address.split('\n')[:3]
            joined = ' '.join(lines).strip()
            return get_country_iso_code(joined)
    return "未検出"

def extract_ship_from(text):
    """送り元を抽出"""
    patterns = [
        r'Ship\s*From[:\s]*\n([^\n]+(?:\n[^\n]+){0,3})',
        r'送り元[:\s]*\n([^\n]+(?:\n[^\n]+){0,3})',
        r'Shipper[:\s]*\n([^\n]+(?:\n[^\n]+){0,3})',
        r'From[:\s]*\n([^\n]+(?:\n[^\n]+){0,3})',
    ]
    for pattern in patterns:
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            address = match.group(1).strip()
            lines = address.split('\n')[:3]
            joined = ' '.join(lines).strip()
            return get_country_iso_code(joined) # 送り元も国コードを返すように変更
    return "未検出"

# Geminiモデル一覧取得関数を定義
def get_gemini_models(api_key):
    GEMINI_LIST_MODELS_URL = "https://generativelanguage.googleapis.com/v1beta/models"
    headers = {
        "Content-Type": "application/json"
    }
    params = {
        "key": api_key
    }
    try:
        response = requests.get(GEMINI_LIST_MODELS_URL, headers=headers, params=params)
        response.raise_for_status() # HTTPエラーがあれば例外を発生させる
        models_data = response.json().get('models', [])
        # 'generateContent' メソッドをサポートするモデルのみをフィルタリング
        supported_models = [m for m in models_data if 'generateContent' in m.get('supportedGenerationMethods', [])]
        return supported_models
    except requests.exceptions.RequestException as e:
        st.error(f"Geminiモデル一覧の取得中にエラーが発生しました: {e}")
        return []
    except json.JSONDecodeError:
        st.error("Geminiモデル一覧のレスポンスがJSON形式ではありません。")
        return []

# Claude APIキーをサイドバーから入力
if 'claude_api_key' not in st.session_state:
    st.session_state['claude_api_key'] = ''
if 'gemini_api_key' not in st.session_state:
    st.session_state['gemini_api_key'] = ''
if 'llm_provider' not in st.session_state:
    st.session_state['llm_provider'] = 'Claude'

with st.sidebar:
    st.header("AI API設定")
    st.selectbox("利用するAIモデル", ["Claude", "Gemini"], index=0, key="llm_provider")
    st.text_input("Claude APIキー", value=st.session_state['claude_api_key'], type="password", key="claude_api_key")
    st.text_input("Gemini APIキー", value=st.session_state['gemini_api_key'], type="password", key="gemini_api_key")
    
    if st.session_state['gemini_api_key']:
        if st.button("Geminiモデル一覧を取得", key="get_gemini_models"):
            with st.spinner("Geminiモデル一覧を取得中..."):
                gemini_models_list = get_gemini_models(st.session_state['gemini_api_key'])
                st.session_state['gemini_models'] = [m['name'] for m in gemini_models_list]
                if not st.session_state['gemini_models']:
                    st.error("Geminiモデルが見つかりませんでした。APIキーを確認してください。")
    
    if 'gemini_models' in st.session_state and st.session_state['gemini_models']:
        st.selectbox("Geminiモデル名", st.session_state['gemini_models'], key="gemini_model")
    else:
        st.session_state['gemini_model'] = "gemini-pro" # デフォルト値

# Claude API呼び出し関数
CLAUDE_API_URL = "https://api.anthropic.com/v1/messages"
def get_invoice_info_from_claude(text):
    prompt = f"""
以下の請求書テキストから、インボイス番号、製造国、送り先の国、送り元の国を抽出してください。
出力は以下のJSON形式で返してください。
{{
  \"インボイス番号\": \"\",
  \"製造国\": \"\",
  \"送り先の国\": \"\",
  \"送り元の国\": \"\"
}}
抽出できない場合は空文字列にしてください。
テキスト:
{text}
"""
    headers = {
        "x-api-key": st.session_state['claude_api_key'],
        "content-type": "application/json",
        "anthropic-version": "2023-06-01" # APIバージョンを指定
    }
    data = {
        "model": "claude-3-opus-20240229", # 最適なモデルを選択
        "max_tokens": 1024, # 十分なトークン数を確保
        "messages": [
            {"role": "user", "content": prompt}
        ]
    }
    try:
        response = requests.post(CLAUDE_API_URL, headers=headers, json=data, timeout=60) # タイムアウト設定
        response.raise_for_status() # HTTPエラーがあれば例外を発生させる

        response_json = response.json()
        
        # Claude 3のレスポンス形式に対応
        content_parts = response_json.get('content', [])
        response_text = ""
        for part in content_parts:
            if part.get('type') == 'text':
                response_text += part.get('text', '')

        match = re.search(r'{.*}', response_text, re.DOTALL)
        if match:
            try:
                return json.loads(match.group(0))
            except json.JSONDecodeError:
                return {"error": "JSONパース失敗", "raw": response_text}
        else:
            return {"error": "JSON抽出失敗", "raw": response_text}
    except requests.exceptions.RequestException as e:
        return {"error": f"APIリクエストエラー: {e}", "raw": str(e)}
    except Exception as e:
        return {"error": f"予期せぬエラー: {e}", "raw": str(e)}

# Gemini API呼び出し関数
# Gemini API呼び出し関数
def get_invoice_info_from_gemini(text):
    model_name = "gemini-2.5-pro"  # 固定モデル名を使用
    GEMINI_API_URL = f"https://generativelanguage.googleapis.com/v1/models/{model_name}:generateContent"
    prompt = f"""
以下の請求書テキストから、インボイス番号、製造国、送り先の国、送り元の国を抽出してください。
出力は以下のJSON形式で返してください。
{{
  \"インボイス番号\": \"\",
  \"製造国\": \"\",
  \"送り先の国\": \"\",
  \"送り元の国\": \"\"
}}
抽出できない場合は空文字列にしてください。
テキスト:
{text}
"""
    headers = {
        "Content-Type": "application/json"
    }
    params = {
        "key": st.session_state['gemini_api_key']
    }
    data = {
        "contents": [{
            "parts": [{
                "text": prompt
            }]
        }],
        "generationConfig": {
            "temperature": 0.2,
            "maxOutputTokens": 1024
        },
        "safetySettings": [{
            "category": "HARM_CATEGORY_DANGEROUS_CONTENT",
            "threshold": "BLOCK_NONE"
        }]
    }
    
    try:
        response = requests.post(GEMINI_API_URL, headers=headers, params=params, json=data, timeout=60) # タイムアウト設定
        response.raise_for_status() # HTTPエラーがあれば例外を発生させる
        
        response_json = response.json()
        
        # Geminiのレスポンスからコンテンツを抽出
        content = response_json.get('candidates', [{}])[0].get('content', {}).get('parts', [{}])[0].get('text', '')

        if content:
            match = re.search(r'{.*}', content, re.DOTALL)
            if match:
                try:
                    return json.loads(match.group(0))
                except json.JSONDecodeError:
                    return {"error": "JSONパース失敗", "raw": content}
            else:
                return {"error": "JSON抽出失敗", "raw": content}
        else:
            return {"error": "返答候補なし", "raw": response.text}
    except requests.exceptions.RequestException as e:
        return {"error": f"APIリクエストエラー: {e}", "raw": str(e)}
    except Exception as e:
        return {"error": f"予期せぬエラー: {e}", "raw": str(e)}

# 結果保存用のセッションステート初期化
if 'results_df' not in st.session_state:
    st.session_state['results_df'] = pd.DataFrame(columns=['ファイル名', 'インボイス番号', '製造国', '送り先の国', '送り元の国', '処理日時'])

# UI
st.title("📋 インボイス情報抽出アプリ")
st.markdown("PDFインボイスまたはJPG画像をアップロードすると主要情報を自動抽出します")
st.markdown("---")

# 前回の結果をクリア
if st.button("結果をクリア"):
    st.session_state['results_df'] = pd.DataFrame(columns=['ファイル名', 'インボイス番号', '製造国', '送り先の国', '送り元の国', '処理日時'])
    st.success("結果をクリアしました")

uploaded_files = st.file_uploader("請求書PDFまたはJPG画像をアップロードしてください（複数可）", type=["pdf", "jpg", "jpeg"], accept_multiple_files=True)

if uploaded_files:
    st.write(f"アップロードされたファイル数: {len(uploaded_files)}個")
    for i, uploaded_file in enumerate(uploaded_files, 1):
        st.write(f"{i}. {uploaded_file.name}")
    
    if st.button("処理開始", type="primary"):
        for uploaded_file in uploaded_files:
            st.markdown("---")
            st.subheader(f"ファイル: {uploaded_file.name}")
            
            with st.spinner(f"{uploaded_file.name} を解析中..."):
                extracted_text = ""
                file_extension = uploaded_file.name.lower().split('.')[-1]
                
                if file_extension == "pdf":
                    extracted_text = extract_text_from_pdf(uploaded_file)
                elif file_extension in ["jpg", "jpeg"]:
                    extracted_text = extract_text_from_jpg(uploaded_file)
                else:
                    st.error(f"{uploaded_file.name}: 未対応のファイル形式です。PDFまたはJPG/JPEGファイルをアップロードしてください。")
                    continue  # 次のファイルへ
            
                if "エラー:" in extracted_text:
                    st.error(f"ファイルからのテキスト抽出中にエラーが発生しました: {extracted_text}")
                    continue  # 次のファイルへ
                
                with st.container():
                    st.subheader(f"{st.session_state['llm_provider']}による抽出情報")
                    
                    result = {}
                    if st.session_state['llm_provider'] == "Claude":
                        if st.session_state['claude_api_key']:
                            result = get_invoice_info_from_claude(extracted_text)
                        else:
                            st.error("Claude APIキーが未入力です。サイドバーから入力してください。")
                            break  # 処理を中断
                    elif st.session_state['llm_provider'] == "Gemini":
                        if st.session_state['gemini_api_key']:
                            result = get_invoice_info_from_gemini(extracted_text)
                        else:
                            st.error("Gemini APIキーが未入力です。サイドバーから入力してください。")
                            break  # 処理を中断
                    
                    if result:
                        if "error" in result:
                            st.error(f"AIによる情報抽出中にエラーが発生しました: {result['error']}")
                            with st.expander("API生レスポンス"):
                                st.json(result.get('raw', ''))
                        else:
                            st.success("情報抽出が完了しました！")
                            # 結果をデータフレームに追加
                            current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                            new_row = pd.DataFrame([{
                                'ファイル名': uploaded_file.name,
                                'インボイス番号': result.get("インボイス番号", "未検出"),
                                '製造国': result.get("製造国", "未検出"),
                                '送り先の国': result.get("送り先の国", "未検出"),
                                '送り元の国': result.get("送り元の国", "未検出"),
                                '処理日時': current_time
                            }])
                            st.session_state['results_df'] = pd.concat([st.session_state['results_df'], new_row], ignore_index=True)
                            
                            # 個別の結果を表示
                            col1, col2 = st.columns(2)
                            with col1:
                                st.text_input("インボイス番号", result.get("インボイス番号", "未検出"), disabled=True, key=f"invoice_no_{uploaded_file.name}")
                                st.text_input("製造国", result.get("製造国", "未検出"), disabled=True, key=f"origin_{uploaded_file.name}")
                            with col2:
                                st.text_input("送り先の国", result.get("送り先の国", "未検出"), disabled=True, key=f"ship_to_{uploaded_file.name}")
                                st.text_input("送り元の国", result.get("送り元の国", "未検出"), disabled=True, key=f"ship_from_{uploaded_file.name}")
                    
                    st.caption(f"処理日時: {current_time}")
                    with st.expander("抽出された全テキストを表示"):
                        st.text_area("フルテキスト", extracted_text, height=200, disabled=True, key=f"text_{uploaded_file.name}")
        
        # 全ファイルの処理が終わった後に結果一覧を表示
        if len(st.session_state['results_df']) > 0:
            st.markdown("---")
            st.subheader("抽出結果一覧")
            st.dataframe(st.session_state['results_df'], use_container_width=True)
            
            # CSVダウンロードボタン
            csv = st.session_state['results_df'].to_csv(index=False)
            st.download_button(
                label="結果をCSVでダウンロード",
                data=csv,
                file_name=f"invoice_extraction_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv",
                mime="text/csv"
            )

久しぶりにMediumのサブスクリプションを復活させました。これを機に、毎日アプリを作って公開していこうと思います。

日本の企業では、まだまだ紙ベースの業務が根強く残っています。典型的な業務フローはこんな感じです：FAXで届いた書類をPDFに変換し、PDFを開いて目視で確認、そこからデータを一つ一つエクセルに転記し、最終的にその情報を基幹システムに手入力する――。想像しただけで時間がかかりそうですよね。

今回作ったのは、このプロセスを効率化するツールです。PDFから必要な情報を自動で抽出し、CSVファイルに出力するところまでを実装しました。

当初はAPIを使わず、Pythonだけで実装しようと試みました。しかし、精度の問題にぶつかりました。結局、信頼性を重視してAPIを活用することに。やはり適切なツールを使うことの重要性を再認識しました。