117 lines
3.2 KiB
Python
117 lines
3.2 KiB
Python
from abc import ABCMeta, abstractmethod
|
|
from collections import deque
|
|
from enum import Enum
|
|
|
|
|
|
class Rank(Enum):
|
|
|
|
OPERATOR = 0
|
|
SUPERVISOR = 1
|
|
DIRECTOR = 2
|
|
|
|
|
|
class Employee(metaclass=ABCMeta):
|
|
|
|
def __init__(self, employee_id, name, rank, call_center):
|
|
self.employee_id = employee_id
|
|
self.name = name
|
|
self.rank = rank
|
|
self.call = None
|
|
self.call_center = call_center
|
|
|
|
def take_call(self, call):
|
|
"""Assume the employee will always successfully take the call."""
|
|
self.call = call
|
|
self.call.employee = self
|
|
self.call.state = CallState.IN_PROGRESS
|
|
|
|
def complete_call(self):
|
|
self.call.state = CallState.COMPLETE
|
|
self.call_center.notify_call_completed(self.call)
|
|
|
|
@abstractmethod
|
|
def escalate_call(self):
|
|
pass
|
|
|
|
def _escalate_call(self):
|
|
self.call.state = CallState.READY
|
|
call = self.call
|
|
self.call = None
|
|
self.call_center.notify_call_escalated(call)
|
|
|
|
|
|
class Operator(Employee):
|
|
|
|
def __init__(self, employee_id, name):
|
|
super(Operator, self).__init__(employee_id, name, Rank.OPERATOR)
|
|
|
|
def escalate_call(self):
|
|
self.call.level = Rank.SUPERVISOR
|
|
self._escalate_call()
|
|
|
|
|
|
class Supervisor(Employee):
|
|
|
|
def __init__(self, employee_id, name):
|
|
super(Operator, self).__init__(employee_id, name, Rank.SUPERVISOR)
|
|
|
|
def escalate_call(self):
|
|
self.call.level = Rank.DIRECTOR
|
|
self._escalate_call()
|
|
|
|
|
|
class Director(Employee):
|
|
|
|
def __init__(self, employee_id, name):
|
|
super(Operator, self).__init__(employee_id, name, Rank.DIRECTOR)
|
|
|
|
def escalate_call(self):
|
|
raise NotImplemented('Directors must be able to handle any call')
|
|
|
|
|
|
class CallState(Enum):
|
|
|
|
READY = 0
|
|
IN_PROGRESS = 1
|
|
COMPLETE = 2
|
|
|
|
|
|
class Call(object):
|
|
|
|
def __init__(self, rank):
|
|
self.state = CallState.READY
|
|
self.rank = rank
|
|
self.employee = None
|
|
|
|
|
|
class CallCenter(object):
|
|
|
|
def __init__(self, operators, supervisors, directors):
|
|
self.operators = operators
|
|
self.supervisors = supervisors
|
|
self.directors = directors
|
|
self.queued_calls = deque()
|
|
|
|
def dispatch_call(self, call):
|
|
if call.rank not in (Rank.OPERATOR, Rank.SUPERVISOR, Rank.DIRECTOR):
|
|
raise ValueError('Invalid call rank: {}'.format(call.rank))
|
|
employee = None
|
|
if call.rank == Rank.OPERATOR:
|
|
employee = self._dispatch_call(call, self.operators)
|
|
if call.rank == Rank.SUPERVISOR or employee is None:
|
|
employee = self._dispatch_call(call, self.supervisors)
|
|
if call.rank == Rank.DIRECTOR or employee is None:
|
|
employee = self._dispatch_call(call, self.directors)
|
|
if employee is None:
|
|
self.queued_calls.append(call)
|
|
|
|
def _dispatch_call(self, call, employees):
|
|
for employee in employees:
|
|
if employee.call is None:
|
|
employee.take_call(call)
|
|
return employee
|
|
return None
|
|
|
|
def notify_call_escalated(self, call): # ...
|
|
def notify_call_completed(self, call): # ...
|
|
def dispatch_queued_call_to_newly_freed_employee(self, call, employee): # ... |