Automating OSHA chemical hygiene inventory checks

import hashlib
import logging
from datetime import date, datetime, timedelta, timezone
from typing import Any

from pydantic import BaseModel, ConfigDict, Field, field_validator

logging.basicConfig(level=logging.INFO, format="%(asctime)s [OSHA-CHK] %(message)s")
logger = logging.getLogger(__name__)


class ContainerRow(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    container_id: str = Field(..., min_length=1, max_length=64)
    cas_number: str = Field(..., alias="cas")
    chemical_name: str = Field(..., min_length=2, max_length=200)
    quantity: float = Field(..., ge=0)
    unit: str = Field(..., pattern=r"^(g|kg|mL|L)$")
    control_present: str | None = None
    received_date: date
    expiry_date: date | None = None
    sds_id: str | None = None

    @field_validator("cas_number")
    @classmethod
    def normalize_and_checksum_cas(cls, v: str) -> str:
        # Strip stray whitespace/prefixes, then verify the CAS check digit:
        # the last digit equals (sum of each preceding digit times its
        # position from the right, starting at 1) mod 10. A bad checksum
        # means a typo that would detach the container from CHP policy.
        digits = v.strip().upper().removeprefix("CAS").strip(" -")
        parts = digits.split("-")
        if len(parts) != 3 or not all(p.isdigit() for p in parts):
            raise ValueError(f"Malformed CAS number: {v!r}")
        body = parts[0] + parts[1]
        check = int(parts[2])
        total = sum(int(d) * i for i, d in enumerate(reversed(body), start=1))
        if total % 10 != check:
            raise ValueError(f"CAS checksum failed: {v!r}")
        return f"{parts[0]}-{parts[1]}-{parts[2]}"


def load_inventory(raw_rows: list[dict[str, Any]]) -> tuple[list[ContainerRow], list[dict]]:
    """Validate every row; return (accepted, rejected) so a bad CAS never
    silently drops a hazardous container from the evaluation."""
    accepted: list[ContainerRow] = []
    rejected: list[dict] = []
    for raw in raw_rows:
        try:
            accepted.append(ContainerRow.model_validate(raw))
        except Exception as exc:  # ValidationError and CAS ValueError
            rejected.append({"row": raw, "error": str(exc)})
            logger.warning("rejected container %s: %s", raw.get("container_id"), exc)
    return accepted, rejected

# Peroxide-former expiry windows in days, keyed by peroxide class.
# Sourced from the CHP — Class I formers (e.g. isopropyl ether) expire fast;
# stabilized Class III formers get the longest window.
PEROXIDE_WINDOWS_DAYS: dict[str, int] = {"I": 90, "II": 180, "III": 365}


def evaluate(container: ContainerRow, rule: dict[str, Any] | None, today: date) -> list[str]:
    """Return reason codes for a single container per 29 CFR 1910.1450."""
    if rule is None:
        return ["unmapped_hazard_class"]  # 1910.1450(e): outside the CHP

    reasons: list[str] = []

    # 1910.1450(d): exposure determination via quantity threshold.
    if container.quantity > rule["threshold_qty"]:
        reasons.append("over_threshold")

    # 1910.1450 App A: particularly hazardous substances need the declared
    # control physically present and attested.
    required = rule.get("control_required")
    if required and container.control_present != required:
        reasons.append("control_missing")

    # Peroxide-former expiry: 3/6/12-month window by peroxide class.
    pclass = rule.get("peroxide_class")
    if pclass:
        window = timedelta(days=PEROXIDE_WINDOWS_DAYS[pclass])
        past_window = today > (container.received_date + window)
        past_stamp = container.expiry_date is not None and today > container.expiry_date
        if container.expiry_date is None or past_stamp or past_window:
            reasons.append("expired")

    # 1910.1200: hazard communication requires a linked SDS.
    if not container.sds_id:
        reasons.append("no_sds")

    return reasons

from sqlalchemy import Column, String, DateTime, create_engine
from sqlalchemy.orm import declarative_base, Session
from sqlalchemy.dialects.postgresql import insert as pg_insert

Base = declarative_base()


class ComplianceLedger(Base):
    __tablename__ = "chp_ledger"
    audit_hash = Column(String(64), primary_key=True)  # natural idempotency key
    container_id = Column(String(64), nullable=False)
    cas_number = Column(String(16), nullable=False)
    reason = Column(String(32), nullable=False)
    matrix_version = Column(String(20), nullable=False)
    recorded_at = Column(DateTime, nullable=False)


def run_checks(
    db_url: str,
    rows: list[dict[str, Any]],
    chp_rules: dict[str, dict[str, Any]],
    matrix_version: str,
) -> dict[str, int]:
    engine = create_engine(db_url, pool_pre_ping=True)
    Base.metadata.create_all(engine)
    today = datetime.now(timezone.utc).date()
    stats = {"recorded": 0, "skipped": 0, "quarantined": 0, "rejected": 0}

    accepted, rejected = load_inventory(rows)
    stats["rejected"] = len(rejected)

    with Session(engine) as session:
        # Sort by container_id so processing order is stable and reproducible.
        for container in sorted(accepted, key=lambda c: c.container_id):
            rule = chp_rules.get(container.cas_number)
            reasons = evaluate(container, rule, today) or ["compliant"]
            if reasons != ["compliant"]:
                stats["quarantined"] += 1

            for reason in reasons:
                audit_hash = hashlib.sha256(
                    f"{container.container_id}:{reason}:{matrix_version}".encode("utf-8")
                ).hexdigest()
                stmt = pg_insert(ComplianceLedger).values(
                    audit_hash=audit_hash,
                    container_id=container.container_id,
                    cas_number=container.cas_number,
                    reason=reason,
                    matrix_version=matrix_version,
                    recorded_at=datetime.now(timezone.utc),
                ).on_conflict_do_nothing(index_elements=["audit_hash"])
                result = session.execute(stmt)
                stats["recorded" if result.rowcount else "skipped"] += 1

        session.commit()

    logger.info("run complete: %s", stats)
    return stats