Claude APIで事故受付の初報を構造化 — 未決案件の台帳へ夜間バッチ

SCHEMA = {
    "type": "object",
    "properties": {
        "accident_date": {
            "type": ["string", "null"],
            "description": "事故日。YYYY-MM-DD。書かれていなければnull",
        },
        "accident_date_basis": {
            "type": "string",
            "enum": ["明記", "換算", "不明"],
            "description": "明記=初報に日付がある / 換算=受付日時から計算した",
        },
        "accident_time_text": {"type": ["string", "null"]},
        "location_text": {"type": ["string", "null"]},
        "parties": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "role": {"type": "string",
                             "enum": ["通報者", "相手方", "同乗者", "目撃者", "その他"]},
                    "name_text": {"type": ["string", "null"]},
                    "vehicle_text": {"type": ["string", "null"]},
                },
                "required": ["role", "name_text", "vehicle_text"],
                "additionalProperties": False,
            },
        },
        "injury": {"type": "string", "enum": ["あり", "なし", "不明"]},
        "injury_detail": {"type": ["string", "null"]},
        "vehicle_status": {"type": "string",
                           "enum": ["走行可能", "走行不可", "不明"]},
        "damage_detail": {"type": ["string", "null"]},
        "police_reported": {"type": "string", "enum": ["済", "未", "不明"]},
        "missing_items": {"type": "array", "items": {"type": "string"}},
    },
    "required": [
        "accident_date", "accident_date_basis", "accident_time_text",
        "location_text", "parties", "injury", "injury_detail",
        "vehicle_status", "damage_detail", "police_reported", "missing_items",
    ],
    "additionalProperties": False,
}

受付日時: 2026-10-27 09:12 / 受付経路: 電話
昨日の夕方6時ごろ、市役所の前の交差点で追突されました。
相手の方は軽トラックで、こちらのバンパーが外れていて自走はできません。
首が少し痛いですが、救急車は呼んでいません。警察にはまだ連絡していません。

SYSTEM = """あなたは事故受付の初報を台帳の項目に分ける担当です。
初報に書かれていることだけを根拠にし、書かれていない項目は null か「不明」にします。
過失の有無、保険金の支払可否、負傷の程度は判断しません。
accident_date は、初報に日付があれば「明記」、「昨日」などから受付日時で換算したら「換算」、
決められなければ null と「不明」にします。
聞き取れていない項目は missing_items に項目名で入れます。"""
 
 
def make_params(receipt: dict) -> dict:
    prompt = (f"受付日時: {receipt['received_at']} / 受付経路: {receipt['channel']}\n"
              f"{receipt['text']}")
    return {
        "model": "claude-sonnet-5-5",
        "max_tokens": 1500,
        "system": SYSTEM,
        "messages": [{"role": "user", "content": prompt}],
        "output_config": {"format": {"type": "json_schema", "schema": SCHEMA}},
    }

{
  "accident_date": "2026-10-26",
  "accident_date_basis": "換算",
  "accident_time_text": "夕方6時ごろ",
  "location_text": "市役所の前の交差点",
  "parties": [
    {"role": "通報者", "name_text": null, "vehicle_text": "バン"},
    {"role": "相手方", "name_text": null, "vehicle_text": "軽トラック"}
  ],
  "injury": "あり",
  "injury_detail": "首が少し痛い。救急車は呼んでいない",
  "vehicle_status": "走行不可",
  "damage_detail": "バンパーが外れている",
  "police_reported": "未",
  "missing_items": ["通報者の氏名", "相手方の氏名と連絡先", "事故の正確な場所"]
}

import json
from pathlib import Path
 
import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request
 
client = anthropic.Anthropic()
 
receipts = [json.loads(line)
            for line in Path("intake/2026-10-27.jsonl").read_text(encoding="utf-8").splitlines()]
 
requests = [
    Request(custom_id=r["receipt_id"],
            params=MessageCreateParamsNonStreaming(**make_params(r)))
    for r in receipts
]
 
batch = client.messages.batches.create(requests=requests)
Path("batch_id.txt").write_text(batch.id, encoding="utf-8")
print(batch.id, batch.processing_status)

import sqlite3
from datetime import date
 
db = sqlite3.connect("open_cases.db")
db.execute("""CREATE TABLE IF NOT EXISTS open_cases (
    receipt_id TEXT PRIMARY KEY,
    status TEXT NOT NULL,
    review_reason TEXT,
    extracted TEXT,
    batch_id TEXT
)""")
 
 
def check(extracted: dict, received_at: str) -> list[str]:
    reasons = []
    try:
        if extracted["accident_date"]:
            d = date.fromisoformat(extracted["accident_date"])
            if d > date.fromisoformat(received_at[:10]):
                reasons.append("事故日が受付日より後")
    except ValueError:
        reasons.append("事故日が日付として不正")
    if extracted["injury"] != "なし":
        reasons.append("負傷あり/不明")
    if extracted["missing_items"]:
        reasons.append("聞き取り漏れあり")
    return reasons
 
 
received = {r["receipt_id"]: r["received_at"] for r in receipts}
batch_id = Path("batch_id.txt").read_text(encoding="utf-8")
 
for res in client.messages.batches.results(batch_id):
    rid = res.custom_id
    if res.result.type != "succeeded":
        db.execute("INSERT OR IGNORE INTO open_cases VALUES (?,?,?,?,?)",
                   (rid, "要再投入", res.result.type, None, batch_id))
        continue
    msg = res.result.message
    if msg.stop_reason != "end_turn":
        db.execute("INSERT OR IGNORE INTO open_cases VALUES (?,?,?,?,?)",
                   (rid, "要再投入", f"stop_reason={msg.stop_reason}", None, batch_id))
        continue
    extracted = json.loads(msg.content[0].text)
    reasons = check(extracted, received[rid])
    status = "要確認" if reasons else "自動抽出"
    db.execute(
        """INSERT INTO open_cases VALUES (?,?,?,?,?)
           ON CONFLICT(receipt_id) DO UPDATE SET
             extracted=excluded.extracted, review_reason=excluded.review_reason,
             status=excluded.status, batch_id=excluded.batch_id
           WHERE open_cases.status IN ('自動抽出','要確認','要再投入')""",
        (rid, status, " / ".join(reasons), json.dumps(extracted, ensure_ascii=False), batch_id),
    )
db.commit()