SDSのPDFからリスクアセスメントの下書きをClaudeで作る — 人が確認する欄の分け方

import base64
import json
import sys
from pathlib import Path
 
import anthropic
 
MODEL = "claude-opus-5-5"
client = anthropic.Anthropic()
 
PROMPT = """このSDS(安全データシート)から、次のJSONだけを出力してください。
ルール:
- SDSに書いてある内容だけを使い、推測や補完をしない
- 各項目に page(PDFビューアーのページ番号)と
  quote(SDSの原文のまま。要約しない)を付ける
- 見つからない項目は空配列にし、not_found に項目名を書く
- hazard には対象(成分名または製品全体)を含める
- ばく露限界値は、機関名をSDSの表記のまま body に入れる
- 保護具は、SDSが挙げているものをそのまま列挙する
 
{"product": "", "supplier": "", "revision_date": "",
 "components": [{"name": "", "cas": "", "content": ""}],
 "ghs": [{"hazard": "", "category": "", "page": 0, "quote": ""}],
 "exposure_limits": [{"body": "", "value": "", "page": 0, "quote": ""}],
 "ppe": [{"item": "", "page": 0, "quote": ""}],
 "not_found": []}
"""
 
 
def extract(pdf: Path) -> dict:
    data = base64.standard_b64encode(pdf.read_bytes()).decode("utf-8")
    msg = client.messages.create(
        model=MODEL,
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": [
                {"type": "document", "source": {
                    "type": "base64",
                    "media_type": "application/pdf",
                    "data": data}},
                {"type": "text", "text": PROMPT},
            ],
        }],
    )
    if msg.stop_reason == "max_tokens":
        raise RuntimeError("出力が途中で切れました")
    text = "".join(b.text for b in msg.content if b.type == "text")
    text = text.strip().removeprefix("```json").removesuffix("```")
    return json.loads(text)
 
 
for pdf in sorted(Path(sys.argv[1]).glob("*.pdf")):
    try:
        out = extract(pdf)
    except (json.JSONDecodeError, RuntimeError) as e:
        print(f"要手作業: {pdf.name} ({e})")
        continue
    pdf.with_suffix(".json").write_text(
        json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"抽出: {pdf.name}")

import json
import re
import subprocess
import sys
from pathlib import Path
 
 
def norm(s: str) -> str:
    return re.sub(r"\s+", "", s)
 
 
def page_texts(pdf: Path) -> list[str]:
    out = subprocess.run(
        ["pdftotext", "-layout", str(pdf), "-"],
        capture_output=True, text=True, check=True).stdout
    pages = out.split("\f")
    if pages and not pages[-1].strip():
        pages.pop()
    return [norm(p) for p in pages]
 
 
def verify(pdf: Path) -> dict:
    rec = json.loads(pdf.with_suffix(".json").read_text(encoding="utf-8"))
    texts = page_texts(pdf)
    rec["text_layer"] = any(texts)
    for key in ("ghs", "exposure_limits", "ppe"):
        for row in rec[key]:
            p = row["page"]
            row["verified"] = (
                rec["text_layer"]
                and 1 <= p <= len(texts)
                and norm(row["quote"]) in texts[p - 1]
            )
    return rec
 
 
for pdf in sorted(Path(sys.argv[1]).glob("*.pdf")):
    if not pdf.with_suffix(".json").exists():
        continue
    pdf.with_suffix(".json").write_text(
        json.dumps(verify(pdf), ensure_ascii=False, indent=2),
        encoding="utf-8")
    print(f"照合: {pdf.name}")

import csv
import json
import sys
from pathlib import Path
 
CLAUDE_COLS = ["製品名", "供給者", "SDS改訂日", "成分(CAS)", "GHS区分",
               "ばく露限界値", "SDS記載の保護具", "根拠ページ",
               "照合結果", "SDSに記載なし"]
HUMAN_COLS = ["作業内容", "取扱量・含有率", "換気", "作業時間・頻度",
              "見積り方法", "リスクレベル", "低減措置", "実施者", "承認日"]
 
 
def row(rec: dict) -> list[str]:
    rows = rec["ghs"] + rec["exposure_limits"] + rec["ppe"]
    ok = rec["text_layer"] and all(r["verified"] for r in rows)
    return [
        rec["product"], rec["supplier"], rec["revision_date"],
        "; ".join(f'{c["name"]}({c["cas"]})' for c in rec["components"]),
        "; ".join(f'{g["hazard"]} {g["category"]}' for g in rec["ghs"]),
        "; ".join(f'{e["body"]} {e["value"]}' for e in rec["exposure_limits"]),
        "; ".join(p["item"] for p in rec["ppe"]),
        ", ".join(sorted({str(r["page"]) for r in rows})),
        "照合済み" if ok else "要目視",
        ", ".join(rec["not_found"]),
    ] + [""] * len(HUMAN_COLS)
 
 
records = [json.loads(Path(p).read_text(encoding="utf-8"))
           for p in sys.argv[2:]]
with open(sys.argv[1], "w", newline="", encoding="utf-8-sig") as f:
    w = csv.writer(f)
    w.writerow(CLAUDE_COLS + HUMAN_COLS)
    w.writerows(row(r) for r in records)