# Def to check inGSettings and update structure to the backward compatibility # !!! ATTENTION: Backward compatibility has been started from v1.1.13 !!! # So you can use config of the orchestrator 1.1.13 in new Orchestrator versions and all will be ok :) (hope it's true) import win32security from . import Orchestrator # For user defs # v1.2.0 Def for old procesor to new processor # Return new activity for the new processor def v1_2_0_ProcessorOld2NewActivityDict(inActivityOld): if inActivityOld["Type"] == "WindowsLogon": lResult = { "Def": Orchestrator.OSCredentialsVerify, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{"inUserStr": inActivityOld["User"],"inPasswordStr":inActivityOld["Password"],"inDomainStr":inActivityOld["Domain"]}, # Args dictionary "ArgGSettings": None, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": None # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } elif inActivityOld["Type"] == "GlobalDictKeyListValueGet": lResult = { "Def": Orchestrator.GSettingsKeyListValueGet, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{"inKeyList": inActivityOld["KeyList"]}, # Args dictionary "ArgGSettings": "inGSettings", # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": None # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } elif inActivityOld["Type"] == "CMDStart": lResult = { "Def": Orchestrator.OSCMD, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inCMDStr": inActivityOld["Command"]}, # Args dictionary "ArgGSettings": None, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": None # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } elif inActivityOld["Type"] == "OrchestratorRestart": lResult = { "Def": Orchestrator.OrchestratorRestart, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {}, # Args dictionary "ArgGSettings": "inGSettings", # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": None # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } elif inActivityOld["Type"] == "OrchestratorSessionSave": lResult = { "Def": Orchestrator.OrchestratorSessionSave, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {}, # Args dictionary "ArgGSettings": "inGSettings", # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": None # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } elif inActivityOld["Type"] == "GlobalDictKeyListValueSet": lResult = { "Def": Orchestrator.GSettingsKeyListValueSet, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inKeyList": inActivityOld["KeyList"], "inValue": inActivityOld["Value"]}, # Args dictionary "ArgGSettings": "inGSettings", # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": None # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } elif inActivityOld["Type"] == "GlobalDictKeyListValueAppend": lResult = { "Def": Orchestrator.GSettingsKeyListValueAppend, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inKeyList": inActivityOld["KeyList"], "inValue": inActivityOld["Value"]}, # Args dictionary "ArgGSettings": "inGSettings", # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": None # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } elif inActivityOld["Type"] == "GlobalDictKeyListValueOperator+": lResult = { "Def": Orchestrator.GSettingsKeyListValueOperatorPlus, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inKeyList": inActivityOld["KeyList"], "inValue": inActivityOld["Value"]}, # Args dictionary "ArgGSettings": None, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": None # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } elif inActivityOld["Type"] == "ProcessStart": lResult = { "Def": Orchestrator.ProcessStart, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inPathStr": inActivityOld["Path"], "inArgList": inActivityOld["ArgList"]}, # Args dictionary "ArgGSettings": None, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": None # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } elif inActivityOld["Type"] == "ProcessStartIfTurnedOff": lResult = { "Def": Orchestrator.ProcessStart, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inPathStr": inActivityOld["Path"], "inArgList": inActivityOld["ArgList"], "inStopProcessNameWOExeStr": inActivityOld["CheckTaskName"]}, # Args dictionary "ArgGSettings": None, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": None # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } elif inActivityOld["Type"] == "ProcessStop": lResult = { "Def": Orchestrator.ProcessStop, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inProcessNameWOExeStr": inActivityOld["Name"], "inCloseForceBool": inActivityOld["FlagForce"], "inUserNameStr": inActivityOld["User"]}, # Args dictionary "ArgGSettings": None, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": None # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } elif inActivityOld["Type"] == "PythonStart": lResult = { "Def": Orchestrator.PythonStart, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inModulePathStr": inActivityOld["ModuleName"], "inDefNameStr": inActivityOld["FunctionName"], "inArgList": inActivityOld["ArgList"], "inArgDict": inActivityOld["ArgDict"] }, # Args dictionary "ArgGSettings": None, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": "inLogger" # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } else: raise Exception(f"BackwardCompatibility up to v1.2.0, old processor: No type {inActivityOld['Type']} has been found in old processor.") return lResult # return the result # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # HERE IS THE MAIN DEF WHICH IS LAUNCHES WHEN START # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def Update(inGSettings): lL = inGSettings["Logger"] # Alias for logger # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # v1.1.13 to v1.1.14 # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # if "Autocleaner" not in inGSettings: # Add "Autocleaner" structure inGSettings["Autocleaner"] = { # Some gurbage is collecting in g settings. So you can configure autocleaner to periodically clear gSettings "IntervalSecFloat": 600.0, # Sec float to periodically clear gsettings } if lL: lL.warning(f"Backward compatibility (v1.1.13 to v1.1.14): Add default 'Autocleaner' structure") # Log about compatibility if "Client" not in inGSettings: # Add "Client" structure inGSettings["Client"] = { # Settings about client web orchestrator "Session":{ # Settings about web session. Session algorythms works only for special requests (URL in ServerSettings) "LifetimeSecFloat": 600.0, # Client Session lifetime in seconds. after this time server will forget about this client session "LifetimeRequestSecFloat": 120.0, # 1 client request lifetime in server in seconds "ControlPanelRefreshIntervalSecFloat": 1.5, # Interval to refresh control panels for session, "TechnicalSessionGUIDCache": { # TEchnical cache. Fills when web browser is requesting #"SessionGUIDStr":{ # Session with some GUID str. On client session guid stored in cookie "SessionGUIDStr" # "InitDatetime": None, # 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": None, # Last client request handler # "UserADStr": None, # User, who connect. None if user is not exists # "DomainADStr": None, # Domain of the user who connect. None if user is not exists #} } } } if lL: lL.warning(f"Backward compatibility (v1.1.13 to v1.1.14): Add default 'Client' structure") # Log about compatibility if "RequestTimeoutSecFloat" not in inGSettings["Server"]: # Add Server > "RequestTimeoutSecFloat" property inGSettings["Server"]["RequestTimeoutSecFloat"] = 300 # Time to handle request in seconds if lL: lL.warning( f"Backward compatibility (v1.1.13 to v1.1.14): Add default 'Server' > 'RequestTimeoutSecFloat' property") # Log about compatibility if "DefSettingsUpdatePathList" not in inGSettings["OrchestratorStart"]: # Add OrchestratorStart > "DefSettingsUpdatePathList" property inGSettings["OrchestratorStart"]["DefSettingsUpdatePathList"] = [] # List of the .py files which should be loaded before init the algorythms if lL: lL.warning(f"Backward compatibility (v1.1.13 to v1.1.14): Add default 'OrchestratorStart' > 'DefSettingsUpdatePathList' property list") # Log about compatibility # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # v1.1.20 to v1.2.0 # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # Update Structure gSettings["Processor"] if "Processor" in inGSettings: # Check if Processor exist del inGSettings["Processor"] # Remove the key if lL: lL.warning(f"Backward compatibility (v1.1.20 to v1.2.0): Remove old structure 'Processor'") # Log about compatibility if "ProcessorDict" not in inGSettings: # Create new ProcessorDict structure inGSettings["ProcessorDict"]={ "ActivityList": [ # List of the activities # { # "Def":"DefAliasTest", # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) # "ArgList":[1,2,3], # Args list # "ArgDict":{"ttt":1,"222":2,"dsd":3} # Args dictionary # "ArgGSettings": # Name of GSettings attribute: str (ArgDict) or index (for ArgList) # }, ], "AliasDefDict": {} , # Storage for def with Str alias. To use it see pyOpenRPA.Orchestrator.ControlPanel "CheckIntervalSecFloat": 1.0, # Interval for check gSettings in ProcessorDict > ActivityList "ExecuteBool": True, # Flag to execute thread processor "ThreadIdInt": None # Fill thread id when processor will be inited } if lL: lL.warning(f"Backward compatibility (v1.1.20 to v1.2.0): Create new structure 'ProcessorDict'") # Log about compatibility if "VersionStr" not in inGSettings: # Create new ProcessorDict structure inGSettings["VersionStr"] = None if lL: lL.warning(f"Backward compatibility (v1.1.20 to v1.2.0): Create new attribute 'VersionStr'") # Log about compatibility