import requests
import time
import json
import xml.etree.ElementTree as ET
import subprocess
import threading
from DocumentAPI import GetDocument
from RMISLibrary import Delta, Carrier, RMISCredentials, Document
from TruckStopDBHelper import *
baseURL = "https://api.rmissecure.com" # Base URL for all RMIS API calls
def Start():
rmisCreds = RMISCredentials(1234, "MyPass123!") # Hardcoded credentials for the example, should be sourced from user in production
while True: # Run indefinitely
while (int(DeltaSummary(rmisCreds)) > 0): # Loop until the Summary endpoint returns 0
deltasToClear = [] # Empty list to store deltas to clear
for delta in DeltaFetch(rmisCreds): # Get a batch of deltas for processing
#Get the carrier details from expanded, then store them in the database, finally add the delta to the list of deltas to clear
carrier = CallExpanded(rmisCreds, delta)
dataBaseObject = GetDBConnection("RMISCarriers")
InsertIntoTable(table="Carriers", dataBaseObject=dataBaseObject, object=carrier, whereFields=['ClientID', 'CarrierRMISID'], whereValues=[carrier.clientID, carrier.CarrierRMISID])
dataBaseObject.sqlConn.close()
deltasToClear.append(Delta(int(delta), str(carrier.lastUpdated))) # Append delta to the list to be cleared
if len(deltasToClear) > 0: # If there are deltas to clear, call the clear function with the list
DeltaClear(rmisCreds, deltasToClear)
time.sleep(300) # Wait 5 minutes before running again
def DeltaSummary(rmisCreds:RMISCredentials):
url = baseURL + "/_c/std/api/DeltaAPI.aspx" # URL for Delta API
payload = {
"clientID": rmisCreds.clientID,
"clientPassword": rmisCreds.clientPassword,
"APIMode": "Summary"
}
headers = {
"accept": "application/json",
"content-type": "application/json"
}
response = requests.post(url, json=payload, headers=headers) # Send a POST request to the RMIS Delta API
data = response.json() # Parse JSON response
# Check if the call was successful
if (data.get("RMISDeltaAPI", {}).get('Header', {}).get('Result') == "ERROR"):
print(data.get("RMISDeltaAPI", {}).get('Header', {}).get('Errors')) # Print errors
return 0
else:
try:
# Extract the value of TotalInsdIDs from the response
total_insdids = data.get('RMISDeltaAPI', {}).get('SUMMARY', {}).get('TotalInsdIDs')
if (str(total_insdids).isdigit()): # Check if the total is a digit
print("\n\nNumber of deltas in queue (Summary): " + total_insdids)
return total_insdids # Return the total number of deltas
else:
print("\n\nNumber of deltas in queue (Summary): 0")
return 0 # Return 0 if not a digit
except json.JSONDecodeError:
print("Error decoding JSON response.") # Handle JSON decoding errors
return None
def DeltaFetch(rmisCreds:RMISCredentials):
url = baseURL + "/_c/std/api/DeltaAPI.aspx" # URL for Delta API
payload = {
"clientID": rmisCreds.clientID,
"clientPassword": rmisCreds.clientPassword,
"APIMode": "Fetch",
"MaxRecs": 50 # Number of records to fetch in each call
}
headers = {
"accept": "application/json",
"content-type": "application/json"
}
response = requests.post(url, json=payload, headers=headers) # Send a POST request to fetch deltas
data = response.json() # Parse JSON response
try:
# Extract the list of deltas (InsdID)
deltaList = data.get('RMISDeltaAPI', {}).get('FETCH', {}).get('InsdID')
return deltaList # Return the list of deltas
except:
print("No Deltas returned in fetch call") # Print message if no deltas
return [] # Return an empty list
def CallExpanded(rmisCreds:RMISCredentials, delta:Delta):
documentList = [] # Empty list to store documents
url = baseURL + "/_c/std/api/ExpandedCarrierAPI.aspx"
url_params = {
"clientID": rmisCreds.clientID,
"pwd": rmisCreds.clientPassword,
"querytype": "insdid",
"queryid": delta, # Carrier (Insured) ID to query
"version": 13, # API version
}
headers = {
"accept": "application/json",
"content-type": "application/json"
}
response = requests.get(url, params=url_params, headers=headers) # Send a GET request to fetch detailed carrier data
try:
# Expanded carrier API returns in XML, so process the data as such
root = ET.fromstring(response.text) # Parse XML response
timeStamp = root.find('.//TimeStamp').text # Find the timestamp from the response
# Check for errors in the response
try:
errors = root.find('.//Errors')
for error in errors:
if ("Carrier does not belong to this client" in error.text):
# This error indicates the carrier has been detached; remove from the queue
print(error.text)
return timeStamp # Return the timestamp
else:
pass # Placeholder for handling other errors
except:
pass # Placeholder for handling exceptions
coverages = root.find('.//Coverages') # Find list of coverages in the XML
if coverages is not None:
for coverage in coverages:
rmisImageId = coverage.find('.//RMISImageID').text # Extract RMISImageID
if (rmisImageId):
documentList.append(Document("CERTIFICATE", rmisImageId, delta)) # Append document to list (Type, image ID, carrier ID)
carrierNode = root.find('.//Carrier')
if carrierNode is not None:
mcNumber = carrierNode.find('.//MCNumber').text
dotNumber = carrierNode.find('.//DOTNumber').text
carrierName = carrierNode.find('.//CompanyName').text
carrier = Carrier()
carrier.carrierName = carrierName
carrier.carrierMC = mcNumber
carrier.carrierDOT = dotNumber
carrier.CarrierRMISID = delta
carrier.clientID = rmisCreds.clientID
carrier.lastUpdated = timeStamp
# Start the document API script in a separate thread
StartDocumentThread(rmisCreds, documentList)
return carrier # Return the timestamp
except json.JSONDecodeError:
print("Error decoding JSON response.") # Handle JSON decoding errors
return None
def StartDocumentThread(rmisCreds:RMISCredentials, documentList:list[Document]):
# Start a new thread for running the document API script
thread = threading.Thread(target=RunDocumentAPIScript, args=(rmisCreds, documentList))
thread.start()
def RunDocumentAPIScript(rmisCreds:RMISCredentials, documentList:list[Document]):
for document in documentList:
try:
# Run the Document API script as a subprocess
#subprocess.run(['python', 'DocumentAPI.py', str(rmisCreds.clientID), rmisCreds.clientPassword, document.documentType, document.documentID, document.carrierID])
GetDocument(rmisCreds=rmisCreds, document=document)
except subprocess.CalledProcessError as e:
print(f"Error getting document {document.documentID}: {e}") # Print error if subprocess fails
def DeltaClear(rmisCreds:RMISCredentials, deltaList:list[Delta]):
url = baseURL + "/_c/std/api/DeltaAPI.aspx" # URL for Delta API
clear_insureds = [
{
"insdID": int(delta.insdID), # Insured ID
"timeStamp": str(delta.timeStamp) # Timestamp
}
for delta in deltaList # Create a list of dictionaries for clearing
]
payload = {
"clientID": rmisCreds.clientID, # Client ID
"clientPassword": rmisCreds.clientPassword, # Client password
"APIMode": "Clear", # Request mode
"ClearInsureds": clear_insureds # List of insureds to clear
}
headers = {
"accept": "application/json", # Expected response format
"content-type": "application/json" # Content type of the request
}
# Send a POST request to clear processed deltas
response = requests.post(url, json=payload, headers=headers)
data = response.json() # Parse JSON response
print(data) # Print the response for verification
print("\n\n") # Print new lines for clarity
Start() # Start the process
All
Pages