I did some refractoring while looking at your code. Changed some lines to show alternatives, but this is how I would set it up using a class.
import arcpy
import os
import threading
import time
import requests
import subprocess
from collections import defaultdict
from infi.systray import SysTrayIcon
from win11toast import toast
class ApplicationWatcher():
def __int__(self):
# set class properties
self.missing = []
self.search_results = defaultdict(dict)
def check_now(self, method):
"""
method to trigger respective data manipulations
"""
# clear the previous iteration values
self.missing.clear()
self.search_results.clear()
# get a token
cityworks_token = self.get_token(os.environ.get("CITYWORKS_UN"), os.environ.get("CITYWORKS_PW"))
# process the parcels depending on the value of method
if method == "update":
self.update_search(cityworks_token)
self.search_results = self.get_issued_permits(cityworks_token)
# Get list of case numbers from the result
issued_permits = [result['CASE_NUMBER'] for result in self.search_results]
# get list of mapped permits
mapped_permits = self.get_mapped_permits()
# Debugging
# print("Issued:")
# for issued in issued_permits:
# print(issued)
# print("Mapped:")
# for mapped in mapped_permits:
# print(mapped)
# Compare the issued permits and mapped permits
if method == "clicked" or method == "":
# compare the issued permits to the mapped permits
self.missing = self.compare(issued_permits, mapped_permits)
elif method == "crosscheck":
# compare the mapped permits to the issued permits
self.missing = self.compare(mapped_permits, issued_permits)
elif method == "duplicates":
# find duplicates in the mapped permits
self.missing = self.duplicates(mapped_permits)
# Get permits needing to be mapped
if method == "clicked" and self.missing or method == "" and self.missing:
if len(self.missing) == 1:
message = "The following permit needs to be mapped:"
else:
message = f"The following {len(self.missing)} permits need to be mapped:"
for permit in self.missing:
message += f"\n{permit}"
self.yeah_toast(message)
elif method == "crosscheck" and self.missing:
if len(self.missing) == 1:
message = "The following permit was found on the map but not in Cityworks:"
else:
message = f"The following {len(self.missing)} permits were found on the map but not in Cityworks:"
for permit in self.missing:
message += f"\n{permit}"
self.yeah_toast(message)
elif method == "duplicates" and self.missing:
if len(self.missing) == 1:
message = "The following permit was found as a duplicate:"
else:
message = f"The following {len(self.missing)} permits were found as duplicates on the map:"
for permit in self.missing:
message += f"\n{permit}"
self.yeah_toast(message)
# Display
if method != "" and not self.missing:
self.no_toast()
def compare(self, base, tester):
return [permit for permit in base if permit not in tester]
# def compare(self, issued, mapped):
# permits = [permit for permit in issued if permit not in mapped]
# return permits
#
# def crosscheck(self, issued, mapped):
# permits = [permit for permit in mapped if permit not in issued]
# return permits
def duplicates(self, mapped):
# permits = []
# comparitor = []
#
# for permit in mapped:
# if permit not in comparitor:
# comparitor.append(permit)
# else:
# permits.append(permit)
# shorten this to a list comprehension by counting the values in the list and returning
# those that are equal or grater than 2
return [x for x in mapped if mapped.count(x) >= 2]
def get_token(self, username, password):
arguments = 'data={"LoginName":"' + username + '","Password":"' + password + '"}'
response = requests.get("https://cityworks.ci.kalispell.mt.us/Cityworks/Services/General/"
"Authentication/Authenticate", params=arguments, timeout=5)
cityworks_token = (response.json()["Value"]['Token'])
return cityworks_token
def get_issued_permits(self, token):
# Search ID for BPISSUED is 994
bp_url = "https://cityworks.ci.kalispell.mt.us/Cityworks/services/Ams/Search/Execute?" \
"data={%22SearchId%22:994}&token=" + token
results = requests.get(bp_url, timeout=10)
return results.json()["Value"]
def get_mapped_permits(self):
building_permits_2022 = r"J:\Austin\Projects\SDE Maintenance\Building.sde\Kalispell.BUILDING.BuildingPermits\Kalispell.BUILDING.BuildingPermits2022"
# the 'for each in permi' conditional is redundant and you can just use the index.
# mapped = [each for permit in arcpy.da.SearchCursor(building_permits_2022, "PERMIT_NUM") for each in permit]
return [permit[0] for permit in arcpy.da.SearchCursor(building_permits_2022, "PERMIT_NUM")]
def no_toast(self):
toast("Nothing to do...")
def open_web_pages(self):
if self.missing:
for case_number in self.missing:
for result in self.search_results:
if result['CASE_NUMBER'] == case_number:
ca_obj_id = int(result['CA_OBJECT_ID'])
ca_case_type_id = int(result['CASE_TYPE_ID'])
subprocess.Popen([r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe",
f"https://cityworks.ci.kalispell.mt.us/Cityworks/CwPermit/UF/Case/Page/CUF.aspx?"
f"CaObjId={ca_obj_id}&CaCaseTypeId={ca_case_type_id}&CaSubTypeId="])
else:
toast("Nothing to show...")
# def parse_results(self, results):
# permits = [result['CASE_NUMBER'] for result in results]
# return permits
def update_search(self, token):
url = f"https://cityworks.ci.kalispell.mt.us/Cityworks/Services/Ams/Search/PllSaved?token={token}"
results = requests.get(url)
result_values = results.json()["Value"]
filtered_results = [result for result in result_values if "BACHURSKI" in result["EmployeeName"]]
display = ""
for each in filtered_results:
display += str(f"SearchId: {each['SearchId']}\nSearchName: {each['SearchName']}\n\n")
toast("New Building Permit", display)
def yeah_toast(self, info):
toast("New Building Permit", info, on_click=lambda args: self.open_web_pages())
if __name__ == "__main__":
# Create instance
app = ApplicationWatcher()
menu_options = (("Check Permits Now", None, lambda args: app.check_now("clicked")),
("Open Permit Web Page(s)", None, app.open_web_pages),
("Crosscheck Permits", None, lambda args: app.check_now("crosscheck")),
("Check for Duplicates", None, lambda args: app.check_now("duplicates")),
("Update Search ID", None, lambda args: app.check_now("update")))
systray = SysTrayIcon("TrayIcon.ico", "Building Permit Monitor", menu_options)
systray.start()
def monitoring():
while True:
app.check_now("")
# Check every 30 minutes.
time.sleep(1800)
threading.Thread(target=monitoring, daemon=True).start()
