ORPA-pyOpenRPA/Sources/pyOpenRPA/Orchestrator/BackwardCompatibility.py

# 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
        }
        if lL: lL.warning(f"Backward compatibility (v1.1.20 to v1.2.0): Create new structure 'ProcessorDict'")  # Log about compatibility