diff --git a/.gitignore b/.gitignore index b812ddc..e2513ca 100644 --- a/.gitignore +++ b/.gitignore @@ -162,5 +162,5 @@ sandbox.py # mac *.DS_Store - +os_reports/ call_flow_data-temp/ \ No newline at end of file diff --git a/odins_spear/methods/get.py b/odins_spear/methods/get.py index 6de2ef8..b8d254f 100644 --- a/odins_spear/methods/get.py +++ b/odins_spear/methods/get.py @@ -23,6 +23,22 @@ def session(self): # ADMINISTRATORS # ADVICE OF CHARGE # ALTERNATE NUMBERS + + def user_alternate_numbers(self, user_id: str): + """Fetches a list of a user/ service such as Auto Attendant, Hunt Group, or Call Centres + alternate numebrs. + + Args: + user_id (str): Target user/ service_user_id + + Returns: + Dict: List of all alternate numbers assigned to the user/ service. + """ + + endpoint = f"/users/alternate-numbers?userId={user_id}" + + return self.requester.get(endpoint) + # ANSWER CONFIRMATION # ALTERNATE USER ID # ANNOUNCEMENTS @@ -63,26 +79,142 @@ def group_call_centers(self, service_provider_id: str, group_id: str): return self.requester.get(endpoint) + def group_call_center(self, service_user_id: str): endpoint = f"/groups/call-centers?serviceUserId={service_user_id}" return self.requester.get(endpoint) + def user_call_center(self, user_id: str): endpoint = f"/users/call-center?userId={user_id}" return self.requester.get(endpoint) - + + + def group_call_center_bounced_calls(self, service_user_id: str): + + endpoint = f"/groups/call-centers/bounced-calls?serviceUserId={service_user_id}" + + return self.requester.get(endpoint) + + + def group_call_center_forced_forwarding(self, service_user_id: str): + + endpoint = f"/groups/call-centers/forced-forwarding?serviceUserId={service_user_id}" + + return self.requester.get(endpoint) + + + def group_call_center_overflow(self, service_user_id): + + endpoint = f"/groups/call-centers/overflow?serviceUserId={service_user_id}" + + return self.requester.get(endpoint) + + + def group_call_center_stranded_calls(self, service_user_id): + + endpoint = f"/groups/call-centers/stranded-calls?serviceUserId={service_user_id}" + + return self.requester.get(endpoint) + + + def group_call_center_stranded_calls_unavailable(self, service_user_id): + + endpoint = f"/groups/call-centers/stranded-calls-unavailable?serviceUserId={service_user_id}" + + return self.requester.get(endpoint) + # CALL CONTROL # CALL FORWARDING ALWAYS -# CALL FORWARDING ALWAYS SECONDARY + + def user_call_forwarding_always(self, user_id: str): + + endpoint = f"/users/call-forwarding-always?userId={user_id}" + + return self.requester.get(endpoint) + + + def bulk_call_forwarding_always(self, service_provider_id: str, group_id: str): + + endpoint = f"/users/call-forwarding-always/bulk?serviceProviderId={service_provider_id}&groupId={group_id}" + + return self.requester.get(endpoint) + + # CALL FORWARDING BUSY + + def user_call_forwarding_busy(self, user_id: str): + + endpoint = f"/users/call-forwarding-busy?userId={user_id}" + + return self.requester.get(endpoint) + + + def bulk_call_forwarding_busy(self, service_provider_id: str, group_id: str): + + endpoint = f"/users/call-forwarding-busy/bulk?serviceProviderId={service_provider_id}&groupId={group_id}" + + return self.requester.get(endpoint) + + # CALL FORWARDING NO ANSWER + + def user_call_forwarding_no_answer(self, user_id: str): + + endpoint = f"/users/call-forwarding-no-answer?userId={user_id}" + + return self.requester.get(endpoint) + + + def bulk_call_forwarding_no_answer(self, service_provider_id: str, group_id: str): + + endpoint = f"/users/call-forwarding-no-answer/bulk?serviceProviderId={service_provider_id}&groupId={group_id}" + + return self.requester.get(endpoint) + + # CALL FORWARDING NOT REACHABLE + + def user_call_forwarding_not_reachable(self, user_id: str): + + endpoint = f"/users/call-forwarding-not-reachable?userId={user_id}" + + return self.requester.get(endpoint) + + + def bulk_call_forwarding_not_reachable(self, service_provider_id: str, group_id: str): + + endpoint = f"/users/call-forwarding-not-reachable/bulk?serviceProviderId={service_provider_id}&groupId={group_id}" + + return self.requester.get(endpoint) + + # CALL FORWARDING SELECTIVE -# CALL FORWARDING SETTINGS + + def user_call_forwarding_selective(self, user_id: str): + + endpoint = f"/users/call-forwarding-selective?userId={user_id}" + + return self.requester.get(endpoint) + + + def user_call_forwarding_selective_criterias(self, user_id: str): + + endpoint = f"/users/call-forwarding-selective/criteria?userId={user_id}" + + return self.requester.get(endpoint) + + + def user_call_forwarding_selective_criteria(self, user_id: str, criteria_name: str): + + endpoint = f"/users/call-forwarding-selective/criteria?criteriaName={criteria_name}&userId={user_id}" + + return self.requester.get(endpoint) + # CALL NOTIFY # CALL PARK # CALL PICKUP @@ -533,6 +665,21 @@ def user_report(self, user_id: str): # ROUTE LIST # ROUTING PROFILE # SCHEDULES + + def group_schedules(self, service_provider_id: str, group_id: str): + + endpoint = f"/groups/schedules?serviceProviderId={service_provider_id}&groupId={group_id}" + + return self.requester.get(endpoint) + + + def group_events(self, service_provider_id: str, group_id: str, name: str, type: str): + + endpoint = f"/groups/events?serviceProviderId={service_provider_id}&groupId={group_id}&name={name}&type={type}" + + return self.requester.get(endpoint) + + # SECURITY CLASSIFICATION # SELECTIVE CALL ACCEPTANCE # SELECTIVE CALL REJECTION @@ -553,6 +700,16 @@ def service_providers(self, reseller_id=None): return self.requester.get(endpoint) + def service_provider(self, service_provider_id: str): + """ + Args: + reseller_id (str): Only list the Service Provider IDs within the specified Reseller. + """ + + endpoint = f"/service-providers?serviceProviderId={service_provider_id}" + + return self.requester.get(endpoint) + # SERVICES def user_services_assigned(self, user_id: str): diff --git a/odins_spear/reporter.py b/odins_spear/reporter.py index af1f0ad..7c2f44e 100644 --- a/odins_spear/reporter.py +++ b/odins_spear/reporter.py @@ -1,6 +1,28 @@ +from . import reports class Reporter: - ''' generates human friendly reports - ''' + """ generates human friendly reports. + """ + def __init__(self, api) -> None: - self.api = api \ No newline at end of file + self.api = api + + + def call_flow(self, service_provider_id: str, group_id: str, number: str, number_type: str, + broadworks_entity_type: str): + """Generates a graphical flow chart of how a call to a specified number routes + through the broadworks system. + + Args: + service_provider_id (str): Service Provider/ Enterprise where group is hosted. + group_id (str): Group ID where target number for call flow is located. + number (str): Target number for call flow. + number_type (str): Type of number, options: "dn": Direct Number, "extension": Extension, "alias": Alias + broadworks_entity_type (str): Broadworks entity type target number is associated with. \ + Options: "auto_attendant": Auto Attendant, "call_center": Call Center, "hunt_group": Hunt Group, "user": User + """ + + return reports.call_flow.main(self.api, service_provider_id, group_id, number, number_type, + broadworks_entity_type) + + \ No newline at end of file diff --git a/odins_spear/reports/__init__.py b/odins_spear/reports/__init__.py new file mode 100644 index 0000000..16c4b6b --- /dev/null +++ b/odins_spear/reports/__init__.py @@ -0,0 +1,5 @@ +__all__ = [ + "call_flow" +] + +from .call_flow import main diff --git a/odins_spear/reports/call_flow.py b/odins_spear/reports/call_flow.py new file mode 100644 index 0000000..d82978e --- /dev/null +++ b/odins_spear/reports/call_flow.py @@ -0,0 +1,140 @@ +import json + +from tqdm import tqdm + +from .report_utils.graphviz_module import GraphvizModule +from .report_utils.helpers import find_entity_with_number_type +from .report_utils.parsing import call_flow_module + + +from odins_spear.store import DataStore +from odins_spear.store import broadwork_entities as bre + +def main(api, service_provider_id: str, group_id: str, number: str, number_type: str, + broadworks_entity_type: str): + + data_store = DataStore() + + print("Start.\n") + print("Fetching Service Provider & Group details.") + # Gather entities + service_provider = bre.ServiceProvider.from_dict(data=api.get.service_provider(service_provider_id)) + group = bre.Group.from_dict(service_provider=service_provider, data=api.get.group(service_provider_id, group_id)) + + data_store.store_objects(service_provider, group) + + auto_attendants = api.get.auto_attendants(service_provider_id, group_id) + for aa in tqdm(auto_attendants, desc=f"Fetching all Auto Attendants."): + auto_attendant = bre.AutoAttendant.from_dict(group=group, data=api.get.auto_attendant(aa['serviceUserId'])) + data_store.auto_attendants.append(auto_attendant) + + print("Fetching all users this may take a couple of minutes, please wait.") + users = api.get.users(service_provider_id, group_id, extended=True) + + # Captures users with the forward fucntionality + call_forward_always_users = [ + item["user"]["userId"] for item in api.get.bulk_call_forwarding_always(service_provider_id, group_id) if item["service"]["assigned"] \ + and item["data"]["isActive"] + ] + + call_forward_busy_users = [ + item["user"]["userId"] for item in api.get.bulk_call_forwarding_busy(service_provider_id, group_id) if item["service"]["assigned"] \ + and item["data"]["isActive"] + ] + + call_forward_no_answer_users = [ + item["user"]["userId"] for item in api.get.bulk_call_forwarding_no_answer(service_provider_id, group_id) if item["service"]["assigned"] \ + and item["data"]["isActive"] + ] + + call_forward_not_reachable = [ + item["user"]["userId"] for item in api.get.bulk_call_forwarding_not_reachable(service_provider_id, group_id) if item["service"]["assigned"] \ + and item["data"]["isActive"] + ] + + for u in tqdm(users, desc=f"Parsing all Users."): + user = bre.User.from_dict(group=group, data=u) + + if user.id in call_forward_always_users: + user.call_forwarding_always = str(api.get.user_call_forwarding_always(user.id)["forwardToPhoneNumber"]) + if user.id in call_forward_busy_users: + user.call_forwarding_busy = str(api.get.user_call_forwarding_busy(user.id)["forwardToPhoneNumber"]) + if user.id in call_forward_no_answer_users: + user.call_forwarding_no_answer = str(api.get.user_call_forwarding_no_answer(user.id)["forwardToPhoneNumber"]) + if user.id in call_forward_not_reachable: + user.call_forwarding_not_reachable = str(api.get.user_call_forwarding_not_reachable(user.id)["forwardToPhoneNumber"]) + + data_store.users.append(user) + + + call_centers = api.get.group_call_centers(service_provider_id, group_id) + for cc in tqdm(call_centers, desc=f"Fetching all Call Centers."): + call_center = bre.CallCenter.from_dict(group= group, data= api.get.group_call_center(cc['serviceUserId'])) + + try: + overflow_settings = api.get.group_call_center_overflow(call_center.service_user_id) + call_center.overflow_calls_action = overflow_settings["action"] + call_center.overflow_calls_transfer_to_phone_number = overflow_settings["transferPhoneNumber"] \ + if call_center.overflow_calls_action == "Transfer" else None + except Exception: + call_center.overflow_calls_action = None + call_center.overflow_calls_transfer_to_phone_number = None + + try: + stranded_calls_settings = api.get.group_call_center_stranded_calls(call_center.service_user_id) + call_center.stranded_call_unavailable_action = stranded_calls_settings["action"] + call_center.stranded_call_unavailable_transfer_to_phone_number = stranded_calls_settings["transferPhoneNumber"] \ + if call_center.stranded_call_unavailable_action == "Transfer" else None + except Exception: + call_center.stranded_call_unavailable_action = None + call_center.stranded_call_unavailable_transfer_to_phone_number = None + + try: + stranded_calls_unavailable_settings = api.get.group_call_center_stranded_calls_unavailable(call_center.service_user_id) + call_center.stranded_call_unavailable_action = stranded_calls_unavailable_settings["action"] + call_center.stranded_call_unavailable_transfer_to_phone_number = stranded_calls_unavailable_settings["transferPhoneNumber"] \ + if call_center.stranded_call_unavailable_action == "Transfer" else None + except Exception: + call_center.stranded_call_unavailable_action = None + call_center.stranded_call_unavailable_transfer_to_phone_number = None + + try: + forced_forwarding_settings = api.get.group_call_center_forced_forwarding(call_center.service_user_id) + call_center.forced_forwarding_enabled = forced_forwarding_settings["enabled"] + call_center.forced_forwarding_forward_to_phone_number = forced_forwarding_settings["forwardToPhoneNumber"] \ + if call_center.forced_forwarding_enabled else None + except Exception: + call_center.forced_forwarding_enabled = False + call_center.forced_forwarding_forward_to_phone_number = None + + data_store.call_centers.append(call_center) + + hunt_groups = api.get.group_hunt_groups(service_provider_id, group_id) + for hg in tqdm(hunt_groups, desc=f"Fetching all Hunt Groups."): + hunt_group = bre.HuntGroup.from_dict(group=group, data= api.get.group_hunt_group(hg['serviceUserId'])) + data_store.hunt_groups.append(hunt_group) + + # locate number using broadworks_entity_type to zone in on correct location + call_flow_start_node = find_entity_with_number_type( + number, + number_type.lower(), + getattr(data_store, broadworks_entity_type + "s") + ) + call_flow_start_node._start_node = True + + # Nodes used in the graph + print("Gathering nodes in flow.") + bre_nodes = call_flow_module(call_flow_start_node, data_store) + + print("Generating report.") + # build, generate, save graph + graph = GraphvizModule( + "./os_reports/" + ) + graph.generate_call_flow_graph( + bre_nodes, + number + ) + print("Saving report.") + graph._save_graph(f"Calls To {number}") + print("\nEnd.") \ No newline at end of file diff --git a/odins_spear/reports/report_utils/__init__.py b/odins_spear/reports/report_utils/__init__.py new file mode 100644 index 0000000..46a9d3d --- /dev/null +++ b/odins_spear/reports/report_utils/__init__.py @@ -0,0 +1,11 @@ +__all__ = [ + "graphviz_module", + "helpers", + "parsing", + "report_entities" +] + +from .graphviz_module import * +from .helpers import * +from .parsing import * +from .report_entities import * \ No newline at end of file diff --git a/odins_spear/reports/report_utils/graphviz_module.py b/odins_spear/reports/report_utils/graphviz_module.py new file mode 100644 index 0000000..a9c9233 --- /dev/null +++ b/odins_spear/reports/report_utils/graphviz_module.py @@ -0,0 +1,157 @@ +import graphviz + +from odins_spear.store import broadwork_entities as bre +from .report_entities import external_number + +class GraphvizModule: + + """ + Colours: + - Black: 020300 + - Orange: FFA400 - Darker: B87700 + - Red: 95190C - Darker: 5E1008 + - Purple: 5B5F97 - Darker: 3D4066 + - Redish: A23E48 - Darker: 68272E + """ + + NODE_STYLING = { + 'start' : { + 'shape': 'record', + 'style': 'filled', + 'margin': '0.2', + 'color': '#020300', + 'fontname': 'Arial', + 'fontcolor': 'white' + }, + 'exit': { + 'shape': 'record', + 'style': 'filled', + 'margin': '0.2', + 'color': '#020300', + 'fontname': 'Arial', + 'fontcolor': 'white' + }, + 'auto_attendant': { + 'shape': 'circle', + 'style': 'filled', + 'margin': '0.2', + 'color': '#5E1008', + 'fillcolor': '#FF0000', + 'fontname': 'Arial', + 'fontcolor': 'white' + }, + 'call_centre': { + 'shape': 'Mrecord', + 'style': 'filled', + 'margin': '0.2', + 'color': '#68272E', + 'fillcolor': '#FFC0CB', + 'fontname': 'Arial', + 'fontcolor': 'white' + }, + 'hunt_group': { + 'shape': 'Mrecord', + 'style': 'filled', + 'margin': '0.2', + 'color': '#3D4066', + 'fillcolor': '#9C1FE9', + 'fontname': 'Arial', + 'fontcolor': 'white' + }, + 'user': { + 'shape': 'Mrecord', + 'style': 'filled', + 'margin': '0.2', + 'color': '#B87700', + 'fillcolor': '#FBA200', + 'fontname': 'Arial', + 'fontcolor': 'white' + }, + + } + EDGE_STYLYING = { + "fontname": "Arial" + } + + def __init__(self, output_directory: str =None): + + self.dot = graphviz.Digraph() + self.output_directory = output_directory + + + def generate_call_flow_graph(self, nodes: list, number: str): + self.dot.attr(name=f"calls_to_{number}", label=number, fontname='Arial', + fontcolor='white', rankdir="LR") + + # build nodes + self.dot.node("Start", "Start", GraphvizModule.NODE_STYLING["start"]) + for n in nodes: + if isinstance(n, bre.User): + self.dot.node(n.id, n.extension, GraphvizModule.NODE_STYLING["user"]) + elif isinstance(n, bre.CallCenter): + self.dot.node(n.service_user_id, n.extension, GraphvizModule.NODE_STYLING["call_centre"]) + elif isinstance(n, bre.HuntGroup): + self.dot.node(n.service_user_id, n.extension, GraphvizModule.NODE_STYLING["hunt_group"]) + elif isinstance(n, bre.AutoAttendant): + self.dot.node(n.service_user_id, n.extension, GraphvizModule.NODE_STYLING["auto_attendant"]) + + # build edges + for n in nodes: + try: + if n._start_node: + try: + self.dot.edge("Start", n.id) + except Exception: + self.dot.edge("Start", n.service_user_id) + except AttributeError: + # node is note the start node + pass + + if isinstance(n, bre.User): + if n.call_forwarding_always: + self._format_edge(n, n.call_forwarding_always, "CFA") + if n.call_forwarding_busy: + self._format_edge(n, n.call_forwarding_busy, "CFB") + if n.call_forwarding_not_reachable: + self._format_edge(n, n.call_forwarding_not_reachable, "CFNR") + + elif isinstance(n, bre.CallCenter): + if n.bounced_calls_enabled: + self._format_edge(n, n.bounced_calls_transfer_to_phone_number, "BCT") + if n.overflow_calls_action: + self._format_edge(n, n.overflow_calls_transfer_to_phone_number, "OF") + if n.stranded_calls_action: + self._format_edge(n, n.stranded_calls_transfer_to_phone_number, "SCF") + if n.stranded_call_unavailable_action: + self._format_edge(n, n.stranded_call_unavailable_transfer_to_phone_number, "USCF") + + elif isinstance(n, bre.HuntGroup): + if n.forward_after_timeout_enabled: + self._format_edge(n, n.no_answer_forward_to_phone_number, "NACF") + if n.call_forward_not_reachable_enabled: + self._format_edge(n, n.call_forward_not_reachable_transfer_to_phone_number, "CFNR") + + elif isinstance(n, bre.AutoAttendant): + for key in n.business_hours_menu.keys: + if "Transfer" in key.action: + self._format_edge(n, key.phone_number, key.number) + + + def _format_edge(self, node_a: str, node_b: str, label: str): + try: + self.dot.edge(node_a.id, node_b.id, label, GraphvizModule.EDGE_STYLYING) + except AttributeError: + try: + self.dot.edge(node_a.id, node_b.service_user_id, label, GraphvizModule.EDGE_STYLYING) + except AttributeError: + try: + self.dot.edge(node_a.service_user_id, node_b.service_user_id, label, GraphvizModule.EDGE_STYLYING) + except AttributeError: + self.dot.edge(node_a.service_user_id, node_b.id, label, GraphvizModule.EDGE_STYLYING) + + + def _save_graph(self, filename: str): + self.dot.render(directory=self.output_directory, filename=filename, + format="svg", cleanup=True).replace('\\', '/') + + \ No newline at end of file diff --git a/odins_spear/reports/report_utils/helpers.py b/odins_spear/reports/report_utils/helpers.py new file mode 100644 index 0000000..0787a19 --- /dev/null +++ b/odins_spear/reports/report_utils/helpers.py @@ -0,0 +1,27 @@ +import re + +def find_entity_with_number_type(number: str, number_type: str, broadwork_entities: list): + """ + Finds the phone number, extension, or alias in list of broadwork entities + and returns that entity once found. + + Args: + number (str): Target number looking for e.g. +1-123456789 or alias 0 + number_type (str): Number type of either phone number, extenion, or alias + broadwork_entities (list): List of broadwork entities number can be assigned to + + Returns: + object: Broadwork entity where number is assigned. + """ + + for entity in broadwork_entities: + if number_type == 'dn' and number in entity.phone_number: + return entity + elif number_type == 'extension' and number in entity.extension: + return entity + elif number_type == 'alias': + for alias in entity.aliases: + if re.search(rf'\b{re.escape(alias)}\b', number): + return entity + + return None \ No newline at end of file diff --git a/odins_spear/reports/report_utils/parsing.py b/odins_spear/reports/report_utils/parsing.py new file mode 100644 index 0000000..7babf62 --- /dev/null +++ b/odins_spear/reports/report_utils/parsing.py @@ -0,0 +1,132 @@ +from odins_spear.store import broadwork_entities as bre +from .report_entities import external_number + +# nodes needed for graph generation +NODES = [] +MAX_LEVEL = 30 + +def call_flow_module(node: object, data_store: object): + + # create mapping for easier searchability when gathering nodes + data_store.build_number_mapping() + + # traverses the enities that forward to eachother and adds all to NODES + _traverse_connecting_entities(node, data_store) + + return NODES + + +def _traverse_connecting_entities(entity: object, data_store: object, visited= None): + if visited is None: + visited = [] + + if entity in visited: + return + else: + visited.append(entity) + + try: + if isinstance(entity, bre.User): + # Checks to see if the number is external first and creates external number object for parsing later + if data_store.number_mapping.get(entity.call_forwarding_always) == None and len(str(entity.call_forwarding_always)) >= 3 and str(entity.call_forwarding_always)[2].isdigit(): + entity.call_forwarding_always = external_number(str(entity.call_forwarding_always)) + else: + entity.call_forwarding_always = data_store.number_mapping.get(entity.call_forwarding_always) + if entity.call_forwarding_always and entity.call_forwarding_always not in visited: + _traverse_connecting_entities(entity.call_forwarding_always, data_store, visited) + + if data_store.number_mapping.get(entity.call_forwarding_busy) == None and len(str(entity.call_forwarding_busy)) >= 3 and str(entity.call_forwarding_busy)[2].isdigit(): + entity.call_forwarding_busy = external_number(str(entity.call_forwarding_busy)) + else: + entity.call_forwarding_busy = data_store.number_mapping.get(entity.call_forwarding_busy) + if entity.call_forwarding_busy and entity.call_forwarding_busy not in visited: + _traverse_connecting_entities(entity.call_forwarding_busy, data_store, visited) + + if data_store.number_mapping.get(entity.call_forwarding_no_answer) == None and len(str(entity.call_forwarding_no_answer)) >= 3 and str(entity.call_forwarding_no_answer)[2].isdigit(): + entity.call_forwarding_no_answer = external_number(str(entity.call_forwarding_no_answer)) + else: + entity.call_forwarding_no_answer = data_store.number_mapping.get(entity.call_forwarding_no_answer) + if entity.call_forwarding_no_answer and entity.call_forwarding_no_answer not in visited: + _traverse_connecting_entities(entity.call_forwarding_no_answer, data_store, visited) + + if data_store.number_mapping.get(entity.call_forwarding_not_reachable) == None and len(str(entity.call_forwarding_not_reachable)) >= 3 and str(entity.call_forwarding_not_reachable)[2].isdigit(): + entity.call_forwarding_not_reachable = external_number(str(entity.call_forwarding_not_reachable)) + else: + entity.call_forwarding_not_reachable = data_store.number_mapping.get(entity.call_forwarding_not_reachable) + if entity.call_forwarding_not_reachable and entity.call_forwarding_not_reachable not in visited: + _traverse_connecting_entities(entity.call_forwarding_not_reachable, data_store, visited) + + if isinstance(entity, bre.CallCenter): + if entity.bounced_calls_enabled: + # Checks to see if the number is external first and creates external number object for parsing later + if data_store.number_mapping.get(str(entity.bounced_calls_transfer_to_phone_number)) == None and len(str(entity.bounced_calls_transfer_to_phone_number)) >= 3 and str(entity.bounced_calls_transfer_to_phone_number)[2].isdigit(): + entity.bounced_calls_transfer_to_phone_number = external_number(str(entity.bounced_calls_transfer_to_phone_number)) + else: + entity.bounced_calls_transfer_to_phone_number = data_store.number_mapping.get(str(entity.bounced_calls_transfer_to_phone_number)) + if entity.bounced_calls_transfer_to_phone_number and entity.bounced_calls_transfer_to_phone_number not in visited: + _traverse_connecting_entities(entity.bounced_calls_transfer_to_phone_number, data_store, visited) + + if entity.overflow_calls_action: + if data_store.number_mapping.get(str(entity.overflow_calls_transfer_to_phone_number)) == None and len(str(entity.call_forwarding_overflow_calls_transfer_to_phone_numberalways)) >= 3 and str(entity.overflow_calls_transfer_to_phone_number)[2].isdigit(): + entity.overflow_calls_transfer_to_phone_number = external_number(str(entity.overflow_calls_transfer_to_phone_number)) + else: + entity.overflow_calls_transfer_to_phone_number = data_store.number_mapping.get(str(entity.overflow_calls_transfer_to_phone_number)) + if entity.overflow_calls_transfer_to_phone_number and entity.overflow_calls_transfer_to_phone_number not in visited: + _traverse_connecting_entities(entity.overflow_calls_transfer_to_phone_number, data_store, visited) + + if entity.stranded_calls_action: + if data_store.number_mapping.get(str(entity.stranded_calls_transfer_to_phone_number)) == None and len(str(entity.stranded_calls_transfer_to_phone_number)) >= 3 and str(entity.stranded_calls_transfer_to_phone_number)[2].isdigit(): + entity.stranded_calls_transfer_to_phone_number = external_number(str(entity.stranded_calls_transfer_to_phone_number)) + else: + entity.stranded_calls_transfer_to_phone_number = data_store.number_mapping.get(str(entity.stranded_calls_transfer_to_phone_number)) + if entity.stranded_calls_transfer_to_phone_number and entity.stranded_calls_transfer_to_phone_number not in visited: + _traverse_connecting_entities(entity.stranded_calls_transfer_to_phone_number, data_store, visited) + + if entity.stranded_call_unavailable_action: + if data_store.number_mapping.get(str(entity.stranded_call_unavailable_transfer_to_phone_number)) == None and len(str(entity.stranded_call_unavailable_transfer_to_phone_number)) >= 3 and str(entity.stranded_call_unavailable_transfer_to_phone_number)[2].isdigit(): + entity.stranded_call_unavailable_transfer_to_phone_number = external_number(str(entity.stranded_call_unavailable_transfer_to_phone_number)) + else: + entity.stranded_call_unavailable_transfer_to_phone_number = data_store.number_mapping.get(str(entity.stranded_call_unavailable_transfer_to_phone_number)) + if entity.stranded_call_unavailable_transfer_to_phone_number and entity.stranded_call_unavailable_transfer_to_phone_number not in visited: + _traverse_connecting_entities(entity.stranded_call_unavailable_transfer_to_phone_number, data_store, visited) + + if isinstance(entity, bre.HuntGroup): + if entity.forward_after_timeout_enabled: + if data_store.number_mapping.get(entity.no_answer_forward_to_phone_number) is None and len(str(entity.no_answer_forward_to_phone_number)) >= 4 and str(entity.no_answer_forward_to_phone_number)[2].isdigit(): + entity.no_answer_forward_to_phone_number = external_number(str(entity.no_answer_forward_to_phone_number)) + else: + entity.no_answer_forward_to_phone_number = data_store.number_mapping.get(str(entity.no_answer_forward_to_phone_number)) + if entity.no_answer_forward_to_phone_number and entity.no_answer_forward_to_phone_number not in visited: + _traverse_connecting_entities(entity.no_answer_forward_to_phone_number, data_store, visited) + + if entity.call_forward_not_reachable_enabled: + if data_store.number_mapping.get(str(entity.call_forward_not_reachable_transfer_to_phone_number)) == None and len(str(entity.call_forward_not_reachable_transfer_to_phone_number)) >= 3 and str(entity.call_forward_not_reachable_transfer_to_phone_number)[2].isdigit(): + entity.call_forward_not_reachable_transfer_to_phone_number = external_number(str(entity.call_forward_not_reachable_transfer_to_phone_number)) + else: + entity.call_forward_not_reachable_transfer_to_phone_number = data_store.number_mapping.get(str(entity.call_forward_not_reachable_transfer_to_phone_number)) + if entity.call_forward_not_reachable_transfer_to_phone_number and entity.call_forward_not_reachable_transfer_to_phone_number not in visited: + _traverse_connecting_entities(entity.call_forward_not_reachable_transfer_to_phone_number, data_store, visited) + + if isinstance(entity, bre.AutoAttendant): + for key in entity.business_hours_menu.keys: + if "Transfer" in key.action: + if data_store.number_mapping.get(str(key.phone_number)) == None and len(str(key.phone_number)) >= 3 and str(key.phone_number)[2].isdigit(): + key.phone_number = external_number(str(key.phone_number)) + else: + key.phone_number = data_store.number_mapping.get(str(key.phone_number)) + if key.phone_number and key.phone_number not in visited: + _traverse_connecting_entities(key.phone_number, data_store, visited) + + for key in entity.after_hours_menu.keys: + if "Transfer" in key.action: + if data_store.number_mapping.get(str(key.phone_number)) == None and len(str(key.phone_number)) >= 3 and str(key.phone_number)[2].isdigit(): + key.phone_number = external_number(str(key.phone_number)) + else: + key.phone_number = data_store.number_mapping.get(str(key.phone_number)) + if key.phone_number and key.phone_number not in visited: + _traverse_connecting_entities(key.phone_number, data_store, visited) + except TypeError: + # this object is further up the tree and a loop has occured - Move one and come back + pass + + NODES.append(entity) # Append outside of the if condition diff --git a/odins_spear/reports/report_utils/report_entities.py b/odins_spear/reports/report_utils/report_entities.py new file mode 100644 index 0000000..94375cb --- /dev/null +++ b/odins_spear/reports/report_utils/report_entities.py @@ -0,0 +1,12 @@ +from dataclasses import dataclass, field +from typing import List, Type + +@dataclass +class call_flow: + name: str + nodes: List = field(default_factory=list) + + +@dataclass +class external_number: + id: str \ No newline at end of file diff --git a/odins_spear/store/broadwork_entities.py b/odins_spear/store/broadwork_entities.py index 8836a70..47f6837 100644 --- a/odins_spear/store/broadwork_entities.py +++ b/odins_spear/store/broadwork_entities.py @@ -1,7 +1,7 @@ -from dataclasses import dataclass, field, InitVar -from typing import List, Type +import json -from odins_spear.utils import generators as gen +from dataclasses import dataclass, field +from typing import List, Type @dataclass(kw_only=True) class ServiceProvider: @@ -9,18 +9,14 @@ class ServiceProvider: name: str groups: List['Group'] = field(default_factory=list) is_enterprise: bool = False - default_domain: str = None - support_email: str = None - contact_name: str = None - contact_number: str = None - contact_email: str = None - address_line1: str = None - city: str = None - state_or_province: str = None - zip_or_postcode: str = None - country: str = None - use_service_provider_language: bool = False - + + @classmethod + def from_dict(cls, data): + return cls( + id= data.get("serviceProviderId"), + name= data.get("serviceProviderId"), + is_enterprise= data.get("isEnterprise") + ) @dataclass(kw_only=True) class Group: @@ -28,22 +24,8 @@ class Group: id: str name: str default_domain: str - user_limit: int = None - user_count: int = None - calling_line_id_name: str = None calling_line_id_phone_number: str = None - calling_line_id_display_phone_number: str = None - time_zone: str = None - time_zone_display_name: str = None - location_dialing_code: str = None - contact_name: str = None - contact_number: str = None - contact_email: str = None - address_line1: str = None - address_line2: str = None - city: str = None - state_or_province: str = None - country: str = None + auto_attendants: List['AutoAttendant'] = field(default_factory=list) trunk_groups: List['TrunkGroup'] = field(default_factory=list) call_centers: List['CallCenter'] = field(default_factory=list) hunt_groups: List['HuntGroup'] = field(default_factory=list) @@ -53,161 +35,223 @@ def __post_init__(self): self.service_provider.groups.append(self) self.default_domain = '@' + self.default_domain - + + @classmethod + def from_dict(cls, service_provider: ServiceProvider, data): + return cls( + service_provider= service_provider, + id= data.get("groupId"), + name= data.get("groupName"), + default_domain= data.get("defaultDomain"), + calling_line_id_phone_number= data.get("callingLineIdPhoneNumber") + ) @dataclass(kw_only=True) class TrunkGroup: - name: str + service_user_id: str group: Type['Group'] - access_device: Type['Device'] users: List['User'] = field(default_factory=list) - allow_termination_to_dtg_identity: bool = False - allow_termination_to_trunk_group_identity: bool = False - allow_unscreened_calls: bool = False - allow_unscreened_emergency_calls: bool = False - capacity_exceeded_trap_initial_calls: int = None - capacity_exceeded_trap_offset_calls: int = None - clid_source_for_screened_calls_policy: str = None - continuous_options_sending_interval_seconds: int = None - enable_bursting: bool = False - enable_network_address_identity: bool = False - failure_options_sending_interval_seconds: int = None - failure_threshold_counter: int = None - include_dtg_identity: bool = False - include_otg_identity_for_network_calls: bool = False - include_trunk_group_identity: bool = False - include_trunk_group_identity_for_network_calls: bool = False - invitation_timeout: int = None - invite_failure_threshold_counter: int = None - invite_failure_threshold_window_seconds: int = None - pilot_user_call_optimization_policy: str = None - pilot_user_calling_line_asserted_identity_policy: str = None - pilot_user_calling_line_identity_for_emergency_calls_policy: str = None - pilot_user_calling_line_identity_for_external_calls_policy: str = None - pilot_user_charge_number_policy: str = None - prefix_enabled: bool = False - require_authentication: bool = False - route_to_peering_domain: bool = False - send_continuous_options_message: bool = False - stateful_rerouting_enabled: bool = False - success_threshold_counter: int = None - use_system_clid_source_for_screened_calls_policy: bool = False - use_system_calling_line_asserted_identity_policy: bool = False - use_system_user_lookup_policy: bool = False - user_lookup_policy: str = None max_active_calls: int = None - max_incoming_calls: int = None - max_outgoing_calls: int = None + bursting_enabled: bool = False + bursting_max_active_calls: bool = False + pilot_user_id: str = None def __post_init__(self): - self.service_provider_id = self.group.service_provider_id.id + self.group.trunk_groups.append(self) + + @classmethod + def from_dict(cls, group: Group, data): + + # gather user IDs to gather user object + user_ids = [agent["userId"] for agent in data["agents"]] + users = _get_user_object_from_id(group, user_ids) + + return cls( + service_user_id= data.get(""), + group= group, + users= users, + max_active_calls= data.get("maxActiveCalls"), + bursting_enabled= data.get("enableBursting"), + bursting_max_active_calls= data.get("burstingMaxActiveCalls"), + pilot_user_id= data.get("pilotUserId") + ) @dataclass(kw_only=True) class AAKey: - key_number: int + number: int action: str description: str = None phone_number: str = None submenu_id: int = None + + @classmethod + def from_dict(cls, data): + return cls( + number= data.get("key"), + action= data.get("action"), + description= data.get("description"), + phone_number= data.get("phoneNumber"), + submenu_id= data.get("submenuId") + ) @dataclass(kw_only=True) class AAMenu: - announcement_selection: str = None enable_first_menu_level_extension_dialing: bool = False keys: List[AAKey] = field(default_factory=list) @dataclass(kw_only=True) class AutoAttendant: + service_user_id: str name: str group: Type['Group'] - calling_line_id_last_name: str = None - calling_line_id_first_name: str = None - hiragana_last_name: str = None - hiragana_first_name: str = None - language: str = None - time_zone: str = None - time_zone_display_name: str = None + extension: str = None + phone_number: str = None aliases: List[str] = field(default_factory= [list]) type: str = None - enable_video: bool = False - extension_dialing_scope: str = None - name_dialing_scope: str = None - name_dialing_entries: str = None business_hours_menu: Type['AAMenu'] = None after_hours_menu: Type['AAMenu'] = None - service_user_id: str = None + def __post_init__(self): - self.serviceProviderId = self.group.ServiceProvider.id + self.group.auto_attendants.append(self) + + @classmethod + def from_dict(cls, group: Group, data): + return cls( + service_user_id=data.get("serviceUserId"), + name=data.get("serviceInstanceProfile").get("name"), + group= group, + extension=data.get("serviceInstanceProfile").get("extension"), + phone_number=data.get("serviceInstanceProfile").get("phoneNumber"), + aliases=data.get("serviceInstanceProfile").get("aliases"), + type=data.get("type"), + business_hours_menu=AAMenu( + enable_first_menu_level_extension_dialing=data.get('businessHoursMenu').get('enableFirstMenuLevelExtensionDialing'), + keys= [AAKey.from_dict(key) for key in data.get('businessHoursMenu').get('keys')]), + after_hours_menu=AAMenu( + enable_first_menu_level_extension_dialing=data.get('afterHoursMenu').get('enableFirstMenuLevelExtensionDialing'), + keys= [AAKey.from_dict(key) for key in data.get('afterHoursMenu').get('keys')]), + ) @dataclass(kw_only=True) class CallCenter: - name: str + service_user_id: str group: Type['Group'] agents: List['User'] = field(default_factory=list) - enable_video: bool = False - allow_caller_to_dial_escape_digit: bool = False - reset_call_statistics_upon_entry_in_queue: bool = True - allow_agent_logoff: bool = True - allow_call_waiting_for_agents: bool = True - play_ringing_when_offering_call: bool = True - external_preferred_audio_codec: str = None - internal_preferred_audio_codec: str = None - enable_reporting: bool = False - allow_calls_to_agents_in_wrap_up: bool = True - override_agent_wrap_up_time: bool = False - enable_automatic_state_change_for_agents: bool = False - force_delivery_of_calls: bool = False + extension: str = None + phone_number: str = None + aliases: List[str] = field(default_factory=list) + name: str = None type: str = None - service_user_id_prefix: str = None - calling_line_id_last_name: str = None - calling_line_id_first_name: str = None - password: str = field(default_factory=gen.generate_password) policy: str = None - routing_type: str = None - queue_length: int = None - escape_digit: str = None - service_user_id: str = None - + + bounced_calls_enabled: bool = False + overflow_calls_action: str = None + overflow_calls_transfer_to_phone_number: bool = False + stranded_calls_action: str = None + stranded_calls_transfer_to_phone_number: bool = False + stranded_call_unavailable_action: str = None + stranded_call_unavailable_transfer_to_phone_number: bool = False + + #NOTE: Not sure which forwarding this is. + forced_forwarding_enabled: bool = False + forced_forwarding_forward_to_phone_number: str = None + + night_service: str = None + holiday_service: str = None + def __post_init__(self): - self.service_provider_id = self.group.service_provider_id.id self.group.call_centers.append(self) + @classmethod + def from_dict(cls, group: Group, data): + + # gather user IDs to gather user object + try: + agent_ids = [agent["userId"] for agent in data["agents"]] + agents = _get_user_object_from_id(group, agent_ids) + except KeyError: + agents = [] + + return cls( + service_user_id= data.get("serviceUserId"), + group= group, + agents= agents, + extension= data.get("serviceInstanceProfile").get("extension"), + phone_number=data.get("serviceInstanceProfile").get("phoneNumber"), + name= data.get("serviceInstanceProfile").get("name"), + aliases= data.get("serviceInstanceProfile").get("aliases"), + type= data.get("type"), + policy= data.get("policy"), + + bounced_calls_enabled= data.get("bouncedCallsEnabled"), + overflow_calls_action= data.get("overFlowCallsAction"), + overflow_calls_transfer_to_phone_number= data.get("overflowCallsTransferToPhoneNumber"), + stranded_calls_action= data.get("strandedCallsAction"), + stranded_calls_transfer_to_phone_number= data.get("strandedCallsTransferToPhoneNumber"), + stranded_call_unavailable_action= data.get("strandedCallUnavailableAction"), + stranded_call_unavailable_transfer_to_phone_number= data.get("strandedCallUnavailableTransferToPhoneNumber"), + + #NOTE: Not sure which forwarding this is. + forced_forwarding_enabled= data.get("forcedForwardingEnabled"), + forced_forwarding_forward_to_phone_number= data.get("forcedForwardingEnabled"), + + night_service= data.get("nightService"), + holiday_service= data.get("holidayService") + ) + @dataclass(kw_only=True) class HuntGroup: + service_user_id: str name: str group: Type['Group'] agents: List['User'] = field(default_factory=list) - calling_line_id_last_name: str = None - calling_line_id_first_name: str = None - hiragana_last_name: str = None - hiragana_first_name: str = None - language: str = None - time_zone: str = None aliases: List[str] = field(default_factory=list) + extension: str = None + phone_number: str = None policy: str = None - hunt_after_no_answer: bool = None - no_answer_number_of_rings: int = None - forward_after_timeout: bool = None + + forward_after_timeout_enabled: bool = False forward_timeout_seconds: int = None - allow_call_waiting_for_agents: bool = None - use_system_hunt_group_clid_setting: bool = None - include_hunt_group_name_in_clid: bool = None - enable_not_reachable_forwarding: bool = None - make_busy_when_not_reachable: bool = None - service_user_id: str = None - service_provider_id: str = None - group_id: str = None + no_answer_number_of_rings: int = None + no_answer_forward_to_phone_number: str = None + + call_forward_not_reachable_enabled: bool = False + call_forward_not_reachable_transfer_to_phone_number: str = None def __post_init__(self): - self.service_provider_id = self.group.service_provider_id.id self.group.hunt_groups.append(self) + @classmethod + def from_dict(cls, group: Group, data): + + # gather user IDs to gather user object + agent_ids = [agent["userId"] for agent in data["agents"]] + agents = _get_user_object_from_id(group, agent_ids) + + return cls( + service_user_id= data.get("serviceUserId"), + name= data.get("serviceInstanceProfile").get("name"), + group= group, + agents = agents, + aliases= data.get("serviceInstanceProfile").get("aliases"), + extension= data.get("serviceInstanceProfile").get("extension"), + phone_number= data.get("serviceInstanceProfile").get("phoneNumber"), + policy= data.get("policy"), + + forward_after_timeout_enabled= data.get("forwardAfterTimeout"), + forward_timeout_seconds= data.get("forwardTimeoutSeconds"), + no_answer_number_of_rings= data.get("noAnswerNumberOfRings"), + no_answer_forward_to_phone_number= data.get("forwardToPhoneNumber"), + + call_forward_not_reachable_enabled= data.get("enableNotReachableForwarding"), + call_forward_not_reachable_transfer_to_phone_number= data.get("notReachableForwardToPhoneNumber") + ) @dataclass(kw_only=True) class User: @@ -216,95 +260,30 @@ class User: first_name: str = None last_name: str = None extension: str = None - access_device_endpoint: Type['Device'] = None - calling_line_id_last_name: str = None - calling_line_id_first_name: str = None - hiragana_last_name: str = None - hiragana_first_name: str = None phone_number: str = None - calling_line_id_phone_number: str = None - password: str = None - department: Type['Department'] = None - department_full_path: str = None - language: str = None - time_zone: str = None - time_zone_display_name: str = None - default_alias: str = None - title: str = None - pager_phone_number: str = None - mobile_phone_number: str = None - email_address: str = None - yahoo_id: str = None - address_location: str = None - address: Type['Address'] = None - country_code: str = None - network_class_of_service: str = None - allow_video: bool = True - domain: str = None - endpoint_type: str = None aliases: List[str] = field(default_factory=list) - trunk_addressing: str = None - alternate_user_id: List[str] = field(default_factory=list) - is_enterprise: bool = False - password_expires_days: int = 2147483647 - service_provider_id: str = None - line_port: str = None + + call_forwarding_always: str = None + call_forwarding_busy: str = None + call_forwarding_no_answer: str = None + call_forwarding_not_reachable: str = None + def __post_init__(self): - self.is_enterprise = self.group.service_provider.is_enterprise - self.service_provider_id = self.group.service_provider.id self.group.users.append(self) - self.id = self.id + self.group.default_domain - self.group_id = self.group.id - -@dataclass(kw_only=True) -class Device: - device_type: str - name: str - group: Type['Group'] - device_level: Type['Group'] - use_custom_user_name_password: bool = True - access_device_credential_name: str = None - access_device_credential_password: str = None - net_address: str = None - port: str = None - outbound_proxy_server_net_address: str = None - stun_server_net_address: str = None - mac_address: str = None - serial_number: str = None - description: str = None - physical_location: str = None - transport_protocol: str = None - profile: str = None - static_registration_capable: str = None - config_type: str = None - protocol_choice: List[str] = field(default_factory=list) - is_ip_address_optional: bool = True - use_domain: bool = True - is_mobility_manager_device: bool = False - device_configuration_option: str = None - static_line_ordering: bool = False - device_type_level: str = None - tags: List[str] = field(default_factory=list) - related_services: List[str] = field(default_factory=list) - protocol: str = None - user_name: str = None - group_id: int = field(init=False) - service_provider_id: int = field(init=False) - - # USER HAS THIS OVER DEVICE - # contacts - # support_visual_device_management - # number_of_ports - # number_of_assigned_ports - # status - # line_port - - def __post_init__(self): - self.group_id = self.group.id - self.service_provider_id = self.group.service_provider.id - + + @classmethod + def from_dict(cls, group: Group, data): + return cls( + group= group, + id= data.get("userId"), + first_name= data.get("firstName"), + last_name= data.get("lastName"), + extension= data.get("extension"), + phone_number= data.get("phoneNumber"), + aliases= data.get("aliases") + ) @dataclass(kw_only=True) class Contact: @@ -312,6 +291,13 @@ class Contact: number: str = None email: str = None + @classmethod + def from_dict(cls, data): + return cls( + name= data.get("name"), + number= data.get("number"), + email= data.get("email"), + ) @dataclass(kw_only=True) class Address: @@ -322,9 +308,31 @@ class Address: zip_or_postal_code: str country: str + @classmethod + def from_dict(cls, data): + return cls( + address_line1= data.get("addressLine1"), + address_line2= data.get("addressLine2"), + city= data.get("city"), + state_or_province= data.get("stateOrProvince"), + zip_or_postal_code= data.get("zipOrPostalCode"), + country= data.get("country") + ) @dataclass(kw_only=True) class Department: service_provider_id: str group_id: str - name: str \ No newline at end of file + name: str + + @classmethod + def from_dict(cls, data): + return cls( + service_provider_id= data.get("serviceProviderId"), + group_id= data.get("groupId"), + name= data.get("name") + ) + +def _get_user_object_from_id(group, user_ids: list): + return list(filter(lambda user: any(user_id in user.id for user_id in user_ids), + group.users)) \ No newline at end of file diff --git a/odins_spear/store/data_store.py b/odins_spear/store/data_store.py index 52a5a43..48bc73d 100644 --- a/odins_spear/store/data_store.py +++ b/odins_spear/store/data_store.py @@ -12,31 +12,62 @@ class DataStore: Singleton design pattern to get store use get_instance(). """ - __instance = None - - @staticmethod - def get_instance(): - if DataStore.__instance is None: - DataStore() - return DataStore.__instance - def __init__(self): - if DataStore.__instance is not None: - raise Exception("Singleton cannot be instantiated more than once!") - else: - self.apis: List[api.Api] = [] - self.service_providers_enterprises: List[bre.ServiceProvider] = [] - self.groups: List[bre.Group] = [] - self.trunk_groups: List[bre.TrunkGroup] = [] - self.auto_attendants: List[bre.AutoAttendant] = [] - self.call_centers: List[bre.CallCenter] = [] - self.hunt_groups: List[bre.HuntGroup] = [] - self.users: List[bre.User] = [] - self.devices: List[bre.Device] = [] - self.other_entities = [] #Non-common or custom objects + self.apis: List[api.Api] = [] + self.service_providers_enterprises: List[bre.ServiceProvider] = [] + self.groups: List[bre.Group] = [] + self.trunk_groups: List[bre.TrunkGroup] = [] + self.auto_attendants: List[bre.AutoAttendant] = [] + self.call_centers: List[bre.CallCenter] = [] + self.hunt_groups: List[bre.HuntGroup] = [] + self.users: List[bre.User] = [] + self.devices: List[bre.Device] = [] + self.other_entities = [] #Non-common or custom objects + - DataStore.__instance = self - + def build_id_mapping(self): + """ + Builds mapping of numbers IDs to entity. + + Example: {test@test.com: User, customID: CallCenter} + """ + + self.id_mapping = {} + + entities = self.auto_attendants + self.call_centers + \ + self.hunt_groups + self.users + + for e in entities: + try: + self.id_mapping[e.id] = e + except Exception: + self.id_mapping[e.service_user_id] = e + + + def build_number_mapping(self): + """ + Builds mapping of numbers (phone numbers, extension, aliases) to entity. + + Example: {101: User, +1-123456789: CallCenter} + """ + import re + + self.number_mapping = {} + + entities = self.auto_attendants + self.call_centers + \ + self.hunt_groups + self.users + + for e in entities: + if e.phone_number: + self.number_mapping[e.phone_number] = e + if e.extension: + self.number_mapping[e.extension] = e + if e.aliases: + for a in e.aliases: + number = re.search(r'\d+', a).group() + self.number_mapping[number] = e + + def get_group_state(api, group: bre.Group) -> None: """ takes in group id and loads group state into broadworks entities. @@ -45,7 +76,7 @@ def get_group_state(api, group: bre.Group) -> None: """ pass - def store_object(self, *entities) -> None: + def store_objects(self, *entities) -> None: """ Takes in objects within the odin_api and custom and stores in lists depending on type. diff --git a/setup.py b/setup.py index 1e261a8..9bf8dae 100644 --- a/setup.py +++ b/setup.py @@ -6,7 +6,7 @@ description='Encapsulates the Odin API easing the use as well as additional features such as find alias', author='Jordan Prescott', author_email='jprescott23@gmail.com', - url='https://github.com/Jordan-Prescott/odin_api', + url='https://github.com/Jordan-Prescott/odins_spear', packages=find_packages(), install_requires=[ "certifi==2023.7.22",