import logging
import re
from typing import Optional
from dateutil import parser
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.models.cores.mappers.base_mapper import BaseMapper
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
from ...exceptions import CoreValidationException
from ..models.extended_demographic import PHONE_TYPE_MAPPING, PhoneType, RM02_FIELDS
from ..queries.demographic_info_query import DemographicInfoQuery
[docs]
class DemographicInfoMapper(BaseMapper):
def __init__(
self,
list_of_queries: list[BaseQuery],
hq_credentials: Optional[HqCredentials] = None,
zone_context: Optional[CoreUser] = None,
is_company=None,
logger: logging.Logger = None,
):
self.logger = logger
self.is_company = is_company
super().__init__(
list_of_queries=list_of_queries,
hq_credentials=hq_credentials,
zone_context=zone_context,
)
[docs]
def parse_returned_queries(
self, list_of_queries: list[BaseQuery]
) -> DemographicInfo:
"""
Parses a list of queries to return demographic information from the core.
:raises CoreValidationException: Business data was submitted in the user's fields
"""
assert len(list_of_queries) == 1
assert isinstance(list_of_queries[0], DemographicInfoQuery), (
"Query must be instance of FISHorizon.queries.DemographicInfoQuery"
)
raw_data = list_of_queries[0].raw_core_response
response = re.sub("HSRM02\\d+\\s+HEDS", "00Q200~", raw_data)
demo_data = response.split("~")
fields = dict([(i, l.strip()) for i, l in enumerate(RM02_FIELDS)]) # noqa: E741
rm02_fields = dict(zip(fields.values(), demo_data[1:]))
self.list_of_queries[0].logger.debug(
f"RM02 Fields from demographic call: {rm02_fields}"
)
first_name = demo_data[41]
middle_name = demo_data[42]
last_name = demo_data[43]
title = ""
try:
dob = demo_data[86]
parser.parse(dob).strftime("%m-%d-%Y")
except ValueError:
dob = ""
ssn = demo_data[21].replace("-", "")
cif = demo_data[14]
address_line_1 = (
demo_data[136] + " " + demo_data[137] if demo_data[136] else demo_data[137]
)
address_line_2 = (
demo_data[138] + " " + demo_data[139] if demo_data[138] else demo_data[139]
)
zipcode = (
demo_data[144] + "-" + demo_data[145] if demo_data[145] else demo_data[144]
)
address = Address(
address_line_1, address_line_2, demo_data[142], demo_data[143], zipcode
)
# If no last name found, consider this as a business customer by assigning BusinessName to last_name for Business Enrollment
if not last_name:
last_name = demo_data[1]
if self.is_company is not None and not self.is_company:
raise CoreValidationException(
"Business data was submitted in the user's fields"
)
emails = []
phones = []
drivers_lic = None
for k, data in enumerate(demo_data):
if data == "DL":
drivers_lic = DriverLicense(demo_data[k + 1], demo_data[k + 2])
if data == "EML":
emails.append(demo_data[k + 1])
if data == "PS":
rm02_fields["passport"] = demo_data[k + 1]
if data in PHONE_TYPE_MAPPING.keys():
if data == "BUS":
found_phone = Phone(
area_code=demo_data[k + 2],
phone_number=demo_data[k + 3].replace("-", ""),
extension=demo_data[k + 4],
phone_type=PhoneType.BUSINESS,
)
else:
found_phone = Phone(
area_code=demo_data[k + 2],
phone_number=demo_data[k + 3].replace("-", ""),
phone_type=PHONE_TYPE_MAPPING.get(data, PhoneType.PERSONAL),
)
phones.append(found_phone)
return DemographicInfo(
dob,
emails,
phones,
[address],
first_name,
last_name,
ssn,
middle_name=middle_name,
title=title,
driver_license=drivers_lic,
primary_cif=cif,
additional_details=rm02_fields,
)