from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from typing import Optional
from pathlib import Path
from q2_sdk.core.exceptions import BadParameterError
from decimal import Decimal
from base64 import b64encode
HERE = Path(__file__).absolute().parent
[docs]
class RemoteDepositRequestType(Enum):
RDC_Validate = 41
RDC_Capture = 42
RDC_TransactionList = 43
RDC_TransactionDetail = 44
[docs]
class RemoteDepositCaptureErrorBitflags(Enum):
FrontImageBad = 0x0001
BackImageBad = 0x0002
AmountBad = 0x0004
CheckNumberBad = 0x0008
AccountNumberBad = 0x0010
[docs]
class RemoteDepositStatus(Enum):
Submitted = 0
Accepted = 1
Rejected = 2
[docs]
@dataclass
class BaseRDCRequest:
accounts: list[RDCAccount]
[docs]
def serialize_account_list_to_xml(self) -> str:
accounts_xml = "".join([account.as_xml_str() for account in self.accounts])
return f"<RequestBase>{accounts_xml}</RequestBase>"
[docs]
@staticmethod
def from_hq_request(inp: dict) -> BaseRDCRequest:
account_requests = inp["HostAccount_Req"]
account_list = []
for account_request in account_requests:
account = RDCAccount(
host_account_id=account_request.attrib.get("HostAccountID", None),
account_number_internal=account_request.attrib.get(
"AccountNumberInternal", None
),
account_number_external=account_request.attrib.get(
"AccountNumberExternal", None
),
cif_internal=account_request.attrib.get("CifInternal", None),
cif_external=account_request.attrib.get("CifExternal", None),
note_cert_number=account_request.attrib.get("NoteCertNumber", None),
host_product_type=account_request.attrib.get("HostProductType", None),
host_product_code=account_request.attrib.get("HostProductCode", None),
branch_id=account_request.attrib.get("BrandId", None),
bank_id=account_request.attrib.get("BankId", None),
transaction_type=account_request.attrib.get("TransactionType", None),
ui_source=account_request.attrib.get("UiSource", None),
time_to_live_in_seconds=account_request.attrib.get(
"TimeToLiveInSeconds", None
),
host_error_code=account_request.attrib.get("HostErrorCode", None),
host_user=account_request.attrib.get("HostUser", None),
host_password=account_request.attrib.get("HostPassword", None),
process_state=account_request.attrib.get("ProcessState", None),
is_external_account=account_request.attrib.get(
"IsExternalAccount", False
),
routing_number=account_request.attrib.get("RoutingNumber", None),
rt_ctrl_event=account_request.attrib.get("RtCtrlEvent", None),
skip_balance_check=account_request.attrib.get(
"SkipBalanceCheck", False
),
hydra_product_code=account_request.attrib.get("HydraProductCode", None),
hydra_product_type_code=account_request.attrib.get(
"HydraProductTypeCode", None
),
process_date=account_request.attrib.get("ProcessDate", None),
host_trace_number=account_request.attrib.get("HostTraceNumber", None),
)
account_list.append(account)
return BaseRDCRequest(accounts=account_list)
[docs]
@dataclass
class RDCAccount:
account_number_internal: str
account_number_external: str
cif_internal: str
cif_external: str
note_cert_number: str
host_product_type: str
host_product_code: str
branch_id: int
bank_id: str
transaction_type: int
ui_source: int
time_to_live_in_seconds: int
host_error_code: int
host_user: str
host_password: str
process_state: RemoteDepositCaptureErrorBitflags
is_external_account: bool
routing_number: str
rt_ctrl_event: str
skip_balance_check: bool
hydra_product_code: str
hydra_product_type_code: str
process_date: str
host_account_id: str
host_trace_number: str
is_valid: bool = True
[docs]
def as_xml_str(self) -> str:
xml_str = f"""
<HostAccount_Req
TransactionID="1"
RDCValid="{1 if self.is_valid else 0}"
AccountNumberInternal="{self.account_number_internal}"
AccountNumberExternal="{self.account_number_external}"
CifInternal="{self.cif_internal}"
CifExternal="{self.cif_external}"
Note_Cert_Number="{self.note_cert_number}"
HostProductType="{self.host_product_type}"
HostProductCode="{self.host_product_code}"
BranchId="{self.branch_id}"
BankId="{self.bank_id}"
TransactionType="{self.transaction_type}"
UiSource="{self.ui_source}"
TimeToLiveInSeconds="{self.time_to_live_in_seconds}"
HostErrorCode="{self.host_error_code}"
ProcessState="{self.process_state}"
IsExternalAccount="{self.is_external_account}"
RoutingNumber="{self.routing_number}"
RtCtrlEvent="{self.rt_ctrl_event}"
SkipBalanceCheck="{self.skip_balance_check}"
HydraProductCode="{self.hydra_product_code}"
HydraProductTypeCode="{self.hydra_product_type_code}"
ProcessDate="{self.process_date}"
HostAccountID="{self.host_account_id}"
/>"""
return " ".join(xml_str.replace("\n", "").split())
[docs]
@dataclass
class CheckImage:
raw: bytes
image_type: ImageType
[docs]
def as_base64(self) -> str:
return b64encode(self.raw).decode()
[docs]
def as_adapter_response(self, transaction_id) -> dict[str, str]:
return {
"ImageData": self.as_base64(),
"ImageType": self.image_type.value,
"TransactionID": self.transaction_id,
}
[docs]
@dataclass
class RDCCaptureRequest(BaseRDCRequest):
selected_account: RDCAccount
amount: str
check_number: int
front_image: CheckImage
back_image: CheckImage
@staticmethod
def from_hq_request(
host_account_req: dict, inp: dict, rdc_data
) -> RDCCaptureRequest:
parsed_request = BaseRDCRequest.from_hq_request(inp)
selected_account = RDCAccount(
account_number_internal=host_account_req["AccountNumberInternal"],
account_number_external=host_account_req["AccountNumberExternal"],
cif_internal=host_account_req["CifInternal"],
cif_external=host_account_req["CifExternal"],
note_cert_number=host_account_req["Note_Cert_Number"],
host_error_code=host_account_req["HostErrorCode"],
process_state=host_account_req["ProcessState"],
rt_ctrl_event=host_account_req["RtCtrlEvent"],
process_date=host_account_req["ProcessDate"],
host_trace_number=host_account_req["HostTraceNumber"],
host_product_type=host_account_req["HostProductType"],
host_product_code=host_account_req["HostProductCode"],
branch_id=host_account_req["BranchId"],
bank_id=host_account_req["BankId"],
transaction_type=host_account_req["TransactionType"],
ui_source=host_account_req["UiSource"],
time_to_live_in_seconds=host_account_req["TimeToLiveInSeconds"],
host_user=host_account_req["HostUser"],
host_password=host_account_req["HostPassword"],
is_external_account=host_account_req["IsExternalAccount"],
routing_number=host_account_req["RoutingNumber"],
skip_balance_check=host_account_req["SkipBalanceCheck"],
hydra_product_code=host_account_req["HydraProductCode"],
hydra_product_type_code=host_account_req["HydraProductTypeCode"],
host_account_id=host_account_req["HostAccountID"],
)
response = RDCCaptureRequest(
accounts=parsed_request.accounts,
selected_account=selected_account,
amount=rdc_data["Amount"],
check_number=rdc_data["CheckNumber"],
front_image=CheckImage(rdc_data["FrontImage"], 3),
back_image=CheckImage(rdc_data["BackImage"], 3),
)
return response
[docs]
@dataclass
class RDCCaptureResponse:
host_transaction_id: str
check_number: Optional[int] = None
end_user_message: Optional[str] = ""
success: Optional[bool] = True
process_state: Optional[RemoteDepositCaptureErrorBitflags] = None
[docs]
@dataclass
class RDCTransactionHistoryListRequest(BaseRDCRequest):
pass
[docs]
@dataclass
class RDCValidateRequest(BaseRDCRequest):
pass
[docs]
@dataclass
class RDCTransactionHistoryDetailRequest(BaseRDCRequest):
transaction_id: int
@staticmethod
def from_hq_request(
inp: dict, transaction_id: int
) -> RDCTransactionHistoryDetailRequest:
response = BaseRDCRequest.from_hq_request(inp)
response.transaction_id = transaction_id
return response
[docs]
@dataclass
class RDCTransactionHistoryDetailResponse:
transaction_id: int
host_ref_number: str
transaction_date: str
transaction_amount: str
transaction_status: RemoteDepositStatus
transaction_description: str
account_number: str
front_image: CheckImage
back_image: CheckImage
check_number: Optional[str] = None
def transaction_as_adapter_response(self) -> dict:
return {
"TransactionID": self.transaction_id,
"TransactionDate": self.transaction_date,
"TxnAmount": self.transaction_amount,
"TxnDescription": self.transaction_description,
"ImageNumber": self.transaction_status.value,
"HostRefNumber": self.host_ref_number,
"AccountNumber": self.account_number,
"CheckNumber": self.check_number,
}
def image_as_adapter_response(self, transaction_id) -> list[dict, dict]:
return [
{
"ImageData": self.front_image.as_base64(),
"ImageType": self.front_image.image_type.value,
"TransactionID": transaction_id,
},
{
"ImageData": self.back_image.as_base64(),
"ImageType": self.back_image.image_type.value,
"TransactionID": transaction_id,
},
]
[docs]
@dataclass
class RDCTransactionResponse:
host_transaction_id: str
transaction_description: str
transaction_date: str
transaction_amount: str
transaction_status: RemoteDepositStatus
account_number: str
def as_adapter_response(self, transaction_id) -> dict:
return {
"TransactionID": transaction_id,
"TransactionDate": self.transaction_date,
"TxnAmount": self.transaction_amount,
"TxnDescription": self.transaction_description,
"ImageNumber": self.transaction_status.value,
"HostRefNumber": self.host_transaction_id,
"AccountNumber": self.account_number,
}
[docs]
@dataclass
class RemoteDepositRequest:
internal_account_number: str
transaction_id: int
transaction_type: RemoteDepositRequestType
external_account_number: Optional[str] = None
cif_internal: Optional[str] = None
cif_external: Optional[str] = None
host_product_type: Optional[str] = None
host_product_code: Optional[str] = None
branch_id: Optional[str] = None
bank_id: Optional[str] = None
ui_source: Optional[int] = None
host_trace_number: Optional[str] = None
process_state: Optional[int] = None # bitflag
is_external_account: Optional[bool] = None
routing_number: Optional[str] = None
realtime_control_binary_data_1: Optional[bytes] = None
realtime_control_binary_data_2: Optional[bytes] = None
hydra_product_code: Optional[str] = None
hydra_product_type_code: Optional[str] = None
host_account_id: Optional[int] = None
@staticmethod
def from_xml_payload_request(inp: dict) -> RemoteDepositRequest:
host_account = inp["HostAccount_Req"][0]
internal_account_number = host_account["AccountNumberInternal"]
external_account_number = host_account["AccountNumberExternal"]
cif_internal = host_account["CifInternal"]
cif_external = host_account["CifExternal"]
transaction_type = host_account["TransactionType"]
transaction_id = host_account["TransactionID"]
host_product_type = host_account["HostProductType"]
host_product_code = host_account["HostProductCode"]
branch_id = host_account["BranchId"]
bank_id = host_account["BankId"]
ui_source = host_account["UiSource"]
host_trace_number = host_account["HostTraceNumber"]
process_state = host_account["ProcessState"]
is_external_account = host_account["IsExternalAccount"]
routing_number = host_account["RoutingNumber"]
realtime_control_binary_data_1 = host_account["RtCtlBinData1"]
realtime_control_binary_data_2 = host_account["RtCtlBinData2"]
hydra_product_code = host_account["HydraProductCode"]
hydra_product_type_code = host_account["HydraProductTypeCode"]
host_account_id = host_account["HostAccountID"]
if not transaction_id:
raise BadParameterError("TransactionID must be provided")
return RemoteDepositRequest(
internal_account_number,
transaction_id,
RemoteDepositRequestType(transaction_type),
external_account_number,
cif_internal,
cif_external,
host_product_type,
host_product_code,
branch_id,
bank_id,
ui_source,
host_trace_number,
process_state,
is_external_account,
routing_number,
realtime_control_binary_data_1,
realtime_control_binary_data_2,
hydra_product_code,
hydra_product_type_code,
host_account_id,
)
[docs]
@dataclass
class RDCValidation:
multi_step: Optional[bool] = False
user_valid: Optional[bool] = True
error_message: Optional[str] = None
limit_messages: Optional[list[str]] = None
host_account_ids: Optional[list[int]] = None
daily_amount: Optional[Decimal] = None
daily_count: Optional[int] = None
daily_amount_remaining: Optional[Decimal] = None
daily_count_remaining: Optional[int] = None
weekly_amount: Optional[Decimal] = None
weekly_count: Optional[int] = None
weekly_amount_remaining: Optional[Decimal] = None
weekly_count_remaining: Optional[int] = None
monthly_amount: Optional[Decimal] = None
monthly_count: Optional[int] = None
monthly_amount_remaining: Optional[Decimal] = None
monthly_count_remaining: Optional[int] = None
item_amount: Optional[Decimal] = None
calendar_day_amount: Optional[Decimal] = None
calendar_day_amount_remaining: Optional[Decimal] = None
calendar_day_count: Optional[int] = None
calendar_day_count_remaining: Optional[int] = None
multi_day_amount: Optional[Decimal] = None
def as_adapter_response(self) -> dict[str, str]:
ret_dict = {
"MultiStep": self.multi_step,
"UserValid": self.user_valid,
"ErrorMessage": self.error_message,
"LimitMessages": self.limit_messages,
"HostAccountIds": self.host_account_ids,
"DailyAmt": self.daily_amount,
"DailyCount": self.daily_count,
"DailyAmtRemain": self.daily_amount_remaining,
"DailyCountRemain": self.daily_count_remaining,
"WeeklyAmt": self.weekly_amount,
"WeeklyCount": self.weekly_count,
"WeeklyAmtRemain": self.weekly_amount_remaining,
"WeeklyCountRemain": self.weekly_count_remaining,
"MonthlyAmt": self.monthly_amount,
"MonthlyCount": self.monthly_count,
"MonthlyAmtRemain": self.monthly_amount_remaining,
"MonthlyCountRemain": self.monthly_count_remaining,
"ItemAmt": self.item_amount,
"CalDailyAmt": self.calendar_day_amount,
"CalDailyAmtRemain": self.calendar_day_amount_remaining,
"CalDailyTransaction": self.calendar_day_count,
"CalDailyTransactionRemain": self.calendar_day_count_remaining,
"MultiDayAmt": self.multi_day_amount,
}
return ret_dict
[docs]
@dataclass
class RDCImage:
raw: bytes
image_type: ImageType
def as_base64(self) -> str:
return b64encode(self.raw).decode()
def as_adapter_response(self, transaction_id) -> dict[str, str]:
return {
"ImageData": self.as_base64(),
"ImageType": self.image_type.value,
"TransactionID": transaction_id,
}
[docs]
class MockImage:
[docs]
@staticmethod
def get_front() -> bytes:
return Path.read_bytes(HERE / "check_front.png")
[docs]
@staticmethod
def get_back() -> bytes:
return Path.read_bytes(HERE / "check_back.png")
[docs]
class ImageType(Enum):
GIF = 0
JPG = 1
BMP = 2
PNG = 3
PDF = 4
TIFF = 5