diff --git a/Sources/pyOpenRPA/Orchestrator/Server.py b/Sources/pyOpenRPA/Orchestrator/Server.py index 1911b883..7c2f4b24 100755 --- a/Sources/pyOpenRPA/Orchestrator/Server.py +++ b/Sources/pyOpenRPA/Orchestrator/Server.py @@ -8,6 +8,7 @@ #from http.client import HTTPException import threading +import typing from pyOpenRPA.Tools import CrossOS @@ -129,6 +130,7 @@ from . import ServerSettings async def BackwardCompatibility(inRequest:Request, inResponse:Response, inAuthTokenStr = None): lHTTPRequest = ServerBC.HTTPRequestOld(inRequest=inRequest, inResponse=inResponse, inAuthTokenStr=inAuthTokenStr) lHTTPRequest.path = inRequest.url.path + #print(f"WEB START: {lHTTPRequest.path}") inBodyStr = await inRequest.body() if inBodyStr == None: inBodyStr = "" else: inBodyStr = inBodyStr.decode("utf8") @@ -139,6 +141,7 @@ async def BackwardCompatibility(inRequest:Request, inResponse:Response, inAuthTo lResult = lHTTPRequest.do_GET(inBodyStr=inBodyStr) elif inRequest.method=="POST": lResult = lHTTPRequest.do_POST(inBodyStr=inBodyStr) + #print(f"WEB STOP: {lHTTPRequest.path}") if lHTTPRequest.OpenRPAResponseDict["Headers"]["Content-type"] != None: return StreamingResponse(io.BytesIO(lResult), media_type=lHTTPRequest.OpenRPAResponseDict["Headers"]["Content-type"]) diff --git a/Sources/pyOpenRPA/Orchestrator/ServerSettings.py b/Sources/pyOpenRPA/Orchestrator/ServerSettings.py index 448415c9..94152b46 100755 --- a/Sources/pyOpenRPA/Orchestrator/ServerSettings.py +++ b/Sources/pyOpenRPA/Orchestrator/ServerSettings.py @@ -53,8 +53,10 @@ def HiddenJSInitGenerate(inRequest, inGSettings): # Generate CP HTML + JSON # Return {"Key":{"",""}} -def HiddenCPDictGenerate(inRequest, inGSettings): - dUAC = inRequest.UACClientCheck # Alias. +def HiddenCPDictGenerate(inAuthTokenStr): + inGSettings = __Orchestrator__.GSettingsGet() + dUAC = lambda inKeyList: __Orchestrator__.WebUserUACCheck(inAuthTokenStr=inAuthTokenStr, inKeyList=inKeyList) + #dUAC = d_uac #inRequest.UACClientCheck # Alias. lUACCPTemplateKeyList=["pyOpenRPADict","CPKeyDict"] lL = inGSettings["Logger"] # Alias for logger # Create result JSON @@ -62,15 +64,15 @@ def HiddenCPDictGenerate(inRequest, inGSettings): lRenderFunctionsRobotDict = inGSettings["ServerDict"]["ControlPanelDict"] for lItemKeyStr in lRenderFunctionsRobotDict: lItemDict = lRenderFunctionsRobotDict[lItemKeyStr] - lUACBool = dUAC(inRoleKeyList=lUACCPTemplateKeyList+[lItemKeyStr]) # Check if render function is applicable User Access Rights (UAC) + lUACBool = dUAC(inKeyList=lUACCPTemplateKeyList+[lItemKeyStr]) # Check if render function is applicable User Access Rights (UAC) if lItemKeyStr=="VersionCheck": lUACBool=True # For backward compatibility for the old fron version which not reload page when new orch version is comming if lUACBool: # Run function if UAC is TRUE lCPItemDict = {"HTMLStr": None, "JSONDict":None} try: # HTML Render - lCPItemDict["HTMLStr"] = lItemDict.OnRefreshHTMLStr(inRequest=inRequest) + lCPItemDict["HTMLStr"] = lItemDict.OnRefreshHTMLStr(inAuthTokenStr=inAuthTokenStr) # JSONGeneratorDef - lCPItemDict["JSONDict"] = lItemDict.OnRefreshJSONDict(inRequest=inRequest) + lCPItemDict["JSONDict"] = lItemDict.OnRefreshJSONDict(inAuthTokenStr=inAuthTokenStr) except Exception as e: lL.exception(f"EXCEPTION WHEN HTML/ JSON RENDER") # Insert CPItemDict in result CPDict @@ -78,14 +80,15 @@ def HiddenCPDictGenerate(inRequest, inGSettings): return lCPDict # Return {"Key":{"",""}} -def HiddenRDPDictGenerate(inRequest, inGSettings): - dUAC = inRequest.UACClientCheck # Alias. +def HiddenRDPDictGenerate(inAuthTokenStr): + inGSettings = __Orchestrator__.GSettingsGet() + dUAC = lambda inKeyList: __Orchestrator__.WebUserUACCheck(inAuthTokenStr=inAuthTokenStr, inKeyList=inKeyList) lUACRDPTemplateKeyList=["pyOpenRPADict","RDPKeyDict"] lRDPDict = {"HandlebarsList":[]} # Iterate throught the RDP list for lRDPSessionKeyStrItem in inGSettings["RobotRDPActive"]["RDPList"]: # Check UAC - if dUAC(inRoleKeyList=lUACRDPTemplateKeyList+[lRDPSessionKeyStrItem]): + if dUAC(inKeyList=lUACRDPTemplateKeyList+[lRDPSessionKeyStrItem]): lRDPConfiguration = inGSettings["RobotRDPActive"]["RDPList"][ lRDPSessionKeyStrItem] # Get the configuration dict lDataItemDict = {"SessionKeyStr": "", "SessionHexStr": "", "IsFullScreenBool": False, @@ -102,15 +105,16 @@ def HiddenRDPDictGenerate(inRequest, inGSettings): return lRDPDict # Return {"HostNameUpperStr;UserUpperStr":{"IsListenBool":True}, "HandlebarsList":[{"HostnameUpperStr":"","UserUpperStr":"","IsListenBool":True}]} -def HiddenAgentDictGenerate(inRequest, inGSettings): - dUAC = inRequest.UACClientCheck # Alias. +def HiddenAgentDictGenerate(inAuthTokenStr): + inGSettings = __Orchestrator__.GSettingsGet() + dUAC = lambda inKeyList: __Orchestrator__.WebUserUACCheck(inAuthTokenStr=inAuthTokenStr, inKeyList=inKeyList) lUACAgentTemplateKeyList=["pyOpenRPADict","AgentKeyDict"] lAgentDict = {"HandlebarsList":[]} # Iterate throught the RDP list for lAgentItemKeyStrItem in inGSettings["AgentDict"]: # Check UAC lKeyStr = f"{lAgentItemKeyStrItem[0]};{lAgentItemKeyStrItem[1]}" # turple ("HostNameUpperStr","UserUpperStr") > Str "HostNameUpperStr;UserUpperStr" - if dUAC(inRoleKeyList=lUACAgentTemplateKeyList+[lKeyStr]): + if dUAC(inKeyList=lUACAgentTemplateKeyList+[lKeyStr]): lDataItemDict = inGSettings["AgentDict"][lAgentItemKeyStrItem] lDataItemAgentDict = copy.deepcopy(lDataItemDict) lDataItemAgentDict["ActivityList"] = [] @@ -128,33 +132,42 @@ def HiddenAgentDictGenerate(inRequest, inGSettings): # Client: mGlobal.pyOpenRPA.ServerDataHashStr # Client: mGlobal.pyOpenRPA.ServerDataDict -def pyOpenRPA_ServerData(inRequest,inGSettings): +import asyncio +from fastapi import Request +@app.post("/orpa/client/server-data",response_class=JSONResponse) +#{"Method": "POST", "URL": "/orpa/client/server-data", "MatchType": "Equal","ResponseDefRequestGlobal": pyOpenRPA_ServerData, "ResponseContentType": "application/json"}, +async def pyOpenRPA_ServerData(inRequest: Request): + inGSettings = __Orchestrator__.GSettingsGet() + print(f"START pyOpenRPA_ServerData") # Extract the hash value from request - lValueStr = inRequest.body + lValueStr = await inRequest.body() + lValueStr= lValueStr.decode("utf8") # Generate ServerDataDict lFlagDoGenerateBool = True while lFlagDoGenerateBool: lServerDataDict = { - "CPDict": HiddenCPDictGenerate(inRequest=inRequest, inGSettings=inGSettings), - "RDPDict": HiddenRDPDictGenerate(inRequest=inRequest, inGSettings=inGSettings), - "AgentDict": HiddenAgentDictGenerate(inRequest=inRequest, inGSettings=inGSettings), - "UserDict": {"UACClientDict": inRequest.OpenRPA["DefUserRoleHierarchyGet"](), "CWDPathStr": os.getcwd(), "VersionStr": inGSettings["VersionStr"]}, - } + "CPDict": None,#HiddenCPDictGenerate(inRequest=inRequest, inGSettings=inGSettings), + "RDPDict": None,#HiddenRDPDictGenerate(inRequest=inRequest, inGSettings=inGSettings), + "AgentDict": None,#HiddenAgentDictGenerate(inRequest=inRequest, inGSettings=inGSettings), + "UserDict": {"UACClientDict": {}, "CWDPathStr": os.getcwd(), "VersionStr": inGSettings["VersionStr"]}, + } # inRequest.OpenRPA["DefUserRoleHierarchyGet"]() # Create JSON lServerDataDictJSONStr = json.dumps(lServerDataDict) # Generate hash lServerDataHashStr = str(hash(lServerDataDictJSONStr)) if lValueStr!=lServerDataHashStr and lServerDataHashStr!= "" and lServerDataHashStr!= None: # Case if Hash is not equal + print(f"lValueStr:{lValueStr}, lServerDataHashStr:{lServerDataHashStr}") lFlagDoGenerateBool = False else: # Case Hashes are equal - time.sleep(inGSettings["Client"]["Session"]["ControlPanelRefreshIntervalSecFloat"]) + await asyncio.sleep(inGSettings["Client"]["Session"]["ControlPanelRefreshIntervalSecFloat"]) # Return the result if Hash is changed lResult = {"HashStr": lServerDataHashStr, "ServerDataDict": lServerDataDict} - inResponseDict = inRequest.OpenRPAResponseDict + #inResponseDict = inRequest.OpenRPAResponseDict # Send message back to client - message = json.dumps(lResult) + #message = json.dumps(lResult) # Write content as utf-8 data - inResponseDict["Body"] = bytes(message, "utf8") + #inResponseDict["Body"] = bytes(message, "utf8") + print(f"STOP pyOpenRPA_ServerData") return lResult # GET @@ -171,7 +184,7 @@ def pyOpenRPA_ServerJSInit(inRequest,inGSettings): # /pyOpenRPA/ServerLog return {"HashStr" , "ServerLogList": ["row 1", "row 2"]} # Client: mGlobal.pyOpenRPA.ServerLogListHashStr # Client: mGlobal.pyOpenRPA.ServerLogList -def pyOpenRPA_ServerLog(inRequest,inGSDict): +async def pyOpenRPA_ServerLog(inRequest,inGSDict): # Extract the hash value from request lValueStr = inRequest.body # Generate ServerDataDict @@ -183,7 +196,7 @@ def pyOpenRPA_ServerLog(inRequest,inGSDict): if lValueStr!=lServerLogListHashStr and lServerLogListHashStr!= "" and lServerLogListHashStr!= None: # Case if Hash is not equal Fix because None can be obtained without JSON decode lFlagDoGenerateBool = False else: # Case Hashes are equal - time.sleep(inGSDict["Client"]["DumpLogListRefreshIntervalSecFloat"]) + asyncio.sleep(inGSDict["Client"]["DumpLogListRefreshIntervalSecFloat"]) # Return the result if Hash is changed lResult = {"HashStr": lServerLogListHashStr, "ServerLogList": lServerLogList} inResponseDict = inRequest.OpenRPAResponseDict @@ -512,9 +525,9 @@ def SettingsUpdate(): #{"Method": "GET", "URL": "/pyOpenRPA_logo.png", "MatchType": "Equal", "ResponseFilePath": os.path.join(lOrchestratorFolder, "..\\Resources\\Web\\pyOpenRPA_logo.png"), "ResponseContentType": "image/png", "UACBool":False, "UseCacheBool": True}, {"Method": "POST", "URL": "/orpa/client/user-role-hierarchy-get", "MatchType": "Equal","ResponseDefRequestGlobal": BackwardCompatibility.v1_2_0_UserRoleHierarchyGet, "ResponseContentType": "application/json"}, # New way of the v.1.2.0 functionallity (all defs by the URL from /pyOpenRPA/...) - {"Method": "POST", "URL": "/orpa/client/server-data", "MatchType": "Equal","ResponseDefRequestGlobal": pyOpenRPA_ServerData, "ResponseContentType": "application/json"}, + #{"Method": "POST", "URL": "/orpa/client/server-data", "MatchType": "Equal","ResponseDefRequestGlobal": pyOpenRPA_ServerData, "ResponseContentType": "application/json"}, {"Method": "GET", "URL": "/orpa/client/server-js-init", "MatchType": "Equal","ResponseDefRequestGlobal": pyOpenRPA_ServerJSInit, "ResponseContentType": "application/javascript"}, - {"Method": "POST", "URL": "/orpa/client/server-log", "MatchType": "Equal","ResponseDefRequestGlobal": pyOpenRPA_ServerLog, "ResponseContentType": "application/json"}, + #{"Method": "POST", "URL": "/orpa/client/server-log", "MatchType": "Equal","ResponseDefRequestGlobal": pyOpenRPA_ServerLog, "ResponseContentType": "application/json"}, #{"Method": "GET", "URL": "/orpa/client/screenshot-get", "MatchType": "Equal", "ResponseDefRequestGlobal": pyOpenRPA_Screenshot, "ResponseContentType": "image/png"}, # API #{"Method": "POST", "URL": "/orpa/api/processor-queue-add", "MatchType": "Equal","ResponseDefRequestGlobal": pyOpenRPA_Processor, "ResponseContentType": "application/json"}, diff --git a/Sources/pyOpenRPA/Orchestrator/__Orchestrator__.py b/Sources/pyOpenRPA/Orchestrator/__Orchestrator__.py index f05e4672..f6f9c5ac 100755 --- a/Sources/pyOpenRPA/Orchestrator/__Orchestrator__.py +++ b/Sources/pyOpenRPA/Orchestrator/__Orchestrator__.py @@ -905,6 +905,151 @@ def UACSuperTokenUpdate(inSuperTokenStr, inGSettings=None): {inSuperTokenStr:{"User":lLoginStr, "Domain":"", "TokenDatetime": datetime.datetime.now(), "FlagDoNotExpire":True}} ) +## GSettings defs + +from . import SettingsTemplate + +GSettings = SettingsTemplate.Create(inModeStr = "BASIC") +# Modules alias for pyOpenRPA.Orchestrator and pyOpenRPA.Orchestrator.__Orchestrator__ +lCurrentModule = sys.modules[__name__] +if __name__ == "pyOpenRPA.Orchestrator" and "pyOpenRPA.Orchestrator.__Orchestrator__" not in sys.modules: + sys.modules["pyOpenRPA.Orchestrator.__Orchestrator__"] = lCurrentModule +if __name__ == "pyOpenRPA.Orchestrator.__Orchestrator__" and "pyOpenRPA.Orchestrator" not in sys.modules: + sys.modules["pyOpenRPA.Orchestrator"] = lCurrentModule + +def GSettingsGet(inGSettings=None): + """L+,W+: Вернуть глобальный словарь настроек Оркестратора. Во время выполнения программы глобальный словарь настроек существует в единственном экземпляре (синглтон) + + :param inGSettings: Дополнительный словарь настроек, который необходимо добавить в основной глобальный словарь настроек Оркестратора (синглтон) + :return: Глобальный словарь настроек GSettings + """ + global GSettings # identify the global variable + # Merge dictionaries if some new dictionary has come + if inGSettings is not None and GSettings is not inGSettings: + GSettings = Dictionary.MergeNoException(in1Dict = inGSettings, in2Dict = GSettings) + return GSettings # Return the result + +def GSettingsKeyListValueSet(inValue, inKeyList=None, inGSettings = None): + """L+,W+: Установить значение из глобального словаря настроек Оркестратора GSettings по списку ключей. + + Пример: Для того, чтобы установить значение для ключа car в словаре {"complex":{"car":"green"}, "simple":"HELLO"}, необходимо передать список ключей: ["complex", "car"] + + :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) + :param inValue: Значение для установки в глобальный словарь настроек Оркестратора GSettings + :param inKeyList: Список ключей, по адресу которого установить значение в GSettings + """ + inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings + if inKeyList is None: inKeyList = [] + lDict = inGSettings + for lItem2 in inKeyList[:-1]: + #Check if key - value exists + if lItem2 in lDict: + pass + else: + lDict[lItem2]={} + lDict=lDict[lItem2] + lDict[inKeyList[-1]] = inValue #Set value + return True + +def GSettingsKeyListValueGet(inKeyList=None, inGSettings = None): + """L+,W+: Получить значение из глобального словаря настроек Оркестратора GSettings по списку ключей. + + :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) + :param inKeyList: Список ключей, по адресу которого получить значение из GSettings + :return: Значение (тип данных определяется форматом хранения в GSettings) + """ + inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings + if inKeyList is None: inKeyList = [] + lDict = inGSettings + for lItem2 in inKeyList[:-1]: + #Check if key - value exists + if lItem2 in lDict: + pass + else: + lDict[lItem2]={} + lDict=lDict[lItem2] + return lDict.get(inKeyList[-1],None) + +def GSettingsKeyListValueAppend(inValue, inKeyList=None, inGSettings = None): + """L+,W+: Применить операцию .append к обьекту, расположенному по адресу списка ключей inKeyList в глобальном словаре настроек Оркестратора GSettings. Пример: Добавить значение в конец списка (list). + + .. code-block:: python + + # ПРИМЕР + from pyOpenRPA import Orchestrator + + Orchestrator.GSettingsKeyListValueAppend( + inGSettings = gSettings, + inValue = "NewValue", + inKeyList=["NewKeyDict","NewKeyList"]): + # result inGSettings: { + # ... another keys in gSettings ..., + # "NewKeyDict":{ + # "NewKeyList":[ + # "NewValue" + # ] + # } + #} + + :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) + :param inValue: Значение для установки в глобальный словарь настроек Оркестратора GSettings + :param inKeyList: Список ключей, по адресу которого выполнить добавление в конец списка (list) + """ + inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings + if inKeyList is None: inKeyList = [] + lDict = inGSettings + for lItem2 in inKeyList[:-1]: + #Check if key - value exists + if lItem2 in lDict: + pass + else: + lDict[lItem2]={} + lDict=lDict[lItem2] + lDict[inKeyList[-1]].append(inValue) #Set value + return True + +def GSettingsKeyListValueOperatorPlus(inValue, inKeyList=None, inGSettings = None): + """L+,W+: Применить оператор сложения (+) к обьекту, расположенному по адресу списка ключей inKeyList в глобальном словаре настроек Оркестратора GSettings. Пример: соединить 2 списка (list). + + .. code-block:: python + + # ПРИМЕР + from pyOpenRPA import Orchestrator + + Orchestrator.GSettingsKeyListValueOperatorPlus( + inGSettings = gSettings, + inValue = [1,2,3], + inKeyList=["NewKeyDict","NewKeyList"]): + # result inGSettings: { + # ... another keys in gSettings ..., + # "NewKeyDict":{ + # "NewKeyList":[ + # "NewValue", + # 1, + # 2, + # 3 + # ] + # } + #} + + :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) + :param inValue: Значение для установки в глобальный словарь настроек Оркестратора GSettings + :param inKeyList: Список ключей, по адресу которого выполнить добавление в конец списка (list) + """ + inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings + if inKeyList is None: inKeyList = [] + lDict = inGSettings + for lItem2 in inKeyList[:-1]: + #Check if key - value exists + if lItem2 in lDict: + pass + else: + lDict[lItem2]={} + lDict=lDict[lItem2] + lDict[inKeyList[-1]] += inValue #Set value + return True + + # # # # # # # # # # # # # # # # # # # # # # # # OrchestratorWeb defs # # # # # # # # # # # # # # # # # # # # # # # @@ -1170,13 +1315,11 @@ def WebRequestGet(): lCurrentThread = threading.current_thread() if hasattr(lCurrentThread, "request"): return lCurrentThread.request - from fastapi import FastAPI def WebAppGet() -> FastAPI: """L+,W+: Вернуть экземпляр веб сервера fastapi.FastAPI (app). Подробнее про app см. https://fastapi.tiangolo.com/tutorial/first-steps/ """ return Server.app - def WebUserLoginGet(inAuthTokenStr: str=None) -> str: """L+,W+: Получить логин авторизованного пользователя. Если авторизация не производилась - вернуть None @@ -1197,22 +1340,24 @@ def WebUserDomainGet(inAuthTokenStr: str=None) -> str: :return: Домен пользователя :rtype: str """ + if inAuthTokenStr is None: return None inGS = GSettingsGet() # Get the global settings return inGS.get("ServerDict", {}).get("AccessUsers", {}).get("AuthTokensDict", {}).get(inAuthTokenStr, {}).get("Domain", None) -def WebUserInfoGet(inRequest=None): - """L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.2, см. WebUserLoginGet, WebUserDomainGet] Информация о пользователе, который отправил HTTP запрос. +def WebUserInfoGet(inAuthTokenStr=None): + """L+,W+: Информация о пользователе, который отправил HTTP запрос. :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя :return: Сведения в формате {"DomainUpperStr": "PYOPENRPA", "UserNameUpperStr": "IVAN.MASLOV"} """ - if inRequest is None: inRequest = WebRequestGet() try: - lDomainUpperStr = inRequest.OpenRPA["Domain"].upper() - lUserUpperStr = inRequest.OpenRPA["User"].upper() - return {"DomainUpperStr": lDomainUpperStr, "UserNameUpperStr": lUserUpperStr} + lResultDict = { + "DomainUpperStr": WebUserDomainGet(inAuthTokenStr=inAuthTokenStr).upper(), + "UserNameUpperStr": WebUserLoginGet(inAuthTokenStr=inAuthTokenStr).upper() + } + return lResultDict except Exception as e: return {"DomainUpperStr": None, "UserNameUpperStr": None} @@ -1270,149 +1415,7 @@ def WebUserUACCheck(inAuthTokenStr:str=None, inKeyList:list=None) -> bool: break # Close the loop return lResult # Return the result -## GSettings defs - -from . import SettingsTemplate - -GSettings = SettingsTemplate.Create(inModeStr = "BASIC") -# Modules alias for pyOpenRPA.Orchestrator and pyOpenRPA.Orchestrator.__Orchestrator__ -lCurrentModule = sys.modules[__name__] -if __name__ == "pyOpenRPA.Orchestrator" and "pyOpenRPA.Orchestrator.__Orchestrator__" not in sys.modules: - sys.modules["pyOpenRPA.Orchestrator.__Orchestrator__"] = lCurrentModule -if __name__ == "pyOpenRPA.Orchestrator.__Orchestrator__" and "pyOpenRPA.Orchestrator" not in sys.modules: - sys.modules["pyOpenRPA.Orchestrator"] = lCurrentModule - -def GSettingsGet(inGSettings=None): - """L+,W+: Вернуть глобальный словарь настроек Оркестратора. Во время выполнения программы глобальный словарь настроек существует в единственном экземпляре (синглтон) - - :param inGSettings: Дополнительный словарь настроек, который необходимо добавить в основной глобальный словарь настроек Оркестратора (синглтон) - :return: Глобальный словарь настроек GSettings - """ - global GSettings # identify the global variable - # Merge dictionaries if some new dictionary has come - if inGSettings is not None and GSettings is not inGSettings: - GSettings = Dictionary.MergeNoException(in1Dict = inGSettings, in2Dict = GSettings) - return GSettings # Return the result - -def GSettingsKeyListValueSet(inValue, inKeyList=None, inGSettings = None): - """L+,W+: Установить значение из глобального словаря настроек Оркестратора GSettings по списку ключей. - - Пример: Для того, чтобы установить значение для ключа car в словаре {"complex":{"car":"green"}, "simple":"HELLO"}, необходимо передать список ключей: ["complex", "car"] - - :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) - :param inValue: Значение для установки в глобальный словарь настроек Оркестратора GSettings - :param inKeyList: Список ключей, по адресу которого установить значение в GSettings - """ - inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings - if inKeyList is None: inKeyList = [] - lDict = inGSettings - for lItem2 in inKeyList[:-1]: - #Check if key - value exists - if lItem2 in lDict: - pass - else: - lDict[lItem2]={} - lDict=lDict[lItem2] - lDict[inKeyList[-1]] = inValue #Set value - return True - -def GSettingsKeyListValueGet(inKeyList=None, inGSettings = None): - """L+,W+: Получить значение из глобального словаря настроек Оркестратора GSettings по списку ключей. - - :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) - :param inKeyList: Список ключей, по адресу которого получить значение из GSettings - :return: Значение (тип данных определяется форматом хранения в GSettings) - """ - inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings - if inKeyList is None: inKeyList = [] - lDict = inGSettings - for lItem2 in inKeyList[:-1]: - #Check if key - value exists - if lItem2 in lDict: - pass - else: - lDict[lItem2]={} - lDict=lDict[lItem2] - return lDict.get(inKeyList[-1],None) - -def GSettingsKeyListValueAppend(inValue, inKeyList=None, inGSettings = None): - """L+,W+: Применить операцию .append к обьекту, расположенному по адресу списка ключей inKeyList в глобальном словаре настроек Оркестратора GSettings. Пример: Добавить значение в конец списка (list). - - .. code-block:: python - - # ПРИМЕР - from pyOpenRPA import Orchestrator - Orchestrator.GSettingsKeyListValueAppend( - inGSettings = gSettings, - inValue = "NewValue", - inKeyList=["NewKeyDict","NewKeyList"]): - # result inGSettings: { - # ... another keys in gSettings ..., - # "NewKeyDict":{ - # "NewKeyList":[ - # "NewValue" - # ] - # } - #} - - :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) - :param inValue: Значение для установки в глобальный словарь настроек Оркестратора GSettings - :param inKeyList: Список ключей, по адресу которого выполнить добавление в конец списка (list) - """ - inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings - if inKeyList is None: inKeyList = [] - lDict = inGSettings - for lItem2 in inKeyList[:-1]: - #Check if key - value exists - if lItem2 in lDict: - pass - else: - lDict[lItem2]={} - lDict=lDict[lItem2] - lDict[inKeyList[-1]].append(inValue) #Set value - return True - -def GSettingsKeyListValueOperatorPlus(inValue, inKeyList=None, inGSettings = None): - """L+,W+: Применить оператор сложения (+) к обьекту, расположенному по адресу списка ключей inKeyList в глобальном словаре настроек Оркестратора GSettings. Пример: соединить 2 списка (list). - - .. code-block:: python - - # ПРИМЕР - from pyOpenRPA import Orchestrator - - Orchestrator.GSettingsKeyListValueOperatorPlus( - inGSettings = gSettings, - inValue = [1,2,3], - inKeyList=["NewKeyDict","NewKeyList"]): - # result inGSettings: { - # ... another keys in gSettings ..., - # "NewKeyDict":{ - # "NewKeyList":[ - # "NewValue", - # 1, - # 2, - # 3 - # ] - # } - #} - - :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) - :param inValue: Значение для установки в глобальный словарь настроек Оркестратора GSettings - :param inKeyList: Список ключей, по адресу которого выполнить добавление в конец списка (list) - """ - inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings - if inKeyList is None: inKeyList = [] - lDict = inGSettings - for lItem2 in inKeyList[:-1]: - #Check if key - value exists - if lItem2 in lDict: - pass - else: - lDict[lItem2]={} - lDict=lDict[lItem2] - lDict[inKeyList[-1]] += inValue #Set value - return True def StorageRobotExists(inRobotNameStr): """L+,W+: Проверить, существует ли ключ inRobotNameStr в хранилище пользовательской информации StorageDict (GSettings > StarageDict)