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

# v1.2.0 Def for old procesor to new processor
# Return new activity for the new processor
def v1_2_0_ProcessorOld2NewActivityDict(inActivityOld):

    # # # # # # # # # # # # # # # # # # # # # # # # # # # #
    # Technical defs for support old operations in old processor
    # # # # # # # # # # # # # # # # # # # # # # # # # # # #
    # {
    #               "Type":"WindowsLogon",
    #               "Domain":"",
    #               "User":"",
    #               "Password":""
    #               # Return "Result": True - user is logged on, False - user is not logged on
    #           }
    def WindowLogon(inUserStr, inPasswordStr, inDomainStr=""):
        #################################
        # Windows logon
        #################################
        try:
            hUser = win32security.LogonUser(
                inUserStr,
                inDomainStr,
                inPasswordStr,
                win32security.LOGON32_LOGON_NETWORK,
                win32security.LOGON32_PROVIDER_DEFAULT
            )
        except win32security.error:
            return False
        else:
            return True
        ###################################
    #           {
    #               "Type": "GlobalDictKeyListValueGet",
    #               "KeyList": ["key1","key2",...]
    #           },
    def GlobalDictKeyListValueGet(inGSettings, inKeyList=[]):
        ###########################################################
        # Обработка команды GlobalDictKeyListValueGet
        ###########################################################
        lResult = None
        lDict = inGSettings
        for lItem2 in inKeyList[:-1]:
            # Check if key - value exists
            if lItem2 in lDict:
                pass
            else:
                lDict[lItem2] = {}
            lDict = lDict[lItem2]
        # Return value
        lResult = lDict.get(inKeyList[-1], None)
        return lResult

    # # # # # # # # # # # # # # # # # # # # # # # # # # # #
    # # # # # # # # # # # # # # # # # # # # # # # # # # # #

    if inActivityOld["Type"] == "WindowsLogon":
        lResult = {
            "Def": WindowLogon, # 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)
        }
    elif inActivityOld["Type"] == "GlobalDictKeyListValueGet":
        lResult = {
            "Def": GlobalDictKeyListValueGet, # 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)
        }
    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

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