You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
219 lines
15 KiB
219 lines
15 KiB
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 |