Pluginy

Z důvodu zákaznických úprav server UCS umožňuje běh Python scriptů jako součást hlavního procesu.

Spuštění (v2.4.0)

1. Při spuštění UCS je prohledán adresář /opt/ucs/server/plugins jsou zpracovány všechny soubory mající příponu .py (kromě souboru __init__.py).
2. Soubor je naimportován a je vyhledána proměnná UCS_PLUGIN. Pokud proměnná v souboru není, pak je přeskočen.
3. Podle stringového obsahu proměnné UCS_PLUGIN se UCS pokusí vytvořit instanci dané třídy. Do konstruktoru je předán argument ucs, což je hlavní třída serveru.
4. Pokud třída obsahuje metodu start, pak je tato metoda zavolána. Tato metoda nesmí být blokující, protože pluginy nejsou automaticky vytvářeny jako thready.
5. Při reloadu pluginu nebo zastavení UCS se zavolá metoda stop, pokud jí plugin obsahuje.

#!/usr/bin/python3

"""Example plugin"""

import threading

from constants import AGENT_AUX
from events import EVENT_AGENT_STATUS

UCS_PLUGIN = 'Example'


class Example(threading.Thread):
  """Example plugin"""

  def __init__(self, ucs):
    super().__init__(self)
    self.name = 'ExamplePluginThread'
    self.ucs = ucs
    self.ucs.addCallback(EVENT_AGENT_STATUS, self.agent_status_changed)

    self.terminate = False
    self.work = threading.Event()

  def agent_status_changed(self, agent, args):
    """Called when agent change status"""
    if agent['status'] == AGENT_AUX and agent['reason'] == 'Oběd':
      self.ucs.log.info(f'Agent {agent["displayname"]} odešel na oběd')

  def run(self):
    """Thread main loop"""
    while True:
      self.work.wait(60)
      self.work.clear()
      if self.terminate:
        return

      agents = self.ucs.agents.getAgents()
      for agent in agents:
        if agent['status'] == AGENT_AUX and agent['reason'] == 'Oběd':
          self.ucs.log.info(f'Agent {agent["displayname"]} je na obědě')

  def stop(self):
    """Called on plugin reload or UCS stop"""
    self.ucs.removeCallback(EVENT_AGENT_STATUS, self.agent_status_changed)
    self.terminate = True
    self.work.set()

API metody (v2.4.0)

Pluginy je možné nahrávat a odebírat za běhu UCS (je vyžadováno oprávnění superuser). Slouží k tomu následující API metody:

• ucs.api.plugins.load(sid, plugin_name) nahraje plugin, resp. pokud plugin již existuje, pak ho zastaví a znovu nahraje.
• ucs.api.plugins.unload(sid, plugin_name) zastaví plugin.

Pozor, pokud si plugin zaregistruje callbacky, je nutné v metodě stop callbacky odregistrovat. V opačném případě na ně zůstanou reference a budou nadále provolávány.

Hook změny stavu agenta (v2.4.0)

Plugin může ovlivnit změnu stavu agenta. UCS poskytuje metodu pro registraci callbacku ucs.agents.addChangeStatusHook(callback) a pro jeho deregistraci ucs.agents.removeChangeStatusHook() při zastavení pluginu. Tuto metodu může využívat pouze jeden plugin, pokud se o registraci pokusí jiný plugin, pak je vyvolána výjimka ProcessingError.

Callbacku jsou předány tři argumenty:

• agent:agents/Agent je instance třídy agenta v okamžiku před změnou stavu.
• status:int je nový status (konstanta AGENT_STATE_ENUM).
• reason:str je důvod pro stav nepřipraven.

Callback musí vrátit status:int, reason:str nebo vyvolat výjimku ProcessingError.

• Pokud nemá dojít k ovlivnění stavu do kterého je agent přepínán, pak callback vrátí hodnoty argumentů status a reason, tak jak mu byly předány.
• Pokud má dojít k přepnutí do jiného stavu a/nebo důvodu nepřipravenosti, pak vrátí hodnoty do kterých má být agent přepnut.
• Pokud má být přepnutí zablokováno chybou, pak může plugin buď vrátit první argument s hodnotou None a druhý argument s důvodem zablokování. Pokud není důvod zablokování předán (bool(reason) is False) pak je nastaven Blocked by hook. To způsobí vyvolání výjimky ProcessingError(reason). Případně lze tuto výjimku vyvolat přímo z callbacku.
• Pokud při zpracování v callbacku dojde k jiné výjimce než ProcessingError, pak je agent přepnut do původně požadovaného stavu a je zalogován backtrace.

"""Agent status change hook example"""

import time

from constants import AGENT_READY, AGENT_AUX
from error import ProcessingError

UCS_PLUGIN = 'StatusHook'


class StatusHook:
  """Agent status change hook example"""

  def __init__(self, ucs):
    self.ucs = ucs
    self.ucs.agents.addChangeStatusHook(self.hook)

  def stop(self):
    """Called on plugin reload or UCS stop"""
    self.ucs.agents.removeChangeStatusHook()

  def hook(self, agent, new_status, new_reason):
    """Called when agent status is about to change"""

    hour = time.localtime().tm_hour

    if new_status == AGENT_AUX and new_reason == 'Oběd':
      if hour < 11:
        return None, 'Na oběd je příliš brzy'

      if hour > 17:
        raise ProcessingError('Už je čas na večeři')

    if agent.status == AGENT_OFFLINE and new_status == AGENT_READY:
      return AGENT_AUX, 'Seznámení se z provozními pokyny'

    return new_status, new_reason

Hook změny sdílených dat (v3.1.0)

Plugin může modifikovat nastavovaná data do shared data. UCS poskytuje metodu pro registraci callbacku ucs.shared.addChangeHook(identificator, callback) a pro jeho deregistraci ucs.agents.removeChangeHook(identificator) při zastavení pluginu. Tuto metodu pro daný identificator může využívat pouze jeden plugin, pokud se o registraci pokusí jiný plugin, pak je vyvolána výjimka ProcessingError.

Callbacku jsou předány tři argumenty:

• identificator:str je identificator měněných shared data.
• old_data:any jsou půvidní data.
• new_data:any jsou nově nastavovaná data.

Callback může vrátit libovolná data nebo vyvolat výjimku ProcessingError, pokud má být změna dat zablokována. Pokud při zpracování v callbacku dojde k jiné výjimce než ProcessingError, pak jsou data aktualizována podle požadavku a je zalogován backtrace s výjimkou v callbacku.

"""Shared data change hook example"""

from error import ProcessingError

UCS_PLUGIN = 'SharedDataHook'


class SharedDataHook:
  """Shared data change hook example"""

  def __init__(self, ucs):
    self.ucs = ucs
    self.ucs.shared.addChangeHook('agent_emails', self.hook)

  def stop(self):
    """Called on plugin reload or UCS stop"""
    self.ucs.shared.removeChangeHook('agent_emails')

  def hook(self, identificator, old_data, new_data):
    """Called when new shared data are about to change"""

    if not isinstance(new_data, dict):
      raise ProcessingError('Only associative array is accepted')

    data = {}
    username2id = {
      username: user_id
      for user_id, username
      in self.ucs.tree.getUsers('username').items()
    }
    for username, count in new_data.items():
      user_id = username2id.get(username)
      if user_id:
        data[user_id] = count
        old_count = old_data.get(user_id)
        if old_count is not None:
          self.ucs.log.dbug('User ID %d email count changed from %s to %s', user_id, old_count, count)
      else:
        self.ucs.log.warn('Unknown username: %s', username)

    return data

API middleware (v3.8.0)

Jednotlivé API requesty je možné obsluhovat pomocí middleware. UCS poskytuje metodu pro registraci middleware ucs.api.addMiddleware(method, middleware) a pro jeho deregistraci ucs.api.removeMiddleware(method, middleware) při zastavení pluginu. Pro jednu API metodu je možné registrovat více middleware. Middleware při volání dostává jako první argument handler (což je samotná metoda API nebo další middleware), který následují argumenty předané jako parametry API volání. Middleware může modifikovat vstupní argumenty, následně musí zavolat předaný handler nebo vyvolat výjimku. Vrací výsledek vestavěné API metody UCS, který může modifikovat.

"""Middleware plugin example"""


from error import InvalidDataInput, PermissionDenied

UCS_PLUGIN = "Middleware"


class Middleware:
    """Middleware plugin example"""

    def __init__(self, ucs):
        self.ucs = ucs
        self.ucs.api.addMiddleware("config.users.get", self.get_user)
        self.ucs.api.addMiddleware("config.users.edit", self.edit_user)
        self.ucs.api.addMiddleware("config.users.delete", self.delete_user)

    def stop(self):
        """Cleanup on UCS shutdown"""
        self.ucs.api.removeMiddleware("config.users.get", self.get_user)
        self.ucs.api.removeMiddleware("config.users.edit", self.edit_user)
        self.ucs.api.removeMiddleware("config.users.delete", self.delete_user)

    def get_user(self, handler, *args):
      """Just log middleware usage"""
      self.ucs.log.dev("Middleware get_user args: %s", args)
      result = handler(*args)
      self.ucs.log.dev("Middleware get_user result: %s", result)
      return result

    def edit_user(self, handler, user, data):
        """Keep username and password"""
        if data["id"] == 1:
            if "username" in data and data["username"] != "admin":
                raise InvalidDataInput({"username": "Admin username can't be changed"})

            if "password" in data and data["password"] and len(data["password"]) < 10:
                raise InvalidDataInput({"password": "Admin password is too short"})

        return handler(user, data)

    def delete_user(self, *args):
        """Disable user deletion"""
        raise PermissionDenied("This UCS forbids user removal")