from copy import deepcopy
import logging
from dateutil import parser
from lxml import etree
from q2_sdk.models.demographic import AddressType, DemographicInfo, PhoneType
from q2_sdk.models.cores.queries.base_query import BaseQuery
from q2_cores import data_helpers
[docs]
class UpdateDemographicQuery(BaseQuery):
"""Updates addresses, First/Last Name, Phones, etc"""
def __init__(
self,
logger: logging.Logger,
cust_number: str,
demographic_info: DemographicInfo,
orig_demo_core_resp: str,
):
self.cust_number = cust_number
self.demographic_info = demographic_info
self.orig_demo_core_resp = etree.fromstring(orig_demo_core_resp)
super().__init__(logger)
[docs]
def make_addr_node(self, existing_node_name, address_type):
# Copying nodes in lxml moves them, so deep copy the original response first
orig_demo_core_resp = deepcopy(self.orig_demo_core_resp)
existing_addr = orig_demo_core_resp.find("{*}" + existing_node_name)
try:
addr = [
x
for x in self.demographic_info.addresses
if x.address_type == address_type
][0]
existing_addr.find("{*}Addr1").text = addr.address_1
existing_addr.find("{*}Addr2").text = addr.address_2
existing_addr.find("{*}City").text = addr.city
existing_addr.find("{*}State").text = addr.state
existing_addr.find("{*}Zip").text = addr.zipcode
except IndexError:
# No residential address sent to update
pass
return existing_addr
[docs]
def build(self):
# Copying nodes in lxml moves them, so deep copy the original response first
orig_demo_core_resp = deepcopy(self.orig_demo_core_resp)
home_phone = ""
work_phone = ""
work_phone_extension = ""
cell_phone = ""
home_phones = [
x for x in self.demographic_info.phones if x.type == PhoneType.PERSONAL
]
work_phones = [
x for x in self.demographic_info.phones if x.type == PhoneType.BUSINESS
]
cell_phones = [
x for x in self.demographic_info.phones if x.type == PhoneType.CELL
]
if home_phones:
phone = home_phones[0]
home_phone = "{} {}-{}".format(
phone.area_code, phone.phone_number[:3], phone.phone_number[3:]
)
if work_phones:
phone = work_phones[0]
work_phone = "{} {}-{}".format(
phone.area_code, phone.phone_number[:3], phone.phone_number[3:]
)
work_phone_extension = phone.extension
if cell_phones:
phone = cell_phones[0]
cell_phone = "{} {}-{}".format(
phone.area_code, phone.phone_number[:3], phone.phone_number[3:]
)
root = etree.Element("CustDetailsMod")
cust_details_mod_rq = etree.SubElement(
root,
"CustDetailsModRq",
nsmap={
"xsi": "http://www.w3.org/2001/XMLSchema-instance",
"xsd": "http://www.w3.org/2001/XMLSchema",
},
)
etree.SubElement(cust_details_mod_rq, "CustNumber").text = self.cust_number
etree.SubElement(
cust_details_mod_rq, "FirstName"
).text = self.demographic_info.first_name
etree.SubElement(
cust_details_mod_rq, "LastName"
).text = self.demographic_info.last_name
etree.SubElement(
cust_details_mod_rq, "MiddleName"
).text = self.demographic_info.middle_name
etree.SubElement(
cust_details_mod_rq, "CustTitle"
).text = self.demographic_info.title
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}TaxIdType"))
etree.SubElement(
cust_details_mod_rq, "TaxId"
).text = self.demographic_info.social_security_number
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}TaxIdCertifyDt"))
existing_birth_dt = orig_demo_core_resp.find("{*}BirthDt")
if self.demographic_info.dob:
existing_birth_dt.text = parser.parse(self.demographic_info.dob).isoformat()
cust_details_mod_rq.append(existing_birth_dt)
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}Gender"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}MaritalStatus"))
etree.SubElement(cust_details_mod_rq, "HomePhone").text = home_phone
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}HomeFax"))
etree.SubElement(cust_details_mod_rq, "WorkPhone").text = work_phone
etree.SubElement(
cust_details_mod_rq, "PhoneExtension"
).text = work_phone_extension
etree.SubElement(cust_details_mod_rq, "CellPhone").text = cell_phone
etree.SubElement(
cust_details_mod_rq, "EmailAddr"
).text = self.demographic_info.emails[0]
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}Memo"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}RelPricingGroup"))
# Driver's License
if self.demographic_info.driver_license:
dl_number = self.demographic_info.driver_license.dl_number
dl_state = self.demographic_info.driver_license.state
else:
dl_number = orig_demo_core_resp.find("{*}DriversLicense").text
dl_state = orig_demo_core_resp.find("{*}LicenseState").text
etree.SubElement(cust_details_mod_rq, "DriversLicense").text = dl_number
etree.SubElement(cust_details_mod_rq, "LicenseState").text = dl_state
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}Dependants"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}DependentAges"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}SchoolYears"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CustType"))
res_addr = self.make_addr_node("ResAddr", AddressType.RESIDENTIAL)
cust_details_mod_rq.append(res_addr)
etree.SubElement(cust_details_mod_rq, "HousingType").text = "UnSpecified"
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}PrevAddr"))
etree.SubElement(cust_details_mod_rq, "PrevHousingType").text = "UnSpecified"
postal_addr = self.make_addr_node("PostalAddr", AddressType.POSTAL)
cust_details_mod_rq.append(postal_addr)
etree.SubElement(cust_details_mod_rq, "AltAddrFlag").text = "false"
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}AltAddr"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}AltAddrStartDt"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}AltAddrEndDt"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}AltAddrRecurringFlag"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}AltAddrExpireDt"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}AcctStatus"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}AcctAccessLevel"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CustInfoFrozenFlag"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CustPswd"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}BillerType"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}BillerRegFlag"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}BillerAcct"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}BillerMemoFlag"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}NewBillPayer"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}BillPresFlag"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}DelinqFlag"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}BondInt"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}BondValue"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}BondCashed"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}PrevYrBondInt"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}PrevYrBondValue"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}PrevYrBondCashed"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}RelPricingGroupCER"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}SecretId"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CustClass"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CustBranch"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CustPayrollDept"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CustMailCode"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CustNameExtension"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}BusinessType"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}TaxExempt"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}PersonnelId"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CompanyId"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CoinDepositAllowFlag"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}OptOutElect"))
tax_name = etree.SubElement(cust_details_mod_rq, "TaxName")
etree.SubElement(tax_name, "TaxNameLast")
etree.SubElement(tax_name, "TaxNameFirst")
etree.SubElement(tax_name, "TaxNameInitials")
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}OpeningMethod"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}DeathDt"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}UserDefinedFieldTable"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CustClassList"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}OvdSvcElection"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}CustLocationCode"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}Estatement"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}JoinDt"))
cust_details_mod_rq.append(orig_demo_core_resp.find("{*}Eligibility"))
for elem in cust_details_mod_rq.getchildren():
if None not in elem.nsmap:
elem.attrib["xmlns"] = "http://harlandfs.com/pl/TransactionExchange"
xml_str = etree.tostring(root).decode("utf8")
return data_helpers.normalize_xml_str(xml_str)
[docs]
def mock_response(self):
return "Unknown what this shape is. Please fill it in later"