import json
import logging
from typing import List, Optional
from dateutil import parser as date_parser
import phonenumbers
from q2_sdk.hq.db.country import Country
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.models.cores.mappers.demographic_info import BaseDemographicInfoMapper
from q2_sdk.models.cores.models.core_user import CoreUser
from q2_sdk.models.cores.queries.base_query import BaseQuery
from q2_sdk.models.demographic import (
Address,
DemographicInfo,
DriverLicense,
Phone,
PhoneType,
)
from ...exceptions import CoreException
from ...models import Address as AddressObj, AddressType, RecordType
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 DemographicInfoMapper(BaseDemographicInfoMapper):
def __init__(
self,
list_of_queries: List[BaseQuery],
hq_credentials: Optional[HqCredentials] = None,
logger: logging.Logger = None,
search_method: SearchMethod = SearchMethod.SSN,
allow_joint_enrollment: bool = False,
core_user: CoreUser = None,
passport_id_type: str = "",
include_phone_and_email: bool = True,
):
self.async_data = None
self.logger = logger
self.search_method = search_method
self.allow_joint_enrollment = allow_joint_enrollment
self.core_user = core_user
self.passport_id_type = passport_id_type
self.include_phone_and_email = include_phone_and_email
super().__init__(list_of_queries=list_of_queries, hq_credentials=hq_credentials)
async def _run_queries(self) -> dict:
await super()._run_queries()
self.async_data = {}
if self.search_method == SearchMethod.DYNAMICPROFILE:
country_obj = Country(self.logger, hq_credentials=self.hq_credentials)
countries = await country_obj.get()
self.async_data = {"countries": countries}
[docs]
def parse_returned_queries(
self, list_of_queries: List[BaseQuery]
) -> DemographicInfo:
"""
Handles the demographic information response from the core
"""
assert len(list_of_queries) == 1 or len(list_of_queries) == 3
assert isinstance(list_of_queries[0], DemographicInfoQuery), (
"Query must be an instance of FISIBSOpenAPI.queries.DemographicInfoQuery"
)
if len(list_of_queries) == 3:
assert isinstance(list_of_queries[1], GetPhoneNumbersQuery), (
"Query must be an instance of FISIBSOpenAPI.queries.GetPhoneNumbersQuery"
)
assert isinstance(list_of_queries[2], GetEmailAddressesQuery), (
"Query must be an instance of FISIBSOpenAPI.queries.GetEmailAddressesQuery"
)
demo_raw_core_resp = list_of_queries[0].raw_core_response
response = (
json.loads(demo_raw_core_resp)
if not isinstance(demo_raw_core_resp, dict)
else demo_raw_core_resp
)
match self.search_method:
case SearchMethod.CIF:
demo_obj = self.build_from_cif_response(response, list_of_queries)
case SearchMethod.ACCOUNT:
demo_obj = self.build_from_acct_response(response)
case SearchMethod.DYNAMICPROFILE:
demo_obj = self.build_from_profile_response(response)
case _:
demo_obj = self.build_from_tax_id_response(response)
return demo_obj
[docs]
def build_from_cif_response(
self, response, list_of_queries=None
) -> DemographicInfo:
"""
Builds the demographic info object using the user's CIF
"""
try:
entity = response["Entity"]["customer"]
except TypeError as err:
raise CoreException(
f"There was a problem with the request to the core. {str(err)}"
)
name_components = entity.get("CIFrstNmeMidInitl", "").split(" ")
first_name = name_components[0]
middle_name = name_components[1] if len(name_components) > 1 else None
last_name = entity.get("CISrnme", "")
mothers_maiden_name = entity.get("CIMothersMdnNme", "")
dob = entity.get("CIBirthdate", "")
tax_id = str(entity.get("CICustTaxNbr", ""))
primary_cif = str(entity["CIApplNbr"]).zfill(11)
passport = get_passport_id_from_profile(entity, self.passport_id_type) or ""
addr_1 = entity["CICurStdAddr1Txt"]
addr_2 = ""
city = entity.get("Cty", "")
state = entity.get("St", "")
zipcode = str(entity.get("ZIP", ""))
country = entity.get("CtryCde", "USA")
record_type = (
RecordType.INTERNATIONAL if country != "USA" else RecordType.DOMESTIC
)
address = AddressObj(
address_1=addr_1,
address_2=addr_2,
city=city,
state=state,
zipcode=zipcode[:5],
country=country,
address_type=AddressType.HOME,
record_type=record_type,
postal_code=zipcode if country != "USA" else "",
province="",
)
phone_response = []
email_response = []
if self.include_phone_and_email:
phone_queries = [list_of_queries[1]]
phone_mapper = GetPhoneNumbersMapper(
phone_queries, hq_credentials=self.hq_credentials
)
phone_response = phone_mapper.parse_returned_queries(phone_queries)
email_queries = [list_of_queries[2]]
email_mapper = GetEmailAddressesMapper(
email_queries, hq_credentials=self.hq_credentials
)
email_response = email_mapper.parse_returned_queries(email_queries)
drivers_license = get_drivers_license_from_entity(entity, state)
entity.update({"passport": str(passport)})
demo_obj = DemographicInfo(
date_of_birth=dob,
list_of_emails=email_response,
list_of_phones=phone_response,
list_of_addresses=[address],
first_name=first_name,
middle_name=middle_name,
last_name=last_name,
ssn=tax_id,
primary_cif=primary_cif,
mothers_maiden_name=mothers_maiden_name,
driver_license=drivers_license,
additional_details=entity,
)
return demo_obj
[docs]
def build_from_acct_response(self, response) -> DemographicInfo:
"""
Builds the demographic info object using the user's AccountNumber and the related AccountType
"""
try:
entities = response["Entity"]["dynamic-related-accountsLst"]
except (KeyError, TypeError):
error_msg = "No account found"
if isinstance(response, dict):
meta_data = response.get("Metadata", {})
error_msg = get_error_message(meta_data, "No account found")
raise CoreException(error_msg)
entity = self._get_required_entity(entities)
dob = entity["CIBirthdate"]
first_name = entity["CICurFrstNmeKeyFld2"]
last_name = entity["CICurLstNmeKeyFld1"]
tax_id = str(entity["CICustTaxNbr"])
mothers_maiden_name = entity.get("CIMothersMaidenNme")
primary_cif = entity["CIRltApplNbr01"].zfill(11)
addr_1 = entity["CICurStdAddr1Txt"]
addr_2 = entity.get("CICurStdAddr2Txt", "")
city = entity.get("Cty", "")
state = entity.get("St")
zipcode = str(entity["ZIP"]) if "ZIP" in entity.keys() else None
country = entity.get("CtryCde", "USA")
address = Address(
address_1=addr_1,
address_2=addr_2,
city=city,
state=state,
zipcode=zipcode,
country=country,
)
drivers_license = get_drivers_license_from_entity(entity, state)
phones = get_phones_from_entity(entity)
demo_obj = DemographicInfo(
date_of_birth=dob,
list_of_emails=[],
list_of_phones=phones,
list_of_addresses=[address],
first_name=first_name,
middle_name=None,
last_name=last_name,
ssn=tax_id,
primary_cif=primary_cif,
mothers_maiden_name=mothers_maiden_name,
driver_license=drivers_license,
)
return demo_obj
def _get_required_entity(self, entities) -> dict:
"""
Grabs the user profile from the core
"""
dob = self.core_user.online_user.demographic_info.dob
submitted_first_name = ""
if self.core_user.online_user.demographic_info.first_name:
submitted_first_name = (
self.core_user.online_user.demographic_info.first_name.lower()
)
elif self.core_user.online_user.first_name:
submitted_first_name = self.core_user.online_user.first_name.lower()
req_entity = None
for entity in entities:
relation_cd = entity["CIEnt2ToEnt1RltCde"]
core_first_name = entity.get("CICurFrstNmeKeyFld2", "").lower()
cleaned_core_dob = None
core_dob = entity.get("CIBirthdate")
cleaned_core_dob = core_dob
if core_dob:
try:
date_obj = date_parser.parse(core_dob)
cleaned_core_dob = date_obj.strftime("%m-%d-%Y")
except ValueError:
cleaned_core_dob = None
if (
self.allow_joint_enrollment
and cleaned_core_dob == dob
and core_first_name == submitted_first_name
):
req_entity = entity
break
elif relation_cd == 901:
req_entity = entity
if req_entity:
return req_entity
else:
raise CoreException(
"An error occurred while getting the user profile from core response."
)
[docs]
def build_from_profile_response(self, response) -> DemographicInfo:
"""
Builds the demographic info object from the dynamic-profile response
"""
try:
entity = response["Entity"]
profile = entity["dynamic-customer-profile"]
except TypeError:
meta_data = response.get("Metadata", {})
error_msg = get_error_message(
meta_data, "There was a problem with the request to the core."
)
raise CoreException(error_msg)
dob = profile.get("CIBirthdate", "")
first_name = profile.get("CICurFrstNmeKeyFld2", "")
last_name = profile.get("CICurLstNmeKeyFld1", "")
middle_name = profile.get("CICurMidNmeKeyFld3", "")
mothers_maiden_name = profile.get("CIMothersMaidenNme", "")
tax_id = str(profile.get("CICustTaxNbr", ""))
primary_cif = str(profile["CICustNbr"]).zfill(11)
passport = get_passport_id_from_profile(profile, self.passport_id_type) or ""
addr_1 = profile["CICurStdAddr1Txt"]
addr_2 = profile.get("CICurStdAddr2Txt", "")
city = profile.get("Cty", "")
state = profile.get("St", "")
country = profile.get("CtryCde", "USA")
zipcode = (
str(profile["ZIP"]) if country == "USA" else str(addr_2).split(" ")[-1]
)
postal_code = zipcode if country != "USA" else ""
record_type = (
RecordType.INTERNATIONAL if country != "USA" else RecordType.DOMESTIC
)
address = AddressObj(
address_1=addr_1,
address_2="",
city=city,
state=state,
zipcode=zipcode[:5],
country=country,
address_type=AddressType.HOME,
record_type=record_type,
postal_code=postal_code,
province="",
)
emails = self._get_emails_from_dynamic_profile(entity)
core_phones = {}
if profile.get("CIPrmyPhNbr"):
core_phones["primary"] = str(profile["CIPrmyPhNbr"])
if profile.get("CIScndyPh"):
core_phones["secondary"] = str(profile["CIScndyPh"])
if profile.get("CIMobilePhNbr"):
core_phones["mobile"] = str(profile["CIMobilePhNbr"])
if profile.get("NonAmericanPhoneNo"):
core_phones["international"] = str(profile["NonAmericanPhoneNo"])
phones = self.build_phone_objects(core_phones)
drivers_license = get_drivers_license_from_entity(profile, state)
profile.update({"passport": str(passport)})
demo_obj = DemographicInfo(
date_of_birth=dob,
list_of_emails=emails,
list_of_phones=phones,
list_of_addresses=[address],
first_name=first_name,
middle_name=middle_name,
last_name=last_name,
ssn=tax_id,
primary_cif=primary_cif,
mothers_maiden_name=mothers_maiden_name,
driver_license=drivers_license,
additional_details=profile,
)
return demo_obj
@staticmethod
def _get_emails_from_dynamic_profile(entity: dict) -> List[str]:
"""
Parses the dynamic-profile response for email addresses
"""
emails = []
if "customer-email-addressLst" not in entity.keys():
return []
for cust_node in entity["customer-email-addressLst"]:
if "email-addressesLst" not in cust_node.keys():
return []
for email_node in cust_node["email-addressesLst"]:
email = email_node["EmailAddr"]
emails.append(email)
return emails
[docs]
@staticmethod
def build_from_tax_id_response(response) -> DemographicInfo:
"""
Builds the demographic info object from the demo call using customer's TaxID
"""
entity = response["Entity"]["customersLst"][0]
dob = entity.get("CIBirthdate", "")
first_name = entity["ElmntFrstNme"]
last_name = entity["SurNme"]
tax_id = str(entity["CICustTaxNbr"])
mothers_maiden_name = entity.get("CIMothersMdnNme")
primary_cif = entity["CICustRtnNbr"].zfill(11)
addr_1 = entity["CICurStdAddr1Txt"]
addr_2 = entity.get("CICurStdAddr2Txt", "")
city = entity["ElmntCity"]
state = entity.get("ElmntState", "")
zipcode = str(entity["ElmntZip"])
country = entity.get("CtryCde", "USA")
address = Address(
address_1=addr_1,
address_2=addr_2,
city=city,
state=state,
zipcode=zipcode,
country=country,
)
drivers_license = get_drivers_license_from_entity(entity, state)
phones = get_phones_from_entity(entity)
demo_obj = DemographicInfo(
date_of_birth=dob,
list_of_emails=[],
list_of_phones=phones,
list_of_addresses=[address],
first_name=first_name,
middle_name=None,
last_name=last_name,
ssn=tax_id,
primary_cif=primary_cif,
mothers_maiden_name=mothers_maiden_name,
driver_license=drivers_license,
)
return demo_obj
[docs]
def build_phone_objects(self, core_phones: dict) -> List[Phone]:
"""
Builds SDK Phone objects using the phones returned from the core
"""
phone_objects = []
for phone_type, core_phone in core_phones.items():
if "+" in core_phone:
parsed_phone = phonenumbers.parse(core_phone)
region_code = phonenumbers.phonenumberutil.region_code_for_number(
parsed_phone
)
countries = self.async_data.get("countries")
country_code = "USA"
for country in countries:
if country.IsoCodeA2 == region_code:
country_code = country.IsoCodeA3
break
phone = Phone(
area_code="",
phone_number=str(parsed_phone.national_number),
phone_type=PhoneType.PERSONAL,
country=country_code,
)
else:
phone = Phone(
area_code=core_phone[:3],
phone_number=core_phone[3:],
phone_type=(
PhoneType.CELL if phone_type == "mobile" else PhoneType.PERSONAL
),
)
phone_objects.append(phone)
return phone_objects
[docs]
def get_phones_from_entity(entity: dict) -> List[Phone]:
"""
Parses the entity element for phone numbers
"""
phones = []
if "CIPrmyPhNbr" in entity.keys():
area_code = str(entity["CIPrmyPhNbr"])[:3]
local_number = str(entity["CIPrmyPhNbr"])[3:]
phones = [
Phone(area_code, local_number, PhoneType.PERSONAL),
]
return phones
[docs]
def get_drivers_license_from_entity(
entity: dict, state: str
) -> Optional[DriverLicense]:
"""
Parses the entity element for a drivers license
"""
drivers_license = None
if "CIDrvrLic" in entity.keys():
drivers_license = DriverLicense(str(entity["CIDrvrLic"]), state)
return drivers_license
[docs]
def get_passport_id_from_profile(profile: dict, passport_id_type: str) -> Optional[str]:
"""
Parses the profile element for a Passport ID
"""
for i in range(1, 6):
nbr_key = f"CIPIDNbr{i}"
type_key = f"CIPIDTyp{i}"
if nbr_key not in profile.keys() or type_key not in profile.keys():
continue
if profile[type_key] == passport_id_type:
return str(profile[nbr_key])
return None
[docs]
def get_error_message(meta_data: dict, default: str) -> str:
"""
Returns the error message from the response if available
:param metadata: Metadata element in the demographic info response
:param default: Message to return if one is not found in the metadata
:return: Error message if available, otherwise default message is returned
"""
if "MsgLst" not in meta_data.keys():
return default
for msg in meta_data["MsgLst"]:
if msg["Text"] not in ("Success", None):
return msg["Text"]
return default