import json
from typing import List, Optional, Dict
from datetime import datetime, time, timedelta, timezone
from geopy.geocoders import Nominatim
from timezonefinder import TimezoneFinder
import pytz
class Pagination:
def __init__(self, pageNumber, pageSize):
self.PageNumber = pageNumber
self.PageSize = pageSize
def to_dict(self):
return {
k:v for k,v in {
"pageNumber": self.PageNumber,
"pageSize": self.PageSize
}.items() if v is not None
}
def to_json(self, indent=4):
return json.dumps(self.to_dict(), indent=indent)
class SearchCriteria:
def __init__(self, name, operator, value, valueto, logicalOperator):
self.name = name
self.operator = operator
self.value = value
self.valueTo = valueto
self.logicalOperator = logicalOperator
def to_dict(self):
return {
k:v for k,v in {
"name": self.name,
"operator": self.operator,
"value": self.value,
"valueTo": self.valueTo,
"logicalOperator": self.logicalOperator
}.items() if v is not None
}
def to_json(self, indent=4):
return json.dumps(self.to_dict(), indent=indent)
class SortCriteria:
def __init__(self, direction, name):
self.name = name
self.direction = direction
def to_dict(self):
return {
k:v for k,v in {
"name": self.name,
"direction": self.direction
}.items() if v is not None
}
def to_json(self, indent=4):
return json.dumps(self.to_dict(), indent=indent)
class Load:
class User:
def __init__(self, id: str, firstName: str, lastName: str, email: str, phone: str, userName: str):
self.id = id
self.firstName = firstName
self.lastName = lastName
self.email = email
self.phone = phone
self.userName = userName
def to_dict(self):
return {
k: (v.to_dict() if hasattr(v, "to_dict") else v)
for k, v in self.__dict__.items() if v is not None
}
class LoadState:
def __init__(self, loadStateDescription: str, loadStateId: int):
self.loadStateDescription = loadStateDescription
self.loadStateId = loadStateId
def to_dict(self):
return {
k: (v.to_dict() if hasattr(v, "to_dict") else v)
for k, v in self.__dict__.items() if v is not None
}
class LoadStateReason:
def __init__(self, loadStateReasonDescription: str, loadStateResonId: int):
self.loadStateReasonDescription = loadStateReasonDescription
self.loadStateReasonId = loadStateResonId
def to_dict(self):
return {
k: (v.to_dict() if hasattr(v, "to_dict") else v)
for k, v in self.__dict__.items() if v is not None
}
class EquipmentAttributes:
def __init__(self, equipmentTypeId: int, equipmentOptions: List[int], transportationModeId: int, otherEquipmentNeeds: Optional[str]):
self.equipmentTypeId = equipmentTypeId
self.equipmentOptions = equipmentOptions
self.transportationModeId = transportationModeId
self.otherEquipmentNeeds = otherEquipmentNeeds
def to_dict(self):
return {
k: (v.to_dict() if hasattr(v, "to_dict") else v)
for k, v in self.__dict__.items() if v is not None
}
class Location:
def __init__(self, id: str, locationName: str, city: str, state: str, streetAddress1: str, streetAddress2: str, countryCode: str, postalCode: Optional[str], latitude: float, longitude: float):
self.id = id
self.locationName = locationName
self.city = city
self.state = state
self.streetAddress1 = streetAddress1
self.streetAddress2 = streetAddress2
self.countryCode = countryCode
self.postalCode = postalCode
self.latitude = latitude
self.longitude = longitude
def to_dict(self):
return {
k: (v.to_dict() if hasattr(v, "to_dict") else v)
for k, v in self.__dict__.items() if v is not None
}
class LoadStop:
def __init__(self, id: str, type: int, sequence: int, earlyDateTime: datetime, lateDateTime: datetime, location: 'Load.Location', contactName: Optional[str], contactPhone: Optional[str], stopNotes: Optional[str], referenceNumbers: List[str]):
self.id = id
self.type = type
self.sequence = sequence
self.earlyDateTime = earlyDateTime
self.lateDateTime = lateDateTime
self.location = location
self.contactName = contactName
self.contactPhone = contactPhone
self.stopNotes = stopNotes
self.referenceNumbers = referenceNumbers
def to_dict(self):
return {k: (v.to_dict() if hasattr(v, "to_dict") else v)
for k, v in self.__dict__.items() if v is not None}
class Currency:
def __init__(self, amount, currencyCode):
self.amount = amount
self.currencyCode = currencyCode
def to_dict(self):
return {
k: (v.to_dict() if hasattr(v, "to_dict") else v)
for k, v in self.__dict__.items() if v is not None
}
class RateAttributes:
def __init__(self, postedAllInRate, tenderAllInRate):
self.postedAllInRate = postedAllInRate
self.tenderAllInRate = tenderAllInRate
def to_dict(self):
return {
k: (v.to_dict() if hasattr(v, "to_dict") else v)
for k, v in self.__dict__.items() if v is not None
}
class Dimensional:
def __init__(self, length: float, width: float, weight: float, height: float, palletCount: Optional[int], pieceCount: Optional[int], cube: Optional[float]):
self.length = length
self.width = width
self.weight = weight
self.height = height
self.palletCount = palletCount
self.pieceCount = pieceCount
self.cube = cube
def to_dict(self):
return {
k: (v.to_dict() if hasattr(v, "to_dict") else v)
for k, v in self.__dict__.items() if v is not None
}
class LoadActionAttributes:
def __init__(self, loadActionId: int, loadActionOption: str):
self.loadActionId = loadActionId
self.loadActionOption = loadActionOption
def to_dict(self):
return {
k: (v.to_dict() if hasattr(v, "to_dict") else v)
for k, v in self.__dict__.items() if v is not None
}
class TermsAndConditions:
def __init__(self, id: Optional[str], name: Optional[str]):
self.id = id
self.name = name
def to_dict(self):
return {
k: (v.to_dict() if hasattr(v, "to_dict") else v)
for k, v in self.__dict__.items() if v is not None
}
def __init__(self,
equipmentAttributes: EquipmentAttributes, # Required
loadStops: List[LoadStop], # Required
loadNumber: str, # Required
rateAttributes: Dict[str, RateAttributes], # Required
dimensional: Optional[Dimensional], # Required
loadActionAttributes: LoadActionAttributes, # Required
createDateTime: str = None,
updateDateTime: str = None,
loadState: LoadState = None,
loadStateReason:LoadStateReason = None,
source:str = None,
loadSource:str = None,
integratorLoadUrl:str = None,
nextRefreshAvailableAt:str = None,
freightClassId:int = None,
distance: float = None,
carrierTenderGroupIds = None,
loadActivity = None,
tenderId: str = None,
carrierName:str = None,
updatedBy: str = None,
legacyLoadId: str = None,
loadId: Optional[str] = None,
postAsUserId: Optional[str] = None,
postAsUser: Optional[User] = None,
createdBy: Optional[str] = None,
commodityId: Optional[int] = None,
loadTrackingRequired: Optional[bool] = None,
customData: Optional[List[Dict[str, str]]] = None,
loadLabel: Optional[str] = None,
tenderNotes: Optional[str] = None,
loadReferenceNumbers: Optional[List[str]] = None, # Conditionally Required
termsAndConditions: Optional[TermsAndConditions] = None, # Conditionally Required
isPrivate: Optional[bool] = None,
note: Optional[str] = None):
self.loadId = loadId
self.loadActivity = loadActivity
self.carrierName = carrierName
self.tenderId = tenderId
self.loadState = loadState
self.loadStateReason = loadStateReason
self.legacyLoadId = legacyLoadId
self.createDateTime = createDateTime
self.updateDateTime = updateDateTime
self.updatedBy = updatedBy
self.postAsUserId = postAsUserId
self.postAsUser = postAsUser
self.createdBy = createdBy
self.equipmentAttributes = equipmentAttributes
self.commodityId = commodityId
self.loadStops = loadStops
self.loadNumber = loadNumber
self.loadTrackingRequired = loadTrackingRequired
self.rateAttributes = rateAttributes
self.customData = customData
self.dimensional = dimensional
self.loadActionAttributes = loadActionAttributes
self.loadLabel = loadLabel
self.tenderNotes = tenderNotes
self.loadReferenceNumbers = loadReferenceNumbers
self.termsAndConditions = termsAndConditions
self.isPrivate = isPrivate
self.note = note
#self.validate_data()
def ValidateLoad(self):
errors = validate_data(self)
if(errors):
return False, errors
return True, None
def to_dict(self):
return {
"loadId": self.loadId,
"postAsUserId": self.postAsUserId,
"postAsUser": self.postAsUser.to_dict() if self.postAsUser else None,
"createdBy": self.createdBy,
"equipmentAttributes": self.equipmentAttributes.to_dict(),
"commodityId": self.commodityId,
"loadStops": [stop.to_dict() for stop in self.loadStops],
"loadNumber": self.loadNumber,
"loadTrackingRequired": self.loadTrackingRequired,
"rateAttributes": self.rateAttributes.to_dict(),
"customData": self.customData,
"dimensional": self.dimensional.to_dict(),
"loadActionAttributes": self.loadActionAttributes.to_dict(),
"loadLabel": self.loadLabel,
"tenderNotes": self.tenderNotes,
"loadReferenceNumbers": self.loadReferenceNumbers,
"termsAndConditions": self.termsAndConditions.to_dict() if self.termsAndConditions else None,
"isPrivate": self.isPrivate,
"note": self.note
}
def validate_data(self):
errors = []
# Run validations
errors.append(validate_stop_locations(self))
errors.append(validate_conditional_requirements(self))
errors.append(validate_stop_dates(self))
errors.append(validate_stop_order(self))
errors.append(validate_dimensional(self))
errors = [error for error in errors if error]
if errors:
return errors
def validate_dimensional(self):
errors = []
dimensional = self.dimensional
if not dimensional:
return errors
if dimensional.length is not None:
if not (1 <= dimensional.length <= 999):
errors.append("Length must be between 1 and 999")
if dimensional.width is not None:
if not (1 <= dimensional.width <= 999):
errors.append("Width must be between 1 and 999")
if dimensional.height is not None:
if not (1 <= dimensional.height <= 999):
errors.append("Height must be between 1 and 999")
if dimensional.weight is not None:
if not (1 <= dimensional.weight <= 999999):
errors.append("Weight must be between 1 and 999,999")
if dimensional.palletCount is not None:
if not (1 <= dimensional.palletCount <= 99):
errors.append("Pallet count must be between 1 and 99")
if dimensional.pieceCount is not None:
if not (1 <= dimensional.pieceCount <= 99999):
errors.append("Piece count must be between 1 and 99,999")
if dimensional.cube is not None:
if not (1 <= dimensional.cube <= 9999999):
errors.append("Cube must be between 1 and 9,999,999")
return errors
def validate_stop_locations(self):
errors = []
for stop in self.loadStops:
if not ((stop.location.city and stop.location.state) or stop.location.postalCode):
errors.append(f"Stop {stop.sequence} must have either a city/state or postal code.")
if (stop.location.city and stop.location.state and stop.location.postalCode):
stopValid, message = validate_city_state_zip(city=stop.location.city, state=stop.location.state, zipCode=stop.location.postalCode)
if (not stopValid):
errors.append(f"Stop {stop.sequence}: {message}")
elif (stop.location.city and stop.location.state):
stopValid, message = validate_city_state_zip(city=stop.location.city, state=stop.location.state)
if (not stopValid):
errors.append(f"Stop {stop.sequence}: {message}")
elif (stop.location.postalCode):
stopValid, message = validate_city_state_zip(zipCode=stop.location.postalCode)
if (not stopValid):
errors.append("Stop {stop.sequence}: {message}")
if len(self.loadStops) < 2:
errors.append("Load must have at least 2 stops.")
has_type_1 = any(stop.type == 1 for stop in self.loadStops)
has_type_2 = any(stop.type == 2 for stop in self.loadStops)
if not (has_type_1 and has_type_2):
errors.append("Load must have at least one stop of type 1 (origin) and one stop of type 2 (destination).")
return errors
def validate_conditional_requirements(self):
errors = []
"""Ensures LoadReferenceNumbers and TermsAndConditions are set if LoadActionID is 1."""
if self.loadActionAttributes.loadActionId == 1:
if not self.loadReferenceNumbers:
errors.append("Load Reference Numbers is required when LoadActionID is set to 1.")
if not self.termsAndConditions:
errors.append("Terms And Conditions is required when LoadActionID is set to 1.")
return errors
def validate_stop_dates(self):
errors = []
"""Ensures that all stop dates are in the future."""
now = datetime.now(timezone.utc)
stopTimeZone = None
for stop in self.loadStops:
try:
if (stop.location.postalCode):
stopTimeZone = get_timezone_from_location(zip=stop.location.postalCode, country=stop.location.countryCode)
if (stop.location.city and stop.location.state):
stopTimeZone = get_timezone_from_location(city=stop.location.city, state=stop.location.state, country=stop.location.countryCode)
except Exception as e:
pass
if stopTimeZone:
earlyDateTime = datetime.fromisoformat(stop.earlyDateTime)
lateDateTime = datetime.fromisoformat(stop.lateDateTime)
if len(str(stop.earlyDateTime)) == 10:
earlyDateTime = earlyDateTime.replace(hour=00, minute=00, second=00)
stop.earlyDateTime = earlyDateTime.strftime("%Y-%m-%dT%H:%M:%S")
if len(str(stop.lateDateTime)) == 10:
lateDateTime = lateDateTime.replace(hour=23, minute=59, second=59)
stop.lateDateTime = lateDateTime.strftime("%Y-%m-%dT%H:%M:%S")
earlyDateTime = stopTimeZone.localize(earlyDateTime)
lateDateTime = stopTimeZone.localize(lateDateTime)
earlyDateTime = earlyDateTime.astimezone(timezone.utc)
lateDateTime = lateDateTime.astimezone(timezone.utc)
if earlyDateTime < now:
errors.append(f"Early Date Time for stop {stop.sequence} is in the past: {stop.earlyDateTime}")
if lateDateTime < now:
errors.append(f"Late Date Time for stop {stop.sequence} is in the past. {stop.lateDateTime}")
if (earlyDateTime > now + timedelta(days=45)):
errors.append(f"Early Date Time for stop {stop.sequence} can not be over 45 days in the future: {stop.earlyDateTime}")
if (lateDateTime > now + timedelta(days=45)):
errors.append(f"Late Date Time for stop {stop.sequence} can not be over 45 days in the future: {stop.earlyDateTime}")
if (earlyDateTime > lateDateTime):
errors.append(f"Early Date Time for stop {stop.sequence} can not be after the Late Date Time: {stop.earlyDateTime}, {stop.lateDateTime}")
else:
errors.append(f"Unable to determine stop time zone for stop {stop.sequence}\n")
return errors
def validate_stop_order(self):
errors = []
"""Ensures that load stops are ordered by ascending earlyDateTime."""
for i in range(len(self.loadStops) - 1):
if self.loadStops[i].earlyDateTime > self.loadStops[i + 1].earlyDateTime:
errors.append(f"Load stops are not in ascending order:")
for stop in self.loadStops:
errors.append(f"\tStop: {stop.sequence}: Early Time: {stop.earlyDateTime}")
return errors
def validate_city_state_zip(city = None, state = None, zipCode = None, country = "USA"):
# Initialize the geolocator with a specific user-agent
geolocator = Nominatim(user_agent="validateCityStatesZips")
if (zipCode):
location = geolocator.geocode(f"{zipCode}, {country}", addressdetails=True)
elif (city and state):
location = geolocator.geocode(f"{city}, {state}, {country}", addressdetails=True)
# Check if a location was found
if not location:
return False, "Invalid Location. Check for valid city/state and/or zip code"
address = location.raw.get('address', "") # Convert to lowercase for case-insensitive comparison
"""If Zip provided, use that only, then verify city/state against what is returned"""
city_valid = state_valid = zip_valid = False
if (city and state and zipCode):
zip_valid = True
city_valid = (address.get('city', '').lower() == city.lower() or
address.get('town', '').lower() == city.lower() or
address.get('village', '').lower() == city.lower() or
address.get('municipality', '').lower()) if city else True
stateCode = address.get('ISO3166-2-lvl4').split('-')[1]
if state == stateCode:
state_valid=True
elif (city and state):
zip_valid = True
city_valid = (address.get('city', '').lower() == city.lower() or
address.get('town', '').lower() == city.lower() or
address.get('village', '').lower() == city.lower() or
address.get('municipality', '').lower() == city.lower()) if city else True
stateCode = address.get('ISO3166-2-lvl4').split('-')[1]
if state == stateCode:
state_valid=True
elif (zipCode):
zip_valid = True
city_valid = True
state_valid=True
if city_valid and state_valid and zip_valid:
return True, f"Valid location: {location.address}"
else:
return False, f"Invalid Location. Closest match: {location.address}\n"
def get_timezone_from_location(city: str = None, state: str = None, zip: str = None, country:str = "USA"):
# Step 1: Get latitude and longitude
geolocator = Nominatim(user_agent="timezone_finder")
location = None
if (city and state and zip):
location = geolocator.geocode(f"{city}, {state}, {zip}")
elif (city and state):
location = geolocator.geocode(f"{city}, {state}")
elif (zip):
location = geolocator.geocode(f'{zip}, {country}')
if not location:
raise ValueError("Location not found. Please check city/state.")
# Step 2: Find the timezone using latitude and longitude
tf = TimezoneFinder()
timezone_str = tf.timezone_at(lng=location.longitude, lat=location.latitude)
if not timezone_str:
raise ValueError("Timezone not found for this location.")
# Step 3: Convert a datetime to the local timezone
local_timezone = pytz.timezone(timezone_str)
return local_timezone
def get_city_state_from_zip(zip_code: str, country: str = "USA"):
# Initialize the geolocator
geolocator = Nominatim(user_agent="cityFinder")
# Get the location data for the zip code
location = geolocator.geocode(f'{zip_code}, {country}')
if not location:
raise ValueError("Location not found. Please check the zip code.")
# Extract city and state from the address
# Extract city and state from display_name
display_name = location.raw.get('display_name')
if not display_name:
raise ValueError("Display name not found in location data.")
# Split display_name by commas and extract the city and state
components = display_name.split(", ")
if len(components) < 3:
raise ValueError("Could not parse city and state from display name.")
# Assuming display_name format: "Zip, City, County, State, Country"
city = components[1]
state = components[3]
return city, state