import json from inspect import signature # For detect count of def args #ControlPanelDict from desktopmagic.screengrab_win32 import ( getDisplayRects, saveScreenToBmp, saveRectToBmp, getScreenAsImage, getRectAsImage, getDisplaysAsImages) from http import cookies import uuid # generate UUID4 import time # sleep functions import datetime # datetime functions import threading # Multi-threading # /Orchestrator/RobotRDPActive/ControlPanelDictGet def RobotRDPActive_ControlPanelDictGet(inRequest,inGlobalDict): inResponseDict = inRequest.OpenRPAResponseDict lResultDict = { "DataList":[ # {"SessionKeyStr":"", "SessionHexStr": "", "IsFullScreenBool": False, "IsIgnoredBool": False} ] } # Iterate throught the RDP List for lRDPSessionKeyStrItem in inGlobalDict["RobotRDPActive"]["RDPList"]: lRDPConfiguration = inGlobalDict["RobotRDPActive"]["RDPList"][lRDPSessionKeyStrItem] # Get the configuration dict lDataItemDict = {"SessionKeyStr":"", "SessionHexStr": "", "IsFullScreenBool": False, "IsIgnoredBool": False} # Template lDataItemDict["SessionKeyStr"] = lRDPSessionKeyStrItem # Session key str lDataItemDict["SessionHexStr"] = lRDPConfiguration["SessionHex"] # Session Hex lDataItemDict["IsFullScreenBool"] = True if lRDPSessionKeyStrItem == inGlobalDict["RobotRDPActive"]["FullScreenRDPSessionKeyStr"] else False # Check the full screen for rdp window lDataItemDict["IsIgnoredBool"] = lRDPConfiguration["SessionIsIgnoredBool"] # Is ignored lResultDict["DataList"].append(lDataItemDict) # Send message back to client message = json.dumps(lResultDict) # Write content as utf-8 data inResponseDict["Body"] = bytes(message, "utf8") # def to check control panels for selected session def Monitor_ControlPanelDictGet_SessionCheckInit(inRequest,inGlobalDict): lL = inGlobalDict["Logger"] # Alias for logger lLifetimeSecFloat = inGlobalDict["Client"]["Session"]["LifetimeSecFloat"] lLifetimeRequestSecFloat = inGlobalDict["Client"]["Session"]["LifetimeRequestSecFloat"] lControlPanelRefreshIntervalSecFloat = inGlobalDict["Client"]["Session"]["ControlPanelRefreshIntervalSecFloat"] lCookieSessionGUIDStr = None # generate the new GUID # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # Technicaldef - interval check control panels + check actuality of the session by the datetime def TechnicalCheck(): lItemValue = inGlobalDict["Client"]["Session"]["TechnicalSessionGUIDCache"][lCookieSessionGUIDStr] # Lifetime is ok - check control panel lDatasetCurrentBytes = Monitor_ControlPanelDictGet(inRequest,inGlobalDict) # Call the control panel if lDatasetCurrentBytes != lItemValue["DatasetLast"]["ControlPanel"]["Data"]: # Check if dataset is changed lItemValue["DatasetLast"]["ControlPanel"]["Data"] = lDatasetCurrentBytes # Set new datset lItemValue["DatasetLast"]["ControlPanel"]["ReturnBool"] = True # Set flag to return the data # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # Technicaldef - Create new session struct def TechnicalSessionNew(inSessionGUIDStr): lCookieSessionGUIDStr = inSessionGUIDStr # Generate the new GUID lSessionNew = { # Session with some GUID str. On client session guid stored in cookie "SessionGUIDStr" "InitDatetime": datetime.datetime.now(), # Datetime when session GUID was created "DatasetLast": { "ControlPanel": { "Data": None, # Struct to check with new iterations. None if starts "ReturnBool": False # flag to return, close request and return data as json } }, "ClientRequestHandler": inRequest, # Last client request handler "UserADStr": inRequest.OpenRPA["User"], # User, who connect. None if user is not exists "DomainADStr": inRequest.OpenRPA["Domain"], # Domain of the user who connect. None if user is not exists } inGlobalDict["Client"]["Session"]["TechnicalSessionGUIDCache"][lCookieSessionGUIDStr] = lSessionNew # Set new session in dict inRequest.OpenRPAResponseDict["SetCookies"]["SessionGUIDStr"] = lCookieSessionGUIDStr # Set SessionGUIDStr in cookies if lL: lL.info(f"New session GUID is created. GUID {lCookieSessionGUIDStr}") return lCookieSessionGUIDStr # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # lCreateNewSessionBool = False # Flag to create new session structure # step 1 - get cookie SessionGUIDStr lSessionGUIDStr = inRequest.headers.get("SessionGUIDStr", None) if lSessionGUIDStr is not None: # Check if GUID session is ok #inRequest.OpenRPAResponseDict["StatusCode"] = 301 #inRequest.OpenRPAResponseDict["Headers"]["Location"] = "/" #if lL: lL.info(f"GUID is detected - send HTTP 301 to refresh page") lCookieSessionGUIDStr = lSessionGUIDStr # Get the existing GUID if lSessionGUIDStr not in inGlobalDict["Client"]["Session"]["TechnicalSessionGUIDCache"]: lCookieSessionGUIDStr= TechnicalSessionNew(inSessionGUIDStr = lSessionGUIDStr) # Create new session else: # Update the datetime of the request session inGlobalDict["Client"]["Session"]["TechnicalSessionGUIDCache"][lCookieSessionGUIDStr]["InitDatetime"]=datetime.datetime.now() else: lCookieSessionGUIDStr = TechnicalSessionNew(inSessionGUIDStr = lSessionGUIDStr) # Create new session # Init the RobotRDPActive in another thread #lThreadCheckCPInterval = threading.Thread(target=TechnicalIntervalCheck) #lThreadCheckCPInterval.daemon = True # Run the thread in daemon mode. #lThreadCheckCPInterval.start() # Start the thread execution. # Step 2 - interval check if data is exist lTimeStartSecFloat = time.time() lDoWhileBool = True # Flag to iterate throught the lifetime of the request while lDoWhileBool: #print(lTechnicalSessionGUIDCache) #print(lCookieSessionGUIDStr) if lCookieSessionGUIDStr in inGlobalDict["Client"]["Session"]["TechnicalSessionGUIDCache"]: lItemValue = inGlobalDict["Client"]["Session"]["TechnicalSessionGUIDCache"][lCookieSessionGUIDStr] if (time.time() - lTimeStartSecFloat) >= lLifetimeRequestSecFloat: # Check if lifetime client request is over or has no key if lL: lL.debug(f"Client request lifetime is over") lDoWhileBool = False # Stop the iterations if lDoWhileBool: TechnicalCheck() # Calculate the CP if lItemValue["DatasetLast"]["ControlPanel"]["ReturnBool"] == True: # Return data if data flag it True lDatasetCurrentBytes = lItemValue["DatasetLast"]["ControlPanel"]["Data"] # Set new dataset inResponseDict = inRequest.OpenRPAResponseDict inResponseDict["Body"] = lDatasetCurrentBytes lItemValue["DatasetLast"]["ControlPanel"]["ReturnBool"] = False # Set flag that data was returned lDoWhileBool = False # Stop the iterations else: lCookieSessionGUIDStr = TechnicalSessionNew(inSessionGUIDStr = lCookieSessionGUIDStr) # Create new session if lDoWhileBool: # Sleep if we wait hte next iteration time.sleep(lControlPanelRefreshIntervalSecFloat) # Sleep to the next iteration def Monitor_ControlPanelDictGet(inRequest,inGlobalDict): inResponseDict = inRequest.OpenRPAResponseDict lL = inGlobalDict["Logger"] # Alias for logger # Create result JSON lResultJSON = {"RenderRobotList": [], "RenderRDPList": []} lRenderFunctionsRobotList = inGlobalDict["ControlPanelDict"]["RobotList"] for lItem in lRenderFunctionsRobotList: lUACBool = True # Check if render function is applicable User Access Rights (UAC) if inGlobalDict["Server"]["AccessUsers"]["FlagCredentialsAsk"] is True: lUserRights = inGlobalDict["Server"]["AccessUsers"]["RuleDomainUserDict"][(inRequest.OpenRPA["Domain"].upper(),inRequest.OpenRPA["User"].upper())] if len(lUserRights["ControlPanelKeyAllowedList"]) > 0 and lItem["KeyStr"] not in lUserRights["ControlPanelKeyAllowedList"]: lUACBool = False # UAC Check is not passed - False for user if lUACBool: # Run function if UAC is TRUE # Выполнить вызов и записать результат # Call def (inRequest, inGSettings) or def (inGSettings) lItemResultDict = None lDEFSignature = signature(lItem["RenderFunction"]) # Get signature of the def lDEFARGLen = len(lDEFSignature.parameters.keys()) # get count of the def args try: if lDEFARGLen == 1: # def (inGSettings) lItemResultDict = lItem["RenderFunction"](inGlobalDict) elif lDEFARGLen == 2: # def (inRequest, inGSettings) lItemResultDict = lItem["RenderFunction"](inRequest, inGlobalDict) elif lDEFARGLen == 0: # def () lItemResultDict = lItem["RenderFunction"]() # RunFunction lResultJSON["RenderRobotList"].append(lItemResultDict) except Exception as e: if lL: lL.exception(f"Error in control panel. CP item {lItem}. Exception is below") # Iterate throught the RDP list for lRDPSessionKeyStrItem in inGlobalDict["RobotRDPActive"]["RDPList"]: lRDPConfiguration = inGlobalDict["RobotRDPActive"]["RDPList"][ lRDPSessionKeyStrItem] # Get the configuration dict lDataItemDict = {"SessionKeyStr": "", "SessionHexStr": "", "IsFullScreenBool": False, "IsIgnoredBool": False} # Template lDataItemDict["SessionKeyStr"] = lRDPSessionKeyStrItem # Session key str lDataItemDict["SessionHexStr"] = lRDPConfiguration["SessionHex"] # Session Hex lDataItemDict["IsFullScreenBool"] = True if lRDPSessionKeyStrItem == inGlobalDict["RobotRDPActive"][ "FullScreenRDPSessionKeyStr"] else False # Check the full screen for rdp window lDataItemDict["IsIgnoredBool"] = lRDPConfiguration["SessionIsIgnoredBool"] # Is ignored lResultJSON["RenderRDPList"].append(lDataItemDict) # Send message back to client message = json.dumps(lResultJSON) # Write content as utf-8 data #inResponseDict["Body"] = bytes(message, "utf8") return bytes(message, "utf8") # UserAccess get rights hierarchy dict in json def UserRoleHierarchyGet(inRequest,inGlobalDict): inResponseDict = inRequest.OpenRPAResponseDict # Create result JSON lResultDict = inRequest.OpenRPA["DefUserRoleHierarchyGet"]() # Get the User Role Hierarchy list # Send message back to client message = json.dumps(lResultDict) # Write content as utf-8 data inResponseDict["Body"] = bytes(message, "utf8") def GetScreenshot(inRequest,inGlobalDict): # Get Screenshot def SaveScreenshot(inFilePath): # grab fullscreen # Save the entire virtual screen as a PNG lScreenshot = getScreenAsImage() lScreenshot.save('screenshot.png', format='png') # lScreenshot = ScreenshotSecondScreen.grab_screen() # save image file # lScreenshot.save('screenshot.png') # Сохранить файл на диск SaveScreenshot("Screenshot.png") lFileObject = open("Screenshot.png", "rb") # Write content as utf-8 data inRequest.OpenRPAResponseDict["Body"] = lFileObject.read() # Закрыть файловый объект lFileObject.close() def SettingsUpdate(inGlobalConfiguration): import os import pyOpenRPA.Orchestrator lOrchestratorFolder = "\\".join(pyOpenRPA.Orchestrator.__file__.split("\\")[:-1]) lURLList = \ [ #List of available URLs with the orchestrator server #{ # "Method":"GET|POST", # "URL": "/index", #URL of the request # "MatchType": "", #"BeginWith|Contains|Equal|EqualCase", # "ResponseFilePath": "", #Absolute or relative path # "ResponseFolderPath": "", #Absolute or relative path # "ResponseContentType": "", #HTTP Content-type # "ResponseDefRequestGlobal": None #Function with str result #} #Orchestrator basic dependencies {"Method":"GET", "URL": "/", "MatchType": "EqualCase", "ResponseFilePath": os.path.join(lOrchestratorFolder, "Web\\Index.xhtml"), "ResponseContentType": "text/html"}, {"Method":"GET", "URL": "/Index.js", "MatchType": "EqualCase", "ResponseFilePath": os.path.join(lOrchestratorFolder, "Web\\Index.js"), "ResponseContentType": "text/javascript"}, {"Method":"GET", "URL": "/3rdParty/Semantic-UI-CSS-master/semantic.min.css", "MatchType": "EqualCase", "ResponseFilePath": os.path.join(lOrchestratorFolder, "..\\Resources\\Web\\Semantic-UI-CSS-master\\semantic.min.css"), "ResponseContentType": "text/css"}, {"Method":"GET", "URL": "/3rdParty/Semantic-UI-CSS-master/semantic.min.js", "MatchType": "EqualCase", "ResponseFilePath": os.path.join(lOrchestratorFolder, "..\\Resources\\Web\\Semantic-UI-CSS-master\\semantic.min.js"), "ResponseContentType": "application/javascript"}, {"Method":"GET", "URL": "/3rdParty/jQuery/jquery-3.1.1.min.js", "MatchType": "EqualCase", "ResponseFilePath": os.path.join(lOrchestratorFolder, "..\\Resources\\Web\\jQuery\\jquery-3.1.1.min.js"), "ResponseContentType": "application/javascript"}, {"Method":"GET", "URL": "/3rdParty/Google/LatoItalic.css", "MatchType": "EqualCase", "ResponseFilePath": os.path.join(lOrchestratorFolder, "..\\Resources\\Web\\Google\\LatoItalic.css"), "ResponseContentType": "font/css"}, {"Method":"GET", "URL": "/3rdParty/Semantic-UI-CSS-master/themes/default/assets/fonts/icons.woff2", "MatchType": "EqualCase", "ResponseFilePath": os.path.join(lOrchestratorFolder, "..\\Resources\\Web\\Semantic-UI-CSS-master\\themes\\default\\assets\\fonts\\icons.woff2"), "ResponseContentType": "font/woff2"}, {"Method":"GET", "URL": "/favicon.ico", "MatchType": "EqualCase", "ResponseFilePath": os.path.join(lOrchestratorFolder, "Web\\favicon.ico"), "ResponseContentType": "image/x-icon"}, {"Method":"GET", "URL": "/3rdParty/Handlebars/handlebars-v4.1.2.js", "MatchType": "EqualCase", "ResponseFilePath": os.path.join(lOrchestratorFolder, "..\\Resources\\Web\\Handlebars\\handlebars-v4.1.2.js"), "ResponseContentType": "application/javascript"}, {"Method": "GET", "URL": "/Monitor/ControlPanelDictGet", "MatchType": "Equal", "ResponseDefRequestGlobal": Monitor_ControlPanelDictGet_SessionCheckInit, "ResponseContentType": "application/json"}, {"Method": "GET", "URL": "/GetScreenshot", "MatchType": "BeginWith", "ResponseDefRequestGlobal": GetScreenshot, "ResponseContentType": "image/png"}, {"Method": "GET", "URL": "/pyOpenRPA_logo.png", "MatchType": "Equal", "ResponseFilePath": os.path.join(lOrchestratorFolder, "..\\Resources\\Web\\pyOpenRPA_logo.png"), "ResponseContentType": "image/png"}, {"Method": "POST", "URL": "/Orchestrator/UserRoleHierarchyGet", "MatchType": "Equal","ResponseDefRequestGlobal": UserRoleHierarchyGet, "ResponseContentType": "application/json"} ] inGlobalConfiguration["Server"]["URLList"]=inGlobalConfiguration["Server"]["URLList"]+lURLList return inGlobalConfiguration