"""MCS-150 Official PDF Form Filler. Fills the official FMCSA MCS-150/150B/150C fillable PDF forms using intake data from the order. Produces a ready-to-fax or electronically submit PDF. Forms stored at: docs/MCS-150 Form.pdf — standard (289 fields) docs/MCS-150B Form.pdf — hazmat safety permit (349 fields) docs/MCS-150C Form.pdf — intermodal equipment (33 fields) Usage: from scripts.document_gen.templates.mcs150_pdf_filler import fill_mcs150 pdf_path = fill_mcs150(intake_data, order_number="CO-12345") """ from __future__ import annotations import logging import os import tempfile from datetime import datetime from pathlib import Path from copy import copy LOG = logging.getLogger("document_gen.mcs150_pdf_filler") try: from pypdf import PdfReader, PdfWriter from pypdf.generic import NameObject, BooleanObject, TextStringObject except ImportError: LOG.warning("pypdf not installed — MCS-150 PDF filling unavailable") PdfReader = None # Path to the official forms DOCS_DIR = Path(__file__).resolve().parent.parent.parent.parent / "docs" FORMS = { "mcs150": DOCS_DIR / "MCS-150 Form.pdf", "mcs150b": DOCS_DIR / "MCS-150B Form.pdf", "mcs150c": DOCS_DIR / "MCS-150C Form.pdf", } # ── Field mappings ──────────────────────────────────────────────────── # Question 22: Carrier Operation (checkboxes) CARRIER_OP_MAP = { "authorized_for_hire": "22aBox", "exempt_for_hire": "22bBox", "private_property": "22cBox", "private_passengers": "22dBox", "us_mail": "22eBox", } # Question 23: Entity Type (checkboxes) ENTITY_TYPE_MAP = { "sole_proprietorship": "23aBox", "partnership": "23bBox", "corporation": "23cBox", "llc": "23dBox", "other": "23kBox", } # Question 24: Cargo Types (checkboxes — a through z, aa through dd) CARGO_TYPE_MAP = { "general": "24aBox", "household": "24bBox", "metal": "24cBox", "motor_vehicles": "24dBox", "drivetow": "24eBox", "logs": "24fBox", "building_materials": "24gBox", "mobile_homes": "24hBox", "machinery": "24iBox", "fresh_produce": "24jBox", "liquids": "24kBox", "intermodal": "24lBox", "passengers": "24mBox", "oilfield": "24nBox", "livestock": "24oBox", "grain": "24pBox", "coal": "24qBox", "meat": "24rBox", "garbage": "24sBox", "chemicals": "24tBox", "commodities_dry": "24uBox", "refrigerated": "24vBox", "beverages": "24wBox", "paper": "24xBox", "utilities": "24yBox", "farm_supplies": "24zBox", "construction": "24aaBox", "water_well": "24bbBox", "other": "24ccBox", } def determine_form_type(intake: dict) -> str: """Determine which MCS-150 form to use. Returns 'mcs150', 'mcs150b', or 'mcs150c'. """ if intake.get("is_intermodal_equipment_provider"): return "mcs150c" if intake.get("hazmat") == "yes" and intake.get("needs_hmsp"): return "mcs150b" return "mcs150" def fill_mcs150(intake: dict, order_number: str = "") -> str: """Fill the official MCS-150 PDF form. Args: intake: Dict with all MCS-150 fields from intake form. order_number: Order number for filename. Returns: Path to the filled PDF. """ if PdfReader is None: raise ImportError("pypdf not installed") form_type = determine_form_type(intake) form_path = FORMS[form_type] if not form_path.exists(): raise FileNotFoundError(f"MCS-150 form not found: {form_path}") reader = PdfReader(str(form_path)) writer = PdfWriter() writer.clone_document_from_reader(reader) # Build field values field_updates = {} # ── Text fields ────────────────────────────────────────────────── field_updates["1bizName"] = intake.get("legal_name", "") field_updates["2dbaName"] = intake.get("dba_name", "") field_updates["3principalStreet"] = intake.get("address_street", "") field_updates["4principalCity"] = intake.get("address_city", "") field_updates["5principalState"] = intake.get("address_state", "") field_updates["6principalZip"] = intake.get("address_zip", "") field_updates["13bizPhone"] = intake.get("phone", "") field_updates["14cellPhone"] = intake.get("cell_phone", "") field_updates["15faxNumber"] = intake.get("fax", "") field_updates["16usdotNumber"] = intake.get("dot_number", "") field_updates["usdotNumber"] = intake.get("dot_number", "") # duplicate field field_updates["17mcmxNumber"] = intake.get("mc_number", "") field_updates["19irsNumber"] = intake.get("ein", "") field_updates["20eMail"] = intake.get("email", "") field_updates["21carrierMileage"] = str(intake.get("annual_miles", "")) # Mailing address (if different) if intake.get("mailing_street"): field_updates["8mailStreet"] = intake.get("mailing_street", "") field_updates["9mailCity"] = intake.get("mailing_city", "") field_updates["10mailState"] = intake.get("mailing_state", "") field_updates["11mailZip"] = intake.get("mailing_zip", "") # Fleet/drivers field_updates["totalDrivers"] = str(intake.get("drivers", "")) field_updates["totalCDL"] = str(intake.get("cdl_drivers", intake.get("drivers", ""))) # Vehicle counts — straight trucks and tractors are most common power_units = intake.get("power_units", "") vehicle_type = intake.get("primary_vehicle_type", "straight") if vehicle_type == "tractor": field_updates["tractorOwn"] = str(power_units) else: field_updates["straightOwn"] = str(power_units) # Officers field_updates["officerName1"] = intake.get("signer_name", "") field_updates["officerTitle1"] = intake.get("signer_title", "") # Certification field_updates["certifyName"] = intake.get("signer_name", "") field_updates["certifyTitle"] = intake.get("signer_title", "") field_updates["certifyDate"] = datetime.now().strftime("%m/%d/%Y") # Interstate/intrastate mileage interstate = intake.get("interstate_intrastate", "") if interstate == "interstate": field_updates["interWithin"] = str(intake.get("annual_miles", "")) elif interstate in ("intrastate_hazmat", "intrastate_non_hazmat"): field_updates["intraWithin"] = str(intake.get("annual_miles", "")) # ── Checkbox fields ────────────────────────────────────────────── checkbox_on = {} # Carrier operation carrier_op = intake.get("carrier_operation", "") if carrier_op in CARRIER_OP_MAP: checkbox_on[CARRIER_OP_MAP[carrier_op]] = True # Entity type entity_type = intake.get("entity_type", "") if entity_type in ENTITY_TYPE_MAP: checkbox_on[ENTITY_TYPE_MAP[entity_type]] = True # Cargo types for cargo in intake.get("cargo_types", []): if cargo in CARGO_TYPE_MAP: checkbox_on[CARGO_TYPE_MAP[cargo]] = True # Reason for filing — biennial update checkbox_on["Reason Button"] = True # Biennial update # Certify box checkbox_on["certifyBox"] = True # ── Apply fields to PDF ────────────────────────────────────────── # Update text fields writer.update_page_form_field_values( writer.pages[0], {k: v for k, v in field_updates.items() if v}, auto_regenerate=False, ) # For multi-page forms, try updating all pages for page_idx in range(len(writer.pages)): try: writer.update_page_form_field_values( writer.pages[page_idx], {k: v for k, v in field_updates.items() if v}, auto_regenerate=False, ) except Exception: pass # Apply checkbox fields for field_name, checked in checkbox_on.items(): if checked: try: # Find the field across all pages and set it for page in writer.pages: if "/Annots" in page: for annot in page["/Annots"]: obj = annot.get_object() if obj.get("/T") and str(obj["/T"]) == field_name: obj.update({ NameObject("/V"): NameObject("/Yes"), NameObject("/AS"): NameObject("/Yes"), }) break except Exception as e: LOG.debug("Checkbox %s set failed: %s", field_name, e) # Save work_dir = tempfile.mkdtemp(prefix="pw_mcs150_") dot = intake.get("dot_number", "unknown") date_str = datetime.now().strftime("%Y%m%d") filename = f"MCS150_DOT{dot}_{date_str}_filled.pdf" filepath = os.path.join(work_dir, filename) with open(filepath, "wb") as f: writer.write(f) LOG.info("Filled MCS-150 (%s) → %s", form_type, filepath) return filepath if __name__ == "__main__": test_intake = { "legal_name": "ADAMS LUMBER INC", "dba_name": "Adams Trucking", "dot_number": "1157913", "mc_number": "MC-456789", "address_street": "123 Timber Lane", "address_city": "Portland", "address_state": "OR", "address_zip": "97201", "phone": "(503) 555-1234", "email": "mark@adamslumber.com", "entity_type": "corporation", "carrier_operation": "authorized_for_hire", "interstate_intrastate": "interstate", "hazmat": "no", "power_units": "5", "drivers": "6", "annual_miles": "250000", "cargo_types": ["general", "building_materials", "logs"], "signer_name": "Mark Adams", "signer_title": "President", } path = fill_mcs150(test_intake, order_number="CO-TEST123") print(f"Generated: {path}") print(f"Size: {os.path.getsize(path)} bytes")