7e5946d65a
Customer saw the MCS-150 looking blank / 'data covered by the form fields': the values were correctly written to the AcroForm /V, but pypdf left the template's empty /AP appearance streams in place and NeedAppearances was false, so viewers rendered the blank widget over the value. Setting AcroForm /NeedAppearances=true makes viewers regenerate appearances from the values. (The missing signature was a downstream effect of the separate fobj_put MinIO-upload bug, now fixed -- with no PDF in MinIO the anchor extraction + signature stamping both failed.)
302 lines
11 KiB
Python
302 lines
11 KiB
Python
"""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)
|
|
|
|
# Force viewers to (re)generate field appearance streams from the values we
|
|
# set. Without /NeedAppearances, pypdf leaves the template's blank /AP streams
|
|
# in place, so the typed values are present in /V but the viewer renders the
|
|
# empty widget on top -- the data looks missing / "covered up by the form
|
|
# field". Setting NeedAppearances=true on the AcroForm fixes the rendering.
|
|
try:
|
|
catalog = writer._root_object
|
|
if "/AcroForm" in catalog:
|
|
acro = catalog["/AcroForm"]
|
|
acro_obj = acro.get_object() if hasattr(acro, "get_object") else acro
|
|
acro_obj[NameObject("/NeedAppearances")] = BooleanObject(True)
|
|
except Exception as exc:
|
|
LOG.warning("Could not set NeedAppearances: %s", exc)
|
|
|
|
# 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")
|