ORPA-pyOpenRPA/Orchestrator/Demo/SIMPLE_01/config.py

from pyOpenRPA.Tools import CrossOS 
import os
import sys
import time
import threading
from pyOpenRPA import Orchestrator
from pyOpenRPA.Orchestrator.Server import app 
from fastapi import Depends, Body
from fastapi.responses import PlainTextResponse
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
time.sleep(1.5) # Определить очередность в панели управления
cp_path_str = CrossOS.PathJoinList(CrossOS.PathSplitList(__file__)[:-1]) # ПАПКА ПАНЕЛИ УПРАВЛЕНИЯ (АВТОИДЕНТИФИКАЦИЯ ПО ТЕКУЩЕМУ ФАЙЛУ)
# # # # # # # # # # # # # # БЛОК # # # # # ПАРАМЕТРОВ # # # # # НАЧАЛО # # # # # # # # # # # 
# ПАРАМЕТРЫ АГЕНТА
agent_hostname_str = "OPENRPA-00001" # ЕСЛИ НЕ УКАЗЫВАТЬ - ОРКЕСТРАТОР БУДЕТ ЗАПУСКАТЬ ПРОЦЕСС НА СВОЕЙ УЗ
agent_username_str = "IMASL"
# ПАРАМЕТРЫ РОБОТА
robot_logs_path_str = os.path.join(cp_path_str, "..", "..", "Logs") # ПУТЬ К ПАПКЕ С ЛОГАМИ ВИДА гггг_мм_дд.log
robot_key_str = "SIMPLE_01" # КОД РОБОТА (ТЕКСТОВЫЙ ИДЕНТИФИКАТОР В ОБЩЕМ ПРОСТРАНСТВЕ ОРКЕСТРАТОРА)
robot_start_path_str = "notepad" # ПУТЬ К ФАЙЛУ ЗАПУСКА ПРОЦЕССА
robot_process_wo_exe_str = "notepad" # НАИМЕНОВАНИЕ ПРОЦЕССА РОБОТА !БЕЗ УПОМИНАНИЯ РАСШИРЕНИЯ .EXE!
# ПАРАМЕТРЫ ДОПОЛНИТЕЛЬНЫЕ
additional_folder_path_str = os.path.join(cp_path_str, "..", "..", "Readme") # ПУТЬ К ПАПКЕ С ДОПОЛНИТЕЛЬНЫМИ ДОКУМЕНТАМИ - ОТОБРАЖАЮТСЯ НА ПАНЕЛИ ЗАДАЧ
howto_url_str = f"/{robot_key_str}/howto.docx" # URL ССЫЛКА НА ИНСТРУКЦИЮ. НАЧИНАТЬ ОБЯЗАТЕЛЬНО С /
howto_path_str = os.path.join(cp_path_str, "how_to.docx") # ПУТЬ К ИНСТРУКЦИИ НА ДИСКЕ
# # # # # # # # # # # # # # БЛОК # # # # # ПАРАМЕТРОВ # # # # # КОНЕЦ # # # # # # # # # # # 

# ПОДКЛЮЧЕНИЕ ПАПКИ С ПАНЕЛЬЮ УПРАВЛЕНИЯ. ДОСТУПНА БУДЕТ ПО URL f"/{robot_key_str}/folder"
app.mount(f"/{robot_key_str}/folder", StaticFiles(directory=os.path.abspath(cp_path_str)), name=f"{robot_key_str}_folder")

# ПОДКЛЮЧЕНИЕ ИНСТРУКЦИИ ПОЛЬЗОВАТЕЛЯ
@app.get(path=howto_url_str,tags=[f"{robot_key_str}"])
def get_howto():
    return FileResponse(os.path.abspath(howto_path_str))

# ПОДКЛЮЧЕНИЕ ПАПКИ С ЛОГАМИ
app.mount(f"/{robot_key_str}/logs", StaticFiles(directory=os.path.abspath(robot_logs_path_str)), name=f"{robot_key_str}-logs")

# ПОДКЛЮЧЕНИЕ ПАПКИ С ДОП. ДОКУМЕНТАМИ
app.mount(f"/{robot_key_str}/additional", StaticFiles(directory=os.path.abspath(additional_folder_path_str)), name=f"{robot_key_str}-additional")
def additional_files():
    files = os.listdir(additional_folder_path_str)
    return files[::-1]

# ПРОВЕРКА ПРОЦЕССА (ЗАПУЩЕН ИЛИ ОСТАНОВЛЕН) ВНИМАНИЕ! АСИНХРОННАЯ ПРОВЕРКА, ЧТОБЫ НЕ ЗАВИСАЛ ПОТОК ОБРАБОТКИ (СЕТЕВОЕ ВЗАИМОДЕЙСТВИЕ)
robot_is_started_bool = False
def robot_is_started():
    global robot_is_started_bool
    if agent_hostname_str:
        process_list_guid_str = Orchestrator.AgentProcessWOExeUpperUserListGet(
            inHostNameStr=agent_hostname_str, inUserStr=agent_username_str)
        process_list = Orchestrator.AgentActivityItemReturnGet(inGUIDStr=process_list_guid_str, inTimeoutSecFloat=60.0)
        if robot_process_wo_exe_str.upper() in process_list: robot_is_started_bool=True
        else: robot_is_started_bool=False
    else:
        if robot_process_wo_exe_str.upper() in Orchestrator.ProcessListGet()['ProcessWOExeUpperList']: robot_is_started_bool=True
        else: robot_is_started_bool=False

robot_status_interval_sec_float = 3.0 # ИНТЕРВАЛ ПРОВЕРКИ АКТИВНОСТИ ПРОЦЕССА РОБОТА
def loop_robot_is_started():
    global robot_status_interval_sec_float
    while True:
        try:
            robot_is_started()
        except Exception as e:
            pass
        time.sleep(robot_status_interval_sec_float)

threading.Thread(target=loop_robot_is_started).start()

# ФУНКЦИЯ ЗАПУСКА ФАЙЛА
@app.post(path=f"/{robot_key_str}/action/start",tags=[f"{robot_key_str}"])
def robot_start():
    cmd_str = robot_start_path_str
    global robot_is_started_bool
    if agent_hostname_str:
        Orchestrator.AgentOSCMD(inHostNameStr=agent_hostname_str, inUserStr=agent_username_str, 
        inCMDStr=cmd_str, inRunAsyncBool=True, 
        inSendOutputToOrchestratorLogsBool=False, inCaptureBool=False)
    else:
        Orchestrator.OSCMD(inCMDStr=cmd_str,inRunAsyncBool=True)

# ФУНКЦИЯ ОСТАНОВКИ ПРОЦЕССА
@app.post(path=f"/{robot_key_str}/action/stop",tags=[f"{robot_key_str}"])
def robot_stop():
    cmd_str = f"taskkill /f /FI \"USERNAME eq %username%\" /im {robot_process_wo_exe_str}.exe"
    global robot_is_started_bool
    if agent_hostname_str:
        Orchestrator.AgentOSCMD(inHostNameStr=agent_hostname_str, inUserStr=agent_username_str, 
        inCMDStr=cmd_str, inRunAsyncBool=True, 
        inSendOutputToOrchestratorLogsBool=False, inCaptureBool=False)
    else:
        Orchestrator.OSCMD(inCMDStr=cmd_str,inRunAsyncBool=True)
    
# ИНИЦИАЛИЗАЦИЯ ПАНЕЛИ УПРАВЛЕНИЯ РОБОТОМ
cp_manager = Orchestrator.Managers.ControlPanel(inControlPanelNameStr=robot_key_str,
            inRefreshHTMLJinja2TemplatePathStr=os.path.join(cp_path_str, "index.html"), inJinja2TemplateRefreshBool = True)
cp_manager.Jinja2DataUpdateDictSet(inJinja2DataUpdateDict={"config": sys.modules[__name__]} )


# HOTFIX Orchestrator.AgentProcessWOExeUpperUserListGet
def HOTFIX_01(inHostNameStr, inUserStr, inGSettings = None):
    """L-,W+: Получить список процессов, которые выполняется на сессии Агента. Все процессы фиксируются без постфикса .exe, а также в верхнем регистре.

    ПРИМЕР РЕЗУЛЬТАТА, КОТОРЫЙ МОЖНО ПОЛУЧИТЬ ПО ГУИД ЧЕРЕЗ ФУНКЦИЮ AgentActivityItemReturnGet: ["ORCHESTRATOR", "AGENT", "CHROME", "EXPLORER", ...]

    :param inGSettings: Глобальный словарь настроек Оркестратора (синглтон)
    :param inHostNameStr: Наименование хоста, на котором запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления
    :param inUserStr: Наименование пользователя, на графической сессии которого запущен Агент. Наименования подключенных агентов доступно для просмотра в панели управления
    :return: ГУИД (GUID) строка Активности (ActivityItem). Далее можно ожидать результат этой функции по ГУИД с помощью функции AgentActivityItemReturnGet
    """
    inGSettings = Orchestrator.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 Orchestrator.AgentActivityItemAdd(inGSettings=inGSettings, inHostNameStr=inHostNameStr, inUserStr=inUserStr, inActivityItemDict=lActivityItemDict)
Orchestrator.AgentProcessWOExeUpperUserListGet = HOTFIX_01