import logging
from typing import List, Optional
from q2_sdk.hq.db.country import Country
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.models.cores.base_core import BaseCore
from q2_sdk.models.cores.models.core_user import CoreUser
from q2_sdk.models.demographic import Phone
from q2_sdk.models.demographic import DemographicInfo
from .mappers.demographic_info_mapper import DemographicInfoMapper
from .mappers.get_email_addresses_mapper import GetEmailAddressesMapper
from .mappers.get_phone_numbers_mapper import GetPhoneNumbersMapper
from .queries.demographic_info_query import DemographicInfoQuery
from .queries.get_email_addresses_query import GetEmailAddressesQuery
from .queries.get_phone_numbers_query import GetPhoneNumbersQuery
from .utils import SearchMethod
[docs]
class Core(BaseCore):
CONFIG_FILE_NAME = "FISIBSOpenAPI_Core"
OPTIONAL_CONFIGURATIONS = {
"GET_DEMO_DATA_BY_ACCT": False,
"GET_DEMO_DATA_BY_CIF": False,
"INCLUDE_PASSPORT_IN_DEMO_INFO": False,
"INCLUDE_PHONE_AND_EMAIL": True,
"ALLOW_JOINT_ENROLLMENT": False,
"USE_DYNAMIC_PROFILE": False,
"DIGITAL_SIGNATURE": (
"000000000000000000000002000000000000000000000000000000000000000000"
"000000030000000000000000000000000000000000000000000000000000000186"
"186180000000000000000000000000000000400000000000000000000000000000"
"000000000000010000000000000000000000000000006000000408000000000006"
"09014002400000D806EC0181"
),
"PASSPORT_ID_TYPE": "",
}
def __init__(
self,
logger: logging.Logger,
core_user: CoreUser,
hq_credentials: Optional[HqCredentials] = None,
**kwargs,
):
super().__init__(logger, core_user, hq_credentials=hq_credentials, **kwargs)
self.logger = logger
self.customer_number = ""
[docs]
async def get_full_demographic_info(self) -> DemographicInfo:
"""
An attempt will be made to first build and send the request to retrieve demographic information by SSN, CIF, or Account.
If the demographic call doesn't include a passport number, and it's required based on the configuration,
another request is made to the core to grab passport information.
If phone or email information is missing from initial request, additional calls are made to the core to retrieve these values
"""
demo_mapper = await self.build_demographic_info()
demo_obj = await demo_mapper.execute()
# The demo info response by Account doesn't include Passport Number. Make another call if that's required
search_by_acc = getattr(self.config, "GET_DEMO_DATA_BY_ACCT", False)
include_passport = getattr(self.config, "INCLUDE_PASSPORT_IN_DEMO_INFO", False)
if search_by_acc and include_passport:
self.configured_user.customer_primary_cif = demo_obj.primary_cif
search_by_dynamic_prof = getattr(self.config, "USE_DYNAMIC_PROFILE", False)
search_method = (
SearchMethod.DYNAMICPROFILE
if search_by_dynamic_prof
else SearchMethod.CIF
)
demo_mapper = await self.build_demographic_info(search_method)
# Overwriting previous demo obj for more complete data, including passport if available
first_name = demo_obj.first_name
demo_obj = await demo_mapper.execute()
if search_method == SearchMethod.CIF:
demo_obj.first_name = first_name # Use value from search by acc for accuracy. CIF returns a "first name + middle initial" field
elif search_method == SearchMethod.DYNAMICPROFILE:
phones = await self.fix_international_phones(demo_obj.phones)
demo_obj.phones = phones
# Make additional calls to get missing phone and email info
if not demo_obj.phones:
get_phones_mapper = await self.get_phone_numbers(demo_obj.primary_cif)
phones = await get_phones_mapper.execute()
demo_obj.phones = phones
if not demo_obj.emails:
get_emails_mapper = await self.get_email_addresses(demo_obj.primary_cif)
emails = await get_emails_mapper.execute()
demo_obj.emails = emails
# Return DemographicInfo object
return demo_obj
[docs]
async def build_demographic_info(
self,
search_method: Optional[SearchMethod] = None,
signature: Optional[str] = None,
) -> DemographicInfoMapper:
"""
Builds a request to retrieve a user's demographic information from the core using a search method (CIF, SSN, Account, DynamicProfile) and signature if provided.
If a search method isn't provided as a parameter, this function will search by SSN by default.
Note: INCLUDE_PHONE_AND_EMAIL configuration is only applicable to searches by CIF.
"""
# If search method isn't provided, check configs to determine the method. Default to SSN if not set
if not search_method:
search_method = self._get_search_method()
# Grab relevant the user-provided data for the demographic call
match search_method:
case SearchMethod.SSN:
req_val = self.configured_user.ssn
case SearchMethod.CIF | SearchMethod.DYNAMICPROFILE:
req_val = (
self.customer_number
if self.customer_number
else self.configured_user.customer_primary_cif
)
case SearchMethod.ACCOUNT:
req_val = self._get_account_info()
# Check configs to grab additional info
allow_joint_enrollment = getattr(self.config, "ALLOW_JOINT_ENROLLMENT", False)
signature = (
getattr(self.config, "DIGITAL_SIGNATURE", "")
if not signature
else signature
)
passport_id_type = getattr(self.config, "PASSPORT_ID_TYPE", "")
# Build mapper
demo_query = DemographicInfoQuery(
self.logger, req_val, search_method, signature
)
queries = [demo_query]
include_phone_and_email = getattr(self.config, "INCLUDE_PHONE_AND_EMAIL", True)
if search_method == SearchMethod.CIF and include_phone_and_email:
phone_query = GetPhoneNumbersQuery(self.logger, req_val)
email_query = GetEmailAddressesQuery(self.logger, req_val)
queries.extend([phone_query, email_query])
demo_mapper = DemographicInfoMapper(
list_of_queries=queries,
hq_credentials=self.hq_credentials,
logger=self.logger,
search_method=search_method,
allow_joint_enrollment=allow_joint_enrollment,
core_user=self.core_user,
passport_id_type=passport_id_type,
include_phone_and_email=include_phone_and_email,
)
return demo_mapper
def _get_search_method(self):
search_by_acc = getattr(self.config, "GET_DEMO_DATA_BY_ACCT", False)
search_by_cif = getattr(self.config, "GET_DEMO_DATA_BY_CIF", False)
search_method = SearchMethod.SSN
if search_by_acc:
search_method = SearchMethod.ACCOUNT
elif search_by_cif:
search_method = SearchMethod.CIF
return search_method
def _get_account_info(self):
try:
acct_num = self.core_user.account_list[0].acct_number_internal_unmasked
type_code = self.core_user.account_list[0].hydra_product_type_code
acct_type = None
if type_code and acct_num:
if type_code in ("D", "S"):
acct_type = "DP"
elif type_code == "L":
acct_type = "LN"
else:
# We can get account type from the AccountType dict as configurable option, we can take as is.
acct_type = type_code
self.logger.info(
f"We are making core calls with Account Type Product Code: {type_code}"
)
except (IndexError, AttributeError):
acct_num = None
acct_type = None
return acct_num, acct_type
[docs]
async def get_phone_numbers(self, customer_number: str) -> List[Phone]:
"""
Grabs phone numbers linked to the CIF
"""
get_phones_query = GetPhoneNumbersQuery(self.logger, customer_number)
get_phones_mapper = GetPhoneNumbersMapper(
[get_phones_query], self.hq_credentials
)
return get_phones_mapper
[docs]
async def get_email_addresses(self, customer_number: str) -> List[str]:
"""
Grabs email addresses linked to the CIF
"""
get_emails_query = GetEmailAddressesQuery(self.logger, customer_number)
get_emails_mapper = GetEmailAddressesMapper(
[get_emails_query], self.hq_credentials
)
return get_emails_mapper
[docs]
async def fix_international_phones(self, phones: List[Phone]) -> Phone:
"""
Fix area codes and local numbers for international phones
Phone numbers that are returned from the core in the "NonAmericanPhoneNo" field
are provided as a string such as "+12 3 456 7890". We don't know which part of
the number is the area code if that country uses area codes, so we need to
check the Q2_Country table to find the expected patterns and reformat the Phone
object data to conform to those patterns.
:param phones: Phone objects built from core data
:return: Phone objects with fixed area codes and local numbers
"""
phone: Phone
fixed_phones = []
country_obj = Country(self.logger, hq_credentials=self.hq_credentials)
countries = await country_obj.get()
for phone in phones:
# USA phone numbers are already formatted as expected, so we can skip them
if phone.country == "USA":
fixed_phones.append(phone)
continue
for country in countries:
if country.IsoCodeA3 == phone.country:
try:
area_code_length = country.CityAreaCodeEditMask.text.count("0")
area_code = phone.phone_number[:area_code_length]
local_number = phone.phone_number[area_code_length:]
fixed_phone = Phone(
area_code, local_number, phone.type, country=phone.country
)
fixed_phones.append(fixed_phone)
except AttributeError:
# Country has no area code, so the existing phone is correct
fixed_phones.append(phone)
return fixed_phones