OPEN_RPA_PUBLIC
/
ORPA-pyOpenRPA
2
0
Fork 1
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

#Orchestrator New URLList engine - need testing!

Browse Source
dev-linux
Ivan Maslov 5 years ago
parent 0b551d152a
commit ff5c381a2a
2 changed files with 102 additions and 8 deletions
Show Stats Download Patch File Download Diff File

13
Orchestrator/Settings/Settings.py
Unescape View File

@ -103,7 +103,18 @@ def Settings():
"AuthTokensDict": {
#"<AuthToken>":{"User":"", "Domain":"", "TokenDatetime":<Datetime>}
}
}
},
"URLList":[ #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
# "ResponseDefRequestGlobalResponse": None #Function with str result
#}
]
},
"Scheduler": {
"ActivityTimeCheckLoopSeconds":5, #Количество секунд, между циклами проверки действий

97
Sources/pyOpenRPA/Orchestrator/Server.py
Unescape View File

@ -146,11 +146,15 @@ def UserAccessCheckBefore(inMethod, inRequest):
if lAccessRuleItem["Method"].upper() == inMethod:
#check Match type variant: BeginWith
if lAccessRuleItem["MatchType"].upper() == "BEGINWITH":
if inRequest.path.startswith(lAccessRuleItem["URL"]):
lURLPath = inRequest.path
lURLPath = lURLPath.upper()
if lURLPath.startswith(lAccessRuleItem["URL"].upper()):
lResult = HelpGetFlag(lAccessRuleItem, inRequest, mGlobalDict, lUserDict)
#check Match type variant: Contains
# check Match type variant: Contains
elif lAccessRuleItem["MatchType"].upper() == "CONTAINS":
if inRequest.path.contains(lAccessRuleItem["URL"]):
lURLPath = inRequest.path
lURLPath = lURLPath.upper()
if lURLPath.contains(lAccessRuleItem["URL"].upper()):
lResult = HelpGetFlag(lAccessRuleItem, inRequest, mGlobalDict, lUserDict)
# check Match type variant: Equal
elif lAccessRuleItem["MatchType"].upper() == "EQUAL":
@ -174,11 +178,15 @@ def UserAccessCheckBefore(inMethod, inRequest):
if lAccessRuleItem["Method"].upper() == inMethod:
#check Match type variant: BeginWith
if lAccessRuleItem["MatchType"].upper() == "BEGINWITH":
if inRequest.path.startswith(lAccessRuleItem["URL"]):
lURLPath = inRequest.path
lURLPath = lURLPath.upper()
if lURLPath.startswith(lAccessRuleItem["URL"].upper()):
lResult = HelpGetFlag(lAccessRuleItem, inRequest, mGlobalDict, lUserDict)
#check Match type variant: Contains
elif lAccessRuleItem["MatchType"].upper() == "CONTAINS":
if inRequest.path.contains(lAccessRuleItem["URL"]):
lURLPath = inRequest.path
lURLPath = lURLPath.upper()
if lURLPath.contains(lAccessRuleItem["URL"].upper()):
lResult = HelpGetFlag(lAccessRuleItem, inRequest, mGlobalDict, lUserDict)
# check Match type variant: Equal
elif lAccessRuleItem["MatchType"].upper() == "EQUAL":
@ -194,6 +202,57 @@ def UserAccessCheckBefore(inMethod, inRequest):
return lResult
# HTTPRequestHandler class
class testHTTPServer_RequestHandler(BaseHTTPRequestHandler):
#Tech def
#return {"headers":[],"body":"","statuscode":111}
def URLItemCheckDo(self, inURLItem, inMethod, inResponseDict):
###############################
#Tech sub def - do item
################################
def URLItemDo(inURLItem,inRequest,inGlobalDict,inResponseDict):
#Content-type
if "ResponseContentType" in inURLItem:
inResponseDict["Headers"]["Content-type"] = inURLItem["ResponseContentType"]
#If file path is set
if "ResponseFilePath" in inURLItem:
lFileObject = open(inURLItem["ResponseFilePath"], "rb")
# Write content as utf-8 data
inResponseDict["Body"] = lFileObject.read()
# Закрыть файловый объект
lFileObject.close()
#If function is set
if "ResponseDefRequestGlobalResponse" in inURLItem:
inURLItem["ResponseDefRequestGlobalResponse"](inRequest,inGlobalDict,inResponseDict)
# TODO If folder path is set
if "ResponseFolderPath" in inURLItem:
#
pass
##############################################
if inURLItem["Method"].upper() == inMethod.upper():
# check Match type variant: BeginWith
if inURLItem["MatchType"].upper() == "BEGINWITH":
lURLPath = self.path
lURLPath = lURLPath.upper()
if lURLPath.startswith(inURLItem["URL"].upper()):
URLItemDo(inURLItem, self, mGlobalDict,inResponseDict)
return True
# check Match type variant: Contains
elif inURLItem["MatchType"].upper() == "CONTAINS":
lURLPath = self.path
lURLPath = lURLPath.upper()
if lURLPath.contains(inURLItem["URL"].upper()):
URLItemDo(inURLItem, self, mGlobalDict,inResponseDict)
return True
# check Match type variant: Equal
elif inURLItem["MatchType"].upper() == "EQUAL":
if inURLItem["URL"].upper() == self.path.upper():
URLItemDo(inURLItem, self, mGlobalDict,inResponseDict)
return True
# check Match type variant: EqualCase
elif inURLItem["MatchType"].upper() == "EQUALCASE":
if inURLItem["URL"] == self.path:
URLItemDo(inURLItem, self, mGlobalDict,inResponseDict)
return True
return False
#ResponseContentTypeFile
def SendResponseContentTypeFile(self, inContentType, inFilePath):
# Send response status code
@ -208,9 +267,22 @@ class testHTTPServer_RequestHandler(BaseHTTPRequestHandler):
# Write content as utf-8 data
self.wfile.write(lFileObject.read())
#Закрыть файловый объект
lFileObject.close()
# GET
lFileObject.close()
# ResponseContentTypeFile
def ResponseDictSend(self, inResponseDict):
# Send response status code
self.send_response(inResponseDict["StatusCode"])
# Send headers
for lItemKey, lItemValue in inResponseDict["Headers"].items():
self.send_header(lItemKey, lItemValue)
#Close headers section in response
self.end_headers()
# Write content as utf-8 data
self.wfile.write(inResponseDict["Body"])
# GET
def do_GET(self):
# Prepare result dict
lResponseDict = {"Headers": {}, "Set-cookies":{}, "Body": "", "StatusCode": None}
#####################################
#Do authentication
#Check if authentication is turned on
@ -234,6 +306,17 @@ class testHTTPServer_RequestHandler(BaseHTTPRequestHandler):
######################################
if lFlagUserAccess:
lOrchestratorFolder = "\\".join(__file__.split("\\")[:-1])
############################
#New server engine (url from global dict (URLList))
############################
for lURLItem in mGlobalDict["URLList"]:
#Check if all condition are applied
lFlagURLIsApplied=False
lFlagURLIsApplied=self.URLItemCheckDo(lURLItem, "GET", lResponseDict)
if lFlagURLIsApplied:
self.ResponseDictSend(lResponseDict)
return
###################################################
#Мост между файлом и http запросом (новый формат)
if self.path == "/":
self.SendResponseContentTypeFile('text/html', os.path.join(lOrchestratorFolder, "Web\\Index.xhtml"))

Write Preview
Loading…
Cancel
Save