import psutil import datetime def RenderRobotR01(inGlobalConfiguration): #Subheader Variants lSubheaderRunTrueText="Состояние: Работает" lSubheaderRunFalseText="Состояние: Не работает" #Run button #Такое большое количество слэшей связано с тем, что этот текст отправляется сначала в браузер, рендерится там, а потом отправляется на процессор оркестратора lOnClickRunButton="""mGlobal.Controller.CMDRunText("C:\\\\\\\\RPA\\\\\\\\R01_IntegrationOrderOut\\\\\\\\Sources\\\\\\\\R01_IntegrationOrderOut_64_Start.cmd");""" #Force close button lOnClickForceCloseButton="""mGlobal.Controller.CMDRunText("taskkill /F /im Robot_R01.exe");""" #Result template lResultDict={ "HeaderLeftText":"Автозагрузка заявок на расход", "HeaderRightText":"R01", "SubheaderText":lSubheaderRunFalseText, "BodyKeyValueList":[ #Дата запуска: 10:00:09 01.09.2019 {"Key":"Дата запуска","Value":"Не запущен"}, {"Key":"Текущий шаг","Value":"Не запущен"}, {"Key":"Время выполнения шага","Value":"--с."}, {"Key":"Отчет робота","Value":"Скачать"} ], "FooterText":"Дата изменения: 9:38:00 09.10.2019", "FooterButtonX2List":[ {"Text":"Ручной запуск", "Color":"green", "Link":"", "OnClick": lOnClickRunButton}, {"Text":"Безопасная остановка", "Color disabled":"orange", "Link":""} ], "FooterButtonX1List":[ {"Text":"Принудительная остановка", "Color":"red", "Link":"", "OnClick": lOnClickForceCloseButton} ] } #Check if process running if CheckIfProcessRunning("Robot_R01"): lResultDict["SubheaderText"]=lSubheaderRunTrueText #Fill robot info from Storage - R01 if "Robot_R01" in inGlobalConfiguration.get("Storage",{}): lItemDict=inGlobalConfiguration["Storage"]["Robot_R01"] #lResultDict["HeaderLeftText"]=lItemDict["Name"] #lResultDict["HeaderRightText"]=lItemDict["Code"] lResultDict["BodyKeyValueList"][0]["Value"]=lItemDict.get("RunDateTimeString","Не запущен") lResultDict["BodyKeyValueList"][1]["Value"]=lItemDict.get("StepCurrentName","Не запущен") lResultDict["BodyKeyValueList"][2]["Value"]=lItemDict.get("StepCurrentDuration","--с.") else: #Process not running lResultDict["FooterText"]=f'Дата изменения: {datetime.datetime.now().strftime("%H:%M:%S %d.%m.%Y")}' else: #Process not running lResultDict["FooterText"]=f'Дата изменения: {datetime.datetime.now().strftime("%H:%M:%S %d.%m.%Y")}' return lResultDict def CheckIfProcessRunning(processName): ''' Check if there is any running process that contains the given name processName. ''' #Iterate over the all the running process for proc in psutil.process_iter(): try: # Check if process name contains the given name string. if processName.lower() in proc.name().lower(): return True except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): pass return False; #Orchestrator settings def Settings(): mDict = { "Server": { "ListenPort_": "Порт, по которому можно подключиться к демону", "ListenPort": 8081, "ListenURLList": [ { "Description": "Local machine test", "URL_": "Сетевое расположение сервера демона", "URL": "http://127.0.0.1:8081" } ], "AccessUsers": { "FlagCredentialsAsk": True, #Turn on Authentication "FlagNoRulesBlock": True, #Block request, if user has no rules for requested URL (if FlagCredentialAsk is turned on) "UserDict": { #("Domain", "User"): { # "FunctionDict": { # # ("Method|GET|POST", "Match type|BeginWith|Contains", "URL"): def function (inRequest, inGlobalDict, inAuthenticateDict) # } #} }, "AuthTokensDict": { #"":{"User":"", "Domain":"", "TokenDatetime":} } } }, "Scheduler": { "ActivityTimeCheckLoopSeconds":5, #Количество секунд, между циклами проверки действий "ActivityTimeList": [ { "TimeHH:MM": "19:25", #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": "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": [] }, "Processor": { "LogType_CMDStart": True, #LogType_: 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": { "RefreshSeconds": 5, "RobotList": [ { "RenderFunction": RenderRobotR01 } ] }, "FileManager": { "FileURLFilePathDict_help": "https://localhost:8081/filemanager/. All FileURL s must be set in lowercase", "FileURLFilePathDict": { "r01/report.xlsx": "C:\\RPA\\R01_IntegrationOrderOut\\Data\\Reestr_otgruzok.xlsx" } }, "Storage": { "Robot_R01_help": "Robot data storage in orchestrator env", "Robot_R01": {} } } ################################### #Init .py files from Settings folder #################################### #Get file list from Settings folder import os import pdb #lFunction to call in subfiles lSubmoduleFunctionName = "SettingsUpdate" #lSettingsPath = os.path.join(inSettingsFolderPath, "Settings") lSettingsPath = "\\".join(__file__.split("\\")[:-1]) #lSettingsPath = os.path.join(os.getcwd(), "Settings") #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 os.path.join(lSettingsPath, f) != __file__] import importlib.util for lModuleFilePathItem in lFileList: lModuleName = lModuleFilePathItem[0:-3] lFileFullPath = os.path.join(lSettingsPath, lModuleFilePathItem) lTechSpecification = importlib.util.spec_from_file_location(lModuleName, lFileFullPath) lTechModuleFromSpec = importlib.util.module_from_spec(lTechSpecification) lTechSpecificationModuleLoader = lTechSpecification.loader.exec_module(lTechModuleFromSpec) if lSubmoduleFunctionName in dir(lTechModuleFromSpec): #Run SettingUpdate function in submodule getattr(lTechModuleFromSpec, lSubmoduleFunctionName)(mDict) return mDict