Parsing safety data sheets into structured inventory records

Problem statement #

You need a Python routine that reads a vendor safety data sheet, segments it into the 16 GHS sections, pulls out the product identity, CAS number, hazard classification, H-statement codes, signal word, pictograms, and handling controls, validates them into a typed record, versions that record by its revision date, and links it to the physical container on the shelf — so that a re-parse never duplicates a sheet and every hazardous chemical can be proven to have a current, accessible SDS.

This task sits under the Safety Data Sheet Management guide, part of the broader Hazardous Material & Chemical Inventory Compliance practice. The parser is deliberately narrow: it turns one document into one validated record and reports what it could not extract. It does not decide whether a chemical is stored compliantly — it captures, validates, and versions the hazard data that the downstream checks reason over.

Prerequisites #

• Python 3.10+ (modern type hints, datetime.date, str | None).
• Libraries: pydantic>=2.5 for validation, and pdfplumber for extracting text from vendor PDF sheets. Install with pip install "pydantic>=2.5" pdfplumber. Sheets already supplied as plain text can skip the extraction step.
• Environment variables: SDS_SOURCE_DIR — the read-only path where vendor documents land. Fetching sheets from a vendor portal is out of scope here and follows the Security Boundary Configuration for credential scoping.
• Policy config: a version-controlled heading-alias table (mapping vendor section labels to the canonical 16 GHS sections) and a product-alias table (mapping container labels to sheet product names), kept alongside the Safety Data Sheet Management records.

Step-by-step implementation #

The flow is: read the sheet’s text, segment it into 16 sections by heading, extract the fields that matter from Sections 1, 2, 3, 7, and 8, validate into a Pydantic model, version by the Section 16 revision date, and link the container. The revision date and the derived sds_id are what make a re-run safe.

Step 1 — Read the sheet text and segment the 16 sections #

First lift the text out of the vendor document, then split it on the 16 canonical GHS headings. Vendors phrase headings inconsistently (“Section 1: Identification” versus “1. Identification of the substance”), so segmentation matches on the section number and a keyword rather than an exact string. Text that appears before Section 1 or after Section 16’s body is discarded.

import re
from typing import Any

import pdfplumber

# Canonical GHS section anchors. The regex tolerates "Section 1", "1.", "1 -"
# and a following keyword, so vendor-specific phrasing still maps to a number.
SECTION_ANCHORS: dict[int, re.Pattern[str]] = {
    n: re.compile(rf"(?im)^\s*(?:section\s*)?{n}\b[\.\):\-\s]", )
    for n in range(1, 17)
}


def read_text(path: str) -> str:
    """Extract raw text from a vendor SDS PDF."""
    with pdfplumber.open(path) as pdf:
        return "\n".join(page.extract_text() or "" for page in pdf.pages)


def segment_sections(text: str) -> dict[int, str]:
    """Split SDS text into its 16 GHS sections keyed by section number."""
    # Find the start offset of each section heading, in document order.
    starts: list[tuple[int, int]] = []
    for number, anchor in SECTION_ANCHORS.items():
        m = anchor.search(text)
        if m:
            starts.append((m.start(), number))
    starts.sort()

    sections: dict[int, str] = {}
    for i, (offset, number) in enumerate(starts):
        end = starts[i + 1][0] if i + 1 < len(starts) else len(text)
        sections[number] = text[offset:end].strip()
    return sections

Step 2 — Extract identity, CAS, and GHS hazard fields #

With the sections isolated, pull the fields each one is required to carry. Section 1 gives the product identity; Section 3 lists the CAS number in the composition; Section 2 carries the GHS hazard classification, the H-statement codes, the signal word, and the pictograms. The extractors are conservative: they return None or an empty list when a field is genuinely absent rather than guessing, so validation in Step 3 can decide whether the omission is acceptable.

CAS_RE = re.compile(r"\b(\d{2,7}-\d{2}-\d)\b")
H_CODE_RE = re.compile(r"\b(H\d{3}[A-Za-z]?)\b")
SIGNAL_RE = re.compile(r"(?i)\bsignal word\b[:\s]*(danger|warning)")
PICTOGRAM_RE = re.compile(r"(?i)\b(GHS0[1-9])\b")


def extract_product_name(section1: str) -> str | None:
    # Section 1 labels the identifier as "Product name" / "Product identifier".
    m = re.search(r"(?im)^.*product (?:name|identifier)\s*[:\-]\s*(.+)$", section1)
    return m.group(1).strip() if m else None


def extract_cas(section3: str) -> str | None:
    m = CAS_RE.search(section3)
    return m.group(1) if m else None


def extract_hazards(section2: str) -> dict[str, Any]:
    signal = SIGNAL_RE.search(section2)
    return {
        "h_codes": sorted(set(H_CODE_RE.findall(section2))),
        "signal_word": signal.group(1).title() if signal else "None",
        "pictograms": sorted(set(PICTOGRAM_RE.findall(section2))),
    }


def extract_handling(section7: str, section8: str) -> dict[str, Any]:
    # Section 7 storage guidance, Section 8 PPE / exposure controls.
    ppe = re.findall(r"(?i)\b(gloves|goggles|respirator|face shield|lab coat)\b", section8)
    storage = re.search(r"(?im)^.*store(?:d)?\s+(.+)$", section7)
    return {
        "ppe": sorted(set(p.lower() for p in ppe)),
        "storage_class": storage.group(1).strip()[:120] if storage else None,
    }

Step 3 — Validate into a typed SDS record #

The Pydantic model is the boundary that decides whether a parse is trustworthy. It re-checks the CAS check digit, enforces the H### grammar on every hazard code, and restricts the signal word to the GHS-permitted set. A record with an unparseable revision date cannot be versioned, so it is rejected here and routed to quarantine rather than stored with a guessed date.

import hashlib
from datetime import date

from pydantic import BaseModel, ConfigDict, Field, field_validator


class SdsRecord(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True)

    product_name: str = Field(..., min_length=1, max_length=300)
    cas_number: str | None = None
    revision_date: date
    signal_word: str = "None"
    h_codes: list[str] = Field(default_factory=list)
    pictograms: list[str] = Field(default_factory=list)
    ppe: list[str] = Field(default_factory=list)
    storage_class: str | None = None
    source_version: str  # SHA-256 of the parsed document bytes

    @field_validator("cas_number")
    @classmethod
    def cas_checksum(cls, v: str | None) -> str | None:
        # The trailing digit is a checksum; a typo fails before it is stored.
        if v is None:
            return v
        digits = [int(d) for d in v.replace("-", "")]
        check = digits.pop()
        total = sum(i * d for i, d in enumerate(reversed(digits), start=1))
        if total % 10 != check:
            raise ValueError(f"CAS checksum failed for {v!r}")
        return v

    @field_validator("h_codes")
    @classmethod
    def h_grammar(cls, v: list[str]) -> list[str]:
        pat = re.compile(r"^H\d{3}[A-Za-z]?$")
        bad = [c for c in v if not pat.match(c)]
        if bad:
            raise ValueError(f"Malformed H-codes: {bad}")
        return v

    @field_validator("signal_word")
    @classmethod
    def signal_enum(cls, v: str) -> str:
        if v not in {"Danger", "Warning", "None"}:
            raise ValueError(f"Illegal signal word {v!r}")
        return v

    @property
    def sds_id(self) -> str:
        key = f"{self.product_name.lower()}|{self.revision_date.isoformat()}"
        return hashlib.sha256(key.encode("utf-8")).hexdigest()[:24]

Step 4 — Extract the revision date and version the record #

The revision date lives in Section 16 and defines which sheet is current. Vendors write it many ways, so parse defensively and fail loudly when no date can be found — an unversioned sheet is worse than a quarantined one. The sds_id derived from product plus revision date is the idempotency key: the same sheet re-parsed resolves to the same id.

from datetime import datetime

DATE_PATTERNS = [
    (re.compile(r"(?i)revision date[:\s]*(\d{4}-\d{2}-\d{2})"), "%Y-%m-%d"),
    (re.compile(r"(?i)revision date[:\s]*(\d{2}/\d{2}/\d{4})"), "%m/%d/%Y"),
    (re.compile(r"(?i)(?:revised|version date)[:\s]*(\d{1,2}\s+\w+\s+\d{4})"), "%d %B %Y"),
]


def extract_revision_date(section16: str) -> date | None:
    for pattern, fmt in DATE_PATTERNS:
        m = pattern.search(section16)
        if m:
            try:
                return datetime.strptime(m.group(1), fmt).date()
            except ValueError:
                continue
    return None  # caller quarantines: a sheet with no revision cannot be versioned


def build_record(sections: dict[str, str], raw_bytes: bytes) -> SdsRecord:
    hazards = extract_hazards(sections.get(2, ""))
    handling = extract_handling(sections.get(7, ""), sections.get(8, ""))
    revision = extract_revision_date(sections.get(16, ""))
    if revision is None:
        raise ValueError("No revision date in Section 16; cannot version.")
    return SdsRecord(
        product_name=extract_product_name(sections.get(1, "")) or "",
        cas_number=extract_cas(sections.get(3, "")),
        revision_date=revision,
        source_version=hashlib.sha256(raw_bytes).hexdigest(),
        **hazards,
        **handling,
    )

Step 5 — Link the container and flag stale or missing sheets #

Finally, bind the container to its current sheet. A container whose product has no sheet, or whose linked sheet predates the newest revision on file, is flagged. This is the check that satisfies the accessibility requirement of the Hazard Communication Standard: every hazardous container must resolve to a current SDS or appear on the gap list.

def link_container(container: dict[str, Any], record: SdsRecord,
                   newest_on_file: date, stale_days: int = 365 * 3) -> dict[str, Any]:
    """Attach the SDS to a container, returning a gap entry if one applies."""
    age = (datetime.now(tz=None).date() - record.revision_date).days
    container["current_sds_id"] = record.sds_id
    if record.revision_date < newest_on_file:
        return {"container_id": container["container_id"], "reason": "stale_sds"}
    if age > stale_days:
        return {"container_id": container["container_id"], "reason": "aged_sds"}
    return {}

The validated record then feeds the Automating OSHA Chemical Hygiene Inventory Checks routine, which reads the hazard fields to evaluate storage segregation — it never re-parses the sheet.

Schema and field reference #

Verification #

1. Segment count. Assert len(segment_sections(text)) == 16 for a well-formed sheet. Fewer sections means a heading was missed and the alias table needs extending — a partial parse must never be stored.
2. Reproduce the identity. Re-derive sds_id from product_name and revision_date and confirm it equals the stored key; a mismatch means the versioning inputs changed after ingest.
3. CAS round-trip. Feed a known-good CAS number (e.g. water, 7732-18-5) and a corrupted one through the validator; the good one passes and the corrupted one raises ValueError.
4. Idempotent re-parse. Parse the same document twice. Both runs must yield the same sds_id and source_version, and the upsert must write nothing on the second pass.

Troubleshooting #

• Heading layouts vary by manufacturer. One vendor writes “SECTION 2 — HAZARD IDENTIFICATION”, another “2. Hazards Identification”, a third folds the number into a running header. Match on the section number plus a keyword rather than an exact string, and keep a version-controlled alias table so a new vendor phrasing is a config change, not a code change. When segmentation yields fewer than 16 sections, quarantine the sheet for review rather than storing a partial record.
• CAS checksum failures on OCR’d sheets. A scanned SDS run through OCR frequently transposes digits, and the CAS check digit will catch it — which is the point. Do not disable the checksum to force the record in; re-OCR the page at higher resolution or confirm the number against the product identity, because a wrong CAS silently mis-classifies the chemical everywhere downstream.
• Ambiguous or missing revision dates. Section 16 may carry both a “revision date” and a “print date”, or a locale-specific format the patterns miss. Prefer the labeled revision date, extend DATE_PATTERNS for new formats, and never fall back to the file’s modification time — that would let a re-downloaded sheet appear newer than the manufacturer’s actual revision and quietly supersede the correct record.

Frequently asked questions #

Related #

• Up to the parent topic: Safety Data Sheet Management
• Hazardous Material & Chemical Inventory Compliance — the compliance domain this parser serves
• Automating OSHA Chemical Hygiene Inventory Checks — the checks that read these hazard records
• Tracking RCRA Hazardous Waste Accumulation Dates in Python — the disposal side of the same containers