Claude APIで決算書PDFから勘定科目を抽出する — 貸借の検算で審査用スプレッドへ

from enum import Enum
 
from pydantic import BaseModel
 
BS_ITEMS = [
    "現金預金", "売上債権", "棚卸資産", "その他流動資産", "流動資産合計",
    "有形固定資産", "無形固定資産", "投資その他の資産", "固定資産合計",
    "繰延資産", "資産合計",
    "仕入債務", "短期借入金", "その他流動負債", "流動負債合計",
    "長期借入金", "その他固定負債", "固定負債合計", "負債合計",
    "純資産合計", "負債純資産合計",
]
PL_ITEMS = [
    "売上高", "売上原価", "売上総利益", "販売費及び一般管理費", "営業利益",
    "営業外収益", "営業外費用", "経常利益", "特別利益", "特別損失",
    "税引前当期純利益", "法人税等", "当期純利益",
]
Item = Enum("Item", {n: n for n in BS_ITEMS + PL_ITEMS}, type=str)
Unit = Enum("Unit", {n: n for n in ["円", "千円", "百万円"]}, type=str)
UNIT_YEN = {"円": 1, "千円": 1_000, "百万円": 1_000_000}
 
 
class Line(BaseModel):
    printed_label: str  # 決算書に印字された科目名そのまま
    item: Item          # 科目マスタ上の科目。小計・合計の行もそのまま
    amount: int         # 表の単位のまま。△や(-)は負の数で返す
    source_page: int    # PDFビューアー上のページ番号
 
 
class Period(BaseModel):
    label: str          # 当期 / 前期
    year_end: str       # 決算期末日 YYYY-MM-DD
    balance_sheet: list[Line]
    income_statement: list[Line]
 
 
class Statement(BaseModel):
    company: str
    unit: Unit          # 表紙または表の見出しに書かれた単位
    periods: list[Period]

import base64
from pathlib import Path
 
import anthropic
 
client = anthropic.Anthropic()
MODEL = "claude-sonnet-5-5"
 
SYSTEM = """あなたは決算書PDFを財務スプレッドの科目に起こす担当です。
PDFに印字された数字だけを根拠にし、計算で補った数字は返しません。
金額は表の単位のままの整数で返し、△やカッコ付きの金額は負の数にします。
貸倒引当金などの控除項目は、対応する科目に含めてマイナスで返します。
マスタに該当がない科目は「その他流動資産」などその他の科目に寄せます。
ページ番号はPDFビューアー上のページ番号で答えます。"""
 
PROMPT = (
    "貸借対照表と損益計算書の全行をperiodsに入れてください。"
    "当期と前期の両方の列がある場合は、2つのPeriodに分けます。"
    "小計・合計の行も、印字されている通りに1行として返します。"
)
 
 
def extract_statement(pdf_path: Path) -> Statement:
    data = base64.standard_b64encode(pdf_path.read_bytes()).decode("utf-8")
    response = client.messages.parse(
        model=MODEL,
        max_tokens=16000,
        system=SYSTEM,
        messages=[{
            "role": "user",
            "content": [
                {"type": "document",
                 "source": {"type": "base64",
                            "media_type": "application/pdf",
                            "data": data}},
                {"type": "text", "text": PROMPT},
            ],
        }],
        output_format=Statement,
    )
    if response.stop_reason != "end_turn":
        raise RuntimeError(f"{pdf_path.name}: {response.stop_reason}")
    return response.parsed_output

from collections import defaultdict
 
RULES = {
    "流動資産合計": [(1, "現金預金"), (1, "売上債権"), (1, "棚卸資産"),
                    (1, "その他流動資産")],
    "固定資産合計": [(1, "有形固定資産"), (1, "無形固定資産"),
                    (1, "投資その他の資産")],
    "資産合計": [(1, "流動資産合計"), (1, "固定資産合計"), (1, "繰延資産")],
    "流動負債合計": [(1, "仕入債務"), (1, "短期借入金"), (1, "その他流動負債")],
    "固定負債合計": [(1, "長期借入金"), (1, "その他固定負債")],
    "負債合計": [(1, "流動負債合計"), (1, "固定負債合計")],
    "負債純資産合計": [(1, "負債合計"), (1, "純資産合計")],
    "売上総利益": [(1, "売上高"), (-1, "売上原価")],
    "営業利益": [(1, "売上総利益"), (-1, "販売費及び一般管理費")],
    "経常利益": [(1, "営業利益"), (1, "営業外収益"), (-1, "営業外費用")],
    "税引前当期純利益": [(1, "経常利益"), (1, "特別利益"), (-1, "特別損失")],
    "当期純利益": [(1, "税引前当期純利益"), (-1, "法人税等")],
}
 
 
def sum_by_item(lines) -> dict[str, int]:
    d: dict[str, int] = defaultdict(int)
    for ln in lines:
        d[ln.item.value] += ln.amount
    return d
 
 
def verify(st: Statement) -> list[str]:
    problems = []
    for p in st.periods:
        v = sum_by_item(p.balance_sheet + p.income_statement)
        for total, terms in RULES.items():
            if total not in v:
                continue
            calc = sum(sign * v.get(name, 0) for sign, name in terms)
            tol = len(terms)  # 端数処理の幅。構成科目1つにつき1単位
            if abs(calc - v[total]) > tol:
                problems.append(
                    f"{p.label} {total}: 印字 {v[total]:,} / 積み上げ {calc:,}")
        if v.get("資産合計") != v.get("負債純資産合計"):
            problems.append(
                f"{p.label}: 資産合計 {v.get('資産合計')} と"
                f"負債純資産合計 {v.get('負債純資産合計')} が不一致")
    return problems

from openpyxl import load_workbook
 
 
def collect(statements: list[Statement]):
    """決算期末日 -> {科目: 千円}。前期列と当期列の食い違いも拾う。"""
    by_end: dict[str, dict[str, int]] = {}
    diffs = []
    for st in statements:
        factor = UNIT_YEN[st.unit.value]
        for p in st.periods:
            v = sum_by_item(p.balance_sheet + p.income_statement)
            vals = {k: round(a * factor / 1000) for k, a in v.items()}
            if p.year_end in by_end:
                for k, a in vals.items():
                    old = by_end[p.year_end].get(k)
                    if old is not None and abs(old - a) > 1:
                        diffs.append(f"{p.year_end} {k}: {old:,} → {a:,}")
            by_end.setdefault(p.year_end, {}).update(vals)
    return by_end, diffs
 
 
def write_spread(statements, template, out):
    by_end, diffs = collect(statements)
    wb = load_workbook(template)
    ws = wb["スプレッド"]
    rows = {ws.cell(r, 1).value: r for r in range(2, ws.max_row + 1)}
    for col, end in enumerate(sorted(by_end), start=2):
        ws.cell(1, col, end)
        for item, amount in by_end[end].items():
            if item not in rows:
                raise KeyError(f"テンプレートに科目の行がありません: {item}")
            ws.cell(rows[item], col, amount)
    wb.save(out)
    return diffs

statements = []
for pdf in sorted(Path("kessansho").glob("*.pdf")):
    st = extract_statement(pdf)
    for msg in verify(st):
        print(f"[要確認] {pdf.name}: {msg}")
    statements.append(st)
 
for d in write_spread(statements, "spread_template.xlsx", "spread_out.xlsx"):
    print(f"[期間の食い違い] {d}")