from __future__ import annotations
from copy import deepcopy
from lxml import objectify
from q2_sdk.hq.models.online_session import OnlineSession
from q2_sdk.hq.models.online_user import OnlineUser
from q2_sdk.core.http_handlers.adapter_handler import Q2AdapterRequestHandler
from q2_sdk.models.adapters.remote_deposit import (
RDCTransactionHistoryListRequest,
RemoteDepositRequest,
RDCValidateRequest,
RemoteDepositRequestType,
RDCCaptureResponse,
RDCValidation,
RDCCaptureRequest,
RDCTransactionHistoryDetailRequest,
RDCTransactionResponse,
RDCTransactionHistoryDetailResponse,
)
import base64
import json
[docs]
class Q2RemoteDepositRequestHandler(Q2AdapterRequestHandler):
"""
Adapter type for Remote Deposit including mRDC.
Invoked when a user interacts with mobile remote deposit interfaces
"""
async def handle_adapter(self, message, *args, **kwargs):
response_as_dict = deepcopy(message)
response_as_dict["HostAccount_Req"][0]["HostErrorCode"] = 0
host_account_req = response_as_dict["HostAccount_Req"][0]
transaction_type = host_account_req["TransactionType"]
transaction_id = host_account_req["TransactionID"]
xml_payload = base64.b64decode(message["RequestorData"][0]["XmlPayload"])
self.logger.debug(xml_payload)
xml_payload_request = objectify.fromstring(xml_payload)
self._setup_session_user(xml_payload_request, response_as_dict)
self.logger.info(
f"Routing to RemoteDepositRequestType: {RemoteDepositRequestType(transaction_type)}"
)
match RemoteDepositRequestType(transaction_type):
case RemoteDepositRequestType.RDC_TransactionList:
request = RDCTransactionHistoryListRequest.from_hq_request(
xml_payload_request
)
rdc_transactions = await self.get_rdc_transaction_list(request)
response_as_dict["Transaction_Rsp"] = [
x.as_adapter_response(transaction_id) for x in rdc_transactions
]
host_account_req["StatusDescription"] = "Success"
case RemoteDepositRequestType.RDC_TransactionDetail:
request = RDCTransactionHistoryDetailRequest.from_hq_request(
xml_payload_request, transaction_id
)
rdc_transaction = await self.get_rdc_transaction_detail(request)
response_as_dict["Transaction_Rsp"] = [
rdc_transaction.transaction_as_adapter_response()
]
response_as_dict["Image_Rsp"] = (
rdc_transaction.image_as_adapter_response(transaction_id)
)
host_account_req["StatusDescription"] = "Success"
case RemoteDepositRequestType.RDC_Validate:
request = RDCValidateRequest.from_hq_request(xml_payload_request)
rdc_validation = await self.get_validations(request)
validation_json = rdc_validation.as_adapter_response()
response_as_dict["RequestorData"][0]["XmlPayload"] = base64.b64encode(
bytes(request.serialize_account_list_to_xml(), "utf-8")
).decode()
response_as_dict["HostAccount_Req"][0]["RtCtlBinData2"] = (
base64.b64encode(
bytes(json.dumps(validation_json), "utf-8")
).decode()
)
# response_as_dict["HostAccount_Req"][0]["RtCtlBinData2"] = "eyJNdWx0aVN0ZXAiOmZhbHNlLCJVc2VyVmFsaWQiOnRydWUsIkVycm9yTWVzc2FnZSI6bnVsbCwiTGltaXRNZXNzYWdlcyI6WyJDYWxlbmRhciBEYXkgZGVwb3NpdCBhbW91bnQgbGltaXQgJDEwMDAxLjAwLiIsIkNhbGVuZGFyIERheSBkZXBvc2l0IGFtb3VudCBsaW1pdCByZW1haW5pbmc6ICQxMDAwMS4wMC4iLCJDYWxlbmRhciBEYXkgdHJhbnNhY3Rpb24gbGltaXQgJDUuIiwiQ2FsZW5kYXIgRGF5IHRyYW5zYWN0aW9uIGxpbWl0IHJlbWFpbmluZzogJDUuIl0sIkhvc3RBY2NvdW50SWRzIjpudWxsLCJEYWlseUFtdFJlbWFpbiI6bnVsbCwiRGFpbHlDb3VudFJlbWFpbiI6bnVsbCwiV2Vla2x5QW10UmVtYWluIjpudWxsLCJXZWVrbHlDb3VudFJlbWFpbiI6bnVsbCwiTW9udGhseUFtdFJlbWFpbiI6bnVsbCwiTW9udGhseUNvdW50UmVtYWluIjpudWxsLCJEYWlseUFtdCI6bnVsbCwiRGFpbHlDb3VudCI6bnVsbCwiV2Vla2x5QW10IjpudWxsLCJXZWVrbHlDb3VudCI6bnVsbCwiTW9udGhseUFtdCI6bnVsbCwiSXRlbUFtdCI6bnVsbCwiTW9udGhseUNvdW50IjpudWxsLCJDYWxEYWlseUFtdCI6MTAwMDEuMDAsIkNhbERhaWx5QW10UmVtYWluIjoxMDAwMS4wMCwiQ2FsRGFpbHlUcmFuc2FjdGlvbiI6NSwiQ2FsRGFpbHlUcmFuc2FjdGlvblJlbWFpbiI6NSwiTXVsdGlEYXlBbXQiOm51bGx9"
host_account_req["StatusDescription"] = "Success"
case RemoteDepositRequestType.RDC_Capture:
rdc_data = host_account_req["RtCtlBinData1"]
rdc_data_obj = objectify.fromstring(base64.b64decode(rdc_data))
request = RDCCaptureRequest.from_hq_request(
host_account_req, xml_payload_request, rdc_data_obj
)
rdc_transactions = await self.capture_rdc(request)
host_account_req["EndUserMessage"] = rdc_transactions.end_user_message
host_account_req["HostTraceNumber"] = (
rdc_transactions.host_transaction_id
)
if rdc_transactions.check_number:
response_as_dict["Transaction_Rsp"][0] = [
{"CheckNumber": rdc_transactions.check_number}
]
if rdc_transactions.success:
host_account_req["StatusDescription"] = "Success"
else:
response_as_dict["HostAccount_Req"][0]["HostErrorCode"] = 8
host_account_req["StatusDescription"] = "Error"
case _:
self.logger.info(
f"Unknown RDC Request Type Recieved {transaction_type}"
)
host_account_req["HostErrorCode"] = -732 # unknown
dict_to_log = deepcopy(response_as_dict)
if dict_to_log.get("Image_Rsp"):
for image_rsp in dict_to_log["Image_Rsp"]:
image_rsp["ImageData"] = "Redacted"
self.logger.info(dict_to_log)
return response_as_dict
def _setup_session_user(self, element, response_as_dict):
requestor = response_as_dict["RequestorData"][0]
# there is probably a better place for this logic but not sure how specific it is to RDC Adapters
self.online_session = OnlineSession()
self.online_session.aba = response_as_dict["HostAccount_Req"][0]["BankId"]
self.online_session.session_id = element.SessionId
self.online_session.workstation = requestor["TransactionId"]
self.online_session.client_address = requestor["IpAddress"]
self.online_user = OnlineUser()
self.online_user.login_name = requestor["UserData"]
self.online_user.user_id = requestor["UserID"]
self.online_user.customer_id = requestor["CustomerID"]
self.online_user.hq_session_id = element.SessionId
[docs]
async def get_rdc_transaction_list(
self, transaction_list_request: RemoteDepositRequest
) -> list[RDCTransactionResponse]: # pragma: no cover
"""Using the information provided in `transaction_list_request` param, return a list of RDCTransactionResponse objects"""
raise NotImplementedError
[docs]
async def get_rdc_transaction_detail(
self, transaction_detail_request: RDCTransactionHistoryDetailRequest
) -> RDCTransactionHistoryDetailResponse: # pragma: no cover
"""Using the information provided in `transaction_detail_request` param, return a RDCTransactionHistoryDetailResponse object"""
raise NotImplementedError
[docs]
async def capture_rdc(
self, remote_deposit_capture_request: RDCCaptureRequest
) -> RDCCaptureResponse: # pragma: no cover
"""Using the information provided in `remote_deposit_capture_request` param, return a list of RDCSubmitResponse objects"""
raise NotImplementedError
[docs]
async def get_validations(
self, remote_deposit_validation_request: RDCValidateRequest
) -> RDCValidation: # pragma: no cover
"""Using the information provided in `remote_deposit_validation_request` param, return a Validation object"""
raise NotImplementedError