feat(portaswitch): add phone numbers contact search

Author: yakimenko73

app/bss/adapters/__init__.py

@@ -262,6 +262,7 @@ def retrieve_contacts_v2(
             self, session: SessionInfo,
             user: UserInfo,
             search: Optional[str] = None,
+            phone_numbers: List[str] = [],
             page: Optional[int] = 1,
             items_per_page: Optional[int] = 100,
     ) -> tuple[List[ContactInfo], int]:
@@ -428,6 +429,7 @@ def retrieve_contacts_v2(
             self, session: SessionInfo,
             user: UserInfo,
             search: Optional[str] = None,
+            phone_numbers: List[str] = [],
             page: Optional[int] = 1,
             items_per_page: Optional[int] = 100,
     ) -> tuple[List[ContactInfo], int]:

app/bss/adapters/netsapiens.py

@@ -461,6 +461,7 @@ def retrieve_contacts_v2(
             session: SessionInfo,
             user: UserInfo,
             search: Optional[str] = None,
+            phone_numbers: List[str] = [],
             page: Optional[int] = 1,
             items_per_page: Optional[int] = 100,
     ) -> tuple[List[ContactInfo], int]:

app/bss/adapters/portaswitch/adapter.py

@@ -557,6 +557,7 @@ def retrieve_contacts_v2(
             self, session: SessionInfo,
             user: UserInfo,
             search: Optional[str] = None,
+            phone_numbers: List[str] = [],
             page: Optional[int] = 1,
             items_per_page: Optional[int] = 100,
     ) -> tuple[List[ContactInfo], int]:
@@ -583,6 +584,46 @@ def retrieve_contacts_v2(
             i_customer = int(account_info["i_customer"])
             i_account = int(account_info["i_account"])
 
+            # Normalize and prepare phone numbers if provided
+            normalized_phone_numbers: set[str] = set[str]()
+            if phone_numbers:
+                for number in phone_numbers:
+                    if not number:
+                        continue
+                    normalized_phone_numbers.add(number.replace("+", "").strip())
+
+            # If phone_numbers search is provided, ignore generic `search` and use a unified implementation
+            if normalized_phone_numbers:
+                contacts: list[ContactInfo] = []
+                for number in normalized_phone_numbers:
+                    try:
+                        # Exact search by main number (id)
+                        result = self._admin_api.get_account_list(i_customer, id=number)
+                        accounts = result.get("account_list", []) or []
+                        for account in accounts:
+                            contacts.append(Serializer.get_contact_info_by_account(account, account["i_account"]))
+                    except Exception as e:
+                        logging.debug(f"Failed to fetch accounts for phone number {number}: {e}")
+                        continue
+
+                # Add matching custom contacts
+                custom_contacts = [
+                    Serializer.get_contact_info_by_custom_entry(entry)
+                    for entry in self._portaswitch_settings.CONTACTS_CUSTOM
+                ]
+
+                for contact in custom_contacts:
+                    number = contact.numbers.main if contact.numbers and contact.numbers.main else ""
+
+                    if number.replace("+", "").strip() in normalized_phone_numbers:
+                        contacts.append(contact)
+
+                total_count = len(contacts)
+                start_idx = (page - 1) * items_per_page
+                end_idx = start_idx + items_per_page
+
+                return contacts[start_idx:end_idx], total_count
+
             contacts = []
             match self._portaswitch_settings.CONTACTS_SELECTING:
                 case PortaSwitchContactsSelectingMode.EXTENSIONS:

app/main.py

@@ -3,9 +3,9 @@
 import os
 import sys
 from datetime import datetime
-from typing import Optional, Union
+from typing import Optional, Union, List
 
-from fastapi import FastAPI, APIRouter, Depends, Response, Request, Header, Body
+from fastapi import FastAPI, APIRouter, Depends, Response, Request, Header, Body, Query
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from pydantic import conint
 from starlette.responses import StreamingResponse
@@ -591,6 +591,7 @@ def get_user_contact_list(
 )
 def get_user_contact_list_v2(
         search: str = None,
+        phone_numbers: List[str] = Query(default=[]),
         auth_data: HTTPAuthorizationCredentials = Depends(security),
         page: Optional[conint(ge=1)] = 1,
         items_per_page: Optional[conint(ge=1)] = 100,
@@ -615,7 +616,7 @@ def get_user_contact_list_v2(
                             tenant_id=bss.default_id_if_none(x_webtrit_tenant_id))
 
     if Capabilities.extensions in bss_capabilities:
-        contacts, total = bss.retrieve_contacts_v2(session, user, search, page, items_per_page)
+        contacts, total = bss.retrieve_contacts_v2(session, user, search, phone_numbers, page, items_per_page)
 
         return Contacts(items=contacts,
                         pagination=Pagination(