from datetime import datetime
import logging
from typing import Optional, Callable
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.models.cores.base_core import BaseCore
from q2_sdk.models.cores.mappers.demographic_info import BaseDemographicInfoMapper
from q2_sdk.models.cores.models.core_user import CoreUser
from q2_sdk.models.online_transaction import Transaction
from q2_cores.Symitar.queries import mock_responses
from ..Symitar import mappers, queries
[docs]
class Core(BaseCore):
CONFIG_FILE_NAME = "Symitar_Core"
REQUIRED_CONFIGURATIONS = {
"CARD_PREFIX": "1",
"DEVICE_TYPE": "Q2",
"UNIT_NUMBER": "1",
}
OPTIONAL_CONFIGURATIONS = {
"GET_CARD_TYPE_WHITELIST": ["Credit Card"],
"USE_SYMXCHANGE": False,
"SET_MEMBER_NUMBER_AS_DEMO_CIF": False,
"STRIP_ZEROS_FROM_MEMBER_NUMBER": False,
}
def __init__(
self,
logger: logging.Logger,
core_user: CoreUser,
hq_credentials: Optional[HqCredentials] = None,
**kwargs,
):
super().__init__(logger, core_user, hq_credentials=hq_credentials, **kwargs)
self.use_symxchange = getattr(self.config, "USE_SYMXCHANGE", False)
[docs]
async def build_base_query(
self,
repgen_name: str,
procedure: str,
member_number: str = None,
data_attrs: dict = None,
mock_response_string: str = None,
data_file_name: str = "",
rg_params: dict = None,
purge_account_cache: bool = False,
mock_failure=False,
) -> mappers.SymitarBaseMapper:
"""A helper method for constructing an Repgen/PowerOn Symconnect request string.
:param repgen_name: Repgen/PowerOn name that will be called
:param procedure: Value that will be passed to RGSTATE
:param member_number: Symitar account number, defaults to None. If none will use self.configured_user.customer_primary_cif
:param data_attrs: Key/Value pairs to be saved into a letter file e.g. first_name=Joe, defaults to None
:param mock_response_string: The response to emulate what Symitar would respond with, defaults to None
:param data_file_name: Name of the letter file that will be created if using data_attrs, defaults to "".
If used RGUSERCHR1 will also contain the letter file name.
:param rg_params: RGUSERCHR1 through RGUSERCHR5 parameters, defaults to None
:param purge_account_cache: Will tell the adapter to dump it's cache for this account, defaults to False
:param mock_failure: Simulate a Core failure
:return: A mapper that can be executed on
"""
member_number = (
self.configured_user.customer_primary_cif
if not member_number
else member_number
)
query = queries.SymitarBaseQuery(
self.logger,
member_number,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
repgen_name,
procedure,
data_attrs=data_attrs,
mock_response_string=mock_response_string,
data_file_name=data_file_name,
rg_params=rg_params,
purge_account_cache=purge_account_cache,
mock_failure=mock_failure,
use_symxchange=self.use_symxchange,
)
if self.use_symxchange:
return mappers.SymXchangeBaseMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
else:
return mappers.SymitarBaseMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def build_demographic_info(self) -> BaseDemographicInfoMapper:
set_member_number_as_demo_cif = getattr(
self.config, "SET_MEMBER_NUMBER_AS_DEMO_CIF", False
)
strip_zeroes_from_cif = getattr(
self.config, "STRIP_ZEROS_FROM_MEMBER_NUMBER", False
)
demographic_query = queries.DemographicInfoQuery(
self.logger,
self.configured_user.customer_primary_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
use_symxchange=self.use_symxchange,
set_member_number_as_demo_cif=set_member_number_as_demo_cif,
strip_zeroes_from_cif=strip_zeroes_from_cif,
)
return mappers.DemographicInfoMapper(
[demographic_query],
hq_credentials=self.hq_credentials,
zone_context=self.core_user,
)
[docs]
async def get_eligible_skip_loans(
self, configs: Optional[dict] = None, cif: str = None
) -> mappers.GetEligibleSkipLoansMapper:
configs = configs if configs else {}
repgen = configs.get("repgen_name", "Q2.POWER.PACK")
startstate = configs.get("start_state", "STARTSTATE")
cif = self.configured_user.customer_primary_cif if not cif else cif
query = queries.SymitarBaseQuery(
self.logger,
cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
repgen,
startstate,
mock_response_string=mock_responses.mock_get_eligible_skip_loans_response(),
use_symxchange=self.use_symxchange,
)
return mappers.GetEligibleSkipLoansMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def skip_loan(
self,
loan_id,
account_id,
skip_fee: int,
repgen: str = "Q2.SKIP.PAY",
procedure: str = "SKIP",
loan_note: str = None,
) -> mappers.SkipLoanMapper:
if not loan_note:
loan_note = self.get_loan_note()
rg_params = {
"JRGUSERCHR1": str(loan_id),
"JRGUSERCHR2": str(account_id),
"JRGUSERCHR3": str(skip_fee),
"JRGUSERCHR4": str(loan_note),
}
query = queries.SymitarBaseQuery(
self.logger,
self.configured_user.customer_primary_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
repgen,
procedure,
rg_params=rg_params,
mock_response_string=mock_responses.mock_skip_loan_response(),
)
return mappers.SkipLoanMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
@staticmethod
def get_loan_note():
now_time = datetime.now()
time_format = "%Y-%m-%dT%H:%M:%S"
timestamp = now_time.strftime(time_format)
note = "Symitar SkipAPay processed on {0}".format(timestamp)
return note
[docs]
async def assess_skip_fee(
self,
loan_id,
account_id,
gl_code,
skip_fee: int,
repgen: str = "Q2.SKIP.PAY",
startstate: str = "STARTSTATE",
) -> mappers.AssessSkipFeeMapper:
query = queries.AssessFeeQuery(
self.logger,
self.configured_user.customer_primary_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
repgen,
startstate,
loan_id,
account_id,
skip_fee,
gl_code,
)
return mappers.AssessSkipFeeMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def build_get_cards(self) -> mappers.GetCardsMapper:
self.logger.warning(
"build_get_cards is deprecated. Please use Core.build_get_all_cards instead."
)
list_of_queries = []
card_accounts = [
x
for x in self.core_user.account_list
if x.product_type_name in self.config.GET_CARD_TYPE_WHITELIST
]
for account in self.core_user.account_list:
if (
account.hydra_product_type_code == "D"
and account.can_deposit
and account.can_view
):
card_accounts.append(account)
unique_cifs = sorted(
set([
x.cif_internal_unmasked
for x in card_accounts
if x.cif_internal_unmasked is not None
])
)
for internal_cif in unique_cifs:
list_of_queries.append(
queries.GetCardsQuery(
self.logger,
internal_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
use_symxchange=self.use_symxchange,
)
)
return mappers.GetCardsMapper(
list_of_queries,
hq_credentials=self.hq_credentials,
zone_context=self.core_user,
)
[docs]
async def build_get_all_cards(self) -> mappers.GetAllCardsMapper:
list_of_queries = []
card_accounts = [
x
for x in self.core_user.account_list
if x.product_type_name in self.config.GET_CARD_TYPE_WHITELIST
]
for account in self.core_user.account_list:
if (
account.hydra_product_type_code == "D"
and account.can_deposit
and account.can_view
):
card_accounts.append(account)
unique_cifs = sorted(
set([
x.cif_internal_unmasked
for x in card_accounts
if x.cif_internal_unmasked is not None
])
)
for internal_cif in unique_cifs:
list_of_queries.append(
queries.GetAllCardsQuery(
self.logger,
internal_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
use_symxchange=self.use_symxchange,
)
)
return mappers.GetAllCardsMapper(
list_of_queries,
hq_credentials=self.hq_credentials,
zone_context=self.core_user,
)
[docs]
async def get_sub_account_list(self) -> mappers.GetSubAccountMapper:
query = queries.GetSubAccountListQuery(
self.logger,
self.configured_user.customer_primary_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
)
return mappers.GetSubAccountMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def get_sub_account_details(
self, share_type_id: int, funding_amount: float
) -> mappers.GetSubAccountDetailsMapper:
rg_params = {"share_type": share_type_id, "funding_amount": funding_amount}
query = queries.GetSubAccountDetailsQuery(
self.logger,
self.configured_user.customer_primary_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
rg_params,
)
return mappers.GetSubAccountDetailsMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def open_sub_account(
self,
share_type_id: int,
funding_amount: float,
share_start_id: int,
data_file_name: str,
withholdings: int = None,
joint_locator: int = None,
) -> mappers.OpenAccountMapper:
rg_params = {
"AMOUNT": funding_amount,
"SHARETYPE": share_type_id,
"STARTID": share_start_id,
}
if withholdings is not None:
rg_params["WITHHOLDINGS"] = withholdings
if joint_locator is not None:
rg_params["JOINTSIGNERSLOCS"] = joint_locator
query = queries.OpenAccountQuery(
self.logger,
self.configured_user.customer_primary_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
data_file_name,
rg_params,
)
return mappers.OpenAccountMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
def get_payment_builder(self, payment_type: str) -> Callable:
supported_payments = {"PrincipalPayment": self.build_principal_payment_string}
payment_method = supported_payments.get(payment_type)
if not payment_method:
self.logger.error(
"The selected payments type ({}) is currently not supported by this core.".format(
payment_type
)
)
raise NotImplementedError(
"{} is not a payment supported by this core".format(payment_type)
)
return payment_method
[docs]
def build_principal_payment_string(
self, transaction_info: Transaction
) -> queries.SymitarPrincipalPaymentQuery:
"""
Builds custom Symitar principal loan payment request to be added to AddFundsTransfer GT Data
"""
query = queries.SymitarPrincipalPaymentQuery(
self.logger,
transaction_info.from_account,
transaction_info.to_account,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
transaction_info.amount,
memo=transaction_info.memo,
)
return query
[docs]
async def get_estatement_enrollment_info(
self, configs: Optional[dict] = None
) -> mappers.EstatementEnrollmentInfoMapper:
if configs:
repgen = configs.get("core_call", "Q2.POWER.PACK")
startstate = configs.get("core_method", "ESTATEMENTSTARTSTATE")
else:
repgen = "Q2.POWER.PACK"
startstate = "ESTATEMENTSTARTSTATE"
query = queries.SymitarBaseQuery(
self.logger,
self.configured_user.customer_primary_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
repgen,
startstate,
rg_params=configs,
mock_response_string=mock_responses.mock_estatement_enrollment_info_response(),
use_symxchange=self.use_symxchange,
)
return mappers.EstatementEnrollmentInfoMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def update_estatement_preferences(
self, configs: Optional[dict] = None
) -> mappers.EstatementUpdateMapper:
if configs:
rg_params = {
"RGUSERNUM2": configs["estatements_enabled_value"],
"RGUSERNUM3": configs["mail_code"],
"RGUSERCHR1": configs["email"],
"RGUSERNUM4": configs["enable_new_statement_notifications"],
}
repgen = configs.get("core_call", "Q2.POWER.PACK")
startstate = configs.get("core_method", "ESTATEMENTUPDATE")
else:
rg_params = {}
repgen = "Q2.POWER.PACK"
startstate = "ESTATEMENTUPDATE"
query = queries.SymitarBaseQuery(
self.logger,
self.configured_user.customer_primary_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
repgen,
startstate,
rg_params=rg_params,
mock_response_string=mock_responses.mock_estatement_enrollment_update_response(),
use_symxchange=self.use_symxchange,
)
return mappers.EstatementUpdateMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def get_account_demographic(
self, repgenname=None, procedure=None
) -> mappers.AccountDemographicMapper:
repgenname = "Q2.POWER.PACK" if not repgenname else repgenname
procedure = "ALLACCOUNTPROFILES" if not procedure else procedure
query = queries.AccountDemographicQuery(
self.logger,
self.configured_user.customer_primary_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
repgenname,
procedure,
use_symxchange=self.use_symxchange,
)
return mappers.AccountDemographicMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def update_account_demographic(
self, repgenname, procedure, data_file, rg_params
) -> mappers.AccountDemographicUpdateMapper:
query = queries.AccountDemographicUpdateQuery(
self.logger,
self.configured_user.customer_primary_cif,
self.config.UNIT_NUMBER,
self.config.DEVICE_TYPE,
self.config.CARD_PREFIX,
repgenname,
procedure,
data_file,
rg_params,
use_symxchange=self.use_symxchange,
)
return mappers.AccountDemographicUpdateMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)