--- updatedAt: 2025-01-06T18:15:41.000Z agentTools: projectIndex: https://developer.truckstop.com/llms.txt --- # Understand comprehensive risk for carriers ```python RiskFactor.py import json import requests import csv from TruckstopAuth import ManageTruckstopTokens from TruckstopLibrary.Auth import TruckstopCredentials from TruckstopLibrary.RiskFactor import RiskFactorReport truckstopUserProd = TruckstopCredentials( "testUser@truckstop.com", "MyPass123!", "Q0FFQjVDNzEtNz...==" ) baseURL = "https://api.truckstop.com/riskfactoranalysis/v1/carriers/" def GetReportByIdentifier(accessToken:str, identifier:str): url = baseURL + str(identifier) headers = { "Authorization": "bearer " + accessToken, "Accept": "application/json", } response = requests.get(url=url, headers=headers) if (response.status_code == 200): try: report = RiskFactorReport.from_dict(json.loads(response.text)) return report, None except Exception as e: return None, str(e) else: errors = json.loads(response.text) errors = errors.get("errors") return None, errors def SearchForCarrier(accessToken:str, email:str = None, phoneNumber:str = None, vin:str = None): params = [] if email is not None: params.append(f"email={email}") if phoneNumber is not None: params.append(f"phone={phoneNumber}") if vin is not None: params.append(f"vin={vin}") url = baseURL + "/search?" + "&".join(params) headers = { "Authorization": "bearer " + accessToken, "Accept": "application/json" } response = requests.get(url=url, headers=headers) if (response.status_code==200): dotNumbers = json.loads(response.text).get("dotNumbers") return dotNumbers, None else: errorText = json.loads(response.text) errorText = errorText.get("errors") return None, errorText def GetBulkReports(accessToken:str, identifiers:list[str]): url = baseURL headers = { "Authorization": "bearer " + accessToken, "Accept": "application/json", } payload = { "carrierIds": identifiers } response = requests.post(url=url, headers=headers, json=payload) if (response.status_code == 200): response = json.loads(response.text) print(f"Found {response.get("resultCount").get("knownCarrierIdCount")} of {response.get("resultCount").get("queriedCarrierIdCount")}") print(f"Carriers IDs not found: " + ', '.join(response.get("resultCount").get("unknownCarrierIds"))) reports = response.get("reports") reportList = [] errorList = [] for report in reports: try: report = RiskFactorReport.from_dict(report) reportList.append(report) except Exception as e: errorList.append(str(e)) return reportList, errorList else: errorText = json.loads(response.text) errorText = errorText.get("errors") return None, errorText def GetCarrierRMISCertificationStatus(accessToken:str, identifier:str): url = baseURL + str(identifier) + "/certification" headers = { "Authorization": "bearer " + accessToken, "Accept": "application/json", } response = requests.get(url=url, headers=headers) if (response.status_code == 200): response = json.loads(response.text) return response, None else: errorText = json.loads(response.text) errorText = errorText.get("errors") return None, errorText # Sample usage token = ManageTruckstopTokens(truckstopCreds=truckstopUserProd, isSandbox=False).AccessToken #TEST 1 print("TEST #1 - Get Single Report") riskFactorReport, error = GetReportByIdentifier(accessToken=token, identifier=2836509) if (riskFactorReport is not None): print (riskFactorReport.reportSummary) else: for k,v in error.items(): for value in v: print (f'{value}') #TEST 2 print ("\n\nTEST #2 - Search for a carrier") dotList, error = SearchForCarrier(accessToken=token, email="RT@AMCARRIERSINC.COM") if dotList is not None: if len(dotList) > 0: for dot in dotList: print(dot) else: print("No results found") else: for k,v in error.items(): for value in v: print (f'{value}') #Test 3 print ("\n\nTest #3 - Get Bulk Reports") reports, errors = GetBulkReports(accessToken=token, identifiers=[1,2836509,"3", "8"]) for report in reports: print ("\n" + report.reportSummary) for error in errors: for k,v in error.items(): for value in v: print (f'{value}') #Test 4 print ("\n\nTest #4 - Get Carrier Cert status") status, error = GetCarrierRMISCertificationStatus(accessToken=token, identifier=2836509) if status is not None: print (f'Certified: {status.get("certified")}') print (f'Date (un)Certified: {status.get("dateCertified")}') else: for k,v in error.items(): for value in v: print (f'{value}') #TEST 5 - Iterate CSV file print ("\n\nTest #5 - Iterate a CSV file") try: riskReports = [] with open("carriers.csv", 'r', newline='') as csvFile: reader = csv.DictReader(csvFile, delimiter=',') # Read CSV file for row in reader: token = ManageTruckstopTokens(truckstopCreds=truckstopUserProd, isSandbox=False).AccessToken assetRiskLevel = None digitalRiskLevel = None contactRiskLevel = None certStatus = None riskFactorReport, error = GetReportByIdentifier(accessToken=token, identifier=row["Identifier"]) if (riskFactorReport is not None): certStatusResponse, certError = GetCarrierRMISCertificationStatus(accessToken=token, identifier=riskFactorReport.identifiers.dotNumber) if (certStatusResponse is not None): certStatus = certStatusResponse.get("certified") for vector in riskFactorReport.riskVectors: if vector.category == "Asset": assetRiskLevel = vector.riskLevel if vector.category == "Digital": digitalRiskLevel = vector.riskLevel if vector.category == "Contact": contactRiskLevel = vector.riskLevel result_row = { **row, "Report Summary": riskFactorReport.reportSummary, "Asset Risk": assetRiskLevel, "Digital Risk": digitalRiskLevel, "Contact Risk": contactRiskLevel, "RMIS Certification": certStatus } riskReports.append(result_row) else: result_row = { **row, "Report Summary": None, "Asset Risk": None, "Digital Risk": None, "Contact Risk": None, "RMIS Certification": None } riskReports.append(result_row) with open("carriers.csv", 'w', newline='') as csvFile: fieldnames = list(riskReports[0].keys()) # Use the keys of the first row as fieldnames writer = csv.DictWriter(csvFile, fieldnames=fieldnames) writer.writeheader() # Write header row to CSV for result in riskReports: writer.writerow(result) # Write each result row to CSV print ("Finished processing CSV file.") except Exception as e: print (str(e)) ``` ```python RiskFactor (library) import json from typing import List class RiskFactorReport: class Identifiers: def __init__(self, dotNumber: str, mcNumbers: List[str], emails: List[str], phoneNumbers: List[str], addresses: List[str], ipAddresses: List[str], vinNumbers: List[str]): self.dotNumber = dotNumber self.mcNumbers = mcNumbers self.emails = emails self.phoneNumbers = phoneNumbers self.addresses = addresses self.ipAddresses = ipAddresses self.vinNumbers = vinNumbers class RiskSubVectors: def __init__(self, riskLevel: str, subCategory: str, contextData: str): self.riskLevel = riskLevel self.subCategory = subCategory self.contextData = contextData class RiskVectors: def __init__(self, riskLevel: str, category: str, riskSubVectors: List['RiskFactorReport.RiskSubVectors'] = []): self.riskLevel = riskLevel self.category = category self.riskSubVectors = riskSubVectors def __init__(self, identifiers: 'RiskFactorReport.Identifiers', reportSummary: str, reportDetails: str, updatedAt: str, riskVectors: List['RiskFactorReport.RiskVectors'] = []): self.identifiers = identifiers self.reportSummary = reportSummary self.reportDetails = reportDetails self.riskVectors = riskVectors self.updatedAt = updatedAt @staticmethod def from_dict(data): """ Recursively deserialize JSON data into RiskFactorReport object and nested classes. """ def parse_identifiers(data): return RiskFactorReport.Identifiers( dotNumber=data.get('dotNumber'), mcNumbers=data.get('mcNumbers', []), emails=data.get('emails', []), phoneNumbers=data.get('phoneNumbers', []), addresses=data.get('addresses', []), ipAddresses=data.get('ipAddresses', []), vinNumbers=data.get('vinNumbers', []) ) def parse_risk_sub_vectors(data): return RiskFactorReport.RiskSubVectors( riskLevel=data.get('riskLevel'), subCategory=data.get('subCategory'), contextData=data.get('contextData') ) def parse_risk_vectors(data): risk_sub_vectors = [parse_risk_sub_vectors(sub_vector) for sub_vector in data.get('riskSubVectors', [])] return RiskFactorReport.RiskVectors( riskLevel=data.get('riskLevel'), category=data.get('category'), riskSubVectors=risk_sub_vectors ) # Deserialize each nested component identifiers = parse_identifiers(data['identifiers']) risk_vectors = [parse_risk_vectors(rv) for rv in data.get('riskVectors', [])] return RiskFactorReport( identifiers=identifiers, reportSummary=data.get('reportSummary'), reportDetails=data.get('reportDetails'), updatedAt=data.get('updatedAt'), riskVectors=risk_vectors ) ``` ```shell carriers.csv Identifier 2836509 MC949137 ``` # Overview This script shows how to obtain risk reports for carriers as well as the carrier's RMIS certification status. # Libraries {/_ shell@ _/} # Getting risk reports You can also use the search endpoint to find a carriers DOT number from # How to get carrier RMIS certification status **NOTE:** This endpoint will only accept the DOT number for the carrier, not the MC Number # Sample use cases This also shows how a CSV file of MC/DOT numbers can be processed to get a report for each.