CU-861mv9d7e: introduce phone_directory contacts selecting mode

Author: yakimenko73

app/bss/adapters/portaswitch/Dockerfile

@@ -66,8 +66,8 @@ ENV PORTASWITCH_CONTACTS_SELECTING="<contacts-selecting>"
 # By default does not apply filter by type
 ENV PORTASWITCH_CONTACTS_SELECTING_EXTENSION_TYPES="Account;Group;Unassigned"
 
-# Defines a list of customer ids for PORTASWITCH_CONTACTS_SELECTING=phonebook mode (optional)
-ENV PORTASWITCH_CONTACTS_SELECTING_PHONEBOOK_CUSTOMER_IDS="8;10;11;13;14"
+# Defines a list of customer ids for PORTASWITCH_CONTACTS_SELECTING=phonebook|phone_directory modes (optional)
+ENV PORTASWITCH_CONTACTS_SELECTING_CUSTOMER_IDS="8;10;11;13;14"
 
 # Ensures that the contacts API filters accounts that do not have an `extension_id` (optional):
 ENV PORTASWITCH_CONTACTS_SKIP_WITHOUT_EXTENSION="True"

app/bss/adapters/portaswitch/adapter.py

@@ -425,27 +425,10 @@ def retrieve_contacts(self, session: SessionInfo, user: UserInfo) -> list[Contac
                                 contacts.append(Serializer.get_contact_info_by_account(account, i_account))
                 case PortaSwitchContactsSelectingMode.PHONEBOOK:
                     phonebook = self._account_api.get_phonebook_list(access_token, 1, 100)['phonebook_rec_list']
-                    number_to_accounts = {}
-                    # Retrieve accounts for a pre-defined list of customers to map them to phonebook records
-                    with ThreadPoolExecutor(max_workers=10) as executor:
-                        future_to_customer_id = {
-                            executor.submit(lambda cid: self._admin_api.get_account_list(int(cid))["account_list"],
-                                            cid): cid
-                            for cid in self._portaswitch_settings.CONTACTS_SELECTING_PHONEBOOK_CUSTOMER_IDS
-                        }
-
-                        for future in as_completed(future_to_customer_id):
-                            try:
-                                accounts = future.result()
-                                for account in accounts:
-                                    number_to_accounts[account["id"]] = account
-                            except Exception as e:
-                                logging.warning(
-                                    f"Error fetching accounts for customer {future_to_customer_id[future]}: {e}")
-
-                    contacts = []
+                    number_to_accounts = self._get_number_to_customer_accounts_map()
+
                     for record in phonebook:
-                        # Normalize phone number by removing '+' prefix
+                        # Normalize phone number by removing the '+' prefix
                         phonebook_record_number = record.get("phone_number").replace("+", "")
                         phonebook_contact_info = Serializer.get_contact_info_by_phonebook_record(record)
 
@@ -463,6 +446,40 @@ def retrieve_contacts(self, session: SessionInfo, user: UserInfo) -> list[Contac
                         if contact.is_current_user is not True:
                             contacts.append(contact)
 
+                case PortaSwitchContactsSelectingMode.PHONE_DIRECTORY:
+                    phone_directories = self._account_api.get_phone_directory_list(access_token, 1, 100)[
+                        'phone_directory_list']
+                    number_to_accounts = self._get_number_to_customer_accounts_map()
+
+                    for directory in phone_directories:
+                        directory_info = self._account_api.get_phone_directory_info(
+                            access_token,
+                            directory['i_ua_config_directory'],
+                            1,
+                            10_000
+                        )['phone_directory_info']
+                        for record in directory_info['directory_records']:
+                            # Normalize phone number by removing the '+' prefix
+                            phone_directory_record_number = record.get("office_number").replace("+", "")
+                            phone_directory_contact_info = Serializer.get_contact_info_by_phone_directory_record(record,
+                                                                                                                 directory_info[
+                                                                                                                     'name'])
+
+                            if account := number_to_accounts.get(phone_directory_record_number):
+                                # If we found a matching account, use its contact info but update with phone directory data
+                                contact = Serializer.get_contact_info_by_account(account, i_account)
+                                contact.first_name = phone_directory_contact_info.first_name
+                                contact.last_name = phone_directory_contact_info.last_name
+                                contact.numbers.main = phone_directory_record_number
+                            else:
+                                # No matching account found, use phone directory contact info as is
+                                contact = phone_directory_contact_info
+                                if contact.numbers.main:
+                                    contact.numbers.main = contact.numbers.main.replace("+", "")
+
+                            if contact.is_current_user is not True:
+                                contacts.append(contact)
+
             # Extend the contact list with custom entries
             contacts.extend([Serializer.get_contact_info_by_custom_entry(entry) for entry in
                              self._portaswitch_settings.CONTACTS_CUSTOM])
@@ -806,3 +823,25 @@ def _check_allowed_addons(self, account_info: dict):
 
         if not allowed_addons.intersection(assigned_addon_names):
             raise WebTritErrorException(403, "Access denied: required add-on not assigned", code="addon_required")
+
+    def _get_number_to_customer_accounts_map(self) -> dict[str, dict]:
+        """Return a mapping of phone numbers to customer accounts."""
+        number_to_accounts = {}
+        # Retrieve accounts for a pre-defined list of customers to map them to phonebook records
+        with ThreadPoolExecutor(max_workers=10) as executor:
+            future_to_customer_id = {
+                executor.submit(lambda cid: self._admin_api.get_account_list(int(cid))["account_list"],
+                                cid): cid
+                for cid in self._portaswitch_settings.CONTACTS_SELECTING_CUSTOMER_IDS
+            }
+
+            for future in as_completed(future_to_customer_id):
+                try:
+                    accounts = future.result()
+                    for account in accounts:
+                        number_to_accounts[account["id"]] = account
+                except Exception as e:
+                    logging.warning(
+                        f"Error fetching accounts for customer {future_to_customer_id[future]}: {e}")
+
+        return number_to_accounts

app/bss/adapters/portaswitch/api/account.py

@@ -263,6 +263,51 @@ def get_phonebook_list(self, access_token: str, page: int, items_per_page: int)
             },
             access_token=access_token)
 
+    def get_phone_directory_list(self, access_token: str, page: int, items_per_page: int) -> dict:
+        """Return the phone directory list of the account.
+
+        Args:
+            access_token (str): Token that authenticates the API user in the PortaBilling API using the account realm.
+            page (int): Shows what page of the phone directory to return.
+            items_per_page (int): Shows the number of items to return.
+
+        Returns:
+            dict: API method execution result containing phone directory items.
+        """
+        return self.__send_request(
+            module="UA",
+            method="get_phone_directory_list",
+            params={
+                "get_total": 1,
+                "limit": items_per_page,
+                "offset": items_per_page * (page - 1),
+            },
+            access_token=access_token)
+
+    def get_phone_directory_info(self, access_token: str, i_ua_config_directory: str, page: int, items_per_page: int) -> dict:
+        """Return the phone directory info of the account.
+
+        Args:
+            access_token (str): Token that authenticates the API user in the PortaBilling API using the account realm.
+            i_ua_config_directory (str): Identifier of the phone directory.
+            page (int): Shows what page of the phone directory items to return.
+            items_per_page (int): Shows the number of items to return.
+
+        Returns:
+            dict: API method execution result containing phone directory items.
+        """
+        return self.__send_request(
+            module="UA",
+            method="get_phone_directory_info",
+            params={
+                "i_ua_config_directory": i_ua_config_directory,
+                "get_total": 1,
+                "get_records": "Y",
+                "limit": items_per_page,
+                "offset": items_per_page * (page - 1),
+            },
+            access_token=access_token)
+
     def get_call_recording(self, recording_id: int, access_token: str) -> tuple[str, Iterator]:
         """Returns the bytes of the call recording file.
 

app/bss/adapters/portaswitch/config.py

@@ -22,7 +22,7 @@ class PortaSwitchSettings(BaseSettings):
     SIGNIN_CREDENTIALS: PortaSwitchSignInCredentialsType = PortaSwitchSignInCredentialsType.SELF_CARE
     CONTACTS_SELECTING: PortaSwitchContactsSelectingMode = PortaSwitchContactsSelectingMode.ACCOUNTS
     CONTACTS_SELECTING_EXTENSION_TYPES: Union[List[PortaSwitchExtensionType], str] = list(PortaSwitchExtensionType)
-    CONTACTS_SELECTING_PHONEBOOK_CUSTOMER_IDS: Union[List[str], str] = []
+    CONTACTS_SELECTING_CUSTOMER_IDS: Union[List[str], str] = []
     CONTACTS_SKIP_WITHOUT_EXTENSION: bool = False
     CONTACTS_CUSTOM: Union[List[dict], str] = []
     HIDE_BALANCE_IN_USER_INFO: Optional[bool] = False
@@ -35,9 +35,9 @@ def decode_contacts_selecting_extension_types(cls, v: Union[List, str]) -> List[
         v = str(v)
         return [PortaSwitchExtensionType(x) for x in v.split(';')] if v else v
 
-    @field_validator("CONTACTS_SELECTING_PHONEBOOK_CUSTOMER_IDS", mode='before')
+    @field_validator("CONTACTS_SELECTING_CUSTOMER_IDS", mode='before')
     @classmethod
-    def decode_contacts_selecting_phonebook_customer_ids(cls, v: Union[List, str]) -> List[str]:
+    def decode_contacts_selecting_customer_ids(cls, v: Union[List, str]) -> List[str]:
         v = str(v)
         return [x.strip() for x in v.split(';')] if v else v
 

app/bss/adapters/portaswitch/serializer.py

@@ -162,6 +162,31 @@ def get_contact_info_by_phonebook_record(phonebook_record_info: dict) -> Contact
             ),
         )
 
+    @staticmethod
+    def get_contact_info_by_phone_directory_record(phone_directory_record_info: dict, company_name: str) -> ContactInfo:
+        """Forms ContactInfo based on the input phonebook_record.
+        Parameters:
+            phone_directory_record_info: dict: The information about the phone directory record to be added to ContactInfo.
+            company_name: str: The name of the phone directory.
+
+        Returns:
+            ContactInfo: The filled structure of ContactInfo.
+        """
+        additional_numbers = [
+            phone_directory_record_info.get("mobile_number"),
+            phone_directory_record_info.get("other_number"),
+        ]
+
+        return ContactInfo(
+            company_name=company_name,
+            first_name=phone_directory_record_info.get("first_name"),
+            last_name=phone_directory_record_info.get("last_name"),
+            numbers=Numbers(
+                main=phone_directory_record_info.get("office_number"),
+                additional=list(filter(lambda x: x not in (None, ""), additional_numbers))
+            ),
+        )
+
     @staticmethod
     def get_contact_info_by_custom_entry(custom_entry: dict) -> ContactInfo:
         """Forms ContactInfo based on the input custom_entry.
@@ -336,7 +361,7 @@ def _xdr_to_call_status(failed: bool, disconnect_cause: int) -> str:
             return "accepted"
         else:
             return "error"
-        
+
     @staticmethod
     def _call_recording_exist(cdr) -> bool:
         bit_flags = cdr["bit_flags"]

app/bss/adapters/portaswitch/types.py

@@ -21,6 +21,7 @@ class PortaSwitchContactsSelectingMode(Enum):
     ACCOUNTS = "accounts"
     EXTENSIONS = "extensions"
     PHONEBOOK = "phonebook"
+    PHONE_DIRECTORY = "phone_directory"
 
     @classmethod
     def _missing_(cls, value):