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

#Settings New Architecture + Server Auth

Browse Source
dev-linux
Ivan Maslov 5 years ago
parent e438b6e0c6
commit 35f761736f
4 changed files with 311 additions and 196 deletions
Show Stats Download Patch File Download Diff File

6
Orchestrator/Orchestrator.py
Unescape View File

@ -21,7 +21,7 @@ logging.basicConfig(filename="Reports\ReportRun_"+datetime.datetime.now().strfti
#Единый глобальный словарь (За основу взять из Settings.py) #Единый глобальный словарь (За основу взять из Settings.py)
global mGlobalDict global mGlobalDict
mGlobalDict = Settings.mDict mGlobalDict = Settings.Settings()
Processor.mGlobalDict = mGlobalDict Processor.mGlobalDict = mGlobalDict
Timer.mGlobalDict = mGlobalDict Timer.mGlobalDict = mGlobalDict
Timer.Processor.mGlobalDict = mGlobalDict Timer.Processor.mGlobalDict = mGlobalDict
@ -38,12 +38,12 @@ lThreadServer = Server.RobotDaemonServer("ServerThread", mGlobalDict)
lThreadServer.start() lThreadServer.start()
#Вечный цикл #Вечный цикл
while True: while True:
lCurrentDateTime=datetime.datetime.now() lCurrentDateTime = datetime.datetime.now()
#Циклический обход правил #Циклический обход правил
lFlagSearchActivityType=True lFlagSearchActivityType=True
for lIndex, lItem in enumerate(mGlobalDict["Scheduler"]["ActivityTimeList"]): for lIndex, lItem in enumerate(mGlobalDict["Scheduler"]["ActivityTimeList"]):
#Проверка дней недели, в рамках которых можно запускать активность #Проверка дней недели, в рамках которых можно запускать активность
lItemWeekdayList=lItem.get("WeekdayList",[0,1,2,3,4,5,6]) lItemWeekdayList=lItem.get("WeekdayList", [0, 1, 2, 3, 4, 5, 6])
if lCurrentDateTime.weekday() in lItemWeekdayList: if lCurrentDateTime.weekday() in lItemWeekdayList:
if lFlagSearchActivityType: if lFlagSearchActivityType:
#Лог #Лог

325
Orchestrator/Server.py
Unescape View File

@ -4,11 +4,16 @@ from threading import Thread
import Processor import Processor
import importlib import importlib
import pdb import pdb
import base64
import uuid
import datetime
from http import cookies
from desktopmagic.screengrab_win32 import ( from desktopmagic.screengrab_win32 import (
getDisplayRects, saveScreenToBmp, saveRectToBmp, getScreenAsImage, getDisplayRects, saveScreenToBmp, saveRectToBmp, getScreenAsImage,
getRectAsImage, getDisplaysAsImages) getRectAsImage, getDisplaysAsImages)
global mGlobalDict global mGlobalDict
def SaveScreenshot(inFilePath): def SaveScreenshot(inFilePath):
# grab fullscreen # grab fullscreen
# Save the entire virtual screen as a PNG # Save the entire virtual screen as a PNG
@ -35,15 +40,87 @@ class RobotDaemonServer(Thread):
print('running server...') print('running server...')
httpd.serve_forever() httpd.serve_forever()
#Authenticate function ()
# return dict
# {
# "Domain": "", #Empty if Auth is not success
# "User": "" #Empty if Auth is not success
# }
def AuthenticateVerify(inRequest):
lResult={"Domain": "", "User": ""}
######################################
#Way 1 - try to find AuthToken
lCookies = cookies.SimpleCookie(inRequest.headers.get("Cookie", ""))
#pdb.set_trace()
if "AuthToken" in lCookies:
lCookieAuthToken = lCookies.get("AuthToken", "").value
if lCookieAuthToken:
#Find AuthToken in GlobalDict
if lCookieAuthToken in mGlobalDict.get("Server", {}).get("AccessUsers", {}).get("AuthTokensDict", {}):
#Auth Token Has Been Founded
lResult["Domain"] = mGlobalDict["Server"]["AccessUsers"]["AuthTokensDict"][lCookieAuthToken]["Domain"]
lResult["User"] = mGlobalDict["Server"]["AccessUsers"]["AuthTokensDict"][lCookieAuthToken]["User"]
#Exit earlier
return lResult
######################################
#Way 2 - try to logon
lHeaderAuthorization = inRequest.headers.get("Authorization", "").split(" ")
if len(lHeaderAuthorization) == 2:
llHeaderAuthorizationDecodedUserPasswordList = base64.b64decode(lHeaderAuthorization[1]).decode("utf-8").split(
":")
lUser = llHeaderAuthorizationDecodedUserPasswordList[0]
lPassword = llHeaderAuthorizationDecodedUserPasswordList[1]
lDomain = None
if "\\" in lUser:
lDomain = lUser.split("\\")[0]
lUser = lUser.split("\\")[1]
#Try to logon - use processor
lLogonResult = Processor.Activity(
{
"Type": "WindowsLogon",
"Domain": lDomain,
"User": lUser,
"Password": lPassword
}
)
#Check result
if lLogonResult["Result"]:
lResult["Domain"] = lLogonResult["Domain"]
lResult["User"] = lLogonResult["User"]
#Create token
lAuthToken=str(uuid.uuid1())
mGlobalDict["Server"]["AccessUsers"]["AuthTokensDict"][lAuthToken] = {}
mGlobalDict["Server"]["AccessUsers"]["AuthTokensDict"][lAuthToken]["Domain"] = lResult["Domain"]
mGlobalDict["Server"]["AccessUsers"]["AuthTokensDict"][lAuthToken]["User"] = lResult["User"]
mGlobalDict["Server"]["AccessUsers"]["AuthTokensDict"][lAuthToken]["TokenDatetime"] = datetime.datetime.now()
#Set-cookie
inRequest.OpenRPA={}
inRequest.OpenRPA["AuthToken"] = lAuthToken
#inRequest.OpenRPAResponse["Set-Cookie"]=[]lResult["Set-Cookie"] = lAuthToken
#pdb.set_trace()
#inRequest.send_header("Set-Cookie:", f"AuthToken={lAuthToken}")
######################################
return lResult
def AuthenticateBlock(inRequest):
# Send response status code
inRequest.send_response(401)
# Send headers
inRequest.send_header('Content-type', 'text/html')
inRequest.send_header('WWW-Authenticate', 'Basic') # Always ask login pass
inRequest.end_headers()
# Write content as utf-8 data
inRequest.wfile.write(bytes("", "utf8"))
# HTTPRequestHandler class # HTTPRequestHandler class
class testHTTPServer_RequestHandler(BaseHTTPRequestHandler): class testHTTPServer_RequestHandler(BaseHTTPRequestHandler):
#ResponseContentTypeFile #ResponseContentTypeFile
def SendResponseContentTypeFile(self,inContentType,inFilePath): def SendResponseContentTypeFile(self, inContentType, inFilePath):
# Send response status code # Send response status code
self.send_response(200) self.send_response(200)
# Send headers # Send headers
self.send_header('Content-type',inContentType) self.send_header('Content-type', inContentType)
#Check if var exist
if hasattr(self, "OpenRPA"):
self.send_header("Set-Cookie", f"AuthToken={self.OpenRPA['AuthToken']}")
self.end_headers() self.end_headers()
lFileObject = open(inFilePath, "rb") lFileObject = open(inFilePath, "rb")
# Write content as utf-8 data # Write content as utf-8 data
@ -52,125 +129,127 @@ class testHTTPServer_RequestHandler(BaseHTTPRequestHandler):
lFileObject.close() lFileObject.close()
# GET # GET
def do_GET(self): def do_GET(self):
#Мост между файлом и http запросом (новый формат) #####################################
if self.path == "/": #Do authentication
self.SendResponseContentTypeFile('text/html',"Web\\Index.xhtml") #Check if authentication is turned on
#Мост между файлом и http запросом (новый формат) #####################################
if self.path == '/3rdParty/Semantic-UI-CSS-master/semantic.min.css': lFlagAccessUserBlock=False
self.SendResponseContentTypeFile('text/css',"..\\Resources\\Web\\Semantic-UI-CSS-master\\semantic.min.css") lAuthenticateDict = {"Domain": "", "User": ""}
#Мост между файлом и http запросом (новый формат) if mGlobalDict.get("Server", {}).get("AccessUsers", {}).get("FlagCredentialsAsk", False):
if self.path == '/3rdParty/Semantic-UI-CSS-master/semantic.min.js': lAuthenticateDict = AuthenticateVerify(self)
self.SendResponseContentTypeFile('application/javascript',"..\\Resources\\Web\\Semantic-UI-CSS-master\\semantic.min.js") if not lAuthenticateDict["User"]:
#Мост между файлом и http запросом (новый формат) lFlagAccessUserBlock=True
if self.path == '/3rdParty/jQuery/jquery-3.1.1.min.js': if lFlagAccessUserBlock:
self.SendResponseContentTypeFile('application/javascript',"..\\Resources\\Web\\jQuery\\jquery-3.1.1.min.js") AuthenticateBlock(self)
#Мост между файлом и http запросом (новый формат) #####################################
if self.path == '/3rdParty/Google/LatoItalic.css': else:
self.SendResponseContentTypeFile('font/css',"..\\Resources\\Web\\Google\\LatoItalic.css") #Мост между файлом и http запросом (новый формат)
#Мост между файлом и http запросом (новый формат) if self.path == "/":
if self.path == '/3rdParty/Semantic-UI-CSS-master/themes/default/assets/fonts/icons.woff2': self.SendResponseContentTypeFile('text/html', "Web\\Index.xhtml")
self.SendResponseContentTypeFile('font/woff2',"..\\Resources\\Web\\Semantic-UI-CSS-master\\themes\\default\\assets\\fonts\\icons.woff2") #Мост между файлом и http запросом (новый формат)
#Мост между файлом и http запросом (новый формат) if self.path == '/3rdParty/Semantic-UI-CSS-master/semantic.min.css':
if self.path == '/favicon.ico': self.SendResponseContentTypeFile('text/css',"..\\Resources\\Web\\Semantic-UI-CSS-master\\semantic.min.css")
self.SendResponseContentTypeFile('image/x-icon',"Web\\favicon.ico") #Мост между файлом и http запросом (новый формат)
#Мост между файлом и http запросом (новый формат) if self.path == '/3rdParty/Semantic-UI-CSS-master/semantic.min.js':
if self.path == '/3rdParty/Handlebars/handlebars-v4.1.2.js': self.SendResponseContentTypeFile('application/javascript',"..\\Resources\\Web\\Semantic-UI-CSS-master\\semantic.min.js")
self.SendResponseContentTypeFile('application/javascript',"..\\Resources\\Web\\Handlebars\\handlebars-v4.1.2.js") #Мост между файлом и http запросом (новый формат)
#Получить скриншот if self.path == '/3rdParty/jQuery/jquery-3.1.1.min.js':
if self.path.split("?")[0] == '/GetScreenshot': self.SendResponseContentTypeFile('application/javascript',"..\\Resources\\Web\\jQuery\\jquery-3.1.1.min.js")
#Сохранить файл на диск #Мост между файлом и http запросом (новый формат)
SaveScreenshot("Screenshot.png") if self.path == '/3rdParty/Google/LatoItalic.css':
self.SendResponseContentTypeFile('image/png',"Screenshot.png") self.SendResponseContentTypeFile('font/css',"..\\Resources\\Web\\Google\\LatoItalic.css")
#Monitor #Мост между файлом и http запросом (новый формат)
if self.path == '/Monitor/JSONDaemonListGet': if self.path == '/3rdParty/Semantic-UI-CSS-master/themes/default/assets/fonts/icons.woff2':
# Send response status code self.SendResponseContentTypeFile('font/woff2',"..\\Resources\\Web\\Semantic-UI-CSS-master\\themes\\default\\assets\\fonts\\icons.woff2")
self.send_response(200) #Мост между файлом и http запросом (новый формат)
# Send headers if self.path == '/favicon.ico':
self.send_header('Content-type','application/json') self.SendResponseContentTypeFile('image/x-icon',"Web\\favicon.ico")
self.end_headers() #Мост между файлом и http запросом (новый формат)
# Send message back to client if self.path == '/3rdParty/Handlebars/handlebars-v4.1.2.js':
message = json.dumps(mGlobalDict) self.SendResponseContentTypeFile('application/javascript',"..\\Resources\\Web\\Handlebars\\handlebars-v4.1.2.js")
# Write content as utf-8 data #Получить скриншот
self.wfile.write(bytes(message, "utf8")) if self.path.split("?")[0] == '/GetScreenshot':
if self.path == '/Monitor/ControlPanelDictGet': #Сохранить файл на диск
# Send response status code SaveScreenshot("Screenshot.png")
self.send_response(200) self.SendResponseContentTypeFile('image/png',"Screenshot.png")
# Send headers #Monitor
self.send_header('Content-type','application/json') if self.path == '/Monitor/JSONDaemonListGet':
self.end_headers() # Send response status code
#Create result JSON self.send_response(200)
lResultJSON={"RenderRobotList":[]} # Send headers
lRenderFunctionsRobotList=mGlobalDict["ControlPanelDict"]["RobotList"] self.send_header('Content-type','application/json')
for lItem in lRenderFunctionsRobotList: self.end_headers()
#Подключить модуль для вызова # Send message back to client
#print(globals()) message = json.dumps(mGlobalDict)
lModuleImportName = lItem.get("RenderFunctionModuleSubfolderName","") # Write content as utf-8 data
if lModuleImportName!="": self.wfile.write(bytes(message, "utf8"))
lModuleImportName = f'{lItem["RenderFunctionModuleSubfolderName"]}.{lItem["RenderFunctionModuleName"]}' if self.path == '/Monitor/ControlPanelDictGet':
else: # Send response status code
lModuleImportName = lItem["RenderFunctionModuleName"] self.send_response(200)
lModule=importlib.import_module(lModuleImportName) # Send headers
#Найти функцию self.send_header('Content-type','application/json')
lFunction=getattr(lModule,lItem["RenderFunctionName"]) self.end_headers()
#Выполнить вызов и записать результат #Create result JSON
lItemResultDict=lFunction(mGlobalDict) lResultJSON={"RenderRobotList":[]}
#RunFunction lRenderFunctionsRobotList=mGlobalDict["ControlPanelDict"]["RobotList"]
lResultJSON["RenderRobotList"].append(lItemResultDict) for lItem in lRenderFunctionsRobotList:
# Send message back to client #Подключить модуль для вызова
message = json.dumps(lResultJSON) #print(globals())
# Write content as utf-8 data lModuleImportName = lItem.get("RenderFunctionModuleSubfolderName","")
self.wfile.write(bytes(message, "utf8")) if lModuleImportName!="":
#Filemanager function lModuleImportName = f'{lItem["RenderFunctionModuleSubfolderName"]}.{lItem["RenderFunctionModuleName"]}'
if self.path.lower().startswith('/filemanager/'): else:
lFileURL=self.path[13:] lModuleImportName = lItem["RenderFunctionModuleName"]
# check if file in FileURL - File Path Mapping Dict lModule=importlib.import_module(lModuleImportName)
if lFileURL.lower() in mGlobalDict["FileManager"]["FileURLFilePathDict"]: #Найти функцию
self.SendResponseContentTypeFile('application/octet-stream',mGlobalDict["FileManager"]["FileURLFilePathDict"][lFileURL]) lFunction=getattr(lModule, lItem["RenderFunctionName"])
# Auth function #Выполнить вызов и записать результат
if self.path.lower().startswith('/auth'): lItemResultDict=lFunction(mGlobalDict)
# Send response status code #RunFunction
self.send_response(401) lResultJSON["RenderRobotList"].append(lItemResultDict)
# Send headers # Send message back to client
self.send_header('Content-type','text/html') message = json.dumps(lResultJSON)
self.send_header('WWW-Authenticate', 'Basic') #Always ask login pass # Write content as utf-8 data
lFlagIsKerberos=False self.wfile.write(bytes(message, "utf8"))
lHeaderAuthorization = self.headers.get("Authorization","").split(" ") #Filemanager function
import base64 if self.path.lower().startswith('/filemanager/'):
if len(lHeaderAuthorization) == 2: lFileURL=self.path[13:]
llHeaderAuthorizationDecodedUserPasswordList = base64.b64decode(lHeaderAuthorization[1]).decode("utf-8").split(":") # check if file in FileURL - File Path Mapping Dict
lUser = llHeaderAuthorizationDecodedUserPasswordList[0] if lFileURL.lower() in mGlobalDict["FileManager"]["FileURLFilePathDict"]:
lPassword = llHeaderAuthorizationDecodedUserPasswordList[1] self.SendResponseContentTypeFile('application/octet-stream',mGlobalDict["FileManager"]["FileURLFilePathDict"][lFileURL])
lDomain = None # Auth function
if "\\" in lUser:
lDomain = lUser.split("\\")[0]
lUser = lUser.split("\\")[1]
#print(f"Header Authorization, domain: {lDomain}, login: {lUser}, password: {lPassword}")
#self.send_header('Vary', 'negotiative')
#self.send_header('TCN', 'choice')
self.end_headers()
# Write content as utf-8 data
self.wfile.write(bytes("", "utf8"))
#pdb.set_trace()
# POST # POST
def do_POST(self): def do_POST(self):
#Централизованная функция получения запросов/отправки #####################################
if self.path == '/Utils/Processor': #Do authentication
#ReadRequest #Check if authentication is turned on
lInputObject={} #####################################
if self.headers.get('Content-Length') is not None: lFlagAccessUserBlock=False
lInputByteArrayLength = int(self.headers.get('Content-Length')) lAuthenticateDict = {"Domain": "", "User": ""}
lInputByteArray=self.rfile.read(lInputByteArrayLength) if mGlobalDict.get("Server", {}).get("AccessUsers", {}).get("FlagCredentialsAsk", False):
#Превращение массива байт в объект lAuthenticateDict = AuthenticateVerify(self)
lInputObject=json.loads(lInputByteArray.decode('utf8')) if not lAuthenticateDict["User"]:
# Send response status code lFlagAccessUserBlock=True
self.send_response(200) if lFlagAccessUserBlock:
# Send headers AuthenticateBlock(self)
self.send_header('Content-type','application/json') #####################################
self.end_headers() else:
# Send message back to client #Централизованная функция получения запросов/отправки
message = json.dumps(Processor.ActivityListOrDict(lInputObject)) if self.path == '/Utils/Processor':
# Write content as utf-8 data #ReadRequest
self.wfile.write(bytes(message, "utf8")) lInputObject={}
return if self.headers.get('Content-Length') is not None:
lInputByteArrayLength = int(self.headers.get('Content-Length'))
#print(ChildProcessReadWaitString(p)) lInputByteArray=self.rfile.read(lInputByteArrayLength)
#Превращение массива байт в объект
lInputObject=json.loads(lInputByteArray.decode('utf8'))
# Send response status code
self.send_response(200)
# Send headers
self.send_header('Content-type','application/json')
self.end_headers()
# Send message back to client
message = json.dumps(Processor.ActivityListOrDict(lInputObject))
# Write content as utf-8 data
self.wfile.write(bytes(message, "utf8"))
return

174
Orchestrator/Settings/Settings.py
Unescape View File

@ -64,77 +64,111 @@ def CheckIfProcessRunning(processName):
return False; return False;
#Orchestrator settings #Orchestrator settings
mDict = { def Settings():
"Server":{ mDict = {
"ListenPort_":"Порт, по которому можно подключиться к демону", "Server": {
"ListenPort":8081, "ListenPort_": "Порт, по которому можно подключиться к демону",
"ListenURLList":[ "ListenPort": 8081,
{ "ListenURLList": [
"Description":"Local machine test", {
"URL_":"Сетевое расположение сервера демона", "Description": "Local machine test",
"URL":"http://127.0.0.1:8081" "URL_": "Сетевое расположение сервера демона",
} "URL": "http://127.0.0.1:8081"
] }
}, ],
"Scheduler":{ "AccessUsers": {
"ActivityTimeCheckLoopSeconds":5, #Количество секунд, между циклами проверки действий "FlagCredentialsAsk": True, #Turn on Authentication
"ActivityTimeList":[ "FlagNoRulesBlock": True, #Block request, if user has no rules for requested URL (if FlagCredentialAsk is turned on)
{ "UserDict": {
"TimeHH:MM":"19:25", #Time [HH:MM] to trigger activity #("Domain", "User"): {
"WeekdayList":[1,2,3], #List of the weekday index when activity is applicable, Default [1,2,3,4,5,6,7] # "FunctionDict": {
"Activity":{ # # ("Method|GET|POST", "Match type|BeginWith|Contains", "URL"): def function (inRequest, inGlobalDict, inAuthenticateDict)
"Type":"ProcessStart", #Activity type # }
"Path":"Notepad", #Executable file path #}
"ArgList": [] #List of the arguments },
"AuthTokensDict": {
#"<AuthToken>":{"User":"", "Domain":"", "TokenDatetime":<Datetime>}
} }
}, }
{ },
"TimeHH:MM":"19:20", #Time [HH:MM] to trigger activity "Scheduler": {
"WeekdayList":[1,2,3], #List of the weekday index when activity is applicable, Default [1,2,3,4,5,6,7] "ActivityTimeCheckLoopSeconds":5, #Количество секунд, между циклами проверки действий
"Activity":{ "ActivityTimeList":[
"Type":"ProcessStop", #Activity type {
"Name":"OpenRPARobotDaemon.exe", #Process name "TimeHH:MM":"19:25", #Time [HH:MM] to trigger activity
"FlagForce":True, #Force process close "WeekdayList":[1,2,3], #List of the weekday index when activity is applicable, Default [1,2,3,4,5,6,7]
"User": "%username%" #Empty, user or %username% "Activity":{
"Type":"ProcessStart", #Activity type
"Path":"Notepad", #Executable file path
"ArgList": [] #List of the arguments
}
},
{
"TimeHH:MM":"19:20", #Time [HH:MM] to trigger activity
"WeekdayList":[1,2,3], #List of the weekday index when activity is applicable, Default [1,2,3,4,5,6,7]
"Activity":{
"Type":"ProcessStop", #Activity type
"Name":"OpenRPARobotDaemon.exe", #Process name
"FlagForce":True, #Force process close
"User": "%username%" #Empty, user or %username%
}
},
{
"TimeHH:MMStart":"19:20", #Time [HH:MM] to trigger activity
"TimeHH:MMStop":"19:20",
"ActivityIntervalSeconds":5,
"WeekdayList":[1,2,3], #List of the weekday index when activity is applicable, Default [1,2,3,4,5,6,7]
"Activity":{
"Type": "PythonStart", #Activity type
"ModuleName": "CheckActivity", #Python function module name
"FunctionName": "test_activity", #Python function name
"ArgList": ["TestArg1","TestArg2"] #Input python function args
}
} }
}, ],
{ "LogList":[]
"TimeHH:MMStart":"19:20", #Time [HH:MM] to trigger activity },
"TimeHH:MMStop":"19:20", "Processor":{
"ActivityIntervalSeconds":5, "LogType_CMDStart":True, #LogType_<Type>: <bool> if Trace for command is selected for False, the tracing will be off for such activity type. Default True
"WeekdayList":[1,2,3], #List of the weekday index when activity is applicable, Default [1,2,3,4,5,6,7] "LogList":[] #List of processor activity, which was executed. Fill list when orchestrator is running
"Activity":{ },
"Type": "PythonStart", #Activity type "ControlPanelDict":{
"ModuleName": "CheckActivity", #Python function module name "RefreshSeconds": 5,
"FunctionName": "test_activity", #Python function name "RobotList": [
"ArgList": ["TestArg1","TestArg2"] #Input python function args {
"RenderFunctionName":"RenderRobotR01",
"RenderFunctionModuleName":"Settings",
"RenderFunctionModuleSubfolderName":"Settings"
} }
]
},
"FileManager": {
"FileURLFilePathDict_help":"https://localhost:8081/filemanager/<file URL>. All FileURL s must be set in lowercase",
"FileURLFilePathDict":{
"r01/report.xlsx":"C:\\RPA\\R01_IntegrationOrderOut\\Data\\Reestr_otgruzok.xlsx"
} }
], },
"LogList":[] "Storage":{
}, "Robot_R01_help":"Robot data storage in orchestrator env",
"Processor":{ "Robot_R01":{}
"LogType_CMDStart":True, #LogType_<Type>: <bool> if Trace for command is selected for False, the tracing will be off for such activity type. Default True }
"LogList":[] #List of processor activity, which was executed. Fill list when orchestrator is running }
}, ###################################
"ControlPanelDict":{ #Init .py files from Settings folder
"RefreshSeconds": 5, ####################################
"RobotList": [ #Get file list from Settings folder
{ import os
"RenderFunctionName":"RenderRobotR01", import pdb
"RenderFunctionModuleName":"Settings", lSettingsPath = os.path.join(os.getcwd(), "Settings")
"RenderFunctionModuleSubfolderName":"Settings" #pdb.set_trace()
} #Lambda function to get files .py from settings folder except Settings.py
] lFileList = [f for f in os.listdir(lSettingsPath) if os.path.isfile(os.path.join(lSettingsPath, f)) and f.split(".")[-1]=="py" and f != "Settings.py"]
}, #TODO Import files as python module and call SettingsUpdate()
"FileManager": { import importlib.util
"FileURLFilePathDict_help":"https://localhost:8081/filemanager/<file URL>. All FileURL s must be set in lowercase", for lModuleFilePathItem in lFileList
"FileURLFilePathDict":{ spec = importlib.util.spec_from_file_location("module.name", "/path/to/file.py")
"r01/report.xlsx":"C:\\RPA\\R01_IntegrationOrderOut\\Data\\Reestr_otgruzok.xlsx" foo = importlib.util.module_from_spec(spec)
} #spec.loader.exec_module(foo)
}, #foo.MyClass()
"Storage":{ #Return Global settings
"Robot_R01_help":"Robot data storage in orchestrator env", return mDict
"Robot_R01":{}
}
}

2
Orchestrator/Settings/Test.py
Unescape View File

@ -0,0 +1,2 @@
def SettingsUpdate():
print("Everything is ok!")
Write Preview
Loading…
Cancel
Save