""" DOT Drug & Alcohol Compliance Program handler ($149). Instant-delivery service: when a motor carrier orders the program we generate a complete, print-ready PDF "binder" and email it to them automatically (no admin step). The binder bundles the written testing policy, program-management instructions, supervisor-training materials and access, EAP / rehab / SAP resources, regulation citations, random-testing instructions, required forms, and recordkeeping guidance. Policy variant (DOT operating administration) selection: For a trucking carrier the program is FMCSA (49 CFR Part 382) — that is the default. If the customer's operation falls under a different DOT mode (FRA, PHMSA, FTA, FAA, USCG) we honor an explicit ``dot_da_mode`` value in the intake data. An optional ``state_dfwp`` value appends a state Drug-Free Workplace addendum. Returns the local PDF path so job_server uploads it to MinIO and the INSTANT_DELIVERY path emails it to the customer. """ from __future__ import annotations import json import logging import os import tempfile from datetime import datetime from .base_handler import BaseServiceHandler LOG = logging.getLogger("workers.services.dot_drug_alcohol") # DOT operating administrations we can build a binder for. FMCSA is the # trucking default; everything else requires an explicit intake override. _VALID_MODES = {"fmcsa", "fra", "phmsa", "fta", "faa", "uscg"} class DrugAlcoholProgramHandler(BaseServiceHandler): """Generate and instant-deliver the DOT D&A Compliance Program binder.""" SERVICE_SLUG = "dot-drug-alcohol" SERVICE_NAME = "DOT Drug & Alcohol Compliance Program" REQUIRES_LLM = False async def process(self, order_data: dict) -> list[str]: order_number = order_data.get("order_number") or order_data.get("name", "") LOG.info("[%s] Building DOT D&A compliance binder", order_number) intake = order_data.get("intake_data") or {} if isinstance(intake, str): try: intake = json.loads(intake) except (TypeError, ValueError): intake = {} carrier_name = ( intake.get("legal_name") or intake.get("entity_name") or order_data.get("customer_name", "") ).strip() dot_number = str(intake.get("dot_number", "")).strip() cdl_drivers = intake.get("cdl_drivers", "") owner_operators = intake.get("owner_operators", "") der_name = (intake.get("der_name") or "").strip() current_provider = (intake.get("current_da_provider") or "").strip() # ── Policy variant (DOT mode) selection ────────────────────────── mode = self._resolve_mode(intake) # Optional state Drug-Free Workplace addendum. Accept either an # explicit flag/state value or derive from the carrier's base state # if the intake marks it as a DFWP state. state_dfwp = self._resolve_state_dfwp(intake) # The C-TPA / consortium that administers the program. If the # customer already has a provider, name it; otherwise default to our # managed consortium. provider_name = current_provider or "Performance West Consortium / C-TPA" # ── Guard: need at minimum the carrier name to personalize ─────── if not carrier_name: LOG.warning( "[%s] No carrier name in intake — pausing for intake", order_number ) self._request_intake(order_data) return [] # ── Generate the binder ────────────────────────────────────────── work_dir = self._safe_work_dir() date_str = datetime.now().strftime("%Y%m%d") safe_name = "".join( c for c in carrier_name if c.isalnum() or c in (" ", "-", "_") ).strip().replace(" ", "_")[:40] or "carrier" out_path = os.path.join( work_dir, f"DOT_DA_Compliance_Binder_{safe_name}_{date_str}.pdf" ) from scripts.document_gen.templates.dot_da_binder_generator import ( generate_da_binder, ) result = generate_da_binder( output_path=out_path, carrier_name=carrier_name, dot_number=dot_number, mode=mode, cdl_drivers=cdl_drivers, owner_operators=owner_operators, der_name=der_name, der_title="Designated Employer Representative (DER)", provider_name=provider_name, state_dfwp=state_dfwp, ) if not result: LOG.error("[%s] D&A binder generation returned no file", order_number) return [] LOG.info( "[%s] D&A binder ready (mode=%s, state_dfwp=%s) -> %s", order_number, mode, state_dfwp or "none", result, ) return [result] # ------------------------------------------------------------------ # # Variant resolution # ------------------------------------------------------------------ # def _resolve_mode(self, intake: dict) -> str: """Pick the DOT operating-administration variant. Trucking carriers are FMCSA (49 CFR Part 382) by default. Honor an explicit ``dot_da_mode`` override for the rare non-FMCSA operation. """ raw = (intake.get("dot_da_mode") or intake.get("dot_mode") or "").lower().strip() if raw in _VALID_MODES: return raw # A few human-friendly aliases customers might supply. aliases = { "trucking": "fmcsa", "motor carrier": "fmcsa", "cdl": "fmcsa", "rail": "fra", "railroad": "fra", "pipeline": "phmsa", "transit": "fta", "bus": "fta", "aviation": "faa", "air": "faa", "maritime": "uscg", "marine": "uscg", "vessel": "uscg", } return aliases.get(raw, "fmcsa") def _resolve_state_dfwp(self, intake: dict) -> str: """Return a state name if a Drug-Free Workplace addendum is wanted.""" explicit = (intake.get("state_dfwp") or "").strip() if explicit: return explicit # If the customer opted into a DFWP add-on, use their base/operating # state. Otherwise omit the addendum (DOT program stands alone). if str(intake.get("include_state_dfwp", "")).lower() in ("1", "true", "yes"): return ( intake.get("base_state") or intake.get("address_state") or "" ).strip() return "" # ------------------------------------------------------------------ # # Helpers # ------------------------------------------------------------------ # def _safe_work_dir(self) -> str: try: return self._make_work_dir() except Exception: return tempfile.mkdtemp(prefix="da_binder_") def _request_intake(self, order_data: dict) -> None: """Best-effort: nudge for intake when carrier name is missing.""" try: fn = getattr(self, "_request_entity_intake", None) if callable(fn): fn(order_data) except Exception as exc: LOG.warning("Could not request intake: %s", exc)