diff --git a/scripts/document_gen/templates/mcs150_pdf_filler.py b/scripts/document_gen/templates/mcs150_pdf_filler.py new file mode 100644 index 0000000..9a9dd01 --- /dev/null +++ b/scripts/document_gen/templates/mcs150_pdf_filler.py @@ -0,0 +1,288 @@ +"""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")