TruckstopLibrary.Load.py

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