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")

Transformace metadat hovoru (v4.38.0)

Metadata předávaná pomocí API metody cdr.metadata lze modifikovat pomocí callbacku. Transformátor metadat je metoda, která akceptuje uniqueid hovoru pro který se nastavují metadata a samotná metadata, která upraví: Callable[[str, dict[str, Any]], None].

Metoda se registruje pomocí ucs.api.cdr.set_metadata_transformer(callback) a ruší pomocí ucs.api.cdr.unset_metadata_transformer().

Příklad pluginu, který přetransformuje metadata {"call_code": ["Reklamace", "Pozáruční"]} na {"accountcode": "Reklamace|Pozáruční"}:

"""Transform metadata for CDR"""
from typing import Any

class MetadataTransformer:
    """Transform metadata for CDR"""

    def __init__(self, ucs):
        self.__ucs = ucs
        self.__ucs.api.cdr.set_metadata_transformer(self.transform)

    def stop(self):
        """Remove callback"""
        self.__ucs.api.cdr.unset_metadata_transformer()

    def transform(self, uniqueid: str, metadata: dict[str, Any]) -> None:
        """Transform call_code list to accountcode str"""
        call_code = metadata.pop("call_code", None)
        if call_code is not None:
            metadata["accountcode"] = "|".join(call_code)

Transformace lookupu hovoru (v4.39.0)

Výsledek dohledání jména volajícího, na základě telefonního čísla ze kterého volá, lze modifikovat pomocí callbacku. Transformátor výsledku lookupu je metoda, která akceptuje argumenty:

• lookup_id - lookupu který provedl vyhledání,
• number - telefonní číslo volajícího pro které bylo provedeno vyhledání
• data - výsledek lookupu

Callback musí mít následující signaturu:

Callable[[int, str, dict[str, Any]], dict[str, Any]]

Registrace transformátoru:

ucs.lookups.set_lookup_transformer(callback)

Deregistrace transformátoru:

ucs.lookups.unset_lookup_transformer()

Pokud transformátor selže výjimkou, UCS automaticky vrání původní (nemodifikovaná) data lookupu.

Příklad pluginu

Následující plugin předřadí jménu volajícího otazník, pokud je volající přiřazen k více společnostem. Při tomto výsledku lookupu:

{"name": "Our Customer", "companies": ["First", "Second"]}

bude přetransformováno na:

{"name": "?Our Customer", "companies": ["First", "Second"]}

---

from typing import Any

class LookupTransformer:
    def __init__(self, ucs):
        self.__ucs = ucs
        self.__ucs.lookups.set_lookup_transformer(self.transform)

    def stop(self):
      """Remove callback"""
        self.__ucs.lookups.unset_lookup_transformer()

    def transform(self, lookup_id: int, number: str, data: dict[str, Any]) -> dict[str, Any]:
        """Prefix caller name if they have multiple companies"""
        if not data:
            return {"name": "Dang!"}

        prefix = "?" if len(data["companies"]) > 1 else ""
        data["name"] = f'{prefix}{data["name"]}'
        return data