140 lines
5.0 KiB
Python
140 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Data mapper for NEFT SFTP feed.
|
|
Maps parsed NEFT transactions (dicts) to NEFTOutwardRecord for database insertion.
|
|
- No padding of account numbers (kept as-is, trimmed).
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from typing import Dict, Any, List
|
|
|
|
from logging_config import get_logger
|
|
from db.models import NEFTOutwardRecord
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class NEFTDataMapper:
|
|
"""Maps parsed NEFT transactions to NEFTOutwardRecord objects."""
|
|
|
|
# -------------------------
|
|
# Helpers
|
|
# -------------------------
|
|
|
|
@staticmethod
|
|
def convert_date(date_str: str) -> str:
|
|
"""
|
|
Convert NEFT date (YYYYMMDD) to DDMMYYYY.
|
|
|
|
Input : '20260306'
|
|
Output: '06032026'
|
|
|
|
On error, returns today's date in DDMMYYYY format.
|
|
"""
|
|
try:
|
|
if not date_str or len(date_str.strip()) != 8 or not date_str.isdigit():
|
|
raise ValueError(f"Invalid NEFT date format: {date_str!r}")
|
|
dt = datetime.strptime(date_str, "%Y%m%d")
|
|
return dt.strftime("%d%m%Y")
|
|
except Exception as e:
|
|
logger.error(f"Error converting date '{date_str}': {e}")
|
|
return datetime.now().strftime("%d%m%Y")
|
|
|
|
|
|
|
|
|
|
@staticmethod
|
|
def convert_amount(amount_in: Any) -> Decimal:
|
|
"""
|
|
Convert amount to Decimal and return absolute value.
|
|
Use TXNIND to capture the sign semantics.
|
|
"""
|
|
try:
|
|
if isinstance(amount_in, Decimal):
|
|
val = amount_in
|
|
else:
|
|
txt = (str(amount_in) or '').replace(',', '').strip()
|
|
val = Decimal(txt) if txt else Decimal('0')
|
|
return abs(val)
|
|
except Exception as e:
|
|
logger.error(f"Error converting amount '{amount_in}': {e}")
|
|
return Decimal('0')
|
|
|
|
# -------------------------
|
|
# Mapping
|
|
# -------------------------
|
|
|
|
@classmethod
|
|
def map_transaction(cls, parsed_txn: Dict[str, Any], bankcode: str) -> NEFTOutwardRecord:
|
|
"""
|
|
Map a single parsed NEFT transaction (dict) to NEFTOutwardRecord.
|
|
|
|
Args:
|
|
parsed_txn: Dict emitted by SFTPUtrParser
|
|
bankcode : Bank code for this transaction (mapped to NEFTOutwardRecord.bank_code)
|
|
"""
|
|
try:
|
|
# Amount handling
|
|
amount_in = parsed_txn.get('amount', '0')
|
|
txn_amt = cls.convert_amount(amount_in)
|
|
txnind = 'DR'
|
|
sender_to_reciver_info = ' '
|
|
|
|
# Date handling
|
|
txn_date_raw = parsed_txn.get('tran_date', '') or ''
|
|
txn_date_ddmmyyyy = cls.convert_date(txn_date_raw)
|
|
|
|
# Account numbers: NO padding, just trim
|
|
sender_acct = (parsed_txn.get('remitter_acct_no') or '').strip()
|
|
recvr_acct = (parsed_txn.get('benef_acct_no') or '').strip()
|
|
|
|
|
|
recvr_acct_name = (parsed_txn.get('beneficiary_details') or '').strip()
|
|
|
|
record = NEFTOutwardRecord(
|
|
|
|
bank_code=bankcode,
|
|
txnind=txnind,
|
|
jrnl_id=(parsed_txn.get('journal_no') or '').strip(),
|
|
ref_no=(parsed_txn.get('utr') or '').strip(),
|
|
txn_date=txn_date_ddmmyyyy,
|
|
txn_amt=txn_amt,
|
|
sender_ifsc=(parsed_txn.get('ifsc_sender') or '').strip(),
|
|
reciever_ifsc=(parsed_txn.get('ifsc_recvr') or '').strip(),
|
|
sender_acct_no=sender_acct,
|
|
sender_acct_name=(parsed_txn.get('sender_acct_name') or '').strip(),
|
|
recvr_acct_no=recvr_acct,
|
|
recvr_acct_name=recvr_acct_name,
|
|
reject_code=(parsed_txn.get('reject_code') or '').strip(),
|
|
reject_reason=(parsed_txn.get('reject_reason') or '').strip(),
|
|
benef_address=(parsed_txn.get('benef_address') or '').strip(),
|
|
msg_type=(parsed_txn.get('sub_msg_type') or '').strip(),
|
|
sender_to_reciver_info=sender_to_reciver_info,
|
|
)
|
|
|
|
return record
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error mapping NEFT transaction: {e}", exc_info=True)
|
|
raise
|
|
|
|
@classmethod
|
|
def map_transactions(cls, parsed_transactions: List[Dict[str, Any]], bankcode: str) -> List[NEFTOutwardRecord]:
|
|
"""
|
|
Map a list of parsed NEFT transactions to NEFTOutwardRecord objects.
|
|
|
|
Args:
|
|
parsed_transactions: List of dicts from SFTPUtrParser
|
|
bankcode : Bank code to be applied to each record
|
|
"""
|
|
records: List[NEFTOutwardRecord] = []
|
|
for txn in parsed_transactions:
|
|
try:
|
|
rec = cls.map_transaction(txn, bankcode)
|
|
records.append(rec)
|
|
except Exception as e:
|
|
logger.warning(f"Skipping transaction due to error: {e}")
|
|
logger.info(f"Mapped {len(records)} NEFT transactions for bank {bankcode}")
|
|
return records
|