import logging
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.hq.models.online_user import OnlineUser
from q2_sdk.models.cores.base_core import BaseCore
from q2_sdk.models.cores.models.core_user import CoreUser
from ..exceptions import MissingCoreParameters
from .mappers.demographic_info_mapper import DemographicInfoMapper
from .mappers.initial_search_mapper import InitialSearchMapper
from .queries.demographic_info_query import DemographicInfoQuery
from .queries.initial_search_query import InitialSearchQuery
[docs]
class Core(BaseCore):
"""
Summary Flow:
Uses an initial search system, where an initial search is made to retrieve a unique internal ID for that user.
That ID can then be used for all additional calls. This base implementation provides the ability to retrieve demographic information from the core.
The RM01 Customer Search call retrieves the unique customer identifier (the internal ID number) using the SSN and Last Name to search.
The RM02 Customer Inquiry call is the primary demographic call. It uses the unique customer identifier as the input for the call.
"""
CONFIG_FILE_NAME = "FISHorizon_Core"
OPTIONAL_CONFIGURATIONS = {
"SEARCH_BY_LASTNAME_IF_SSN_MISSING": False,
"SEARCH_BY_CIF": False,
}
def __init__(
self,
logger: logging.Logger,
core_user: CoreUser,
hq_credentials: HqCredentials = None,
**kwargs,
):
super().__init__(logger, core_user, hq_credentials=hq_credentials, **kwargs)
self.tran_code = None
[docs]
async def build_initial_customer_key_search(self) -> InitialSearchMapper:
"""
Make initial search call to retrieve unique customer identifier associated with the SSN.
Note: If an SSN isn't available for the search and the optional configuration ``SEARCH_BY_LASTNAME_IF_SSN_MISSING`` is set to True,
an attempt to retrieve the unique customer identifier by last name and date of birth (if available) is made.
"""
self.configured_user: OnlineUser
ssn = self.configured_user.ssn or ""
dob = ""
last_name = ""
# if ssn is missing, verify if attempt to search with dob and lastname should be made
search_by_last_name = getattr(
self.config, "SEARCH_BY_LASTNAME_IF_SSN_MISSING", False
)
if not ssn and search_by_last_name:
self.logger.debug("Searching by last name...")
last_name = self.configured_user.last_name
user_dob = self.configured_user.demographic_info.dob
if user_dob:
dob = f"{user_dob[6:10]}{user_dob[0:2]}{user_dob[3:5]}" # DOB format must be: YYYYMMDD
if not any([ssn, last_name]):
self.logger.debug(
"SSN and last name not found. Cannot perform initial search"
)
raise MissingCoreParameters("No values found for initial search")
initial_search = InitialSearchQuery(
logger=self.logger, ssn=ssn, last_name=last_name, dob=dob
)
return InitialSearchMapper([initial_search], hq_credentials=self.hq_credentials)
[docs]
async def build_demographic_info(
self, unique_customer_identifier: str = None
) -> DemographicInfoMapper:
"""
Builds a request to retrieve a user's demographic information from the core using a unique identifier.
If a unique identifier is provided, a demographic request is created immediately using the provided value.
If SEARCH_BY_CIF config is set to True, a demographic request is created immediately using the online user's CIF.
Otherwise, an initial search is performed to find the unique customer identifier (using the user's SSN or last name/dob if optional configuration is set).
If the search fails, the user's primary cif is still utilized for the request.
:param unique_customer_identifier: A unique identifier known to Fiserv for the customer, used for the demographic request. Defaults to ``None``
"""
self.logger.debug("Attempting to retrieve user's demographic information...")
if unique_customer_identifier:
self.logger.debug(
"Making demographic request using provided unique customer identifier"
)
return await self._build_demo_mapper(unique_customer_identifier)
search_by_cif = getattr(self.config, "SEARCH_BY_CIF", False)
if search_by_cif:
self.logger.debug("Making demographic request using CIF value")
unique_customer_identifier = self.configured_user.customer_primary_cif
return await self._build_demo_mapper(unique_customer_identifier)
try:
self.logger.debug("Building search for unique customer identifier")
initial_search = await self.build_initial_customer_key_search()
unique_customer_identifier = await initial_search.execute()
except MissingCoreParameters:
unique_customer_identifier = None
self.logger.debug(
f"Initial customer identifer search attempted. FISHorizon unique customer idenfier found: {unique_customer_identifier is not None}"
)
unique_customer_identifier = (
unique_customer_identifier or self.configured_user.customer_primary_cif
)
self.logger.debug(
f"Unique FISHorizon customer identifier: {unique_customer_identifier}"
)
if not unique_customer_identifier:
raise MissingCoreParameters("Customer identifier not found")
return await self._build_demo_mapper(unique_customer_identifier)
async def _build_demo_mapper(self, unique_customer_identifier):
"""
Uses unique customer identifier found to return the demographic information mapper
"""
demographic_query = DemographicInfoQuery(
self.logger, unique_customer_identifier
)
return DemographicInfoMapper(
[demographic_query],
is_company=self.core_user.online_user.is_company,
hq_credentials=self.hq_credentials,
logger=self.logger,
)