OPEN_RPA_PUBLIC
/
ORPA-pyOpenRPA
2
0
Fork 1
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

Перевод в процессе

Browse Source
dev-linux
Ivan Maslov 2 years ago
parent ba49f6682d
commit 5b6f0fa103
1 changed files with 132 additions and 119 deletions
Show Stats Download Patch File Download Diff File

245
Sources/pyOpenRPA/Orchestrator/__Orchestrator__.py
Unescape View File

@ -563,7 +563,7 @@ def OrchestratorScheduleGet() -> schedule:
# Однопоточный schedule # Однопоточный schedule
Orchestrator.OrchestratorScheduleGet().every(5).seconds.do(lProcess.StatusCheckStart) Orchestrator.OrchestratorScheduleGet().every(5).seconds.do(lProcess.StatusCheckStart)
#Многопоточный schedule. См. описание Orchestrator.OrchestratorThreadStart #Многопоточный schedule. cм. описание Orchestrator.OrchestratorThreadStart
Orchestrator.OrchestratorScheduleGet().every(5).seconds.do(Orchestrator.OrchestratorThreadStart,lProcess.StatusCheckStart) Orchestrator.OrchestratorScheduleGet().every(5).seconds.do(Orchestrator.OrchestratorThreadStart,lProcess.StatusCheckStart)
:return: schedule объект :return: schedule объект
@ -574,12 +574,12 @@ def OrchestratorScheduleGet() -> schedule:
def OrchestratorThreadStart(inDef, *inArgList, **inArgDict): def OrchestratorThreadStart(inDef, *inArgList, **inArgDict):
""" """
Execute def in new thread and pass some args with list and dict types Запустить функцию в отдельном потоке. В таком случае получить результат выполнения функции можно только через общие переменные. (Например через GSettings)
:param inDef: Python Def :param inDef: Python функция
:param inArgList: args as list :param inArgList: Список неименованных аргументов функции inDef
:param inArgDict: args as dict :param inArgDict: Словарь именованных аргументов функции inDef
:return: threading.Thread object :return: threading.Thread экземпляр
""" """
lDefThread = threading.Thread(target=inDef,args=inArgList,kwargs=inArgDict) lDefThread = threading.Thread(target=inDef,args=inArgList,kwargs=inArgDict)
lDefThread.start() lDefThread.start()
@ -587,9 +587,9 @@ def OrchestratorThreadStart(inDef, *inArgList, **inArgDict):
def OrchestratorIsAdmin(): def OrchestratorIsAdmin():
""" """
Check if Orchestrator process is running as administrator Проверить, запущен ли Оркестратор с правами администратора. Права администратора нужны Оркестратору только для контроля графической сессии, на которой он запущен. Если эти права выделить индивидуально, то права администратора будут необязательны.
:return: True - run as administrator; False - not as administrator :return: True - Запущен с правами администратора; False - Не запущен с правами администратора
""" """
try: try:
return ctypes.windll.shell32.IsUserAnAdmin() return ctypes.windll.shell32.IsUserAnAdmin()
@ -597,17 +597,20 @@ def OrchestratorIsAdmin():
return False return False
def OrchestratorIsInited() -> bool: def OrchestratorIsInited() -> bool:
"""Check if Orchestrator initial actions were processed """
Проверить, было ли проинициализировано ядро Оркестратора
:return: True - orc is already inited; False - else :return: True - Ядро Оркестратора было проинициализировано; False - Требуется время на инициализацию
:rtype: bool :rtype: bool
""" """
return Core.IsOrchestratorInitialized(inGSettings=GSettingsGet()) return Core.IsOrchestratorInitialized(inGSettings=GSettingsGet())
def OrchestratorInitWait() -> None: def OrchestratorInitWait() -> None:
"""Wait thread while orc will process initial action. """
ATTENTION: DO NOT CALL THIS DEF IN THREAD WHERE ORCHESTRATOR MUST BE INITIALIZED - INFINITE LOOP Ожидать инициализацию ядра Оркестратора
!ВНИМАНИЕ!: НИ В КОЕМ СЛУЧАЕ НЕ ЗАПУСКАТЬ ЭТУ ФУНКЦИЮ В ОСНОВНОМ ПОТОКЕ, ГДЕ ПРОИСХОДИТ ИНИЦИАЛИЗАЦИЯ ЯДРА ОРКЕСТРАТОРА - ВОЗНИКНЕТ ВЕЧНЫЙ ЦИКЛ
""" """
lIntervalSecFloat = 0.5 lIntervalSecFloat = 0.5
while not OrchestratorIsInited(): while not OrchestratorIsInited():
@ -616,9 +619,9 @@ def OrchestratorInitWait() -> None:
def OrchestratorRerunAsAdmin(): def OrchestratorRerunAsAdmin():
""" """
Check if not admin - then rerun orchestrator as administrator Перезапустить Оркестратор с правами локального администратора. Права администратора нужны Оркестратору только для контроля графической сессии, на которой он запущен. Если эти права выделить индивидуально, то права администратора будут необязательны.
:return: True - run as administrator; False - not as administrator :return: True - Запущен с правами администратора; False - Не запущен с правами администратора
""" """
if not OrchestratorIsAdmin(): if not OrchestratorIsAdmin():
# Re-run the program with admin rights # Re-run the program with admin rights
@ -628,20 +631,21 @@ def OrchestratorRerunAsAdmin():
def OrchestratorPySearchInit(inGlobPatternStr, inDefStr = None, inDefArgNameGSettingsStr = None, inAsyncInitBool = False): def OrchestratorPySearchInit(inGlobPatternStr, inDefStr = None, inDefArgNameGSettingsStr = None, inAsyncInitBool = False):
""" """
Search the py files by the glob and do the safe init (in try except). Also add inited module in sys.modules as imported (module name = file name without extension). Выполнить поиск и инициализацию пользовательских .py файлов в Оркестраторе (например панелей управления роботов)
You can init CP in async way!
Добавляет инициализированный модуль в пространство sys.modules как imported (имя модуля = имя файла без расширения).
.. code-block:: python .. code-block:: python
# USAGE VAR 1 (without the def auto call) # ВАРИАНТ ИСПОЛЬЗОВАНИЯ 1 (инициализация модуля py без вызова каких-либо функций внутри)
# Autoinit control panels starts with CP_ # автоинициализация всех .py файлов, с префиксом CP_, которые расположены в папке ControlPanel
Orchestrator.OrchestratorPySearchInit(inGlobPatternStr="ControlPanel\\CP_*.py") Orchestrator.OrchestratorPySearchInit(inGlobPatternStr="ControlPanel\\CP_*.py")
# USAGE VAR 2 (with the def auto call) - for the backward compatibility CP for the Orchestrator ver. < 1.2.7 # ВАРИАНТ ИСПОЛЬЗОВАНИЯ 2 (инициализация модуля py с вызовом функции внутри) - преимущественно для обратной совместимости старых версий панелей управления < 1.2.7
# Autoinit control panels starts with CP_ # автоинициализация всех .py файлов, с префиксом CP_, которые расположены в папке ControlPanel
Orchestrator.OrchestratorPySearchInit(inGlobPatternStr="ControlPanel\\CP_*.py", inDefStr="SettingsUpdate", inDefArgNameGSettingsStr="inGSettings") Orchestrator.OrchestratorPySearchInit(inGlobPatternStr="ControlPanel\\CP_*.py", inDefStr="SettingsUpdate", inDefArgNameGSettingsStr="inGSettings")
# INFO: The code above will replace the code below # ДЛЯ СПРАВКИ & ИСТОРИИ: Код выше позволил отказаться от блока кода ниже для каждого .py файла
## !!! For Relative import !!! CP Version Check ## !!! For Relative import !!! CP Version Check
try: try:
sys.path.insert(0,os.path.abspath(os.path.join(r""))) sys.path.insert(0,os.path.abspath(os.path.join(r"")))
@ -650,11 +654,11 @@ def OrchestratorPySearchInit(inGlobPatternStr, inDefStr = None, inDefArgNameGSet
except Exception as e: except Exception as e:
gSettings["Logger"].exception(f"Exception when init CP. See below.") gSettings["Logger"].exception(f"Exception when init CP. See below.")
:param inGlobPatternStr: example"..\\*\\*\\*X64*.cmd" :param inGlobPatternStr: Пример "..\\*\\*\\*_CP*.py"
:param inDefStr: OPTIONAL The string name of the def. For backward compatibility if you need to auto call some def from initialized module :param inDefStr: ОПЦИОНАЛЬНО Строковое наименование функции. Преимущественно для обратной совместимости
:param inDefArgNameGSettingsStr: OPTIONAL The name of the GSettings argument in def (if exists) :param inDefArgNameGSettingsStr: ОПЦИОНАЛЬНО Наименование аргумента, в который требуется передать GSettings (если необходимо)
:param inAsyncInitBool: OPTIONAL True - init py modules in many threads - parallel execution. False (default) - sequence execution :param inAsyncInitBool: ОПЦИОНАЛЬНО True - Инициализация py модулей в отдельных параллельных потоках - псевдопараллельное выполнение. False - последовательная инициализация
:return: { "ModuleNameStr":{"PyPathStr": "", "Module": ...}, ...} :return: Сведения об инициализированных модулях в структуре: { "ModuleNameStr":{"PyPathStr": "", "Module": ...}, ...}
""" """
# # # # # # # # # # # # # # # #
@ -698,16 +702,19 @@ def OrchestratorPySearchInit(inGlobPatternStr, inDefStr = None, inDefArgNameGSet
def OrchestratorSessionSave(inGSettings=None): def OrchestratorSessionSave(inGSettings=None):
""" """
Orchestrator session save in file Сохранить состояние Оркестратора (для дальнейшего восстановления в случае перезапуска):
(from version 1.2.7)
- RDP сессий, которые контролирует Оркестратор
- Хранилища DataStorage в глобальном словаре настроек GSettings. DataStorage поддерживает хранение объектов Python
(до версии 1.2.7)
_SessionLast_GSettings.pickle (binary) _SessionLast_GSettings.pickle (binary)
(above the version 1.2.7) (начиная с версии 1.2.7)
_SessionLast_RDPList.json (encoding = "utf-8") _SessionLast_RDPList.json (encoding = "utf-8")
_SessionLast_StorageDict.pickle (binary) _SessionLast_StorageDict.pickle (binary)
:param inGSettings: Global settings dict (singleton) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон)
:return: True every time
""" """
inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings
lL = inGSettings["Logger"] lL = inGSettings["Logger"]
@ -740,15 +747,19 @@ def OrchestratorSessionSave(inGSettings=None):
def OrchestratorSessionRestore(inGSettings=None): def OrchestratorSessionRestore(inGSettings=None):
""" """
Check _SessioLast... files in working directory. if exist - load into gsettings Восстановить состояние Оркестратора, если ранее состояние Оркестратора было сохранено с помощью функции OrchestratorSessionSave:
(from version 1.2.7)
- RDP сессий, которые контролирует Оркестратор
- Хранилища DataStorage в глобальном словаре настроек GSettings. DataStorage поддерживает хранение объектов Python
(до версии 1.2.7)
_SessionLast_GSettings.pickle (binary) _SessionLast_GSettings.pickle (binary)
(above the version 1.2.7)
(начиная с версии 1.2.7)
_SessionLast_RDPList.json (encoding = "utf-8") _SessionLast_RDPList.json (encoding = "utf-8")
_SessionLast_StorageDict.pickle (binary) _SessionLast_StorageDict.pickle (binary)
:param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон)
:return:
""" """
inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings
lL = inGSettings.get("Logger",None) lL = inGSettings.get("Logger",None)
@ -785,33 +796,47 @@ def OrchestratorSessionRestore(inGSettings=None):
def UACKeyListCheck(inRequest, inRoleKeyList) -> bool: def UACKeyListCheck(inRequest, inRoleKeyList) -> bool:
""" """
Check is client is has access for the key list Проверить права доступа для пользователя запроса по списку ключей до права.
:param inRequest: request handler (from http.server import BaseHTTPRequestHandler) .. code-block:: python
:param inRoleKeyList:
:return: bool # ВАРИАНТ ИСПОЛЬЗОВАНИЯ 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) return inRequest.UACClientCheck(inRoleKeyList=inRoleKeyList)
def UACUserDictGet(inRequest) -> dict: def UACUserDictGet(inRequest) -> dict:
""" """
Return user UAC hierarchy dict of the inRequest object. Empty dict - superuser access Вернуть UAC (User Access Control) словарб доступов для пользователя, который отправил запрос. Пустой словарь - супердоступ (доступ ко всему)
:param inRequest: request handler (from http.server import BaseHTTPRequestHandler) :param inRequest: Экземпляр request (from http.server import BaseHTTPRequestHandler)
:return: user UAC hierarchy dict :return: Словарь доступов пользователя. Пустой словарь - супердоступ (доступ ко всему)
""" """
return inRequest.UserRoleHierarchyGet() # get the Hierarchy return inRequest.UserRoleHierarchyGet() # get the Hierarchy
def UACUpdate(inADLoginStr, inADStr="", inADIsDefaultBool=True, inURLList=None, inRoleHierarchyAllowedDict=None, inGSettings = None): def UACUpdate(inADLoginStr, inADStr="", inADIsDefaultBool=True, inURLList=None, inRoleHierarchyAllowedDict=None, inGSettings = None):
""" """
Update user access (UAC) Дообогащение словаря доступа UAC пользователя inADStr\\inADLoginStr. Если ранее словарь не устанавливался, то по умолчанию он {}. Далее идет дообогащение теми ключами, которые присутствуют в inRoleHierarchyAllowedDict
:param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон)
:param inADLoginStr: :param inADLoginStr: Логин пользователя
:param inADStr: :param inADStr: Домен пользователя. Если пусто - локальный компьютер или домен по-умолчанию, который настроен в ОС
:param inADIsDefaultBool: :param inADIsDefaultBool: True - домен настроен по умолчанию; False - домен требуется обязательно указывать
:param inURLList: :param inURLList: Список разрешенных URL для пользователя. Для добавления URL рекомендуем использовать функции WebURLConnectDef, WebURLConnectFile, WebURLConnectFolder
:param inRoleHierarchyAllowedDict: Формат: {
"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 inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings
lUserTurple = (inADStr.upper(),inADLoginStr.upper()) # Create turple key for inGSettings["ServerDict"]["AccessUsers"]["RuleDomainUserDict"] lUserTurple = (inADStr.upper(),inADLoginStr.upper()) # Create turple key for inGSettings["ServerDict"]["AccessUsers"]["RuleDomainUserDict"]
@ -840,10 +865,10 @@ def UACUpdate(inADLoginStr, inADStr="", inADIsDefaultBool=True, inURLList=None,
def UACSuperTokenUpdate(inSuperTokenStr, inGSettings=None): def UACSuperTokenUpdate(inSuperTokenStr, inGSettings=None):
""" """
Add supertoken for the all access (it is need for the robot communication without human) Добавить супертокен (полный доступ). Супертокены используются для подключения к Оркестратору по API
:param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон)
:param inSuperTokenStr: :param inSuperTokenStr: Кодовая строковая комбинация, которая будет предоставлять доступ роботу / агенту для взаимодействия с Оркестратором по API
""" """
inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings
lLoginStr = "SUPERTOKEN" lLoginStr = "SUPERTOKEN"
@ -855,31 +880,26 @@ def UACSuperTokenUpdate(inSuperTokenStr, inGSettings=None):
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# OrchestratorWeb defs # OrchestratorWeb defs
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
def WebURLIndexChange(inURLIndexStr:str ="/"): def WebURLIndexChange(inURLIndexStr:str ="/"):
"""Change the index page of the orchestrator if you dont want the '/' (main) path """
Изменить адрес главной страницы Оркестратора. По умолчанию '/'
:param inURLIndexStr: New url for the index page of the orchestrator, defaults to "/" :param inURLIndexStr: Новый адрес главной страницы Оркестратора.
:type inURLIndexStr: str, optional :type inURLIndexStr: str, опционально
""" """
GSettingsGet()["ServerDict"]["URLIndexStr"] = inURLIndexStr GSettingsGet()["ServerDict"]["URLIndexStr"] = inURLIndexStr
def WebURLConnectDef(inMethodStr, inURLStr, inMatchTypeStr, inDef, inContentTypeStr="application/octet-stream", inGSettings = None, inUACBool = None): def WebURLConnectDef(inMethodStr, inURLStr, inMatchTypeStr, inDef, inContentTypeStr="application/octet-stream", inGSettings = None, inUACBool = None):
""" """
Connect URL to DEF Подключить функцию Python к URL.
"inMethodStr":"GET|POST",
"inURLStr": "/index", #URL of the request
"inMatchTypeStr": "", #"BeginWith|Contains|Equal|EqualCase",
"inContentTypeStr": "", #HTTP Content-type
"inDef": None #Function with str result
: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: Глобальный словарь настроек Оркестратора (синглтон) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон)
:param inMethodStr: "GET|POST",
:param inURLStr: example "/index", #URL of the request
:param inMatchTypeStr: #"BeginWith|Contains|Equal|EqualCase",
:param inDef: def arg allowed list: 2:[inRequest, inGSettings], 1: [inRequest], 0: []
:param inContentTypeStr: default: "application/octet-stream"
:param inUACBool: default: None; True - check user access before do this URL item. None - get Server flag if ask user
""" """
inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings
lURLItemDict = { lURLItemDict = {
@ -898,20 +918,15 @@ def WebURLConnectDef(inMethodStr, inURLStr, inMatchTypeStr, inDef, inContentType
def WebURLConnectFolder(inMethodStr, inURLStr, inMatchTypeStr, inFolderPathStr, inGSettings = None, inUACBool = None, inUseCacheBool= False): def WebURLConnectFolder(inMethodStr, inURLStr, inMatchTypeStr, inFolderPathStr, inGSettings = None, inUACBool = None, inUseCacheBool= False):
""" """
Connect URL to Folder Подключить папку к URL.
"inMethodStr":"GET|POST",
"inURLStr": "/Folder/", #URL of the request
"inMatchTypeStr": "", #"BeginWith|Contains|Equal|EqualCase",
"inFolderPathStr": "", #Absolute or relative path
"inUACBool"
:param inMethodStr: Метод доступа по URL "GET" || "POST"
:param inURLStr: URL адрес. Пример "/index"
:param inMatchTypeStr: Тип соответсвия строки URL с inURLStr: "BeginWith" || "Contains" || "Equal" || "EqualCase" || "EqualNoParam"
:param inFolderPathStr: Путь к папке на диске, в которой искать файл и возвращать пользователю по HTTP
:param inUACBool: True - Выполнять проверку прав доступа пользователя перед отправкой ответа; False - не выполнять проверку прав доступа пользователя
:param inUseCacheBool: True - выполнить кэширование страницы, чтобы в следющих запросах открыть быстрее; False - не кэшировать
:param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон)
:param inMethodStr:
:param inURLStr:
:param inMatchTypeStr:
:param inFolderPathStr:
:param inUACBool: default: None; True - check user access before do this URL item. None - get Server flag if ask user
:param inUseCacheBool: True - cache this page - dont open ever
""" """
inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings
# Check if last symbol is "/" - append if not exist # Check if last symbol is "/" - append if not exist
@ -934,20 +949,18 @@ def WebURLConnectFolder(inMethodStr, inURLStr, inMatchTypeStr, inFolderPathStr,
def WebURLConnectFile(inMethodStr, inURLStr, inMatchTypeStr, inFilePathStr, inContentTypeStr=None, inGSettings = None, inUACBool = None, inUseCacheBool = False): def WebURLConnectFile(inMethodStr, inURLStr, inMatchTypeStr, inFilePathStr, inContentTypeStr=None, inGSettings = None, inUACBool = None, inUseCacheBool = False):
""" """
Connect URL to File
"inMethodStr":"GET|POST",
"inURLStr": "/index", #URL of the request
"inMatchTypeStr": "", #"BeginWith|Contains|Equal|EqualCase",
"inFolderPathStr": "", #Absolute or relative path
Подключить файл к 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: Глобальный словарь настроек Оркестратора (синглтон) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон)
:param inMethodStr:
:param inURLStr:
:param inMatchTypeStr:
:param inFilePathStr:
:param inContentTypeStr: If none - autodetect
:param inUACBool: default: None; True - check user access before do this URL item. None - get Server flag if ask user
:param inUseCacheBool: True - cache this page - dont open ever
""" """
inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings
lURLItemDict = { lURLItemDict = {
@ -965,13 +978,13 @@ def WebURLConnectFile(inMethodStr, inURLStr, inMatchTypeStr, inFilePathStr, inCo
def WebListenCreate(inServerKeyStr="Default", inAddressStr="", inPortInt=80, inCertFilePEMPathStr=None, inKeyFilePathStr=None, inGSettings = None): def WebListenCreate(inServerKeyStr="Default", inAddressStr="", inPortInt=80, inCertFilePEMPathStr=None, inKeyFilePathStr=None, inGSettings = None):
""" """
Create listen interface for the web server Настроить веб-сервер Оркестратора.
:param inAddressStr: IP адрес для прослушивания. Если "", то прослушивать запросы со всех сетевых карт. Если "127.0.0.1", то слушать запросы только с той ОС, на которой работает Оркестратор
:param inPortInt: Номер порта для прослушивания. Если HTTP - 80; Если HTTPS - 443. По умолчанию 80. Допускается установка других портов
:param inCertFilePEMPathStr: Путь файлу сертификата, сгенерированного в .pem (base64) формате. Обязателен при использовании защищенного HTTPS/SSL соединения.
:param inKeyFilePathStr: Путь к файлу закрытого ключа в base64 формате
:param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон)
:param inAddressStr: IP interface to listen
:param inPortInt: Port int to listen for HTTP default is 80; for HTTPS default is 443
:param inCertFilePEMPathStr: Path to .pem (base 64) certificate. Required for SSL connection. ATTENTION - do not use certificate with password
:param inKeyFilePathStr: Path to the private key file
:return: :return:
""" """
inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings
@ -986,13 +999,13 @@ def WebListenCreate(inServerKeyStr="Default", inAddressStr="", inPortInt=80, inC
def WebCPUpdate(inCPKeyStr, inHTMLRenderDef=None, inJSONGeneratorDef=None, inJSInitGeneratorDef=None, inGSettings = None): def WebCPUpdate(inCPKeyStr, inHTMLRenderDef=None, inJSONGeneratorDef=None, inJSInitGeneratorDef=None, inGSettings = None):
""" """
Add control panel HTML, JSON generator or JS when page init Добавить панель управления робота в Оркестратор. Для панели управления открыт доступ для использования функции-генератора HTML, генератора JSON данных, генератора JS кода.
:param inGSettings: Глобальный словарь настроек Оркестратора (синглтон) :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон)
:param inCPKeyStr: :param inCPKeyStr: Текстовый ключ панели управления. Ключ для каждой панели управления должен быть уникальным!
:param inHTMLRenderDef: :param inHTMLRenderDef: Функция Python для генерации HTML кода. Для использования Jinja2 шаблонов HTML см. pyOpenRPA.Orchestrator.Managers.ControlPanel
:param inJSONGeneratorDef: :param inJSONGeneratorDef: Функция Python для генерации JSON контейнера для отправки на клиентскую часть Оркестратора
:param inJSInitGeneratorDef: :param inJSInitGeneratorDef: Функция Python для генерации JS кода для отправки на клиентскую часть Оркестратора
""" """
lCPManager = Managers.ControlPanel(inControlPanelNameStr=inCPKeyStr, inRefreshHTMLJinja2TemplatePathStr=None) lCPManager = Managers.ControlPanel(inControlPanelNameStr=inCPKeyStr, inRefreshHTMLJinja2TemplatePathStr=None)
# CASE HTMLRender # CASE HTMLRender
@ -1005,11 +1018,11 @@ def WebCPUpdate(inCPKeyStr, inHTMLRenderDef=None, inJSONGeneratorDef=None, inJSI
def WebAuditMessageCreate(inRequest=None, inOperationCodeStr="-", inMessageStr="-"): def WebAuditMessageCreate(inRequest=None, inOperationCodeStr="-", inMessageStr="-"):
""" """
Create message string with request user details (IP, Login etc...). Very actual for IT security in big company. Создание сообщения ИТ аудита с такими сведениями как (Домен, IP, логин и тд.). Данная функция особенно актуальна в том случае, если требуется реализовать дополнительные меры контроля ИТ системы.
.. code-block:: python .. code-block:: python
# USAGE # ПРИМЕР
from pyOpenRPA import Orchestrator from pyOpenRPA import Orchestrator
lWebAuditMessageStr = Orchestrator.WebAuditMessageCreate( lWebAuditMessageStr = Orchestrator.WebAuditMessageCreate(
@ -1017,13 +1030,13 @@ def WebAuditMessageCreate(inRequest=None, inOperationCodeStr="-", inMessageStr="
inOperationCodeStr = "OP_CODE_1", inOperationCodeStr = "OP_CODE_1",
inMessageStr="Success"): inMessageStr="Success"):
# Log the WebAudit message # Логгирование сообщения
lLogger.info(lWebAuditMessageStr) lLogger.info(lWebAuditMessageStr)
:param inRequest: HTTP request handler. Optional if call def from request thread :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:param inOperationCodeStr: operation code in string format (actual for IT audit in control panels) :param inOperationCodeStr: Код операции, который принят в компании в соответствии с регламентными процедурами
:param inMessageStr: additional message after :param inMessageStr: Дополнительное сообщение, которое необходимо отправить в сообщение об ИТ аудите
:return: format "WebAudit :: DOMAIN\\USER@101.121.123.12 :: operation code :: message" :return: Формат сообщения: "WebAudit :: DOMAIN\\USER@101.121.123.12 :: operation code :: message"
""" """
try: try:
if inRequest is None: inRequest = WebRequestGet() if inRequest is None: inRequest = WebRequestGet()
@ -1039,10 +1052,10 @@ def WebAuditMessageCreate(inRequest=None, inOperationCodeStr="-", inMessageStr="
def WebRequestParseBodyBytes(inRequest=None): def WebRequestParseBodyBytes(inRequest=None):
""" """
Extract the body in bytes from the request Извлечь данные в байт виде из тела (body) HTTP запроса.
:param inRequest: inRequest from the server. Optional if call def from request thread :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:return: Bytes or None :return: Строка байт b'' или None (если тело запроса было пустым)
""" """
if inRequest is None: inRequest = WebRequestGet() if inRequest is None: inRequest = WebRequestGet()
lBodyBytes=None lBodyBytes=None
@ -1053,20 +1066,20 @@ def WebRequestParseBodyBytes(inRequest=None):
def WebRequestParseBodyStr(inRequest=None): def WebRequestParseBodyStr(inRequest=None):
""" """
Extract the body in str from the request Извлечь данные в виде строки из тела (body) HTTP запроса.
:param inRequest: inRequest from the server. Optional if call def from request thread :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:return: str or None :return: Текстовая строка '' или None (если тело запроса было пустым)
""" """
if inRequest is None: inRequest = WebRequestGet() if inRequest is None: inRequest = WebRequestGet()
return WebRequestParseBodyBytes(inRequest=inRequest).decode('utf-8') return WebRequestParseBodyBytes(inRequest=inRequest).decode('utf-8')
def WebRequestParseBodyJSON(inRequest=None): def WebRequestParseBodyJSON(inRequest=None):
""" """
Extract the body in dict/list from the request Извлечь из тела (body) запроса HTTP JSON данные и преобразовать в Dict / List структуры языка Python.
:param inRequest: inRequest from the server. Optional if call def from request thread :param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:return: dict or list :return: Словарь (dict), список (list) или None (если тело запроса было пустым)
""" """
if inRequest is None: inRequest = WebRequestGet() if inRequest is None: inRequest = WebRequestGet()
return json.loads(WebRequestParseBodyStr(inRequest=inRequest)) return json.loads(WebRequestParseBodyStr(inRequest=inRequest))

Write Preview
Loading…
Cancel
Save