Исходный код pyOpenRPA.Orchestrator.__Orchestrator__

from re import I
import subprocess, json, psutil, time, os
import typing
import keyboard
from pyOpenRPA.Tools import CrossOS
if CrossOS.IS_WINDOWS_BOOL: import win32security #CrossOS
if CrossOS.IS_LINUX_BOOL: from simplepam import authenticate #CrossOS
import sys, base64, logging, ctypes, copy #Get input argument
import pickle
import inspect
import schedule
#from partd import Server

from . import Server
from . import Timer
from . import Processor
from . import BackwardCompatibility # Backward compatibility from v1.1.13
from . import Core
from . import Managers
from ..Tools import License
from ..Utils import Dictionary

if CrossOS.IS_WINDOWS_BOOL: from subprocess import CREATE_NEW_CONSOLE

from .Utils import LoggerHandlerDumpLogList
from ..Tools import Debugger

# ATTENTION! HERE IS NO Relative import because it will be imported dynamically
# All function check the flag SessionIsWindowResponsibleBool == True else no cammand is processed
# All functions can return None, Bool or Dict { "IsSuccessful": True }
from .RobotRDPActive import CMDStr # Create CMD Strings
from .RobotRDPActive import Connector # RDP API

#from .Settings import Settings
import importlib
from importlib import util
import threading # Multi-threading for RobotRDPActive
from .RobotRDPActive import RobotRDPActive #Start robot rdp active
from .RobotScreenActive import Monitor #Start robot screen active
from . import SettingsTemplate # Settings template
import uuid # Generate uuid
import datetime # datetime
import math
import glob # search the files
import urllib

from . import ServerSettings

from fastapi import FastAPI, Depends

#Единый глобальный словарь (За основу взять из Settings.py)
gSettingsDict = None

# AGENT DEFS

[документация]def AgentActivityItemAdd(inHostNameStr, inUserStr, inActivityItemDict, inGSettings=None): """L+,W+: Добавить активность в словарь активностей выбранного Агента :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inActivityItemDict: Активность (ActivityItem). См. функцию ProcessorActivityitemCreate :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lActivityItemDict = copy.deepcopy(inActivityItemDict) # Add GUIDStr if not exist lGUIDStr = None if "GUIDStr" not in lActivityItemDict: lGUIDStr = str(uuid.uuid4()) # generate new GUID lActivityItemDict["GUIDStr"] = lGUIDStr else: lGUIDStr = lActivityItemDict["GUIDStr"] # Add CreatedByDatetime lActivityItemDict["CreatedByDatetime"] = datetime.datetime.now() # Main alg lAgentDictItemKeyTurple = (inHostNameStr.upper(),inUserStr.upper()) if lAgentDictItemKeyTurple not in inGSettings["AgentDict"]: inGSettings["AgentDict"][lAgentDictItemKeyTurple] = SettingsTemplate.__AgentDictItemCreate__() lThisAgentDict = inGSettings["AgentDict"][lAgentDictItemKeyTurple] lThisAgentDict["ActivityList"].append(lActivityItemDict) # Return the result return lGUIDStr
[документация]def AgentActivityItemExists(inHostNameStr, inUserStr, inGUIDStr, inGSettings = None): """L+,W+: Выполнить проверку, что активность (ActivityItem) была отправлена на сторону Агента. :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inGUIDStr: ГУИД (GUID) активности (ActivityItem) :return: True - Активность присутствует ; False - Активность еще не была отправлена на сторону Агента """ # Check if GUID is exists in dict - has been recieved inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings # Main alg lAgentDictItemKeyTurple = (inHostNameStr.upper(),inUserStr.upper()) lResultBool = False if lAgentDictItemKeyTurple in inGSettings["AgentDict"]: for lActivityItem in inGSettings["AgentDict"][lAgentDictItemKeyTurple]["ActivityList"]: if inGUIDStr == lActivityItem.get("GUIDStr",None): lResultBool = True break return lResultBool
[документация]def AgentActivityItemReturnExists(inGUIDStr, inGSettings = None): """L+,W+: Выполнить проверку, что активность (ActivityItem) была выполнена на стороне Агента и результат был получен на стороне Оркестратора. :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inGUIDStr: ГУИД (GUID) активности (ActivityItem) :return: True - Активность присутствует; False - Активность еще не была выполнена на стороне Агента """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings # Check if GUID is exists in dict - has been recieved return inGUIDStr in inGSettings["AgentActivityReturnDict"]
[документация]def AgentActivityItemReturnGet(inGUIDStr, inCheckIntervalSecFloat = 0.5, inGSettings=None, inTimeoutSecFloat=None): """L+,W+: Ожидает появления результата по активности (ActivityItem). Возвращает результат выполнения активности. !ВНИМАНИЕ! Замораживает поток, пока не будет получен результат. !ВНИМАНИЕ! Запускать следует после того как будет инициализировано ядро Оркестратора (см. функцию OrchestratorInitWait), иначе будет инициирована ошибка. :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inGUIDStr: ГУИД (GUID) активности (ActivityItem) :param inCheckIntervalSecFloat: Интервал в секундах, с какой частотой выполнять проверку результата. По умолчанию 0.5 :param inTimeoutSecFloat: Время ожидания ответа. Если ответ не поступил, генерация исключения Exception. :return: Результат выполнения активности. !ВНИМАНИЕ! Возвращаются только то результаты, которые могут быть интерпретированы в JSON формате. """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lTimeStampSecFloat = time.time() #Check if Orchestrator has been initialized - else raise exception if Core.IsOrchestratorInitialized(inGSettings=inGSettings) == True: # Wait while result will not come here lLoopBool = True while lLoopBool: if not AgentActivityItemReturnExists(inGSettings=inGSettings, inGUIDStr=inGUIDStr): time.sleep(inCheckIntervalSecFloat) else: lLoopBool=False if (inTimeoutSecFloat is not None) and (time.time() - lTimeStampSecFloat) >= inTimeoutSecFloat: raise Exception(f"Orchestrator.AgentActivityItemReturnGet !ВНИМАНИЕ! ПРЕВЫШЕНО ВРЕМЯ ОЖИДАНИЯ ОТВЕТА") # Return the result return inGSettings["AgentActivityReturnDict"][inGUIDStr]["Return"] else: raise Exception(f"Orchestrator.AgentActivityItemReturnGet !ВНИМАНИЕ! ФУНКЦИЯ МОЖЕТ БЫТЬ ЗАПУЩЕНА ТОЛЬКО ПОСЛЕ ИНИЦИАЛИЗАЦИИ!")
[документация]def AgentOSCMD(inHostNameStr, inUserStr, inCMDStr, inRunAsyncBool=True, inSendOutputToOrchestratorLogsBool=True, inCMDEncodingStr="cp1251", inGSettings=None, inCaptureBool=True): """L+,W+: Отправка команды командной строки на сессию, где работает pyOpenRPA.Agent. Результат выполнения команды можно выводить в лог оркестратора. :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inCMDStr: Команда для исполнения на стороне сессии Агента :param inRunAsyncBool: True - Агент не ожидает окончания выполнения команды. !ВНИМАНИЕ! Логирование в такой ситуации будет невозможно; False - Агент ожидает окончания выполнения операции. :param inSendOutputToOrchestratorLogsBool: True - отправлять весь вывод от команды в логи Оркестратора; Flase - Не отправлять; Default True :param inCMDEncodingStr: Кодировка DOS среды, в которой исполняется команда. Если некорректно установить кодировку - русские символы будут испорчены. По умолчанию установлена "cp1251" :param inCaptureBool: True - не запускать приложение как отдельное. Результат выполнения команды будет выводиться в окне Агента (если окно Агента присутствует на экране). False - команда будет запущена в отдельном DOS окне. :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lActivityItemDict = { "Def":"OSCMD", # def alias (look pyOpeRPA.Agent gSettings["ProcessorDict"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{"inCMDStr":inCMDStr,"inRunAsyncBool":inRunAsyncBool, "inSendOutputToOrchestratorLogsBool": inSendOutputToOrchestratorLogsBool, "inCMDEncodingStr": inCMDEncodingStr, "inCaptureBool":inCaptureBool}, # 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) } #Send item in AgentDict for the futher data transmition return AgentActivityItemAdd(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inActivityItemDict=lActivityItemDict)
[документация]def AgentOSLogoff(inHostNameStr, inUserStr): """L+,W+: Выполнить операцию logoff на стороне пользователя. :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet """ inGSettings = GSettingsGet() # Set the global settings lCMDStr = "shutdown /l" if CrossOS.IS_LINUX_BOOL: lCMDStr="logout" lActivityItemDict = { "Def":"OSCMD", # def alias (look pyOpeRPA.Agent gSettings["ProcessorDict"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{"inCMDStr":lCMDStr,"inRunAsyncBool":False, "inSendOutputToOrchestratorLogsBool": True, "inCMDEncodingStr": "cp1251"}, # 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) } #Send item in AgentDict for the futher data transmition return AgentActivityItemAdd(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inActivityItemDict=lActivityItemDict)
[документация]def AgentOSFileSend(inHostNameStr, inUserStr, inOrchestratorFilePathStr, inAgentFilePathStr, inGSettings = None): """L+,W+: Отправить файл по адресу inOrchestratorFilePathStr со стороны Оркестратора и сохранить по адресу inAgentFilePathStr на стороне Агента. Поддерживает передачу крупных файлов (более 2-х Гб.). Функция является синхронной - не закончит свое выполнение, пока файл не будет передан полностью. !ВНИМАНИЕ - ПОТОКОБЕЗОПАСНАЯ! Вы можете вызвать эту функцию до инициализации ядра Оркестратора. Оркестратор добавит эту функцию в процессорную очередь на исполение. Если вам нужен результат функции, то необходимо сначала убедиться в том, что ядро Оркестратора было инициализированно (см. функцию OrchestratorInitWait). :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inOrchestratorFilePathStr: Полный путь к передаваемому файлу на стороне Оркестратора. :param inAgentFilePathStr: Полный путь к локации, в которую требуется сохранить передаваемый файл. :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings # Check thread if inGSettings["ServerDict"]["ServerThread"] is None: if inGSettings["Logger"]: inGSettings["Logger"].warning(f"AgentOSFileSend run before server init - activity will be append in the processor queue.") lResult = { "Def": AgentOSFileSend, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inHostNameStr":inHostNameStr, "inUserStr":inUserStr, "inOrchestratorFilePathStr":inOrchestratorFilePathStr, "inAgentFilePathStr": inAgentFilePathStr}, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lResult) else: # In processor - do execution lActivityItemCheckIntervalSecFloat = inGSettings["ServerDict"]["AgentFileChunkCheckIntervalSecFloat"] # Get the chunk limit lChunkByteSizeInt = inGSettings["ServerDict"]["AgentFileChunkBytesSizeInt"] lL = inGSettings.get("Logger",None) # Open the file and get the size (in bytes) lFile = open(inOrchestratorFilePathStr,"rb") lFileSizeBytesInt = lFile.seek(0,2) lFile.seek(0) #import pdb #pdb.set_trace() lChunkCountInt = math.ceil(lFileSizeBytesInt/lChunkByteSizeInt) if lL: lL.info(f"О2А: Старт передачи крупного бинарного файла по частям. Общее количество частей: {lChunkCountInt}, от (сторона оркестратора): {inOrchestratorFilePathStr}, на (сторона агента): {inAgentFilePathStr}") for lChunkNumberInt in range(lChunkCountInt): # Read chunk lFileChunkBytes = lFile.read(lChunkByteSizeInt) # Convert to base64 lFileChunkBase64Str = base64.b64encode(lFileChunkBytes).decode("utf-8") # Send chunk if lChunkNumberInt == 0: lActivityItemGUIDStr = AgentOSFileBinaryDataBase64StrCreate(inGSettings=inGSettings,inHostNameStr=inHostNameStr, inUserStr=inUserStr,inFilePathStr=inAgentFilePathStr, inFileDataBase64Str=lFileChunkBase64Str) else: lActivityItemGUIDStr = AgentOSFileBinaryDataBase64StrAppend(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inFilePathStr=inAgentFilePathStr, inFileDataBase64Str=lFileChunkBase64Str) # Wait for the activity will be deleted while AgentActivityItemExists(inGSettings=inGSettings,inHostNameStr=inHostNameStr,inUserStr=inUserStr,inGUIDStr=lActivityItemGUIDStr): time.sleep(lActivityItemCheckIntervalSecFloat) if lL: lL.debug( f"О2А: БИНАРНАЯ ОТПРАВКА: Номер текущей части: {lChunkNumberInt}") if lL: lL.info( f"О2А: БИНАРНАЯ ОТПРАВКА: Отправка завершена успешно") # Close the file lFile.close()
[документация]def AgentOSFileBinaryDataBytesCreate(inHostNameStr, inUserStr, inFilePathStr, inFileDataBytes, inGSettings=None): """L+,W+: Создать бинарный файл, который будет расположен по адресу inFilePathStr на стороне Агента с содержимым inFileDataBytes :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inFilePathStr: Полный путь к сохраняемому файлу на стороне Агента. :param inFileDataBytes: Строка байт (b'') для отправки в создаваемый файл на стороне Агента. :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lFileDataBase64Str = base64.b64encode(inFileDataBytes).decode("utf-8") lActivityItemDict = { "Def":"OSFileBinaryDataBase64StrCreate", # def alias (look pyOpeRPA.Agent gSettings["ProcessorDict"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{"inFilePathStr":inFilePathStr,"inFileDataBase64Str":lFileDataBase64Str}, # 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) } #Send item in AgentDict for the futher data transmition return AgentActivityItemAdd(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inActivityItemDict=lActivityItemDict)
[документация]def AgentOSFileBinaryDataBase64StrCreate(inHostNameStr, inUserStr, inFilePathStr, inFileDataBase64Str, inGSettings=None): """L+,W+: Создать бинарный файл, который будет расположен по адресу inFilePathStr на стороне Агента с содержимым, декодированным с формата base64: inFileDataBase64Str :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inFilePathStr: Полный путь к сохраняемому файлу на стороне Агента. :param inFileDataBase64Str: Строка в формате base64 для отправки в создаваемый файл на стороне Агента. :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lActivityItemDict = { "Def":"OSFileBinaryDataBase64StrCreate", # def alias (look pyOpeRPA.Agent gSettings["ProcessorDict"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{"inFilePathStr":inFilePathStr,"inFileDataBase64Str":inFileDataBase64Str}, # 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) } #Send item in AgentDict for the futher data transmition return AgentActivityItemAdd(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inActivityItemDict=lActivityItemDict)
[документация]def AgentOSFileBinaryDataBase64StrAppend(inHostNameStr, inUserStr, inFilePathStr, inFileDataBase64Str, inGSettings = None): """L+,W+: Добавить бинарную информацию в существующий бинарный файл, который будет расположен по адресу inFilePathStr на стороне Агента с содержимым, декодированным с формата base64: inFileDataBase64Str :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inFilePathStr: Полный путь к сохраняемому файлу на стороне Агента. :param inFileDataBase64Str: Строка в формате base64 для отправки в создаваемый файл на стороне Агента. :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lActivityItemDict = { "Def":"OSFileBinaryDataBase64StrAppend", # def alias (look pyOpeRPA.Agent gSettings["ProcessorDict"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{"inFilePathStr":inFilePathStr,"inFileDataBase64Str":inFileDataBase64Str}, # 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) } #Send item in AgentDict for the futher data transmition return AgentActivityItemAdd(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inActivityItemDict=lActivityItemDict)
# Send text file to Agent (string)
[документация]def AgentOSFileTextDataStrCreate(inHostNameStr, inUserStr, inFilePathStr, inFileDataStr, inEncodingStr = "utf-8",inGSettings=None): """L+,W+: Создать текстовый файл, который будет расположен по адресу inFilePathStr на стороне Агента с содержимым inFileDataStr в кодировке inEncodingStr :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inFilePathStr: Полный путь к сохраняемому файлу на стороне Агента. :param inFileDataStr: Строка для отправки в создаваемый файл на стороне Агента. :param inEncodingStr: Кодировка текстового файла. По умолчанию utf-8 :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lActivityItemDict = { "Def":"OSFileTextDataStrCreate", # def alias (look pyOpeRPA.Agent gSettings["ProcessorDict"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{"inFilePathStr":inFilePathStr,"inFileDataStr":inFileDataStr, "inEncodingStr": inEncodingStr}, # 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) } #Send item in AgentDict for the futher data transmition return AgentActivityItemAdd(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inActivityItemDict=lActivityItemDict)
[документация]def AgentOSFileBinaryDataBase64StrReceive(inHostNameStr, inUserStr, inFilePathStr, inGSettings = None): """L+,W+: Выполнить чтение бинарного файла и получить содержимое в формате base64 (строка) :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inFilePathStr: Путь к бинарному файлу на чтение на стороне Агента :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lActivityItemDict = { "Def":"OSFileBinaryDataBase64StrReceive", # def alias (look pyOpeRPA.Agent gSettings["ProcessorDict"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{"inFilePathStr":inFilePathStr}, # 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) } #Send item in AgentDict for the futher data transmition return AgentActivityItemAdd(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inActivityItemDict=lActivityItemDict)
[документация]def AgentOSFileBinaryDataReceive(inHostNameStr, inUserStr, inFilePathStr): """L+,W+: Чтение бинарного файла на стороне Агента по адресу inFilePathStr. !ВНИМАНИЕ - СИНХРОННАЯ! Функция не завершится, пока не будет получен результат чтения на стороне Агента. :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inFilePathStr: Путь к бинарному файлу, который требуется прочитать на стороне Агента :return: Строка байт (b'') - содержимое бинарного файла """ lFileDataBytes = None inGSettings = GSettingsGet() # Set the global settings # Check thread if OrchestratorIsInited() == False: if inGSettings["Logger"]: inGSettings["Logger"].warning(f"AgentOSFileBinaryDataReceive run before orc init - activity will be append in the processor queue.") lResult = { "Def": AgentOSFileBinaryDataReceive, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inHostNameStr":inHostNameStr, "inUserStr":inUserStr, "inFilePathStr":inFilePathStr}, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lResult) else: # In processor - do execution lActivityItemDict = { "Def":"OSFileBinaryDataBase64StrReceive", # def alias (look pyOpeRPA.Agent gSettings["ProcessorDict"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{"inFilePathStr":inFilePathStr}, # 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) } #Send item in AgentDict for the futher data transmition lGUIDStr = AgentActivityItemAdd(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inActivityItemDict=lActivityItemDict) lFileBase64Str = AgentActivityItemReturnGet(inGUIDStr=lGUIDStr) if lFileBase64Str is not None: lFileDataBytes = base64.b64decode(lFileBase64Str) return lFileDataBytes
[документация]def AgentOSFileTextDataStrReceive(inHostNameStr, inUserStr, inFilePathStr, inEncodingStr="utf-8", inGSettings = None): """L+,W+: Чтение текстового файла на стороне Агента по адресу inFilePathStr. По ГИУД с помощью функции AgentActivityItemReturnGet можно будет получить текстовую строку данных, которые были расположены в файле. !ВНИМАНИЕ - АСИНХРОННАЯ! Функция завершится сразу, не дожидаясь окончания выполнения операции на стороне Агента. :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inFilePathStr: Путь к бинарному файлу, который требуется прочитать на стороне Агента :param inEncodingStr: Кодировка текстового файла. По умолчанию utf-8 :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lActivityItemDict = { "Def":"OSFileTextDataStrReceive", # def alias (look pyOpeRPA.Agent gSettings["ProcessorDict"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{"inFilePathStr":inFilePathStr, "inEncodingStr": inEncodingStr}, # 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) } #Send item in AgentDict for the futher data transmition return AgentActivityItemAdd(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inActivityItemDict=lActivityItemDict)
[документация]def AgentProcessWOExeUpperUserListGet(inHostNameStr, inUserStr, inGSettings = None): """L-,W+: Получить список процессов, которые выполняется на сессии Агента. Все процессы фиксируются без постфикса .exe, а также в верхнем регистре. ПРИМЕР РЕЗУЛЬТАТА, КОТОРЫЙ МОЖНО ПОЛУЧИТЬ ПО ГУИД ЧЕРЕЗ ФУНКЦИЮ AgentActivityItemReturnGet: ["ORCHESTRATOR", "AGENT", "CHROME", "EXPLORER", ...] :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lActivityItemDict = { "Def":"ProcessWOExeUpperUserListGet", # def alias (look pyOpeRPA.Agent gSettings["ProcessorDict"]["AliasDefDict"]) "ArgList":[], # Args list "ArgDict":{}, # 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) } #Send item in AgentDict for the futher data transmition return AgentActivityItemAdd(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inActivityItemDict=lActivityItemDict)
# OS DEFS
[документация]def OSLogoff(): """L+,W+: Выполнить отключение сессии, на которой выполняется Оркестратор. :return: """ if CrossOS.IS_WINDOWS_BOOL: os.system("shutdown /l") elif CrossOS.IS_LINUX_BOOL: os.system("logout")
[документация]def OSCredentialsVerify(inUserStr, inPasswordStr, inDomainStr=""): ## """L+,W+: Выполнить верификацию доменного (локального) пользователя по паре логин/пароль :param inUserStr: Наименование пользователя :param inPasswordStr: Пароль :param inDomainStr: Домен. Если домена нет - не указывать или "" :return: True - Учетные данные верны; False - Учетные данные представлены некорректно """ if CrossOS.IS_WINDOWS_BOOL: try: hUser = win32security.LogonUser( inUserStr,inDomainStr, inPasswordStr, win32security.LOGON32_LOGON_NETWORK, win32security.LOGON32_PROVIDER_DEFAULT ) except win32security.error: return False else: return True if CrossOS.IS_LINUX_BOOL: return authenticate(inUserStr, inPasswordStr, service='login', encoding='utf-8', resetcred=True)
[документация]def OSRemotePCRestart(inHostStr, inForceBool=True, inLogger = None): """L-,W+: Отправить сигнал на удаленную перезагрузку операционной системы. !ВНИМАНИЕ! Перезапуск будет принят, если учетная запись имеет полномочия на перезапуск на соответсвующей машине. :param inHostStr: Имя хоста, который требуется перезагрузить :param inForceBool: True - принудительная перезагрузка; False - мягкая перезагрузка (дождаться окончания выполнения всех операций). По умолчанию True :param inLogger: Логгер, в который отправлять информацию о результате выполнения команды :return: """ if inLogger is None: inLogger = OrchestratorLoggerGet() lCMDStr = f"powershell -Command Restart-Computer -ComputerName {inHostStr}" if inForceBool == True: lCMDStr = lCMDStr + " -Force" OSCMD(inCMDStr=lCMDStr,inLogger=inLogger)
[документация]def OSRestart(inForceBool=True, inLogger = None): """L+,W+: Отправить сигнал на перезагрузку операционной системы. !ВНИМАНИЕ! Перезапуск будет принят, если учетная запись имеет полномочия на перезапуск на соответсвующей машине. :param inForceBool: True - принудительная перезагрузка; False - мягкая перезагрузка (дождаться окончания выполнения всех операций). По умолчанию True :param inLogger: Логгер, в который отправлять информацию о результате выполнения команды :return: """ if inLogger is None: inLogger = OrchestratorLoggerGet() if CrossOS.IS_WINDOWS_BOOL: lCMDStr = f"shutdown -r" if inForceBool == True: lCMDStr = lCMDStr + " -f" elif CrossOS.IS_LINUX_BOOL: lCMDStr = f"reboot" if inForceBool == True: lCMDStr = lCMDStr + " -f" OSCMD(inCMDStr=lCMDStr,inLogger=inLogger)
[документация]def OSCMD(inCMDStr, inRunAsyncBool=True, inLogger = None): """L-,W+: Отправить команду на выполнение на сессию, где выполняется Оркестратор. :param inCMDStr: Команда на отправку :param inRunAsyncBool: True - выполнить команду в асинхронном режиме (не дожидаться окончания выполнения программы и не захватывать результат выполнения); False - Ждать окончания выполнения и захватывать результат :param inLogger: Логгер, в который отправлять информацию о результате выполнения команды :return: Строка результата выполнения команды. Если inRunAsyncBool = False """ if inLogger is None: inLogger = OrchestratorLoggerGet() lResultStr = "" # New feature if inRunAsyncBool == True: inCMDStr = f"start {inCMDStr}" # Subdef to listen OS result def _CMDRunAndListenLogs(inCMDStr, inLogger): lResultStr = "" lOSCMDKeyStr = str(uuid.uuid4())[0:4].upper() # CROSS OS if CrossOS.IS_WINDOWS_BOOL: lCMDProcess = subprocess.Popen(f'cmd /c {inCMDStr}', stdout=subprocess.PIPE, stderr=subprocess.STDOUT, creationflags=CREATE_NEW_CONSOLE) if CrossOS.IS_LINUX_BOOL: lCMDProcess = subprocess.Popen(f'cmd /c {inCMDStr}', stdout=subprocess.PIPE, stderr=subprocess.STDOUT) if inLogger: lListenBool = True inLogger.info(f"{lOSCMDKeyStr}: # # # # Терминальная сессия запущена успешно # # # # ") inLogger.info(f"{lOSCMDKeyStr}: {inCMDStr}") while lListenBool: lOutputLineBytes = lCMDProcess.stdout.readline() if lOutputLineBytes == b"": lListenBool = False lStr = lOutputLineBytes.decode('cp866') if lStr.endswith("\n"): lStr = lStr[:-1] inLogger.info(f"{lOSCMDKeyStr}: {lStr}") lResultStr+=lStr inLogger.info(f"{lOSCMDKeyStr}: # # # # Терминальная сессия завершена успешно # # # # ") return lResultStr # New call if inRunAsyncBool: lThread = threading.Thread(target=_CMDRunAndListenLogs, kwargs={"inCMDStr":inCMDStr, "inLogger":inLogger}) lThread.setName("OSCMD_ACTIVITY") lThread.start() lResultStr="Список ActivityList был запущен в асинхронном режиме - захватить содержимое невозможно" else: lResultStr = _CMDRunAndListenLogs(inCMDStr=inCMDStr, inLogger=inLogger) #lCMDCode = "cmd /c " + inCMDStr #subprocess.Popen(lCMDCode) #lResultCMDRun = 1 # os.system(lCMDCode) return lResultStr
[документация]def OrchestratorRestart(inGSettings=None): """L+,W+: Перезапуск Оркестратора с сохранением информации о запущенных RDP сессиях. :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings OrchestratorSessionSave(inGSettings=inGSettings) # Dump RDP List in file json if inGSettings is not None: lL = inGSettings["Logger"] if lL: lL.info(f"Выполнить перезапуск оркестратора") # Restart session if CrossOS.IS_WINDOWS_BOOL: os.execl(sys.executable, f'"{os.path.abspath(__file__)}"', *sys.argv) if CrossOS.IS_LINUX_BOOL: os.execl(sys.executable, sys.executable, *sys.argv) sys.exit(0)
[документация]def OrchestratorLoggerGet() -> logging.Logger: """L+,W+: Получить логгер Оркестратора :return: Логгер """ return GSettingsGet().get("Logger",None)
[документация]def OrchestratorScheduleGet() -> schedule: """L+,W+: Базовый объект расписания, который можно использовать для запуска / остановки роботов. Подробнее про объект schedule и его примеры использования см. по адресу: schedule.readthedocs.io .. code-block:: python # Однопоточный schedule Orchestrator.OrchestratorScheduleGet().every(5).seconds.do(lProcess.StatusCheckStart) #Многопоточный schedule. cм. описание Orchestrator.OrchestratorThreadStart Orchestrator.OrchestratorScheduleGet().every(5).seconds.do(Orchestrator.OrchestratorThreadStart,lProcess.StatusCheckStart) :return: schedule объект """ if GSettingsGet().get("SchedulerDict",{}).get("Schedule",None) is None: GSettingsGet()["SchedulerDict"]["Schedule"]=schedule return GSettingsGet().get("SchedulerDict",{}).get("Schedule",None)
[документация]def OrchestratorThreadStart(inDef, *inArgList, **inArgDict): """L+,W+: Запустить функцию в отдельном потоке. В таком случае получить результат выполнения функции можно только через общие переменные. (Например через GSettings) :param inDef: Python функция :param inArgList: Список неименованных аргументов функции inDef :param inArgDict: Словарь именованных аргументов функции inDef :return: threading.Thread экземпляр """ lDefThread = threading.Thread(target=inDef,args=inArgList,kwargs=inArgDict) lDefThread.setName(f"ORCHESTRATOR_THREAD_{str(inDef).upper()}") lDefThread.start() return lDefThread
[документация]def OrchestratorIsAdmin(): """L+,W+: Проверить, запущен ли Оркестратор с правами администратора. Права администратора нужны Оркестратору только для контроля графической сессии, на которой он запущен. Если эти права выделить индивидуально, то права администратора будут необязательны. :return: True - Запущен с правами администратора; False - Не запущен с правами администратора """ if CrossOS.IS_WINDOWS_BOOL: try: return ctypes.windll.shell32.IsUserAnAdmin() except: return False elif CrossOS.IS_LINUX_BOOL: return os.getuid()==0 else: return True
[документация]def OrchestratorIsCredentialsAsk(): """L+,W+: Проверить, активирована ли авторизация при переходе к Оркестратору. :return: True - Активирована; False - Деактивирована """ inGSettings = GSettingsGet() return inGSettings["ServerDict"]["AccessUsers"]["FlagCredentialsAsk"]
[документация]def OrchestratorIsInited() -> bool: """L+,W+: Проверить, было ли проинициализировано ядро Оркестратора :return: True - Ядро Оркестратора было проинициализировано; False - Требуется время на инициализацию :rtype: bool """ return Core.IsOrchestratorInitialized(inGSettings=GSettingsGet())
[документация]def OrchestratorInitWait() -> None: """L+,W+: Ожидать инициализацию ядра Оркестратора !ВНИМАНИЕ!: НИ В КОЕМ СЛУЧАЕ НЕ ЗАПУСКАТЬ ЭТУ ФУНКЦИЮ В ОСНОВНОМ ПОТОКЕ, ГДЕ ПРОИСХОДИТ ИНИЦИАЛИЗАЦИЯ ЯДРА ОРКЕСТРАТОРА - ВОЗНИКНЕТ ВЕЧНЫЙ ЦИКЛ """ lIntervalSecFloat = 0.5 while not OrchestratorIsInited(): time.sleep(lIntervalSecFloat)
[документация]def OrchestratorRerunAsAdmin(): """L-,W+: Перезапустить Оркестратор с правами локального администратора. Права администратора нужны Оркестратору только для контроля графической сессии, на которой он запущен. Если эти права выделить индивидуально, то права администратора будут необязательны. :return: True - Запущен с правами администратора; False - Не запущен с правами администратора """ if not OrchestratorIsAdmin(): if "PYTHON_CONFIGURE" in os.environ: # Re-run the program with admin rights ctypes.windll.shell32.ShellExecuteW(None, "runas", "cmd", " ".join(["/C", "cd /d", f'"{os.getcwd()}"', "&&",f'"{os.environ["PYTHON_CONFIGURE"]}"',sys.executable]+sys.argv), None, 1) else: # Re-run the program with admin rights ctypes.windll.shell32.ShellExecuteW(None, "runas", "cmd", " ".join(["/C", "cd /d", f'"{os.getcwd()}"', "&&", sys.executable,sys.argv]), None, 1) else: print(f"Уже запущено с правами администратора!")
[документация]def OrchestratorPySearchInit(inGlobPatternStr, inDefStr = None, inDefArgNameGSettingsStr = None, inAsyncInitBool = False, inPackageLevelInt = 0): """L+,W+: Выполнить поиск и инициализацию пользовательских .py файлов в Оркестраторе (например панелей управления роботов) Добавляет инициализированный модуль в пространство sys.modules как imported (имя модуля = имя файла без расширения). ВНИМАНИЕ! ПРЕОБРАЗУЕТ ПУТИ МЕЖДУ WINDOWS И LINUX НОТАЦИЯМИ .. code-block:: python # ВАРИАНТ ИСПОЛЬЗОВАНИЯ 1 (инициализация модуля py без вызова каких-либо функций внутри) # автоинициализация всех .py файлов, с префиксом CP_, которые расположены в папке ControlPanel Orchestrator.OrchestratorPySearchInit(inGlobPatternStr="ControlPanel\\CP_*.py") # ВАРИАНТ ИСПОЛЬЗОВАНИЯ 2 (инициализация модуля py с вызовом функции внутри) - преимущественно для обратной совместимости старых версий панелей управления < 1.2.7 # автоинициализация всех .py файлов, с префиксом CP_, которые расположены в папке ControlPanel Orchestrator.OrchestratorPySearchInit(inGlobPatternStr="ControlPanel\\CP_*.py", inDefStr="SettingsUpdate", inDefArgNameGSettingsStr="inGSettings") # ДЛЯ СПРАВКИ & ИСТОРИИ: Код выше позволил отказаться от блока кода ниже для каждого .py файла ## !!! For Relative import !!! CP Version Check try: sys.path.insert(0,os.path.abspath(os.path.join(r""))) from ControlPanel import CP_VersionCheck CP_VersionCheck.SettingsUpdate(inGSettings=gSettings) except Exception as e: gSettings["Logger"].exception(f"Exception when init CP. See below.") :param inGlobPatternStr: Пример "..\\*\\*\\*_CP*.py" :param inDefStr: ОПЦИОНАЛЬНО Строковое наименование функции. Преимущественно для обратной совместимости :param inDefArgNameGSettingsStr: ОПЦИОНАЛЬНО Наименование аргумента, в который требуется передать GSettings (если необходимо) :param inAsyncInitBool: ОПЦИОНАЛЬНО True - Инициализация py модулей в отдельных параллельных потоках - псевдопараллельное выполнение. False - последовательная инициализация :param inPackageLevelInt: ОПЦИОНАЛЬНО Уровень вложенности модуля в пакет. По умолчанию 0 (не является модулем пакета) :return: Сведения об инициализированных модулях в структуре: { "ModuleNameStr":{"PyPathStr": "", "Module": ...}, ...} """ inGlobPatternStr = CrossOS.PathStr(inGlobPatternStr) # # # # # # # # def __execute__(inResultDict, inPyPathItemStr, inDefStr = None, inDefArgNameGSettingsStr = None, inPackageLevelInt=0): try: lPyPathItemStr = inPyPathItemStr CrossOS.PathSplitList(inPathStr=lPyPathItemStr)[-1*inPackageLevelInt-1:-1] lModuleNameStr = os.path.basename(lPyPathItemStr)[0:-3] if inPackageLevelInt==0: lTechSpecification = importlib.util.spec_from_file_location(lModuleNameStr, lPyPathItemStr) lTechModuleFromSpec = importlib.util.module_from_spec(lTechSpecification) sys.modules[lModuleNameStr] = lTechModuleFromSpec # Add initialized module in sys - python will not init this module enought lTechSpecificationModuleLoader = lTechSpecification.loader.exec_module(lTechModuleFromSpec) else: lPrePackagePathStr=CrossOS.PathStr("\\".join(CrossOS.PathSplitList(inPathStr=lPyPathItemStr)[:-1-inPackageLevelInt])) lPackageNameStr=".".join(CrossOS.PathSplitList(inPathStr=lPyPathItemStr)[-inPackageLevelInt-1:-1]) lModuleNameStr=os.path.basename(lPyPathItemStr)[0:-3] sys.path.insert(0,lPrePackagePathStr) lTechSpecification = importlib.import_module(f".{lModuleNameStr}", lPackageNameStr) lTechModuleFromSpec=sys.modules[f"{lPackageNameStr}.{lModuleNameStr}"] lModuleNameStr= f"{lPackageNameStr}.{lModuleNameStr}" lItemDict = {"ModuleNameStr": lModuleNameStr, "PyPathStr": lPyPathItemStr, "Module": lTechModuleFromSpec} if lL: lL.info(f"Модуль .py {lModuleNameStr} был успешно инициализирован. Путь к файлу: {lPyPathItemStr}") inResultDict[lModuleNameStr]=lItemDict # Backward compatibility to call def with gsettings when init if inDefStr is not None and inDefStr != "": lDef = getattr(lTechModuleFromSpec, inDefStr) lArgDict = {} if inDefArgNameGSettingsStr is not None and inDefArgNameGSettingsStr != "": lArgDict = {inDefArgNameGSettingsStr:GSettingsGet()} lDef(**lArgDict) except Exception as e: if lL: lL.exception(f"Ошибка при инициализации .py файла: {os.path.abspath(lPyPathItemStr)}") # # # # # # # # lResultDict = {} inGlobPatternStr = CrossOS.PathStr(inPathStr=inGlobPatternStr) lPyPathStrList = glob.glob(inGlobPatternStr) # get the file list lL = OrchestratorLoggerGet() # get the logger for lPyPathItemStr in lPyPathStrList: if inAsyncInitBool == True: # ASYNC EXECUTION lThreadInit = threading.Thread(target=__execute__,kwargs={ "inResultDict":lResultDict, "inPyPathItemStr": lPyPathItemStr, "inDefStr": inDefStr, "inDefArgNameGSettingsStr": inDefArgNameGSettingsStr, "inPackageLevelInt":inPackageLevelInt}, daemon=True) lThreadInit.setName("PY_SEARCH_MODULE_INIT") lThreadInit.start() else: # SYNC EXECUTION __execute__(inResultDict=lResultDict, inPyPathItemStr=lPyPathItemStr, inDefStr = inDefStr, inDefArgNameGSettingsStr = inDefArgNameGSettingsStr, inPackageLevelInt=inPackageLevelInt) return lResultDict
[документация]def OrchestratorSessionSave(inGSettings=None): """L+,W+: Сохранить состояние Оркестратора (для дальнейшего восстановления в случае перезапуска): - RDP сессий, которые контролирует Оркестратор - Хранилища DataStorage в глобальном словаре настроек GSettings. DataStorage поддерживает хранение объектов Python (до версии 1.2.7) _SessionLast_GSettings.pickle (binary) (начиная с версии 1.2.7) _SessionLast_RDPList.json (encoding = "utf-8") _SessionLast_StorageDict.pickle (binary) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lL = inGSettings["Logger"] try: # Dump RDP List in file json #lFile = open("_SessionLast_RDPList.json", "w", encoding="utf-8") #lFile.write(json.dumps(inGSettings["RobotRDPActive"]["RDPList"])) # dump json to file #lFile.close() # Close the file #if inGSettings is not None: # if lL: lL.info( # f"Orchestrator has dump the RDP list before the restart.") ## _SessionLast_StorageDict.pickle (binary) #if "StorageDict" in inGSettings: # with open('_SessionLast_StorageDict.pickle', 'wb') as lFile: # pickle.dump(inGSettings["StorageDict"], lFile) # if lL: lL.info( # f"Orchestrator has dump the StorageDict before the restart.") #SessionLast lDumpDict = {"StorageDict":inGSettings["StorageDict"], "ManagersProcessDict":inGSettings["ManagersProcessDict"], "RobotRDPActive":{"RDPList": inGSettings["RobotRDPActive"]["RDPList"]}} with open('_SessionLast_GSettings.pickle', 'wb') as lFile: pickle.dump(lDumpDict, lFile) if lL: lL.info( f"Оркестратор сохранил в рабочую директорию список РДП соединений и словарь DataStorage. При следующем запуске оркестратора информация востановится с диска") except Exception as e: if lL: lL.exception(f"Произошла ошибка при сохранении данных в файл") return True
[документация]def OrchestratorSessionRestore(inGSettings=None): """L+,W+: Восстановить состояние Оркестратора, если ранее состояние Оркестратора было сохранено с помощью функции OrchestratorSessionSave: - RDP сессий, которые контролирует Оркестратор - Хранилища DataStorage в глобальном словаре настроек GSettings. DataStorage поддерживает хранение объектов Python (до версии 1.2.7) _SessionLast_GSettings.pickle (binary) (начиная с версии 1.2.7) _SessionLast_RDPList.json (encoding = "utf-8") _SessionLast_StorageDict.pickle (binary) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lL = inGSettings.get("Logger",None) # _SessionLast_RDPList.json (encoding = "utf-8") if os.path.exists("_SessionLast_RDPList.json"): lFile = open("_SessionLast_RDPList.json", "r", encoding="utf-8") lSessionLastRDPList = json.loads(lFile.read()) lFile.close() # Close the file os.remove("_SessionLast_RDPList.json") # remove the temp file inGSettings["RobotRDPActive"]["RDPList"] = lSessionLastRDPList # Set the last session dict if lL: lL.warning(f"Список РДП был восстановлен из прошлой сессии оркестратора") # _SessionLast_StorageDict.pickle (binary) if os.path.exists("_SessionLast_StorageDict.pickle"): if "StorageDict" not in inGSettings: inGSettings["StorageDict"] = {} with open('_SessionLast_StorageDict.pickle', 'rb') as lFile: lStorageDictDumpDict = pickle.load(lFile) Dictionary.MergeNoException(in1Dict=inGSettings["StorageDict"], in2Dict=lStorageDictDumpDict) # Merge dict 2 into dict 1 if lL: lL.warning(f"Словарь StorageDict был восстановлен из прошлой сессии оркестратора") os.remove("_SessionLast_StorageDict.pickle") # remove the temp file # _SessionLast_Gsettings.pickle (binary) if os.path.exists("_SessionLast_GSettings.pickle"): if "StorageDict" not in inGSettings: inGSettings["StorageDict"] = {} if "ManagersProcessDict" not in inGSettings: inGSettings["ManagersProcessDict"] = {} with open('_SessionLast_GSettings.pickle', 'rb') as lFile: lStorageDictDumpDict = pickle.load(lFile) Dictionary.MergeNoException(in1Dict=inGSettings, in2Dict=lStorageDictDumpDict) # Merge dict 2 into dict 1 if lL: lL.warning(f"Словарь GSettings был восстановлен из прошлой сессии оркестратора") os.remove("_SessionLast_GSettings.pickle") # remove the temp file
[документация]def UACKeyListCheck(inRequest, inRoleKeyList) -> bool: """L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.2, см. WebUserUACCheck] Проверить права доступа для пользователя запроса по списку ключей до права. .. code-block:: python # ВАРИАНТ ИСПОЛЬЗОВАНИЯ 1 (инициализация модуля py без вызова каких-либо функций внутри) # автоинициализация всех .py файлов, с префиксом CP_, которые расположены в папке ControlPanel Orchestrator.UACKeyListCheck(inRequest=lRequest, inRoleKeyList=["ROBOT1","DISPLAY_DASHBOARD"]) :param inRequest: Экземпляр request (from http.server import BaseHTTPRequestHandler) :param inRoleKeyList: Список ключей, права доступа к которому требуется проверить :return: True - Пользователь обладает соответствующим правом; False - Пользователь не обладает соответствующим правом """ return inRequest.UACClientCheck(inRoleKeyList=inRoleKeyList)
[документация]def UACUserDictGet(inRequest) -> dict: """L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.2, см. WebUserUACHierarchyGet] Вернуть UAC (User Access Control) словарь доступов для пользователя, который отправил запрос. Пустой словарь - супердоступ (доступ ко всему) :param inRequest: Экземпляр request (from http.server import BaseHTTPRequestHandler) :return: Словарь доступов пользователя. Пустой словарь - супердоступ (доступ ко всему) """ return inRequest.UserRoleHierarchyGet() # get the Hierarchy
[документация]def UACUpdate(inADLoginStr, inADStr="", inADIsDefaultBool=True, inURLList=None, inRoleHierarchyAllowedDict=None, inGSettings = None): """L+,W+: Дообогащение словаря доступа UAC пользователя inADStr\\inADLoginStr. Если ранее словарь не устанавливался, то по умолчанию он {}. Далее идет дообогащение теми ключами, которые присутствуют в inRoleHierarchyAllowedDict :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inADLoginStr: Логин пользователя :param inADStr: Домен пользователя. Если пусто - локальный компьютер или домен по-умолчанию, который настроен в ОС :param inADIsDefaultBool: True - домен настроен по умолчанию; False - домен требуется обязательно указывать :param inURLList: Список разрешенных URL для пользователя. Для добавления URL рекомендуем использовать функции WebURLConnectDef, WebURLConnectFile, WebURLConnectFolder Формат: { "Method": "GET" || "POST", "URL": "/GetScreenshot", "MatchType": "BeginWith" || "Equal" || "EqualCase" || "Contains" || "EqualNoParam", "ResponseDefRequestGlobal": Функция python || "ResponseFilePath": Путь к файлу || "ResponseFolderPath": Путь к папке, в которой искать файлы, "ResponseContentType": пример MIME "image/png", "UACBool":False - не выполнять проверку прав доступа по запросу, "UseCacheBool": True - кэшировать ответ}, :param inRoleHierarchyAllowedDict: Словарь доступа пользователя UAC. Пустой словарь - полный доступ """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lUserTurple = (inADStr.upper(),inADLoginStr.upper()) # Create turple key for inGSettings["ServerDict"]["AccessUsers"]["RuleDomainUserDict"] if inURLList is None: inURLList = [] # Check if None if inRoleHierarchyAllowedDict is None: inRoleHierarchyAllowedDict = {} # Check if None # Get the old URLList try: inURLList += inGSettings["ServerDict"]["AccessUsers"]["RuleDomainUserDict"][lUserTurple]["MethodMatchURLBeforeList"] except: pass # Check RoleHierarchyAllowedDict in gSettings for the old role hierarchy - include in result. if lUserTurple in inGSettings["ServerDict"]["AccessUsers"]["RuleDomainUserDict"] and "RoleHierarchyAllowedDict" in inGSettings["ServerDict"]["AccessUsers"]["RuleDomainUserDict"][lUserTurple]: lRoleHierarchyAllowedOLDDict = inGSettings["ServerDict"]["AccessUsers"]["RuleDomainUserDict"][lUserTurple]["RoleHierarchyAllowedDict"] Dictionary.Merge(in1Dict=inRoleHierarchyAllowedDict, in2Dict=lRoleHierarchyAllowedOLDDict) # Merge dict 2 into dict 1 # Create Access item lRuleDomainUserDict = { "MethodMatchURLBeforeList": inURLList, "RoleHierarchyAllowedDict": inRoleHierarchyAllowedDict } # Case add domain + user inGSettings["ServerDict"]["AccessUsers"]["RuleDomainUserDict"].update({(inADStr.upper(),inADLoginStr.upper()):lRuleDomainUserDict}) if inADIsDefaultBool: # Case add default domain + user inGSettings["ServerDict"]["AccessUsers"]["RuleDomainUserDict"].update({("",inADLoginStr.upper()):lRuleDomainUserDict})
[документация]def UACSuperTokenUpdate(inSuperTokenStr, inGSettings=None): """L+,W+: Добавить супертокен (полный доступ). Супертокены используются для подключения к Оркестратору по API :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inSuperTokenStr: Кодовая строковая комбинация, которая будет предоставлять доступ роботу / агенту для взаимодействия с Оркестратором по API """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lLoginStr = "SUPERTOKEN" UACUpdate(inGSettings=inGSettings, inADLoginStr=lLoginStr) inGSettings["ServerDict"]["AccessUsers"]["AuthTokensDict"].update( {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 # # # # # # # # # # # # # # # # # # # # # # #
[документация]def WebUserLoginGet(inAuthTokenStr: str=None) -> str: """L+,W+: Получить логин авторизованного пользователя. Если авторизация не производилась - вернуть None :param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен) :type inAuthTokenStr: str, опционально :return: Логин пользователя :rtype: str """ isCredentialAsk = OrchestratorIsCredentialsAsk() if isCredentialAsk: if inAuthTokenStr is None: raise ConnectionError("Не удается получить токен для авторизации") else: if inAuthTokenStr is None: return None inGS = GSettingsGet() # Get the global settings return inGS.get("ServerDict", {}).get("AccessUsers", {}).get("AuthTokensDict", {}).get(inAuthTokenStr, {}).get("User", None)
[документация]def WebUserDomainGet(inAuthTokenStr: str=None) -> str: """L+,W+: Получить домен авторизованного пользователя. Если авторизация не производилась - вернуть None :param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен) :type inAuthTokenStr: str, опционально :return: Домен пользователя :rtype: str """ isCredentialAsk = OrchestratorIsCredentialsAsk() if isCredentialAsk: if inAuthTokenStr is None: raise ConnectionError("Не удается получить токен для авторизации") else: 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(inAuthTokenStr=None): """L+,W+: Информация о пользователе, который отправил HTTP запрос. :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя :return: Сведения в формате {"DomainUpperStr": "PYOPENRPA", "UserNameUpperStr": "IVAN.MASLOV"} """ try: lResultDict = { "DomainUpperStr": WebUserDomainGet(inAuthTokenStr=inAuthTokenStr).upper(), "UserNameUpperStr": WebUserLoginGet(inAuthTokenStr=inAuthTokenStr).upper() } return lResultDict except Exception as e: return {"DomainUpperStr": None, "UserNameUpperStr": None}
[документация]def WebUserIsSuperToken(inAuthTokenStr: str=None): """L+,W+: [ИЗМЕНЕНИЕ В 1.3.1] Проверить, авторизован ли HTTP запрос с помощью супер токена (токен, который не истекает). :param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен) :type inAuthTokenStr: str, опционально :return: True - является супертокеном; False - не является супертокеном; None - авторизация не производилась """ if inAuthTokenStr is None: return None inGSettings = GSettingsGet() # Get the global settings lIsSuperTokenBool = False # Get Flag is supertoken (True|False) lIsSuperTokenBool = inGSettings.get("ServerDict", {}).get("AccessUsers", {}).get("AuthTokensDict", {}).get(inAuthTokenStr, {}).get("FlagDoNotExpire", False) return lIsSuperTokenBool
[документация]def WebUserUACHierarchyGet(inAuthTokenStr: str=None) -> dict: """L+,W+: [ИЗМЕНЕНИЕ В 1.3.1] Вернуть словарь доступа UAC в отношении пользователя, который выполнил HTTP запрос inRequest :param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен) :type inAuthTokenStr: str, опционально :return: UAC словарь доступа или {}, что означает полный доступ """ isCredentialAsk = OrchestratorIsCredentialsAsk() if isCredentialAsk: if inAuthTokenStr is None: raise ConnectionError("Не удается получить токен для авторизации") else: if inAuthTokenStr is None: return {} lDomainUpperStr = WebUserDomainGet(inAuthTokenStr=inAuthTokenStr).upper() lUserUpperStr = WebUserLoginGet(inAuthTokenStr=inAuthTokenStr).upper() if lUserUpperStr is None: return {} else: return GSettingsGet().get("ServerDict", {}).get("AccessUsers", {}).get("RuleDomainUserDict", {}).get((lDomainUpperStr, lUserUpperStr), {}).get("RoleHierarchyAllowedDict", {})
[документация]def WebUserUACCheck(inAuthTokenStr:str=None, inKeyList:list=None) -> bool: """L+,W+: Проверить UAC доступ списка ключей для пользователя :param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен) :type inAuthTokenStr: str, опционально :return: True - доступ имеется, False - доступа нет :rtype: bool """ isCredentialAsk = OrchestratorIsCredentialsAsk() # Если авторизации не происходило - супердоступ if isCredentialAsk: if inAuthTokenStr is None: return False else: if inAuthTokenStr is None: return True lResult = True # Init flag lRoleHierarchyDict = WebUserUACHierarchyGet(inAuthTokenStr=inAuthTokenStr) # get the Hierarchy # Try to get value from key list lKeyValue = lRoleHierarchyDict # Init the base for lItem in inKeyList: if type(lKeyValue) is dict: if lItem in lKeyValue: # Has key lKeyValue = lKeyValue[lItem] # Get the value and go to the next loop iteration else: # Else branch - true or false if len(lKeyValue)>0: # False - if Dict has some elements lResult = False # Set the False Flag else: lResult = True # Set the True flag break # Stop the loop else: # Has element with no detalization - return True lResult = True # Set the flag break # Close the loop return lResult # Return the result
[документация]def WebURLIndexChange(inURLIndexStr:str ="/"): """L+,W+: Изменить адрес главной страницы Оркестратора. По умолчанию '/' :param inURLIndexStr: Новый адрес главной страницы Оркестратора. :type inURLIndexStr: str, опционально """ GSettingsGet()["ServerDict"]["URLIndexStr"] = inURLIndexStr
[документация]def WebURLConnectDef(inMethodStr, inURLStr, inMatchTypeStr, inDef, inContentTypeStr="application/octet-stream", inGSettings = None, inUACBool = None): """L+,W+: Подключить функцию Python к URL. :param inMethodStr: Метод доступа по URL "GET" || "POST" :param inURLStr: URL адрес. Пример "/index" :param inMatchTypeStr: Тип соответсвия строки URL с inURLStr: "BeginWith" || "Contains" || "Equal" || "EqualCase" || "EqualNoParam" :param inDef: Функция Python. Допускаются функции, которые принимают следующие наборы параметров: 2:[inRequest, inGSettings], 1: [inRequest], 0: [] :param inContentTypeStr: МИМЕ тип. По умолчанию: "application/octet-stream" :param inUACBool: True - Выполнять проверку прав доступа пользователя перед отправкой ответа; False - не выполнять проверку прав доступа пользователя :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lURLItemDict = { "Method": inMethodStr.upper(), "URL": inURLStr, # URL of the request "MatchType": inMatchTypeStr, # "BeginWith|Contains|Equal|EqualCase", # "ResponseFilePath": "", #Absolute or relative path #"ResponseFolderPath": "C:\Abs\Archive\scopeSrcUL\OpenRPA\Orchestrator\Settings", # Absolute or relative path "ResponseContentType": inContentTypeStr, #HTTP Content-type "ResponseDefRequestGlobal": inDef, #Function with str result "UACBool": inUACBool } inGSettings["ServerDict"]["URLList"].append(lURLItemDict) Server.BCURLUpdate()
[документация]def WebURLConnectFolder(inMethodStr, inURLStr, inMatchTypeStr, inFolderPathStr, inExceptionFlagBool=False, inGSettings = None, inUACBool = None, inUseCacheBool= False): """L+,W+: Подключить папку к URL. :param inMethodStr: Метод доступа по URL "GET" || "POST" :param inURLStr: URL адрес. Пример "/index" :param inMatchTypeStr: Тип соответсвия строки URL с inURLStr: "BeginWith" || "Contains" || "Equal" || "EqualCase" || "EqualNoParam" :param inFolderPathStr: Путь к папке на диске, в которой искать файл и возвращать пользователю по HTTP :param inExceptionFlagBool: Флаг на обработку ошибки. True - показывать ошибку в терминале (остановка инициализации), False - не показывать :param inUACBool: True - Выполнять проверку прав доступа пользователя перед отправкой ответа; False - не выполнять проверку прав доступа пользователя :param inUseCacheBool: True - выполнить кэширование страницы, чтобы в следющих запросах открыть быстрее; False - не кэшировать :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings # Check if last symbol is "/" - append if not exist lFolderPathStr = os.path.abspath(inFolderPathStr) if lFolderPathStr[-1]!="/":lFolderPathStr+="/" # Prepare URLItem lURLItemDict = { "Method": inMethodStr.upper(), "URL": inURLStr, # URL of the request "MatchType": inMatchTypeStr, # "BeginWith|Contains|Equal|EqualCase", # "ResponseFilePath": "", #Absolute or relative path "ResponseFolderPath": lFolderPathStr, # Absolute or relative path "ResponseContentType": None, #HTTP Content-type #"ResponseDefRequestGlobal": inDef #Function with str result "UACBool": inUACBool, "UseCacheBool": inUseCacheBool } inGSettings["ServerDict"]["URLList"].append(lURLItemDict) Server.BCURLUpdate(inExceptionFlagBool)
[документация]def WebURLConnectFile(inMethodStr, inURLStr, inMatchTypeStr, inFilePathStr, inContentTypeStr=None, inGSettings = None, inUACBool = None, inUseCacheBool = False): """L+,W+: Подключить файл к URL. :param inMethodStr: Метод доступа по URL "GET" || "POST" :param inURLStr: URL адрес. Пример "/index" :param inMatchTypeStr: Тип соответсвия строки URL с inURLStr: "BeginWith" || "Contains" || "Equal" || "EqualCase" || "EqualNoParam" :param inFilePathStr: Путь к файлу на диске, который возвращать пользователю по HTTP :param inContentTypeStr: МИМЕ тип. Если None - выполнить автоматическое определение :param inUACBool: True - Выполнять проверку прав доступа пользователя перед отправкой ответа; False - не выполнять проверку прав доступа пользователя :param inUseCacheBool: True - выполнить кэширование страницы, чтобы в следющих запросах открыть быстрее; False - не кэшировать :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lURLItemDict = { "Method": inMethodStr.upper(), "URL": inURLStr, # URL of the request "MatchType": inMatchTypeStr, # "BeginWith|Contains|Equal|EqualCase", "ResponseFilePath": os.path.abspath(inFilePathStr), #Absolute or relative path #"ResponseFolderPath": os.path.abspath(inFilePathStr), # Absolute or relative path "ResponseContentType": inContentTypeStr, #HTTP Content-type #"ResponseDefRequestGlobal": inDef #Function with str result "UACBool":inUACBool, "UseCacheBool": inUseCacheBool } inGSettings["ServerDict"]["URLList"].append(lURLItemDict) Server.BCURLUpdate()
[документация]def WebListenCreate(inServerKeyStr="Default", inAddressStr="0.0.0.0", inPortInt=1024, inCertFilePEMPathStr=None, inKeyFilePathStr=None, inGSettings = None): """L+,W+: Настроить веб-сервер Оркестратора. :param inAddressStr: IP адрес для прослушивания. Если "0.0.0.0", то прослушивать запросы со всех сетевых карт. Если "127.0.0.1", то слушать запросы только с той ОС, на которой работает Оркестратор :param inPortInt: Номер порта для прослушивания. Если HTTP - 80; Если HTTPS - 443. По умолчанию 1024 (Связано с тем, что в линукс можно устанавливать порты выше 1000). Допускается установка других портов :param inCertFilePEMPathStr: Путь файлу сертификата, сгенерированного в .pem (base64) формате. Обязателен при использовании защищенного HTTPS/SSL соединения. :param inKeyFilePathStr: Путь к файлу закрытого ключа в base64 формате :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: """ inCertFilePEMPathStr=CrossOS.PathStr(inPathStr=inCertFilePEMPathStr) inKeyFilePathStr=CrossOS.PathStr(inPathStr=inKeyFilePathStr) inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings inGSettings["ServerDict"]["ListenDict"][inServerKeyStr]={ "AddressStr":inAddressStr, "PortInt":inPortInt, "CertFilePEMPathStr":inCertFilePEMPathStr, "KeyFilePathStr":inKeyFilePathStr, "ServerInstance": None }
[документация]def WebCPUpdate(inCPKeyStr, inHTMLRenderDef=None, inJSONGeneratorDef=None, inJSInitGeneratorDef=None, inGSettings = None): """L+,W+: Добавить панель управления робота в Оркестратор. Для панели управления открыт доступ для использования функции-генератора HTML, генератора JSON данных, генератора JS кода. :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inCPKeyStr: Текстовый ключ панели управления. Ключ для каждой панели управления должен быть уникальным! :param inHTMLRenderDef: Функция Python для генерации HTML кода. Для использования Jinja2 шаблонов HTML см. pyOpenRPA.Orchestrator.Managers.ControlPanel :param inJSONGeneratorDef: Функция Python для генерации JSON контейнера для отправки на клиентскую часть Оркестратора :param inJSInitGeneratorDef: Функция Python для генерации JS кода для отправки на клиентскую часть Оркестратора """ lCPManager = Managers.ControlPanel(inControlPanelNameStr=inCPKeyStr, inRefreshHTMLJinja2TemplatePathStr=None) # CASE HTMLRender if inHTMLRenderDef is not None: lCPManager.mBackwardCompatibilityHTMLDef = inHTMLRenderDef # CASE JSONGenerator if inJSONGeneratorDef is not None: lCPManager.mBackwardCompatibilityJSONDef = inJSONGeneratorDef # CASE JSInitGeneratorDef if inJSInitGeneratorDef is not None: lCPManager.mBackwardCompatibilityJSDef = inJSInitGeneratorDef
[документация]def WebRequestHostGet(inRequest) -> str: """L+,W+: Получить наименование хоста, с которого поступил запрос :param inRequest: Экземпляр fastapi.Request, по умолчанию None :type inRequest: fastapi.Request, опционально :return: Наименование хоста :rtype: str """ return inRequest.client.host
[документация]def WebAuditMessageCreate(inAuthTokenStr:str = None, inHostStr:str=None, inOperationCodeStr:str="-", inMessageStr:str="-"): """L+,W+: [ИЗМЕНЕНИЕ В 1.3.1] Создание сообщения ИТ аудита с такими сведениями как (Домен, IP, логин и тд.). Данная функция особенно актуальна в том случае, если требуется реализовать дополнительные меры контроля ИТ системы. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator lWebAuditMessageStr = Orchestrator.WebAuditMessageCreate( inRequest = lRequest, inOperationCodeStr = "OP_CODE_1", inMessageStr="Success"): # Логгирование сообщения lLogger.info(lWebAuditMessageStr) :param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен) :type inAuthTokenStr: str, опционально :param inHostStr: IP адрес хоста пользователя / бота, по умолчанию None (не установлен) :type inHostStr: str, опционально :param inOperationCodeStr: Код операции, который принят в компании в соответствии с регламентными процедурами :param inMessageStr: Дополнительное сообщение, которое необходимо отправить в сообщение об ИТ аудите :return: Формат сообщения: "WebAudit :: DOMAIN\\USER@101.121.123.12 :: operation code :: message" """ try: lDomainUpperStr = WebUserDomainGet(inAuthTokenStr=inAuthTokenStr).upper() lUserLoginStr = WebUserLoginGet(inAuthTokenStr=inAuthTokenStr).upper() lResultStr = f"ВебАудит :: {lDomainUpperStr}\\\\{lUserLoginStr}@{inHostStr} :: {inOperationCodeStr} :: {inMessageStr}" except Exception as e: print(str(e)) # Has no logger - must be dead alg branch lResultStr = inMessageStr return lResultStr
[документация]def WebRequestParseBodyBytes(inRequest=None): """L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Извлечь данные в байт виде из тела (body) HTTP запроса. :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя :return: Строка байт b'' или None (если тело запроса было пустым) """ if inRequest is None: inRequest = WebRequestGet() lBodyBytes=None return inRequest.body.encode("utf8")
[документация]def WebRequestParseBodyStr(inRequest=None): """L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Извлечь данные в виде строки из тела (body) HTTP запроса. :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя :return: Текстовая строка '' или None (если тело запроса было пустым) """ if inRequest is None: inRequest = WebRequestGet() return inRequest.body
[документация]def WebRequestParseBodyJSON(inRequest=None): """L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Извлечь из тела (body) запроса HTTP JSON данные и преобразовать в Dict / List структуры языка Python. :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя :return: Словарь (dict), список (list) или None (если тело запроса было пустым) """ if inRequest is None: inRequest = WebRequestGet() return json.loads(WebRequestParseBodyStr(inRequest=inRequest))
[документация]def WebRequestParsePath(inRequest=None): """L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Извлечь декодированный URL путь из HTTP запроса пользователя в формате строки. :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя :return: str, пример: /pyOpenRPA/Debugging/DefHelper """ if inRequest is None: inRequest = WebRequestGet() return urllib.parse.unquote(inRequest.path)
[документация]def WebRequestParseFile(inRequest=None): """L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Извлечь файл (наименование + содержимое в виде строки байт b'') из HTTP запроса пользователя. :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя :return: Кортеж формата (FileNameStr, FileBodyBytes) or (None, None), если файл не был обнаружен """ if inRequest is None: inRequest = WebRequestGet() lResultTurple=(None,None) if inRequest.headers.get('Content-Length') is not None: lInputByteArray = WebRequestParseBodyBytes(inRequest=inRequest) #print(f"BODY:ftInputByteArrayl") # Extract bytes data lBoundaryStr = str(inRequest.headers.get('Content-Type')) lBoundaryStr = lBoundaryStr[lBoundaryStr.index("boundary=")+9:] # get the boundary key #print(LBoundoryStr) lSplit = lInputByteArray.split('\r\n\r\n') lDelimiterRNRNIndex = lInputByteArray.index(b'\r\n\r\n') #print(LSplit) # Get file name lSplit0 = lInputByteArray[:lDelimiterRNRNIndex].split(b'\r\n')[1] lFileNameBytes = lSplit0[lSplit0.index(b'filename="')+10:-1] lFileNameStr = lFileNameBytes.decode("utf-8") # File data bytes lFileDataBytes = lInputByteArray[lDelimiterRNRNIndex+4:] lFileDataBytes = lFileDataBytes[:lFileDataBytes.index(b"\r\n--"+lBoundaryStr.encode("utf-8"))] lResultTurple = (lFileNameStr, lFileDataBytes) return lResultTurple
[документация]def WebRequestResponseSend(inResponeStr, inRequest=None, inContentTypeStr: str = None, inHeadersDict: dict = None): """L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Установить ответ на HTTP запрос пользователя. :param inResponeStr: Тело ответа (строка) :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя :param inContentTypeStr: МИМЕ тип. Пример: "html/text" :param inHeadersDict: Словарь (dict) ключей, которые добавить в headers HTTP ответа на запрос пользователя """ if inRequest is None: inRequest = WebRequestGet() inRequest.OpenRPAResponseDict["Body"] = bytes(inResponeStr, "utf8") if inHeadersDict is not None: inRequest.OpenRPAResponseDict["Headers"].update(inHeadersDict) if inContentTypeStr is not None: inRequest.OpenRPAResponseDict["Headers"]["Content-type"] = inContentTypeStr
[документация]def WebRequestGet(): """L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.2] Вернуть экземпляр HTTP запроса, если функция вызвана в потоке, который был порожден для отработки HTTP запроса пользователя. """ lCurrentThread = threading.current_thread() if hasattr(lCurrentThread, "request"): return lCurrentThread.request
[документация]def WebAppGet() -> FastAPI: """L+,W+: Вернуть экземпляр веб сервера fastapi.FastAPI (app). Подробнее про app см. https://fastapi.tiangolo.com/tutorial/first-steps/ """ return Server.app
[документация]def WebAuthDefGet(): """Вернуть функцию авторизации пользователя. Функция может использоваться для доменной авторизации. .. code-block:: python # ПРИМЕР Если требуется авторизация пользователя (получить inAuthTokenStr) from fastapi import Request from fastapi.responses import JSONResponse, HTMLResponse from pyOpenRPA import Orchestrator @app.post("/url/to/def",response_class=JSONResponse) async def some_def(inRequest:Request, inAuthTokenStr:str=Depends(Orchestrator.WebAuthDefGet())): l_input_dict = await inRequest.json() if lValueStr == None or lValueStr == b"": lValueStr="" else: lValueStr = lValueStr.decode("utf8") :return: Функция авторизации :rtype: def """ return ServerSettings.IdentifyAuthorize
[документация]def StorageRobotExists(inRobotNameStr): """L+,W+: Проверить, существует ли ключ inRobotNameStr в хранилище пользовательской информации StorageDict (GSettings > StarageDict) :param inRobotNameStr: Наименование (ключ) робота. !ВНИМАНИЕ! Наименование чувствительно к регистру :return: True - ключ робота присутствует в хранилище; False - отсутствует """ return inRobotNameStr in GSettingsGet()["StorageDict"]
[документация]def StorageRobotGet(inRobotNameStr): """L+,W+: Получить содержимое по ключу робота inRobotNameStr в хранилище пользовательской информации StorageDict (GSettings > StarageDict) :param inRobotNameStr: Наименование (ключ) робота. !ВНИМАНИЕ! Наименование чувствительно к регистру :return: Значение или структура, которая расположена по адресу (GSettings > StarageDict > inRobotNameStr) """ if inRobotNameStr not in GSettingsGet()["StorageDict"]: GSettingsGet()["StorageDict"][inRobotNameStr]={} return GSettingsGet()["StorageDict"][inRobotNameStr]
[документация]def ProcessorAliasDefCreate(inDef, inAliasStr=None, inGSettings = None): """L+,W+: Создать синоним (текстовый ключ) для инициации выполнения функции в том случае, если запрос на выполнения пришел из вне (передача функций невозможна). Старая версия. Новую версию см. ActivityItemDefAliasCreate .. code-block:: python # USAGE from pyOpenRPA import Orchestrator def TestDef(): pass lAliasStr = Orchestrator.ProcessorAliasDefCreate( inGSettings = gSettings, inDef = TestDef, inAliasStr="TestDefAlias") # Now you can call TestDef by the alias from var lAliasStr with help of ActivityItem (key Def = lAliasStr) :param inDef: функция Python :param inAliasStr: Строковый ключ (синоним), который можно будет использовать в Активности (ActivityItem) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: Строковый ключ, который был назначен. Ключ может быть изменен, если входящий текстовый ключ был уже занят. """ return ActivityItemDefAliasCreate(inDef=inDef, inAliasStr=inAliasStr, inGSettings = inGSettings)
[документация]def ProcessorAliasDefUpdate(inDef, inAliasStr, inGSettings = None): """L+,W+: Обновить синоним (текстовый ключ) для инициации выполнения функции в том случае, если запрос на выполнения пришел из вне (передача функций невозможна). Старая версия. Новую версию см. ActivityItemDefAliasUpdate .. code-block:: python # USAGE from pyOpenRPA import Orchestrator def TestDef(): pass Orchestrator.ProcessorAliasDefUpdate( inGSettings = gSettings, inDef = TestDef, inAliasStr="TestDefAlias") # Now you can call TestDef by the alias "TestDefAlias" with help of ActivityItem (key Def = "TestDefAlias") :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inDef: функция Python :param inAliasStr: Строковый ключ (синоним), который можно будет использовать в Активности (ActivityItem) :return: Строковый ключ, который был назначен. Ключ будет тем же, если входящий текстовый ключ был уже занят. """ return ActivityItemDefAliasUpdate(inDef=inDef, inAliasStr=inAliasStr, inGSettings = inGSettings)
# ActivityItem defs
[документация]def ActivityItemHelperDefList(inDefQueryStr=None): """L+,W+: Получить список синонимов (текстовых ключей), доступных для использования в Активностях (ActivityItem). :param inDefStr: Часть текстового ключ (начало / середина / конец) :return: Список доступных ключей в формате: ["ActivityItemDefAliasUpdate", "ActivityItemDefAliasCreate", etc...] """ lResultList = [] if inDefQueryStr is not None: # do search alg for lKeyStr in GSettingsGet()["ProcessorDict"]["AliasDefDict"]: if inDefQueryStr.upper() in lKeyStr.upper(): lResultList.append(lKeyStr) else: for lKeyStr in GSettingsGet()["ProcessorDict"]["AliasDefDict"]: lResultList.append(lKeyStr) return lResultList
[документация]def ActivityItemHelperDefAutofill(inDef): """L+,W+: Анализ аргументов функции по синониму (текстовому ключу). :param inDef: Часть текстового ключ (начало / середина / конец) :return: Преднастроенная структура активности (ActivityItem) { "Def": None, "ArgList": [], "ArgDict": {}, "ArgGSettingsStr": None, "ArgLoggerStr": None } """ lResultDict = { "Def": None, "ArgList": [], "ArgDict": {}, "ArgGSettingsStr": None, "ArgLoggerStr": None } lResultDict["Def"] = inDef lGS = GSettingsGet() if inDef in lGS["ProcessorDict"]["AliasDefDict"]: lDefSignature = inspect.signature(lGS["ProcessorDict"]["AliasDefDict"][inDef]) for lItemKeyStr in lDefSignature.parameters: lItemValue = lDefSignature.parameters[lItemKeyStr] # Check if arg name contains "GSetting" or "Logger" if "GSETTING" in lItemKeyStr.upper(): lResultDict["ArgGSettingsStr"] = lItemKeyStr elif "LOGGER" in lItemKeyStr.upper(): lResultDict["ArgLoggerStr"] = lItemKeyStr else: if lItemValue.default is inspect._empty: lResultDict["ArgDict"][lItemKeyStr] = None else: lResultDict["ArgDict"][lItemKeyStr] = lItemValue.default return lResultDict
[документация]def ActivityItemCreate(inDef, inArgList=None, inArgDict=None, inArgGSettingsStr=None, inArgLoggerStr=None, inGUIDStr = None, inThreadBool = False): """L+,W+: Создать Активность (ActivityItem). Активность можно использовать в ProcessorActivityItemAppend или в Processor.ActivityListExecute или в функциях работы с Агентами. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator # ВАРИАНТ 1 def TestDef(inArg1Str, inGSettings, inLogger): pass lActivityItem = Orchestrator.ActivityItemCreate( inDef = TestDef, inArgList=[], inArgDict={"inArg1Str": "ArgValueStr"}, inArgGSettingsStr = "inGSettings", inArgLoggerStr = "inLogger") # lActivityItem: # { # "Def":TestDef, # "ArgList":inArgList, # "ArgDict":inArgDict, # "ArgGSettings": "inArgGSettings", # "ArgLogger": "inLogger" # } # ВАРИАНТ 2 def TestDef(inArg1Str): pass Orchestrator.ActivityItemDefAliasUpdate( inGSettings = gSettings, inDef = TestDef, inAliasStr="TestDefAlias") lActivityItem = Orchestrator.ActivityItemCreate( inDef = "TestDefAlias", inArgList=[], inArgDict={"inArg1Str": "ArgValueStr"}, inArgGSettingsStr = None, inArgLoggerStr = None) # lActivityItem: # { # "Def":"TestDefAlias", # "ArgList":inArgList, # "ArgDict":inArgDict, # "ArgGSettings": None, # "ArgLogger": None # } :param inDef: Функция Python или синоним (текстовый ключ) :param inArgList: Список (list) неименованных аргументов к функции :param inArgDict: Словарь (dict) именованных аргументов к функции :param inArgGSettingsStr: Текстовое наименование аргумента GSettings (если требуется передавать) :param inArgLoggerStr: Текстовое наименование аргумента logger (если требуется передавать) :param inGUIDStr: ГУИД идентификатор активности (ActivityItem). Если ГУИД не указан, то он будет сгенерирован автоматически :param inThreadBool: True - выполнить ActivityItem в новом потоке; False - выполнить последовательно в общем потоке процессорной очереди :return: lActivityItemDict= { "Def":inDef, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList":inArgList, # Args list "ArgDict":inArgDict, # Args dictionary "ArgGSettings": inArgGSettingsStr, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": inArgLoggerStr, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "GUIDStr": inGUIDStr, "ThreadBool": inThreadBool } """ # Work about GUID in Activity items if inGUIDStr is None: inGUIDStr = str(uuid.uuid4()) # generate new GUID if inArgList is None: inArgList=[] if inArgDict is None: inArgDict={} lActivityItemDict= { "Def":inDef, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList":inArgList, # Args list "ArgDict":inArgDict, # Args dictionary "ArgGSettings": inArgGSettingsStr, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": inArgLoggerStr, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "GUIDStr": inGUIDStr, "ThreadBool": inThreadBool } return lActivityItemDict
[документация]def ActivityItemDefAliasCreate(inDef, inAliasStr=None, inGSettings = None): """L+,W+: Создать синоним (текстовый ключ) для инициации выполнения функции в том случае, если запрос на выполнения пришел из вне (передача функций невозможна). .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator def TestDef(): pass lAliasStr = Orchestrator.ActivityItemDefAliasCreate( inGSettings = gSettings, inDef = TestDef, inAliasStr="TestDefAlias") # Now you can call TestDef by the alias from var lAliasStr with help of ActivityItem (key Def = lAliasStr) :param inDef: функция Python :param inAliasStr: Строковый ключ (синоним), который можно будет использовать в Активности (ActivityItem) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: Строковый ключ, который был назначен. Ключ может быть изменен, если входящий текстовый ключ был уже занят. """ #TODO Pay attention - New alias can be used too - need to create more complex algorythm to create new alias! inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lL = inGSettings["Logger"] if inAliasStr is None: inAliasStr = str(inDef) # Check if key is not exists if inAliasStr in inGSettings["ProcessorDict"]["AliasDefDict"]: inAliasStr = str(inDef) if lL: lL.warning(f"Orchestrator.ProcessorAliasDefCreate: Alias {inAliasStr} already exists in alias dictionary. Another alias will be generated and returned") inGSettings["ProcessorDict"]["AliasDefDict"][inAliasStr] = inDef return inAliasStr
[документация]def ActivityItemDefAliasModulesLoad(): """L+,W+: Загрузить все функции из импортированных модулей sys.modules в ActivityItem синонимы - полезно для отладки со стороны панели управления. """ lL = OrchestratorLoggerGet() lL.info(f"Синонимы функций: старт инициализации sys.modules") lSysModulesSnapshot = copy.copy(sys.modules) # Actual when start from jupyter for lModuleItemStr in lSysModulesSnapshot: lModuleItem = lSysModulesSnapshot[lModuleItemStr] for lDefItemStr in dir(lModuleItem): try: lDefItem = getattr(lModuleItem,lDefItemStr) if callable(lDefItem) and not lDefItemStr.startswith("_"): ActivityItemDefAliasCreate(inDef=lDefItem, inAliasStr=f"{lModuleItemStr}.{lDefItemStr}") except ModuleNotFoundError: pass lL.info(f"Синонимы функций: окончание инициализации sys.modules")
[документация]def ActivityItemDefAliasUpdate(inDef, inAliasStr, inGSettings = None): """L+,W+: Обновить синоним (текстовый ключ) для инициации выполнения функции в том случае, если запрос на выполнения пришел из вне (передача функций невозможна). .. code-block:: python # USAGE from pyOpenRPA import Orchestrator def TestDef(): pass Orchestrator.ActivityItemDefAliasUpdate( inGSettings = gSettings, inDef = TestDef, inAliasStr="TestDefAlias") # Now you can call TestDef by the alias "TestDefAlias" with help of ActivityItem (key Def = "TestDefAlias") :param inDef: функция Python :param inAliasStr: Строковый ключ (синоним), который можно будет использовать в Активности (ActivityItem) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: Строковый ключ, который был назначен. Ключ будет тем же, если входящий текстовый ключ был уже занят. """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings if callable(inDef): inGSettings["ProcessorDict"]["AliasDefDict"][inAliasStr] = inDef else: raise Exception(f"pyOpenRPA Exception: You can't use Orchestrator.ActivityItemDefAliasUpdate with arg 'inDef' string value. inDef is '{inDef}', inAliasStr is '{inAliasStr}'") return inAliasStr
[документация]def ProcessorActivityItemCreate(inDef, inArgList=None, inArgDict=None, inArgGSettingsStr=None, inArgLoggerStr=None, inGUIDStr = None, inThreadBool = False): """L+,W+: Создать Активность (ActivityItem). Активность можно использовать в ProcessorActivityItemAppend или в Processor.ActivityListExecute или в функциях работы с Агентами. Старая версия. Новую версию см. в ActivityItemCreate .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator # ВАРИАНТ 1 def TestDef(inArg1Str, inGSettings, inLogger): pass lActivityItem = Orchestrator.ProcessorActivityItemCreate( inDef = TestDef, inArgList=[], inArgDict={"inArg1Str": "ArgValueStr"}, inArgGSettingsStr = "inGSettings", inArgLoggerStr = "inLogger") # lActivityItem: # { # "Def":TestDef, # "ArgList":inArgList, # "ArgDict":inArgDict, # "ArgGSettings": "inArgGSettings", # "ArgLogger": "inLogger" # } # ВАРИАНТ 2 def TestDef(inArg1Str): pass Orchestrator.ProcessorAliasDefUpdate( inGSettings = gSettings, inDef = TestDef, inAliasStr="TestDefAlias") lActivityItem = Orchestrator.ProcessorActivityItemCreate( inDef = "TestDefAlias", inArgList=[], inArgDict={"inArg1Str": "ArgValueStr"}, inArgGSettingsStr = None, inArgLoggerStr = None) # lActivityItem: # { # "Def":"TestDefAlias", # "ArgList":inArgList, # "ArgDict":inArgDict, # "ArgGSettings": None, # "ArgLogger": None # } :param inDef: Функция Python или синоним (текстовый ключ) :param inArgList: Список (list) неименованных аргументов к функции :param inArgDict: Словарь (dict) именованных аргументов к функции :param inArgGSettingsStr: Текстовое наименование аргумента GSettings (если требуется передавать) :param inArgLoggerStr: Текстовое наименование аргумента logger (если требуется передавать) :param inGUIDStr: ГУИД идентификатор активности (ActivityItem). Если ГУИД не указан, то он будет сгенерирован автоматически :param inThreadBool: True - выполнить ActivityItem в новом потоке; False - выполнить последовательно в общем потоке процессорной очереди :return: lActivityItemDict= { "Def":inDef, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList":inArgList, # Args list "ArgDict":inArgDict, # Args dictionary "ArgGSettings": inArgGSettingsStr, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": inArgLoggerStr, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "GUIDStr": inGUIDStr, "ThreadBool": inThreadBool } """ return ActivityItemCreate(inDef=inDef, inArgList=inArgList, inArgDict=inArgDict, inArgGSettingsStr=inArgGSettingsStr, inArgLoggerStr=inArgLoggerStr, inGUIDStr=inGUIDStr, inThreadBool=inThreadBool)
[документация]def ProcessorActivityItemAppend(inGSettings = None, inDef=None, inArgList=None, inArgDict=None, inArgGSettingsStr=None, inArgLoggerStr=None, inActivityItemDict=None): """L+,W+: Добавить активность (ActivityItem) в процессорную очередь. .. code-block:: python # USAGE from pyOpenRPA import Orchestrator # EXAMPLE 1 def TestDef(inArg1Str, inGSettings, inLogger): pass lActivityItem = Orchestrator.ProcessorActivityItemAppend( inGSettings = gSettingsDict, inDef = TestDef, inArgList=[], inArgDict={"inArg1Str": "ArgValueStr"}, inArgGSettingsStr = "inGSettings", inArgLoggerStr = "inLogger") # Activity have been already append in the processor queue # EXAMPLE 2 def TestDef(inArg1Str): pass Orchestrator.ProcessorAliasDefUpdate( inGSettings = gSettings, inDef = TestDef, inAliasStr="TestDefAlias") lActivityItem = Orchestrator.ProcessorActivityItemCreate( inDef = "TestDefAlias", inArgList=[], inArgDict={"inArg1Str": "ArgValueStr"}, inArgGSettingsStr = None, inArgLoggerStr = None) Orchestrator.ProcessorActivityItemAppend( inGSettings = gSettingsDict, inActivityItemDict = lActivityItem) # Activity have been already append in the processor queue :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inDef: Функция Python или синоним (текстовый ключ) :param inArgList: Список (list) неименованных аргументов к функции :param inArgDict: Словарь (dict) именованных аргументов к функции :param inArgGSettingsStr: Текстовое наименование аргумента GSettings (если требуется передавать) :param inArgLoggerStr: Текстовое наименование аргумента logger (если требуется передавать) :param inGUIDStr: ГУИД идентификатор активности (ActivityItem). Если ГУИД не указан, то он будет сгенерирован автоматически :param inThreadBool: True - выполнить ActivityItem в новом потоке; False - выполнить последовательно в общем потоке процессорной очереди :param inActivityItemDict: Альтернативный вариант заполнения, если уже имеется Активность (ActivityItem). В таком случае не требуется заполнять аргументы inDef, inArgList, inArgDict, inArgGSettingsStr, inArgLoggerStr :return ГУИД активности (ActivityItem) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings if inActivityItemDict is None: if inArgList is None: inArgList=[] if inArgDict is None: inArgDict={} if inDef is None: raise Exception(f"pyOpenRPA Exception: ProcessorActivityItemAppend need inDef arg if you dont use inActivityItemDict") lActivityList=[ { "Def":inDef, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList":inArgList, # Args list "ArgDict":inArgDict, # Args dictionary "ArgGSettings": inArgGSettingsStr, # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": inArgLoggerStr # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } ] else: lActivityList = [inActivityItemDict] # Work about GUID in Activity items lGUIDStr = None for lItemDict in lActivityList: # Add GUIDStr if not exist if "GUIDStr" not in lItemDict: lGUIDStr = str(uuid.uuid4()) # generate new GUID lItemDict["GUIDStr"] = lGUIDStr # Add activity list in ProcessorDict inGSettings["ProcessorDict"]["ActivityList"]+=lActivityList return lGUIDStr
## Process defs
[документация]def ProcessIsStarted(inProcessNameWOExeStr): # Check if process is started """L-,W+: Проверить, запущен ли процесс, который в наименовании содержит inProcessNameWOExeStr. !ВНИМАНИЕ! ПРОВЕРЯЕТ ВСЕ ПРОЦЕССЫ ОПЕРАЦИОННОЙ СИСТЕМЫ. И ДРУГИХ ПОЛЬЗОВАТЕЛЬСКИХ СЕССИЙ, ЕСЛИ ОНИ ЗАПУЩЕНЫ НА ЭТОЙ МАШИНЕ. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator lProcessIsStartedBool = Orchestrator.ProcessIsStarted(inProcessNameWOExeStr = "notepad") # lProcessIsStartedBool is True - notepad.exe is running on the Orchestrator machine :param inProcessNameWOExeStr: Наименование процесса без расширения .exe (WO = WithOut = Без) Пример: Для проверки процесса блокнота нужно указывать "notepad", не "notepad.exe" :return: True - Процесс запущен на ОС ; False - Процесс не запущен на ОС, где работает Оркестратор """ #Iterate over the all the running process for proc in psutil.process_iter(): try: # Check if process name contains the given name string. if inProcessNameWOExeStr.lower() in proc.name().lower(): return True except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): pass return False
[документация]def ProcessStart(inPathStr, inArgList, inStopProcessNameWOExeStr=None): """L-,W+: Запуск процесса на сессии Оркестратора, если на ОС не запущен процесс inStopProcessNameWOExeStr. !ВНИМАНИЕ! ПРИ ПРОВЕРКЕ РАНЕЕ ЗАПУЩЕННЫХ ПРОЦЕССОВ ПРОВЕРЯЕТ ВСЕ ПРОЦЕССЫ ОПЕРАЦИОННОЙ СИСТЕМЫ. И ДРУГИХ ПОЛЬЗОВАТЕЛЬСКИХ СЕССИЙ, ЕСЛИ ОНИ ЗАПУЩЕНЫ НА ЭТОЙ МАШИНЕ. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator Orchestrator.ProcessStart( inPathStr = "notepad" inArgList = [] inStopProcessNameWOExeStr = "notepad") # notepad.exe will be started if no notepad.exe is active on the machine :param inPathStr: Путь к файлу запускаемого процесса :param inArgList: Список аргументов, передаваемых в запускаемый процесс. Пример: ["test.txt"] :param inStopProcessNameWOExeStr: Доп. контроль: Не запускать процесс, если обнаружен запущенный процесс под наименованием inStopProcessNameWOExeStr без расширения .exe (WO = WithOut = Без) """ lStartProcessBool = True if inStopProcessNameWOExeStr is not None: #Check if process running lCheckTaskName = inStopProcessNameWOExeStr if len(lCheckTaskName)>4: if lCheckTaskName[-4:].upper() != ".EXE": lCheckTaskName = lCheckTaskName+".exe" else: lCheckTaskName = lCheckTaskName+".exe" #Check if process exist if not ProcessIsStarted(inProcessNameWOExeStr=lCheckTaskName): lStartProcessBool=True if lStartProcessBool == True: # Start if flag is true lItemArgs=[inPathStr] if inArgList is None: inArgList = [] # 2021 02 22 Minor fix default value lItemArgs.extend(inArgList) subprocess.Popen(lItemArgs,shell=True)
[документация]def ProcessStop(inProcessNameWOExeStr, inCloseForceBool, inUserNameStr = "%username%"): """L-,W+: Остановить процесс на ОС, где работает Оркестратор, под учетной записью inUserNameStr. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator Orchestrator.ProcessStop( inProcessNameWOExeStr = "notepad" inCloseForceBool = True inUserNameStr = "USER_99") # Will close process "notepad.exe" on the user session "USER_99" (!ATTENTION! if process not exists no exceptions will be raised) :param inProcessNameWOExeStr: Наименование процесса без расширения .exe (WO = WithOut = Без). Пример: Для проверки процесса блокнота нужно указывать "notepad", не "notepad.exe" :param inCloseForceBool: True - принудительно завершить процесс. False - отправить сигнал на безопасное отключение (!ВНИМАНИЕ! - ОС не позволяет отправлять сигнал на безопасное отключение на другую сесиию - только на той, где работает Оркестратор) :param inUserNameStr: Наименование пользователя, под именем которого искать процесс для остановки. По умолчанию "%username%" - искать процесс на текущей сессии """ # Support input arg if with .exe lProcessNameWExeStr = inProcessNameWOExeStr if len(lProcessNameWExeStr) > 4: if lProcessNameWExeStr[-4:].upper() != ".EXE": lProcessNameWExeStr = lProcessNameWExeStr + ".exe" else: lProcessNameWExeStr = lProcessNameWExeStr + ".exe" # Flag Force lActivityCloseCommand = 'taskkill /im ' + lProcessNameWExeStr if inCloseForceBool == True: lActivityCloseCommand += " /F" # None - all users, %username% - current user, another str - another user if inUserNameStr is not None: lActivityCloseCommand += f' /fi "username eq {inUserNameStr}"' # Kill process os.system(lActivityCloseCommand)
[документация]def ProcessListGet(inProcessNameWOExeList=None): """L-,W+: Вернуть список процессов, запущенных на ОС, где работает Оркестратор. В списке отсортировать процессы по количеству используемой оперативной памяти. Также можно указать перечень процессов, которые интересуют - функция будет показывать активные из них. !ВНИМАНИЕ! ДЛЯ ПОЛУЧЕНИЯ СПИСКА ВСЕХ ПРОЦЕССОВ ОС НЕОБХОДИМО ЗАПУСКАТЬ ОРКЕСТРАТОР С ПРАВАМИ АДМИНИСТРАТОРА. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator lProcessList = Orchestrator.ProcessListGet() # Return the list of the process on the machine. # !ATTENTION! RUn orchestrator as administrator to get all process list on the machine. :param inProcessNameWOExeList: Список процессов, среди которых искать активные. Если параметр не указывать - вернет перечень всех доступных процессов :return: Сведения о запущенных процессах в следующем формате { "ProcessWOExeList": ["notepad","..."], "ProcessWOExeUpperList": ["NOTEPAD","..."], "ProcessDetailList": [ { 'pid': 412, # Идентификатор процесса 'username': "DESKTOP\\USER", 'name': 'notepad.exe', 'vms': 13.77767775, # В Мб 'NameWOExeUpperStr': 'NOTEPAD', 'NameWOExeStr': "'notepad'"}, {...}] """ if inProcessNameWOExeList is None: inProcessNameWOExeList = [] lMapUPPERInput = {} # Mapping for processes WO exe lResult = {"ProcessWOExeList":[], "ProcessWOExeUpperList":[],"ProcessDetailList":[]} # Create updated list for quick check lProcessNameWOExeList = [] for lItem in inProcessNameWOExeList: if lItem is not None: lProcessNameWOExeList.append(f"{lItem.upper()}.EXE") lMapUPPERInput[f"{lItem.upper()}.EXE"]= lItem # Iterate over the list for proc in psutil.process_iter(): try: # Fetch process details as dict pinfo = proc.as_dict(attrs=['pid', 'name', 'username']) pinfo['vms'] = proc.memory_info().vms / (1024 * 1024) pinfo['NameWOExeUpperStr'] = pinfo['name'][:-4].upper() # Add if empty inProcessNameWOExeList or if process in inProcessNameWOExeList if len(lProcessNameWOExeList)==0 or pinfo['name'].upper() in lProcessNameWOExeList: try: # 2021 02 22 Minor fix if not admin rights pinfo['NameWOExeStr'] = lMapUPPERInput[pinfo['name'].upper()] except Exception as e: pinfo['NameWOExeStr'] = pinfo['name'][:-4] lResult["ProcessDetailList"].append(pinfo) # Append dict to list lResult["ProcessWOExeList"].append(pinfo['NameWOExeStr']) lResult["ProcessWOExeUpperList"].append(pinfo['NameWOExeUpperStr']) except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): pass return lResult
[документация]def ProcessDefIntervalCall(inDef, inIntervalSecFloat, inIntervalAsyncBool=False, inDefArgList=None, inDefArgDict=None, inDefArgGSettingsNameStr=None, inDefArgLoggerNameStr=None, inExecuteInNewThreadBool=True, inLogger=None, inGSettings = None): """L+,W+: Периодический вызов функции Python. :param inDef: Функция Python, которую потребуется периодически вызывать :param inIntervalSecFloat: Интервал между вызовами функции в сек. :param inIntervalAsyncBool: False - ожидать интервал inIntervalSecFloat только после окончания выполнения предыдущей итерации; True - Ожидать интервал сразу после запуска итерации :param inDefArgList: Список (list) неименованных аргументов для передачи в функцию. По умолчанию None :param inDefArgDict: Словарь (dict) именованных аргументов для передачи в функцию. По умолчанию None :param inDefArgGSettingsNameStr: Наименование аргумента глобального словаря настроек Оркестратора GSettings (опционально) :param inDefArgLoggerNameStr: Наименование аргумента логгера Оркестратора (опционально) :param inExecuteInNewThreadBool: True - периодический вызов выполнять в отдельном потоке (не останавливать выполнение текущего потока); False - Выполнение периодического вызова в текущем потоке, в котором была вызвана функция ProcessDefIntervalCall. По умолчанию: True :param inLogger: Логгер для фиксации сообщений выполнения функции (опционально) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings if inLogger is None: inLogger = OrchestratorLoggerGet() #Some edits on start if inDefArgDict is None: inDefArgDict = {} if inDefArgList is None: inDefArgList = [] # Check if inDefArgLogger is set and inLogger is exist if inDefArgLoggerNameStr=="": inDefArgLoggerNameStr=None if inDefArgGSettingsNameStr=="": inDefArgGSettingsNameStr=None if inDefArgLoggerNameStr is not None and not inLogger: raise Exception(f"__Orchestrator__.ProcessDefIntervalCall: Вызываемая функция требует передачи логгера, а он не указан") # Check thread if not Core.IsProcessorThread(inGSettings=inGSettings): if inGSettings["Logger"]: inGSettings["Logger"].warning(f"__Orchestrator__.ProcessDefIntervalCall: функция вызвана не из процессорной очереди - активность будет перемещена в процессорную очередь") lProcessorActivityDict = { "Def": ProcessDefIntervalCall, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inDef": inDef, "inIntervalSecFloat": inIntervalSecFloat, "inIntervalAsyncBool":inIntervalAsyncBool, "inDefArgList": inDefArgList, "inDefArgDict": inDefArgDict, "inDefArgGSettingsNameStr":inDefArgGSettingsNameStr, "inDefArgLoggerNameStr": inDefArgLoggerNameStr, "inExecuteInNewThreadBool": inExecuteInNewThreadBool}, # Args dictionary "ArgGSettings": "inGSettings", # Name of GSettings attribute: str (ArgDict) or index (for ArgList) "ArgLogger": "inLogger" # Name of GSettings attribute: str (ArgDict) or index (for ArgList) } inGSettings["ProcessorDict"]["ActivityList"].append(lProcessorActivityDict) else: # Internal def to execute periodically def __Execute__(inGSettings, inDef, inIntervalSecFloat, inIntervalAsyncBool, inDefArgList, inDefArgDict, inLogger, inDefArgGSettingsNameStr, inDefArgLoggerNameStr): if inLogger: inLogger.info(f"__Orchestrator__.ProcessDefIntervalCall: вызов функции по циклу инициализирован. Функция: {str(inDef)}") # Prepare gSettings and logger args if inDefArgGSettingsNameStr is not None: inDefArgDict[inDefArgGSettingsNameStr] = inGSettings if inDefArgLoggerNameStr is not None: inDefArgDict[inDefArgLoggerNameStr] = inLogger while True: try: # Call async if needed if inIntervalAsyncBool == False: # Case wait result then wait inDef(*inDefArgList, **inDefArgDict) else: # Case dont wait result - run sleep then new iteration (use many threads) lThread2 = threading.Thread(target=inDef, args=inDefArgList, kwargs=inDefArgDict) lThread2.setName("INTERVAL_CALL_DEF") lThread2.start() except Exception as e: if inLogger: inLogger.exception( f"ProcessDefIntervalCall: Interval call has been failed. Traceback is below. Code will sleep for the next call") # Sleep interval time.sleep(inIntervalSecFloat) # Check to call in new thread if inExecuteInNewThreadBool: lThread = threading.Thread(target=__Execute__, kwargs={"inGSettings":inGSettings, "inDef": inDef, "inIntervalSecFloat": inIntervalSecFloat, "inIntervalAsyncBool": inIntervalAsyncBool, "inDefArgList": inDefArgList, "inDefArgDict": inDefArgDict, "inLogger": inLogger, "inDefArgGSettingsNameStr":inDefArgGSettingsNameStr , "inDefArgLoggerNameStr":inDefArgLoggerNameStr}) lThread.setName("INTERVAL_CALL_EXECUTOR") lThread.start() else: __Execute__(inGSettings=inGSettings, inDef=inDef, inIntervalSecFloat=inIntervalSecFloat, inIntervalAsyncBool=inIntervalAsyncBool, inDefArgList=inDefArgList, inDefArgDict=inDefArgDict, inLogger=inLogger, inDefArgGSettingsNameStr=inDefArgGSettingsNameStr , inDefArgLoggerNameStr=inDefArgLoggerNameStr)
# Python def - start module function
[документация]def PythonStart(inModulePathStr, inDefNameStr, inArgList=None, inArgDict=None, inLogger = None): """L+,W+: Импорт модуля и выполнение функции в процессе Оркестратора. .. note:: Импорт модуля inModulePathStr будет происходить каждый раз в вызовом функции PythonStart. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator Orchestrator.PythonStart( inModulePathStr="ModuleToCall.py", # inModulePathStr: Working Directory\\ModuleToCall.py inDefNameStr="TestDef") # Import module in Orchestrator process and call def "TestDef" from module "ModuleToCall.py" :param inModulePathStr: Абсолютный или относительный путь (относительно рабочей директории процесса Оркестратора), по которому расположен импортируемый .py файл :param inDefNameStr: Наименование функции в модуле .py :param inArgList: Список (list) неименованных аргументов для передачи в функцию. По умолчанию None :param inArgDict: Словарь (dict) именованных аргументов для передачи в функцию. По умолчанию None :param inLogger: Логгер для фиксации сообщений выполнения функции (опционально) """ inModulePathStr=CrossOS.PathStr(inPathStr=inModulePathStr) if inLogger is None: inLogger = OrchestratorLoggerGet() if inArgList is None: inArgList=[] if inArgDict is None: inArgDict={} try: lModule=importlib.import_module(inModulePathStr) #Подключить модуль для вызова lFunction=getattr(lModule,inDefNameStr) #Найти функцию return lFunction(*inArgList,**inArgDict) except Exception as e: if inLogger: inLogger.exception("Loop activity error: module/function not founded")
# # # # # # # # # # # # # # # # # # # # # # # # Scheduler # # # # # # # # # # # # # # # # # # # # # # #
[документация]def SchedulerActivityTimeAddWeekly(inTimeHHMMStr="23:55:", inWeekdayList=None, inActivityList=None, inGSettings = None): """L+,W+: Добавить активность по расписанию. Допускается указание времени и дней недели, в которые производить запуск. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator # ВАРИАНТ 1 def TestDef(inArg1Str): pass lActivityItem = Orchestrator.ProcessorActivityItemCreate( inDef = TestDef, inArgList=[], inArgDict={"inArg1Str": "ArgValueStr"}, inArgGSettingsStr = None, inArgLoggerStr = None) Orchestrator.SchedulerActivityTimeAddWeekly( inGSettings = gSettingsDict, inTimeHHMMStr = "04:34", inWeekdayList=[2,3,4], inActivityList = [lActivityItem]) # Activity will be executed at 04:34 Wednesday (2), thursday (3), friday (4) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inTimeHHMMStr: Время запуска активности. Допускаются значения от "00:00" до "23:59". Example: "05:29" :param inWeekdayList: Список (list) дней недели, в которые выполнять запуск список активностей (ActivityItem list). 0 (понедельник) - 6 (воскресенье). Пример: [0,1,2,3,4]. По умолчанию устанавливается каждый день [0,1,2,3,4,5,6] :param inActivityList: Список активностей (ActivityItem list) на исполнение """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings if inWeekdayList is None: inWeekdayList=[0,1,2,3,4,5,6] if inActivityList is None: inActivityList=[] Processor.__ActivityListVerify__(inActivityList=inActivityList) # DO VERIFICATION FOR THE inActivityList lActivityTimeItemDict = { "TimeHH:MMStr": inTimeHHMMStr, # Time [HH:MM] to trigger activity "WeekdayList": inWeekdayList, # List of the weekday index when activity is applicable, Default [1,2,3,4,5,6,7] "ActivityList": inActivityList, "GUID": None # # Will be filled in Orchestrator automatically - is needed for detect activity completion } inGSettings["SchedulerDict"]["ActivityTimeList"].append(lActivityTimeItemDict)
# # # # # # # # # # # # # # # # # # # # # # # # RDPSession # # # # # # # # # # # # # # # # # # # # # # #
[документация]def RDPTemplateCreate(inLoginStr, inPasswordStr, inHostStr="127.0.0.1", inPortInt = 3389, inWidthPXInt = 1680, inHeightPXInt = 1050, inUseBothMonitorBool = False, inDepthBitInt = 32, inSharedDriveList=None, inRedirectClipboardBool=True): """L-,W+: Создать шаблон подключения RDP (dict). Данный шаблон далее можно использовать в Orchestrator.RDPSessionConnect .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator lRDPItemDict = Orchestrator.RDPTemplateCreate( inLoginStr = "USER_99", inPasswordStr = "USER_PASS_HERE", inHostStr="127.0.0.1", inPortInt = 3389, inWidthPXInt = 1680, inHeightPXInt = 1050, inUseBothMonitorBool = False, inDepthBitInt = 32, inSharedDriveList=None, inRedirectClipboardBool=False) :param inLoginStr: Логин учетной записи, на которую будет происходить вход по RDP :param inPasswordStr: Пароль учетной записи, на которую будет происходить вход по RDP. !ВНИМАНИЕ! Пароль нигде не сохраняется - конфиденциальность обеспечена :param inHostStr: Имя хоста, к которому планируется подключение по RDP. Пример "77.77.22.22" :param inPortInt: RDP порт, по которому будет происходить подключение. По умолчанию 3389 (стандартный порт RDP) :param inWidthPXInt: Разрешение ширины RDP графической сессии в пикселях. По умолчанию 1680 :param inHeightPXInt: Разрешение высоты RDP графической сессии в пикселях. По умолчанию 1050 :param inUseBothMonitorBool: True - Использовать все подключенные мониторы на RDP сессии; False - Использовать только один монитор на подключенной RDP сессии :param inDepthBitInt: Глубина цвета на удаленной RDP графической сессии. Допустимые варианты: 32 || 24 || 16 || 15. По умолчанию 32 :param inSharedDriveList: Перечень общих дисков, доступ к которым предоставить на сторону удаленной RDP сессии. Пример: ["c", "d"] :param inRedirectClipboardBool: True - Синхронизировать буфер обмена между сессией Оркестратора и удаленной RDP сессией; False - Не синхронизировать буфер обмена. По умолчанию True (в целях обратной совместимости). !ВНИМАНИЕ! Для учетных записей роботов мы рекомендуем не синхронизировать буфер обмена, так как это может привести к ошибкам роботов, которые работают с клавиатурой и буфером обмена :return: { "Host": inHostStr, # Адрес хоста, пример "77.77.22.22" "Port": str(inPortInt), # RDP порт, пример "3389" "Login": inLoginStr, # Логин УЗ, пример "test" "Password": inPasswordStr, # Пароль УЗ, пример "test" "Screen": { "Width": inWidthPXInt, # Разрешение ширины RDP графической сессии в пикселях. По умолчанию 1680 "Height": inHeightPXInt, Разрешение высоты RDP графической сессии в пикселях. По умолчанию 1050 "FlagUseAllMonitors": inUseBothMonitorBool, "DepthBit": str(inDepthBitInt) }, "SharedDriveList": inSharedDriveList, "RedirectClipboardBool": True, ###### PROGRAM VARIABLE ############ "SessionHex": "77777sdfsdf77777dsfdfsf77777777", "SessionIsWindowExistBool": False, "SessionIsWindowResponsibleBool": False, "SessionIsIgnoredBool": False } """ if inSharedDriveList is None: inSharedDriveList = ["c"] if inPortInt is None: inPortInt = 3389 if inRedirectClipboardBool is None: inRedirectClipboardBool = True lRDPTemplateDict= { # Init the configuration item "Host": inHostStr, # Host address, example "77.77.22.22" "Port": str(inPortInt), # RDP Port, example "3389" "Login": inLoginStr, # Login, example "test" "Password": inPasswordStr, # Password, example "test" "Screen": { "Width": inWidthPXInt, # Width of the remote desktop in pixels, example 1680 "Height": inHeightPXInt, # Height of the remote desktop in pixels, example 1050 # "640x480" or "1680x1050" or "FullScreen". If Resolution not exists set full screen, example "FlagUseAllMonitors": inUseBothMonitorBool, # True or False, example False "DepthBit": str(inDepthBitInt) # "32" or "24" or "16" or "15", example "32" }, "SharedDriveList": inSharedDriveList, # List of the Root sesion hard drives, example ["c"], "RedirectClipboardBool": inRedirectClipboardBool, # True - share clipboard to RDP; False - else ###### Will updated in program ############ "SessionHex": "77777sdfsdf77777dsfdfsf77777777", # Hex is created when robot runs, example "" "SessionIsWindowExistBool": False, # Flag if the RDP window is exist, old name "FlagSessionIsActive". Check every n seconds , example False "SessionIsWindowResponsibleBool": False, # Flag if RDP window is responsible (recieve commands). Check every nn seconds. If window is Responsible - window is Exist too , example False "SessionIsIgnoredBool": False # Flag to ignore RDP window False - dont ignore, True - ignore, example False } return lRDPTemplateDict
# TODO Search dublicates in GSettings RDPlist ! # Return list if dublicates def RDPSessionDublicatesResolve(inGSettings): pass #for lItemKeyStr in inGSettings["RobotRDPActive"]["RDPList"]: # lItemDict = inGSettings["RobotRDPActive"]["RDPList"][lItemKeyStr]
[документация]def RDPSessionConnect(inRDPSessionKeyStr, inRDPTemplateDict=None, inHostStr=None, inPortStr=None, inLoginStr=None, inPasswordStr=None, inGSettings = None, inRedirectClipboardBool=True): """L-,W+: Выполнить подключение к RDP и следить за стабильностью соединения со стороны Оркестратора. !ВНИМАНИЕ! - Подключение будет проигнорировано, если соединение по таком RDP ключу уже было инициализировано ранее. 2 способа использования функции: ВАРИАНТ 1 (ОСНОВНОЙ): inGSettings, inRDPSessionKeyStr, inRDPTemplateDict. Для получения inRDPTemplateDict см. функцию Orchestrator.RDPTemplateCreate ВАРИАНТ 2 (ОБРАТНАЯ СОВМЕСТИМОСТЬ ДО ВЕРСИИ 1.1.20): inGSettings, inRDPSessionKeyStr, inHostStr, inPortStr, inLoginStr, inPasswordStr .. code-block:: python # ПРИМЕР (ВАРИАНТ 1) from pyOpenRPA import Orchestrator lRDPItemDict = Orchestrator.RDPTemplateCreate( inLoginStr = "USER_99", inPasswordStr = "USER_PASS_HERE", inHostStr="127.0.0.1", inPortInt = 3389, inWidthPXInt = 1680, inHeightPXInt = 1050, inUseBothMonitorBool = False, inDepthBitInt = 32, inSharedDriveList=None) Orchestrator.RDPSessionConnect( inGSettings = gSettings, inRDPSessionKeyStr = "RDPKey", inRDPTemplateDict = lRDPItemDict) # Orchestrator will create RDP session by the lRDPItemDict configuration :param inRDPSessionKeyStr: Ключ RDP сессии - необходим для дальнейшей идентификации :param inRDPTemplateDict: Конфигурационный словарь (dict) RDP подключения (см. функцию Orchestrator.RDPTemplateCreate) :param inLoginStr: Логин учетной записи, на которую будет происходить вход по RDP. Обратная совместимость до версии v 1.1.20. Мы рекомендуем использовать inRDPTemplateDict (см. функцию Orchestrator.RDPTemplateCreate) :param inPasswordStr: Пароль учетной записи, на которую будет происходить вход по RDP. !ВНИМАНИЕ! Пароль нигде не сохраняется - конфиденциальность обеспечена. Обратная совместимость до версии v 1.1.20. Мы рекомендуем использовать inRDPTemplateDict (см. функцию Orchestrator.RDPTemplateCreate) :param inHostStr: Имя хоста, к которому планируется подключение по RDP. Пример "77.77.22.22". Обратная совместимость до версии v 1.1.20. Мы рекомендуем использовать inRDPTemplateDict (см. функцию Orchestrator.RDPTemplateCreate) :param inPortInt: RDP порт, по которому будет происходить подключение. По умолчанию 3389 (стандартный порт RDP). Обратная совместимость до версии v 1.1.20. Мы рекомендуем использовать inRDPTemplateDict (см. функцию Orchestrator.RDPTemplateCreate) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inRedirectClipboardBool: True - Синхронизировать буфер обмена между сессией Оркестратора и удаленной RDP сессией; False - Не синхронизировать буфер обмена. По умолчанию True (в целях обратной совместимости). !ВНИМАНИЕ! Для учетных записей роботов мы рекомендуем не синхронизировать буфер обмена, так как это может привести к ошибкам роботов, которые работают с клавиатурой и буфером обмена """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings # Check thread if not Core.IsProcessorThread(inGSettings=inGSettings): if inGSettings["Logger"]: inGSettings["Logger"].warning(f"RDP def was called not from processor queue - activity will be append in the processor queue.") lResult = { "Def": RDPSessionConnect, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inRDPSessionKeyStr": inRDPSessionKeyStr, "inRDPTemplateDict":inRDPTemplateDict, "inHostStr": inHostStr, "inPortStr": inPortStr, "inLoginStr": inLoginStr, "inPasswordStr": inPasswordStr, "inRedirectClipboardBool": inRedirectClipboardBool}, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lResult) else: # In processor - do execution # Var 1 - if RDPTemplateDict is input lRDPConfigurationItem=inRDPTemplateDict # Var 2 - backward compatibility if lRDPConfigurationItem is None: lRDPConfigurationItem = RDPTemplateCreate(inLoginStr=inLoginStr, inPasswordStr=inPasswordStr, inHostStr=inHostStr, inPortInt = int(inPortStr), inRedirectClipboardBool=inRedirectClipboardBool) # ATTENTION - dont connect if RDP session is exist # Start the connect if inRDPSessionKeyStr not in inGSettings["RobotRDPActive"]["RDPList"]: inGSettings["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr] = lRDPConfigurationItem # Add item in RDPList Connector.Session(lRDPConfigurationItem) # Create the RDP session Connector.SystemRDPWarningClickOk() # Click all warning messages else: if inGSettings["Logger"]: inGSettings["Logger"].warning(f"RDP session was not created because it is alredy exists in the RDPList. Use RDPSessionReconnect if you want to update RDP configuration.") return True
[документация]def RDPSessionDisconnect(inRDPSessionKeyStr, inBreakTriggerProcessWOExeList = None, inGSettings = None): """L-,W+: Выполнить отключение RDP сессии и прекратить мониторить его активность. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator Orchestrator.RDPSessionDisconnect( inGSettings = gSettings, inRDPSessionKeyStr = "RDPKey") # Orchestrator will disconnect RDP session and will stop to monitoring current RDP :param inRDPSessionKeyStr: Ключ RDP сессии - необходим для дальнейшей идентификации :param inBreakTriggerProcessWOExeList: Список процессов, наличие которых в диспетчере задач приведет к прерыванию задачи по остановке RDP соединения. Пример ["notepad"] :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings if inBreakTriggerProcessWOExeList is None: inBreakTriggerProcessWOExeList = [] # Check thread if not Core.IsProcessorThread(inGSettings=inGSettings): if inGSettings["Logger"]: inGSettings["Logger"].warning(f"RDP def was called not from processor queue - activity will be append in the processor queue.") lResult = { "Def": RDPSessionDisconnect, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inRDPSessionKeyStr": inRDPSessionKeyStr, "inBreakTriggerProcessWOExeList": inBreakTriggerProcessWOExeList }, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lResult) else: # In processor - do execution lSessionHex = inGSettings["RobotRDPActive"]["RDPList"].get(inRDPSessionKeyStr,{}).get("SessionHex", None) if lSessionHex: lProcessListResult = {"ProcessWOExeList":[],"ProcessDetailList":[]} if len(inBreakTriggerProcessWOExeList) > 0: lProcessListResult = ProcessListGet(inProcessNameWOExeList=inBreakTriggerProcessWOExeList) # Run the task manager monitor if len(lProcessListResult["ProcessWOExeList"]) == 0: # Start disconnect if no process exist inGSettings["RobotRDPActive"]["RDPList"].pop(inRDPSessionKeyStr,None) Connector.SessionClose(inSessionHexStr=lSessionHex) Connector.SystemRDPWarningClickOk() # Click all warning messages return True
[документация]def RDPSessionReconnect(inRDPSessionKeyStr, inRDPTemplateDict=None, inGSettings = None): """L-,W+: Выполнить переподключение RDP сессии и продолжить мониторить его активность. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator lRDPItemDict = Orchestrator.RDPTemplateCreate( inLoginStr = "USER_99", inPasswordStr = "USER_PASS_HERE", inHostStr="127.0.0.1", inPortInt = 3389, inWidthPXInt = 1680, inHeightPXInt = 1050, inUseBothMonitorBool = False, inDepthBitInt = 32, inSharedDriveList=None) Orchestrator.RDPSessionReconnect( inGSettings = gSettings, inRDPSessionKeyStr = "RDPKey", inRDPTemplateDict = inRDPTemplateDict) # Orchestrator will reconnect RDP session and will continue to monitoring current RDP :param inRDPSessionKeyStr: Ключ RDP сессии - необходим для дальнейшей идентификации :param inRDPTemplateDict: Конфигурационный словарь (dict) RDP подключения (см. функцию Orchestrator.RDPTemplateCreate) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings # Check thread if not Core.IsProcessorThread(inGSettings=inGSettings): if inGSettings["Logger"]: inGSettings["Logger"].warning(f"RDP def was called not from processor queue - activity will be append in the processor queue.") lResult = { "Def": RDPSessionReconnect, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inRDPSessionKeyStr": inRDPSessionKeyStr, "inRDPTemplateDict":inRDPTemplateDict }, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lResult) else: lRDPConfigurationItem = inGSettings["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr] RDPSessionDisconnect(inGSettings = inGSettings, inRDPSessionKeyStr=inRDPSessionKeyStr) # Disconnect the RDP 2021 02 22 minor fix by Ivan Maslov # Replace Configuration item if inRDPTemplateDict exists if inRDPTemplateDict is not None: lRDPConfigurationItem=inRDPTemplateDict # Add item in RDPList inGSettings["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr] = lRDPConfigurationItem # Create the RDP session Connector.Session(lRDPConfigurationItem) return True
[документация]def RDPSessionMonitorStop(inRDPSessionKeyStr, inGSettings = None): """L-,W+: Прекратить мониторить активность RDP соединения со стороны Оркестратора. Данная функция не уничтожает активное RDP соединение. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator Orchestrator.RDPSessionMonitorStop( inGSettings = gSettings, inRDPSessionKeyStr = "RDPKey") # Orchestrator will stop the RDP monitoring :param inRDPSessionKeyStr: Ключ RDP сессии - необходим для дальнейшей идентификации :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lResult = True inGSettings["RobotRDPActive"]["RDPList"].pop(inRDPSessionKeyStr,None) # Remove item from RDPList return lResult
[документация]def RDPSessionLogoff(inRDPSessionKeyStr, inBreakTriggerProcessWOExeList = None, inGSettings = None): """L-,W+: Выполнить отключение (logoff) на RDP сессии и прекратить мониторить активность со стороны Оркестратора. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator Orchestrator.RDPSessionLogoff( inGSettings = gSettings, inRDPSessionKeyStr = "RDPKey", inBreakTriggerProcessWOExeList = ['Notepad']) # Orchestrator will logoff the RDP session :param inRDPSessionKeyStr: Ключ RDP сессии - необходим для дальнейшей идентификации :param inBreakTriggerProcessWOExeList: Список процессов, наличие которых в диспетчере задач приведет к прерыванию задачи по остановке RDP соединения. Пример ["notepad"] :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: True - Отключение прошло успешно; False - были зафиксированы ошибки при отключении. """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings if inBreakTriggerProcessWOExeList is None: inBreakTriggerProcessWOExeList = [] lResult = True # Check thread if not Core.IsProcessorThread(inGSettings=inGSettings): if inGSettings["Logger"]: inGSettings["Logger"].warning(f"RDP def was called not from processor queue - activity will be append in the processor queue.") lResult = { "Def": RDPSessionLogoff, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inRDPSessionKeyStr": inRDPSessionKeyStr, "inBreakTriggerProcessWOExeList": inBreakTriggerProcessWOExeList }, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lResult) else: lCMDStr = "shutdown -L" # CMD logoff command # Calculate the session Hex lSessionHex = inGSettings["RobotRDPActive"]["RDPList"].get(inRDPSessionKeyStr,{}).get("SessionHex", None) if lSessionHex: lProcessListResult = {"ProcessWOExeList":[],"ProcessDetailList":[]} if len(inBreakTriggerProcessWOExeList) > 0: lProcessListResult = ProcessListGet(inProcessNameWOExeList=inBreakTriggerProcessWOExeList) # Run the task manager monitor if len(lProcessListResult["ProcessWOExeList"]) == 0: # Start logoff if no process exist # Run CMD - dont crosscheck because CMD dont return value to the clipboard when logoff Connector.SessionCMDRun(inSessionHex=lSessionHex, inCMDCommandStr=lCMDStr, inModeStr="RUN", inLogger=inGSettings["Logger"], inRDPConfigurationItem=inGSettings["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr]) inGSettings["RobotRDPActive"]["RDPList"].pop(inRDPSessionKeyStr,None) # Remove item from RDPList return lResult
def RDPSessionResponsibilityCheck(inRDPSessionKeyStr, inGSettings = None): inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings # Check thread if not Core.IsProcessorThread(inGSettings=inGSettings): if inGSettings["Logger"]: inGSettings["Logger"].warning(f"RDP def was called not from processor queue - activity will be append in the processor queue.") lResult = { "Def": RDPSessionResponsibilityCheck, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inRDPSessionKeyStr": inRDPSessionKeyStr }, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lResult) else: lRDPConfigurationItem = inGSettings["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr] # Get the alias # set the fullscreen # ATTENTION!!! Session hex can be updated!!! Connector.SessionScreenFull(inSessionHex=lRDPConfigurationItem["SessionHex"], inLogger=inGSettings["Logger"], inRDPConfigurationItem=inGSettings["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr]) time.sleep(1) # Check RDP responsibility lDoCheckResponsibilityBool = True lDoCheckResponsibilityCountMax = 20 lDoCheckResponsibilityCountCurrent = 0 while lDoCheckResponsibilityBool: # Check if counter is exceed - raise exception if lDoCheckResponsibilityCountCurrent >= lDoCheckResponsibilityCountMax: pass #raise ConnectorExceptions.SessionWindowNotResponsibleError("Error when initialize the RDP session - RDP window is not responding!") # Check responding lDoCheckResponsibilityBool = not Connector.SystemRDPIsResponsible(inSessionHexStr = lRDPConfigurationItem["SessionHex"]) # Wait if is not responding if lDoCheckResponsibilityBool: time.sleep(3) # increase the couter lDoCheckResponsibilityCountCurrent+=1 return True
[документация]def RDPSessionProcessStartIfNotRunning(inRDPSessionKeyStr, inProcessNameWEXEStr, inFilePathStr, inFlagGetAbsPathBool=True, inGSettings = None): """L-,W+: Выполнить запуск процесса на RDP сессии через графические UI инструменты (без Агента). !ВНИМАНИЕ! Данная функция может работать нестабильно из-за использования графических элементов UI при работе с RDP. Мы рекомендуем использовать конструкцию взаимодействия Оркестратора с Агентом. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator Orchestrator.RDPSessionProcessStartIfNotRunning( inGSettings = gSettings, inRDPSessionKeyStr = "RDPKey", inProcessNameWEXEStr = 'Notepad.exe', inFilePathStr = "path\\to\the\\executable\\file.exe" inFlagGetAbsPathBool = True) # Orchestrator will start the process in RDP session :param inRDPSessionKeyStr: Ключ RDP сессии - необходим для дальнейшей идентификации :param inProcessNameWEXEStr: Наименование процесса с расширением .exe (WEXE - WITH EXE - С EXE). Параметр позволяет не допустить повторного запуска процесса, если он уже был запущен. Example: "Notepad.exe" :param inFilePathStr: Путь к файлу запуска процесса на стороне RDP сессии :param inFlagGetAbsPathBool: True - Преобразовать относительный путь inFilePathStr в абсолютный с учетом рабочей директории Оркестратора; False - Не выполнять преобразований :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings # Check thread lResult = True if not Core.IsProcessorThread(inGSettings=inGSettings): if inGSettings["Logger"]: inGSettings["Logger"].warning(f"RDP def was called not from processor queue - activity will be append in the processor queue.") lActivityItem = { "Def": RDPSessionProcessStartIfNotRunning, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inRDPSessionKeyStr": inRDPSessionKeyStr, "inProcessNameWEXEStr": inProcessNameWEXEStr, "inFilePathStr": inFilePathStr, "inFlagGetAbsPathBool": inFlagGetAbsPathBool }, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lActivityItem) else: lCMDStr = CMDStr.ProcessStartIfNotRunning(inProcessNameWEXEStr, inFilePathStr, inFlagGetAbsPath= inFlagGetAbsPathBool) # Calculate the session Hex lSessionHex = inGSettings["RobotRDPActive"]["RDPList"].get(inRDPSessionKeyStr,{}).get("SessionHex", None) # Run CMD if lSessionHex: Connector.SessionCMDRun(inSessionHex=lSessionHex, inCMDCommandStr=lCMDStr, inModeStr="CROSSCHECK", inLogger=inGSettings["Logger"], inRDPConfigurationItem=inGSettings["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr]) return lResult
[документация]def RDPSessionCMDRun(inRDPSessionKeyStr, inCMDStr, inModeStr="CROSSCHECK", inGSettings = None): """L-,W+: Отправить CMD команду на удаленную сесиию через RDP окно (без Агента). !ВНИМАНИЕ! Данная функция может работать нестабильно из-за использования графических элементов UI при работе с RDP. Мы рекомендуем использовать конструкцию взаимодействия Оркестратора с Агентом. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator lResultDict = Orchestrator.RDPSessionCMDRun( inGSettings = gSettings, inRDPSessionKeyStr = "RDPKey", inModeStr = 'LISTEN') # Orchestrator will send CMD to RDP and return the result (see return section) :param inRDPSessionKeyStr: Ключ RDP сессии - необходим для дальнейшей идентификации :param inCMDStr: Команда CMD, которую отправить на удаленную сессию :param inModeStr: По умолчанию "CROSSCHECK". Варианты: "LISTEN" - Получить результат выполнения операции. Внимание! необходим общий буфер обмена с сессией Оркестратора! "CROSSCHECK" - Выполнить проверку, что операция была выполнена (без получения результата отработки CMD команды). Внимание! необходим общий буфер обмена с сессией Оркестратора! "RUN" - Не получать результат и не выполнять проверку :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :return: Результат выполнения операции в соответсвии с параметрами инициализации функции. Выходная структура: { "OutStr": <> # Результат выполнения CMD (если inModeStr = "LISTEN") "IsResponsibleBool": True|False # True - RDP приняло команду; False - обратная связь не была получена (если inModeStr = "CROSSCHECK") } """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings lResult = { "OutStr": None, # Result string "IsResponsibleBool": False # Flag is RDP is responsible - works only when inModeStr = CROSSCHECK } # Check thread if not Core.IsProcessorThread(inGSettings=inGSettings): if inGSettings["Logger"]: inGSettings["Logger"].warning(f"RDP def was called not from processor queue - activity will be append in the processor queue.") lProcessorActivityDict = { "Def": RDPSessionCMDRun, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inRDPSessionKeyStr": inRDPSessionKeyStr, "inCMDStr": inCMDStr, "inModeStr": inModeStr }, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lProcessorActivityDict) else: #lResult = True # Calculate the session Hex lSessionHex = inGSettings["RobotRDPActive"]["RDPList"].get(inRDPSessionKeyStr,{}).get("SessionHex", None) # Run CMD if lSessionHex: lResult = Connector.SessionCMDRun(inSessionHex=lSessionHex, inCMDCommandStr=inCMDStr, inModeStr=inModeStr, inLogger=inGSettings["Logger"], inRDPConfigurationItem=inGSettings["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr]) return lResult
[документация]def RDPSessionProcessStop(inRDPSessionKeyStr, inProcessNameWEXEStr, inFlagForceCloseBool, inGSettings = None): """L-,W+: Отправка CMD команды в RDP окне на остановку процесса (без Агента). !ВНИМАНИЕ! Данная функция может работать нестабильно из-за использования графических элементов UI при работе с RDP. Мы рекомендуем использовать конструкцию взаимодействия Оркестратора с Агентом. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator lResultDict = Orchestrator.RDPSessionProcessStop( inGSettings = gSettings, inRDPSessionKeyStr = "RDPKey", inProcessNameWEXEStr = 'notepad.exe', inFlagForceCloseBool = True) # Orchestrator will send CMD to RDP and return the result (see return section) :param inRDPSessionKeyStr: Ключ RDP сессии - необходим для дальнейшей идентификации :param inProcessNameWEXEStr: Наименование процесса, который требуется выключить с расширением .exe (WEXE - WITH EXE - С EXE). Пример: "Notepad.exe" :param inFlagForceCloseBool: True - Принудительное отключение. False - Отправка сигнала на безопасное отключение (если процесс поддерживает) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings # Check thread if not Core.IsProcessorThread(inGSettings=inGSettings): if inGSettings["Logger"]: inGSettings["Logger"].warning(f"RDP def was called not from processor queue - activity will be append in the processor queue.") lResult = { "Def": RDPSessionProcessStop, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inRDPSessionKeyStr": inRDPSessionKeyStr, "inProcessNameWEXEStr": inProcessNameWEXEStr, "inFlagForceCloseBool": inFlagForceCloseBool }, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lResult) else: lResult = True lCMDStr = f'taskkill /im "{inProcessNameWEXEStr}" /fi "username eq %USERNAME%"' if inFlagForceCloseBool: lCMDStr+= " /F" # Calculate the session Hex lSessionHex = inGSettings["RobotRDPActive"]["RDPList"].get(inRDPSessionKeyStr,{}).get("SessionHex", None) # Run CMD if lSessionHex: Connector.SessionCMDRun(inSessionHex=lSessionHex, inCMDCommandStr=lCMDStr, inModeStr="CROSSCHECK", inLogger=inGSettings["Logger"], inRDPConfigurationItem=inGSettings["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr]) return lResult
[документация]def RDPSessionFileStoredSend(inRDPSessionKeyStr, inHostFilePathStr, inRDPFilePathStr, inGSettings = None): """L-,W+: Отправка файла со стороны Оркестратора на сторону RDP сессии через UI инструменты RDP окна (без Агента). !ВНИМАНИЕ! Данная функция может работать нестабильно из-за использования графических элементов UI при работе с RDP. Мы рекомендуем использовать конструкцию взаимодействия Оркестратора с Агентом. !ВНИМАНИЕ! Для работы функции требуется открыть доступ по RDP к тем дискам, с которых будет производится копирование файла. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator lResultDict = Orchestrator.RDPSessionFileStoredSend( inGSettings = gSettings, inRDPSessionKeyStr = "RDPKey", inHostFilePathStr = "TESTDIR\\Test.py", inRDPFilePathStr = "C:\\RPA\\TESTDIR\\Test.py") # Orchestrator will send CMD to RDP and return the result (see return section) :param inRDPSessionKeyStr: Ключ RDP сессии - необходим для дальнейшей идентификации :param inHostFilePathStr: Откуда взять файл. Относительный или абсолютный путь к файлу на стороне Оркестратора. Пример: "TESTDIR\\Test.py" :param inRDPFilePathStr: Куда скопировать файл. !Абсолютный! путь на стороне RDP сессии. Пример: "C:\\RPA\\TESTDIR\\Test.py" :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings # Check thread if not Core.IsProcessorThread(inGSettings=inGSettings): if inGSettings["Logger"]: inGSettings["Logger"].warning(f"RDP def was called not from processor queue - activity will be append in the processor queue.") lResult = { "Def": RDPSessionFileStoredSend, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inRDPSessionKeyStr": inRDPSessionKeyStr, "inHostFilePathStr": inHostFilePathStr, "inRDPFilePathStr": inRDPFilePathStr }, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lResult) else: lResult = True lCMDStr = CMDStr.FileStoredSend(inHostFilePath = inHostFilePathStr, inRDPFilePath = inRDPFilePathStr) # Calculate the session Hex lSessionHex = inGSettings["RobotRDPActive"]["RDPList"].get(inRDPSessionKeyStr, {}).get("SessionHex", None) #lSessionHex = inGlobalDict["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr]["SessionHex"] # Run CMD if lSessionHex: Connector.SessionCMDRun(inSessionHex=lSessionHex, inCMDCommandStr=lCMDStr, inModeStr="LISTEN", inClipboardTimeoutSec = 120, inLogger=inGSettings["Logger"], inRDPConfigurationItem=inGSettings["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr]) return lResult
[документация]def RDPSessionFileStoredRecieve(inRDPSessionKeyStr, inRDPFilePathStr, inHostFilePathStr, inGSettings = None): """L-,W+: Получение файла со стороны RDP сессии на сторону Оркестратора через UI инструменты RDP окна (без Агента). !ВНИМАНИЕ! Данная функция может работать нестабильно из-за использования графических элементов UI при работе с RDP. Мы рекомендуем использовать конструкцию взаимодействия Оркестратора с Агентом. !ВНИМАНИЕ! Для работы функции требуется открыть доступ по RDP к тем дискам, с которых будет производится копирование файла. .. code-block:: python # ПРИМЕР from pyOpenRPA import Orchestrator lResultDict = Orchestrator.RDPSessionFileStoredRecieve( inGSettings = gSettings, inRDPSessionKeyStr = "RDPKey", inHostFilePathStr = "TESTDIR\\Test.py", inRDPFilePathStr = "C:\\RPA\\TESTDIR\\Test.py") # Orchestrator will send CMD to RDP and return the result (see return section) :param inRDPSessionKeyStr: Ключ RDP сессии - необходим для дальнейшей идентификации :param inRDPFilePathStr: Откуда скопировать файл. !Абсолютный! путь на стороне RDP сессии. Пример: "C:\\RPA\\TESTDIR\\Test.py" :param inHostFilePathStr: Куда скопировать файл. Относительный или абсолютный путь к файлу на стороне Оркестратора. Пример: "TESTDIR\\Test.py" :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) """ inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings # Check thread if not Core.IsProcessorThread(inGSettings=inGSettings): if inGSettings["Logger"]: inGSettings["Logger"].warning(f"RDP def was called not from processor queue - activity will be append in the processor queue.") lResult = { "Def": RDPSessionFileStoredRecieve, # def link or def alias (look gSettings["Processor"]["AliasDefDict"]) "ArgList": [], # Args list "ArgDict": {"inRDPSessionKeyStr": inRDPSessionKeyStr, "inRDPFilePathStr": inRDPFilePathStr, "inHostFilePathStr": inHostFilePathStr }, # 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) } inGSettings["ProcessorDict"]["ActivityList"].append(lResult) else: lResult = True lCMDStr = CMDStr.FileStoredRecieve(inRDPFilePath = inRDPFilePathStr, inHostFilePath = inHostFilePathStr) # Calculate the session Hex lSessionHex = inGSettings["RobotRDPActive"]["RDPList"].get(inRDPSessionKeyStr,{}).get("SessionHex", None) # Run CMD if lSessionHex: Connector.SessionCMDRun(inSessionHex=lSessionHex, inCMDCommandStr=lCMDStr, inModeStr="LISTEN", inClipboardTimeoutSec = 120, inLogger=inGSettings["Logger"], inRDPConfigurationItem=inGSettings["RobotRDPActive"]["RDPList"][inRDPSessionKeyStr]) return lResult
# # # # # # # # # # # # # # # # # # # # # # # # # # # # Start orchestrator # # # # # # # # # # # # # # # # # # # # # # # #HIDDEN Interval gSettings auto cleaner def to clear some garbage. def GSettingsAutocleaner(inGSettings=None): inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings while True: time.sleep(inGSettings["Autocleaner"]["IntervalSecFloat"]) # Wait for the next iteration lL = inGSettings["Logger"] lNowDatetime = datetime.datetime.now() # Get now time # Clean old items in Client > Session > TechnicalSessionGUIDCache lTechnicalSessionGUIDCacheNew = {} for lItemKeyStr in inGSettings["Client"]["Session"]["TechnicalSessionGUIDCache"]: lItemValue = inGSettings["Client"]["Session"]["TechnicalSessionGUIDCache"][lItemKeyStr] if (lNowDatetime - lItemValue["InitDatetime"]).total_seconds() < inGSettings["Client"]["Session"]["LifetimeSecFloat"]: # Add if lifetime is ok lTechnicalSessionGUIDCacheNew[lItemKeyStr]=lItemValue # Lifetime is ok - set else: if lL: lL.debug(f"Client > Session > TechnicalSessionGUIDCache > lItemKeyStr: Lifetime is expired. Remove from gSettings") # Info inGSettings["Client"]["Session"]["TechnicalSessionGUIDCache"] = lTechnicalSessionGUIDCacheNew # Set updated Cache # Clean old items in AgentActivityReturnDict > GUIDStr > ReturnedByDatetime lTechnicalAgentActivityReturnDictNew = {} for lItemKeyStr in inGSettings["AgentActivityReturnDict"]: lItemValue = inGSettings["AgentActivityReturnDict"][lItemKeyStr] if (lNowDatetime - lItemValue["ReturnedByDatetime"]).total_seconds() < inGSettings["Autocleaner"]["AgentActivityReturnLifetimeSecFloat"]: # Add if lifetime is ok lTechnicalAgentActivityReturnDictNew[lItemKeyStr]=lItemValue # Lifetime is ok - set else: if lL: lL.debug(f"AgentActivityReturnDict lItemKeyStr: Lifetime is expired. Remove from gSettings") # Info inGSettings["AgentActivityReturnDict"] = lTechnicalAgentActivityReturnDictNew # Set updated Cache # # # # # # # # # # # # # # # # # # # # # # # # # # from .. import __version__ # Get version from the package def Start(inDumpRestoreBool = True, inRunAsAdministratorBool = True): Orchestrator(inDumpRestoreBool = True, inRunAsAdministratorBool = True)
[документация]def Orchestrator(inGSettings=None, inDumpRestoreBool = True, inRunAsAdministratorBool = True): """L+,W+: Инициализация ядра Оркестратора (всех потоков) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inDumpRestoreBool: True - Восстановить информацию о RDP сессиях и StorageDict; False - не восстанавливать :param inRunAsAdministratorBool: True - Проверить права администратратора и обеспечить их; False - Не обеспечивать права администратора """ lL = inGSettings["Logger"] # https://stackoverflow.com/questions/130763/request-uac-elevation-from-within-a-python-script License.ConsoleVerify() if not OrchestratorIsAdmin() and inRunAsAdministratorBool==True and CrossOS.IS_WINDOWS_BOOL: OrchestratorRerunAsAdmin() else: # Code of your program here inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings #mGlobalDict = Settings.Settings(sys.argv[1]) global gSettingsDict gSettingsDict = inGSettings # Alias for old name in alg inGSettings["VersionStr"] = __version__ #Logger alias lL = gSettingsDict["Logger"] Processor.gSettingsDict = gSettingsDict Timer.gSettingsDict = gSettingsDict Timer.Processor.gSettingsDict = gSettingsDict #Backward compatibility - restore in Orc def if old def if inDumpRestoreBool == True: OrchestratorSessionRestore(inGSettings=inGSettings) # Init SettingsUpdate defs from file list (after RDP restore) lSettingsUpdateFilePathList = gSettingsDict.get("OrchestratorStart", {}).get("DefSettingsUpdatePathList",[]) lSubmoduleFunctionName = "SettingsUpdate" lSettingsPath = "\\".join(os.path.join(os.getcwd(), __file__).split("\\")[:-1]) for lModuleFilePathItem in lSettingsUpdateFilePathList: # Import defs with try catch try: # Try to init - go next if error and log in logger 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)(gSettingsDict) except Exception as e: if lL: lL.exception(f"Ошибка при инициализации .py файлов в оркестраторе '{lModuleFilePathItem}'") # Turn on backward compatibility BackwardCompatibility.Update(inGSettings= gSettingsDict) # Append Orchestrator def to ProcessorDictAlias lModule = sys.modules[__name__] lModuleDefList = dir(lModule) for lItemDefNameStr in lModuleDefList: # Dont append alias for defs Orchestrator and ___deprecated_orchestrator_start__ if lItemDefNameStr not in ["Orchestrator", "___deprecated_orchestrator_start__"]: lItemDef = getattr(lModule,lItemDefNameStr) if callable(lItemDef): inGSettings["ProcessorDict"]["AliasDefDict"][lItemDefNameStr]=lItemDef #Load all defs from sys.modules ActivityItemDefAliasModulesLoad() #Инициализация настроечных параметров gSettingsDict["ServerDict"]["WorkingDirectoryPathStr"] = os.getcwd() # Set working directory in g settings # Init the RobotScreenActive in another thread lRobotScreenActiveThread = threading.Thread(target= Monitor.CheckScreen) lRobotScreenActiveThread.daemon = True # Run the thread in daemon mode. lRobotScreenActiveThread.setName("SCREEN_ACTIVE") lRobotScreenActiveThread.start() # Start the thread execution. if lL: lL.info("Модуль активного рабочего стола инициализирован") #Logging # Init hotkey to restart orchestrator (LEFT CTRL + LEFT ALT + X) keyboard.add_hotkey("ctrl+alt+x", OrchestratorRestart) if lL: lL.info("Хоткей для перезапуска --> CTRL + ALT + X") #Logging # Init the RobotRDPActive in another thread lRobotRDPThreadControlDict = {"ThreadExecuteBool":True} # inThreadControlDict = {"ThreadExecuteBool":True} lRobotRDPActiveThread = threading.Thread(target= RobotRDPActive.RobotRDPActive, kwargs={"inGSettings":gSettingsDict, "inThreadControlDict":lRobotRDPThreadControlDict}) lRobotRDPActiveThread.daemon = True # Run the thread in daemon mode. lRobotRDPActiveThread.setName("RDP_CONNECT") lRobotRDPActiveThread.start() # Start the thread execution. if lL: lL.info("Модуль подключения по РДП инициализированн") #Logging # Init autocleaner in another thread lAutocleanerThread = threading.Thread(target= GSettingsAutocleaner, kwargs={"inGSettings":gSettingsDict}) lAutocleanerThread.daemon = True # Run the thread in daemon mode. lAutocleanerThread.setName("AUTOCLEANER") lAutocleanerThread.start() # Start the thread execution. if lL: lL.info("Модуль автоочистки инициализирован") #Logging # Set flag that orchestrator has been initialized inGSettings["HiddenIsOrchestratorInitializedBool"] = True # Orchestrator start activity if lL: lL.info("Исполнение списка Активностей") #Logging for lActivityItem in gSettingsDict["OrchestratorStart"]["ActivityList"]: # Processor.ActivityListOrDict(lActivityItem) Processor.ActivityListExecute(inGSettings=gSettingsDict,inActivityList=[BackwardCompatibility.v1_2_0_ProcessorOld2NewActivityDict(lActivityItem)]) # Processor thread lProcessorThread = threading.Thread(target= Processor.ProcessorRunSync, kwargs={"inGSettings":gSettingsDict, "inRobotRDPThreadControlDict":lRobotRDPThreadControlDict}) lProcessorThread.daemon = True # Run the thread in daemon mode. lProcessorThread.setName("PROCESSOR") lProcessorThread.start() # Start the thread execution. if lL: lL.info("Модуль процессора инициализирован") #Logging # Processor monitor thread lProcessorMonitorThread = threading.Thread(target= Processor.ProcessorMonitorRunSync, kwargs={"inGSettings":gSettingsDict}) lProcessorMonitorThread.daemon = True # Run the thread in daemon mode. lProcessorMonitorThread.setName("PROCESSOR_MONITOR") lProcessorMonitorThread.start() # Start the thread execution. if lL: lL.info("Модуль контроля процессора инициализирован") #Logging # Scheduler loop lSchedulerThread = threading.Thread(target= __deprecated_orchestrator_loop__) lSchedulerThread.daemon = True # Run the thread in daemon mode. lSchedulerThread.setName("SCHEDULER_OLD") lSchedulerThread.start() # Start the thread execution. if lL: lL.info("Модуль расписания (старая версия) инициализирован") #Logging # Schedule (new) loop lScheduleThread = threading.Thread(target= __schedule_loop__) lScheduleThread.daemon = True # Run the thread in daemon mode. lScheduleThread.setName("SCHEDULER_NEW") lScheduleThread.start() # Start the thread execution. if lL: lL.info("Модуль расписания (новая версия) инициализирован") #Logging # Restore state for process for lProcessKeyTuple in inGSettings["ManagersProcessDict"]: lProcess = inGSettings["ManagersProcessDict"][lProcessKeyTuple] lProcess.StatusCheckIntervalRestore() lThread = threading.Thread(target= lProcess.StatusRestore) lThread.setName("MANAGER_PROCESS_RESTORE") lThread.start() # Init debug thread (run if "init_dubug" file exists) Debugger.LiveDebugCheckThread(inGSettings=GSettingsGet()) #Инициализация сервера (инициализация всех интерфейсов) Server.InitFastAPI() lListenDict = gSettingsDict.get("ServerDict",{}).get("ListenDict",{}) for lItemKeyStr in lListenDict: lItemDict = lListenDict[lItemKeyStr] lItemDict["ServerInstance"]=Server.app Server.InitUvicorn(inHostStr=lItemDict["AddressStr"], inPortInt=lItemDict["PortInt"], inSSLCertPathStr=lItemDict["CertFilePEMPathStr"], inSSLKeyPathStr=lItemDict["KeyFilePathStr"], inSSLPasswordStr=None)
def __schedule_loop__(): while True: schedule.run_pending() time.sleep(3) # Backward compatibility below to 1.2.7 def __deprecated_orchestrator_loop__(): lL = OrchestratorLoggerGet() inGSettings = GSettingsGet() lDaemonLoopSeconds = gSettingsDict["SchedulerDict"]["CheckIntervalSecFloat"] lDaemonActivityLogDict = {} # Словарь отработанных активностей, ключ - кортеж (<activityType>, <datetime>, <processPath || processName>, <processArgs>) lDaemonLastDateTime = datetime.datetime.now() gDaemonActivityLogDictRefreshSecInt = 10 # The second period for clear lDaemonActivityLogDict from old items gDaemonActivityLogDictLastTime = time.time() # The second perioad for clean lDaemonActivityLogDict from old items while True: try: lCurrentDateTime = datetime.datetime.now() # Циклический обход правил lFlagSearchActivityType = True # Periodically clear the lDaemonActivityLogDict if time.time() - gDaemonActivityLogDictLastTime >= gDaemonActivityLogDictRefreshSecInt: gDaemonActivityLogDictLastTime = time.time() # Update the time for lIndex, lItem in enumerate(lDaemonActivityLogDict): if lItem["ActivityEndDateTime"] and lCurrentDateTime <= lItem["ActivityEndDateTime"]: pass # Activity is actual - do not delete now else: # remove the activity - not actual lDaemonActivityLogDict.pop(lIndex, None) lIterationLastDateTime = lDaemonLastDateTime # Get current datetime before iterator (need for iterate all activities in loop) # Iterate throught the activity list for lIndex, lItem in enumerate(gSettingsDict["SchedulerDict"]["ActivityTimeList"]): try: # Prepare GUID of the activity lGUID = None if "GUID" in lItem and lItem["GUID"]: lGUID = lItem["GUID"] else: lGUID = str(uuid.uuid4()) lItem["GUID"] = lGUID # Проверка дней недели, в рамках которых можно запускать активность lItemWeekdayList = lItem.get("WeekdayList", [0, 1, 2, 3, 4, 5, 6]) if lCurrentDateTime.weekday() in lItemWeekdayList: if lFlagSearchActivityType: ####################################################################### # Branch 1 - if has TimeHH:MM ####################################################################### if "TimeHH:MMStr" in lItem: # Вид активности - запуск процесса # Сформировать временной штамп, относительно которого надо будет проверять время # часовой пояс пока не учитываем lActivityDateTime = datetime.datetime.strptime(lItem["TimeHH:MMStr"], "%H:%M") lActivityDateTime = lActivityDateTime.replace(year=lCurrentDateTime.year, month=lCurrentDateTime.month, day=lCurrentDateTime.day) # Убедиться в том, что время наступило if ( lActivityDateTime >= lDaemonLastDateTime and lCurrentDateTime >= lActivityDateTime): # Log info about activity if lL: lL.info( f"Модуль расписания (старая версия):: Активность была инициализирована в отдельном потоке. В целях информационной безопасности параметры недоступны для просмотра") # Logging # Do the activity lThread = threading.Thread(target=Processor.ActivityListExecute, kwargs={"inGSettings": inGSettings, "inActivityList": lItem["ActivityList"]}) lThread.setName("SCHEDULER_OLD_ACTIVITY") lThread.start() lIterationLastDateTime = datetime.datetime.now() # Set the new datetime for the new processor activity except Exception as e: if lL: lL.exception( f"Модуль расписания (старая версия): Ошибка. Элемент ActivityTimeItem: {lItem}") lDaemonLastDateTime = lIterationLastDateTime # Set the new datetime for the new processor activity # Уснуть до следующего прогона time.sleep(lDaemonLoopSeconds) except Exception as e: if lL: lL.exception(f"Модуль расписания (старая версия): Глобальная ошибка инициализации - обратитесь в тех. поддержку pyOpenRPA") # Backward compatibility below to 1.2.0 def __deprecated_orchestrator_start__(): lSubmoduleFunctionName = "Settings" lFileFullPath = sys.argv[1] lModuleName = (lFileFullPath.split("\\")[-1])[0:-3] lTechSpecification = importlib.util.spec_from_file_location(lModuleName, lFileFullPath) lTechModuleFromSpec = importlib.util.module_from_spec(lTechSpecification) lTechSpecificationModuleLoader = lTechSpecification.loader.exec_module(lTechModuleFromSpec) gSettingsDict = None if lSubmoduleFunctionName in dir(lTechModuleFromSpec): # Run SettingUpdate function in submodule gSettingsDict = getattr(lTechModuleFromSpec, lSubmoduleFunctionName)() ################################################# Orchestrator(inGSettings=gSettingsDict) # Call the orchestrator