import typing
from lxml import objectify
from q2_sdk.models.cores.mappers.demographic_info import BaseDemographicInfoMapper
from q2_sdk.models.cores.queries.base_query import BaseQuery
from q2_sdk.models.demographic import (
Address,
DemographicInfo,
Phone,
PhoneType,
DriverLicense,
)
from ...MiserCohesion.queries import DemographicInfoQuery
[docs]
class DemographicInfoMapper(BaseDemographicInfoMapper):
[docs]
@staticmethod
def parse_returned_queries(
list_of_queries: typing.List[BaseQuery],
) -> DemographicInfo:
assert len(list_of_queries) == 1, (
"MiserCohesion only knows how to deal with a single demographicinfo query"
)
assert isinstance(list_of_queries[0], DemographicInfoQuery), (
"Query must be an instance of MiserCohesion.queries.DemographicInfoQuery"
)
response = list_of_queries[0].raw_core_response
root = objectify.fromstring(response)
ssn = root.TaxID.text
dob = root.BirthDate.text
email = root.Email.text
cif = root.CIFNbr.text
phones = []
home_phone_string = root.findtext(".//{*}HomePhone")
if home_phone_string and home_phone_string != "0000000000":
home_phone = Phone.build_from_str(home_phone_string, PhoneType.PERSONAL)
phones.append(home_phone)
business_phone_string = root.findtext(".//{*}BusPhone")
if business_phone_string and business_phone_string != "0000000000":
work_phone = Phone.build_from_str(root.BusPhone.text, PhoneType.BUSINESS)
work_phone.extension = root.findtext(".//{*}BusPhoneExt")
phones.append(work_phone)
cell_phone_string = root.findtext(".//{*}CellPhone")
if cell_phone_string and cell_phone_string != "0000000000":
cell_phone = Phone.build_from_str(root.CellPhone.text, PhoneType.CELL)
phones.append(cell_phone)
first_name = root.FirstName.text
middle_name = root.MiddleName.text
last_name = root.LastName.text
# The address info comes back in nodes that do not have a guaranteed
# order, but have a reference number in a neighboring node:
# <NameAddrCode*>4 refers to Address 1 Line Info
# <NameAddrCode*>6 refers to Address 2 Line Info
addr_code_elems = root.xpath("//*[starts-with(local-name(), 'NameAddrCode')]")
addr1_code_ref_node = [x for x in addr_code_elems if x.pyval == 4][0]
addr1_code_ref = addr1_code_ref_node.tag[-1]
address_1 = root.find("{*}NameAddrLine" + addr1_code_ref).text
addr2_code_ref_node = [x for x in addr_code_elems if x.pyval == 6][0]
addr2_code_ref = addr2_code_ref_node.tag[-1]
addr_line_2 = root.find("{*}NameAddrLine" + addr2_code_ref).text
split_addr_2 = addr_line_2.split()
city = "".join(split_addr_2[:-2])
state_abbrev = split_addr_2[-2]
zipcode = split_addr_2[-1]
address = Address(address_1, "", city, state_abbrev, zipcode)
dl_number = root.findtext(".//{*}DLNbr")
dl_state = root.findtext(".//{*}DLState")
drivers_license = DriverLicense(dl_number, dl_state)
return DemographicInfo(
dob,
[email],
phones,
[address],
first_name,
last_name,
ssn,
middle_name=middle_name,
driver_license=drivers_license,
primary_cif=cif,
)