import logging
from enum import Enum
from typing import NewType
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.alert_details_mapper import AlertDetailsMapper
from .mappers.alert_type_details_mapper import AlertTypeDetailsMapper
from .mappers.alert_types_mapper import AlertTypesMapper
from .mappers.alerts_by_person_serial_mapper import AlertsByPersonSerialMapper
from .mappers.demographic_info_mapper import DemographicInfoMapper
from .mappers.login_verify_mapper import LoginVerifyMapper
from .models import LoginVerifyResponse
from .queries.alert_details_query import AlertDetailsQuery
from .queries.alert_type_details_query import AlertTypeDetailsQuery
from .queries.alert_types_query import AlertTypesQuery
from .queries.alerts_by_person_serial_query import AlertsByPersonSerialQuery
from .queries.demographic_info_query import DemographicInfoQuery
from .queries.login_verify_query import LoginVerifyQuery
from .utils import SearchType
AlertStr = NewType("AlertStr", str)
[docs]
class PersonVerificationBy(Enum):
BY_TIN = "BY_TIN"
BY_ACCOUNT_NUMBER = "BY_ACCOUNT_NUMBER"
[docs]
class Core(BaseCore):
"""
Builds a request to retrieve
- demographic information for a user
- alerts, alert types, and alert type details
REQUIRED_CONFIGURATIONS:
A dictionary containing required values for the core call
- 'CHANNEL_SERIAL' (str): required to get valid data from the Corelation core
OPTIONAL_CONFIGURATIONS:
A dictionary containing optional values for the demographic info core call
- 'PERSON_SERIAL' (int): The person's identifier. Used for login verification
- 'GET_SERIAL_BY_CIF' (bool): Used during login verification to determine if login verification is performed with CIF/MemberNumber or SSN.
- 'SKIP_LOGIN_VERIFY' (str): If set to true, will skip login verification and search value set in PERSON_VERIFICATION_BY to perform search
- 'PERSON_VERIFICATION_BY' (str): Determines what is sent in the demographic info request.
- BY_TIN (str): Makes one demographic info request with the TaxID/SSN
- BY_ACCOUNT_NUMBER (str): Makes one demographic info request with the member number
- 'PERSON_VERIFICATION_ACCOUNT_NUMBER_ZERO_PADDING_LENGTH' (int): This will pad account numbers prior to making the request
- 'ADDR_TYP_PRIORITY_ORDER_FOR_DEMO_INFO' (list): Loops through addresses returned from the core, and select address based on first priority
- 'ALERT_DESCRIPTIONS' (list): Alert descriptions that should prevent enrollment
"""
CONFIG_FILE_NAME = "Corelation_Core"
REQUIRED_CONFIGURATIONS = {
"CHANNEL_SERIAL": "",
}
OPTIONAL_CONFIGURATIONS = {
"PERSON_SERIAL": None,
"GET_SERIAL_BY_CIF": False,
"SKIP_LOGIN_VERIFY": False,
"PERSON_VERIFICATION_BY": "BY_TIN",
"PERSON_VERIFICATION_ACCOUNT_NUMBER_ZERO_PADDING_LENGTH": 10,
"ADDR_TYP_PRIORITY_ORDER_FOR_DEMO_INFO": ["R"],
"ALERT_DESCRIPTIONS": ["deceased", "new address needed"],
}
def __init__(
self,
logger: logging.Logger,
core_user: CoreUser,
hq_credentials: str = None,
person_serial: int = None,
**kwargs,
):
super().__init__(logger, core_user, hq_credentials=hq_credentials, **kwargs)
self.person_serial: int = person_serial
if not person_serial:
self.person_serial = getattr(self.config, "PERSON_SERIAL", None)
[docs]
async def build_demographic_info(self) -> DemographicInfoMapper:
"""
Builds core requests to retrieve demographic information
"""
req_dict = {}
skip_login_verify = getattr(self.config, "SKIP_LOGIN_VERIFY", False)
if skip_login_verify:
self.logger.info("Skipping login verification")
search_type = SearchType.PERSON_VERIFICATION
req_dict = self.get_person_verification_requirements()
else:
self.logger.info("Performing login verification")
search_type = SearchType.LOGIN_VERIFY
if not self.person_serial:
self.logger.info(
"Person serial not provided, initiating core request for serial value"
)
req_dict = await self.get_login_verify_requirements()
else:
self.logger.info("Using provided person serial for search")
req_dict = {"person_serial": self.person_serial}
person_serial = req_dict.get("person_serial")
if not person_serial:
raise MissingCoreParameters("Person serial not found")
return await self._build_demo_mapper(req_dict, search_type)
async def _build_demo_mapper(
self, req_dict: dict, search_type: SearchType
) -> DemographicInfoMapper:
"""
Builds the mapper for the demograhic information core call
"""
demographic_query = DemographicInfoQuery(
self.logger, req_dict=req_dict, search_type=search_type
)
return DemographicInfoMapper(
[demographic_query],
self.configured_user.ssn,
self.configured_user.customer_primary_cif,
hq_credentials=self.hq_credentials,
search_type=search_type,
configs=self.config,
)
[docs]
async def get_login_verify_requirements(self) -> dict:
"""
Build a core request for login verification,
performing the search either by SSN or CIF based on the GET_SERIAL_BY_CIF config.
Result can then be used to build the requirements dictionary for a demographic call
"""
verify_login_by_cif = getattr(self.config, "GET_SERIAL_BY_CIF", False)
if verify_login_by_cif:
self.logger.info("Performing verification with CIF")
search_method = self.configured_user.customer_primary_cif
else:
self.logger.info("Performing verification with SSN")
search_method = self.configured_user.ssn
initial_mapper = self.make_login_verify_xml(
self.config.CHANNEL_SERIAL, search_method
)
response: LoginVerifyResponse = await initial_mapper.execute()
self.person_serial = response.person_serial
req_dict = {"person_serial": self.person_serial}
if self.person_serial:
self.logger.info("Person serial found for demographic request")
return req_dict
[docs]
def get_person_verification_requirements(self) -> dict:
"""
If SKIP_LOGIN_VERIFY config is set to true, it will skip the login verification and
build requirements dictionary for a demographic call which will rely on the
PERSON_VERIFICATION_BY config. Config will determine whether search should be performed
by SSN (BY_TIN) or CIF (BY_ACCOUNT_NUMBER) to retrieve the person serial.
An error will be raised if the person is deceased.
"""
person_verification_by = getattr(
self.config, "PERSON_VERIFICATION_BY", "BY_TIN"
)
enum_dict = {member.value: member for member in PersonVerificationBy}
person_verification_type = enum_dict.get(person_verification_by)
match person_verification_type:
case PersonVerificationBy.BY_TIN:
verification_value = self.configured_user.ssn
case PersonVerificationBy.BY_ACCOUNT_NUMBER:
zero_padding_len = getattr(
self.config,
"PERSON_VERIFICATION_ACCOUNT_NUMBER_ZERO_PADDING_LENGTH",
10,
)
verification_value = self.configured_user.customer_primary_cif.zfill(
zero_padding_len
)
case _:
raise MissingCoreParameters(
'Invalid Values: PERSON_VERIFICATION_BY config can only be set to "BY_TIN" OR "BY_ACCOUNT_NUMBER"'
)
self.logger.info(
f"Making person verification call. Searching by {person_verification_type.value}"
)
req_dict = {
"login_id": verification_value,
"verify_by": person_verification_type.value,
}
return req_dict
[docs]
def make_login_verify_xml(
self, channel_serial: int, member_number: int
) -> LoginVerifyMapper:
"""
Builds a core request to get the result of the login verification. Uses CIF or SSN as the identifier.
:param channel_serial: str, set as a configuration for the FI to get valid information from the Core
:member_number: the CIF or SSN depending on configs set
"""
query = LoginVerifyQuery(self.logger, member_number, channel_serial)
return LoginVerifyMapper(
[query], self.logger, hq_credentials=self.hq_credentials
)
[docs]
def get_alert_types(self) -> AlertTypesMapper:
"""
Gets the alert types available on the core
"""
query = AlertTypesQuery(self.logger)
mapper = AlertTypesMapper([query], self.hq_credentials, self.core_user)
return mapper
[docs]
def get_alert_type_details(
self, target_serials: list[str]
) -> AlertTypeDetailsMapper:
"""
Gets the details of each alert type
:param target_serials: Serial IDs of the alerts to check
"""
query = AlertTypeDetailsQuery(self.logger, target_serials)
mapper = AlertTypeDetailsMapper([query], self.hq_credentials, self.core_user)
return mapper
[docs]
def get_alerts(self, person_serial: str) -> AlertsByPersonSerialMapper:
"""
Gets a user's alerts within Corelation Keystone
:param person_serial: Person serial number
"""
query = AlertsByPersonSerialQuery(self.logger, person_serial=person_serial)
mapper = AlertsByPersonSerialMapper(
[query], self.hq_credentials, self.core_user
)
return mapper
[docs]
def get_alert_details(
self, target_serials: AlertStr, alert_type_details: dict
) -> AlertDetailsMapper:
"""
Gets the details of each of the user's alerts
:param target_serials: Serial IDs of the alerts to check. Expects a string representation of the Alert object
:param alert_type_details: Details for each alert type
"""
query = AlertDetailsQuery(self.logger, target_serials=target_serials)
mapper = AlertDetailsMapper(
[query], self.hq_credentials, self.core_user, alert_type_details
)
return mapper