132 lines
3.7 KiB
Python
132 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Data mapper for UPI Switch feed.
|
|
Maps parsed UPI transactions (dicts) to UPIInwardRecord for database insertion.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from typing import Dict, Any, List
|
|
|
|
from logging_config import get_logger
|
|
from db.models import UPIInwardRecord
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class UPIDataMapper:
|
|
"""Maps parsed UPI transactions to UPIInwardRecord objects."""
|
|
|
|
# -------------------------
|
|
# Helpers
|
|
# -------------------------
|
|
|
|
@staticmethod
|
|
def convert_date(date_str: str) -> str:
|
|
"""
|
|
Convert date from YYYY-MM-DD → DDMMYYYY
|
|
"""
|
|
try:
|
|
if not date_str:
|
|
return datetime.now().strftime("%d%m%Y")
|
|
|
|
dt = datetime.strptime(date_str.strip(), "%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 safely to Decimal
|
|
"""
|
|
try:
|
|
if isinstance(amount_in, Decimal):
|
|
return amount_in
|
|
|
|
txt = (str(amount_in) or "").replace(",", "").strip()
|
|
return Decimal(txt) if txt else Decimal("0")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error converting amount '{amount_in}': {e}")
|
|
return Decimal("0")
|
|
|
|
@staticmethod
|
|
def normalize_status(status: str) -> str:
|
|
"""
|
|
Normalize status values
|
|
"""
|
|
try:
|
|
if not status:
|
|
return ""
|
|
|
|
s = status.strip()
|
|
|
|
if s.upper() == "SUCCESS":
|
|
return "SUCCESS"
|
|
elif s.upper() == "FAILURE":
|
|
return "FAILED"
|
|
else:
|
|
return s
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error normalizing status '{status}': {e}")
|
|
return status
|
|
|
|
# -------------------------
|
|
# Mapping
|
|
# -------------------------
|
|
|
|
@classmethod
|
|
def map_transaction(
|
|
cls,
|
|
parsed_txn: Dict[str, Any],
|
|
bankcode: str
|
|
) -> UPIInwardRecord:
|
|
"""
|
|
Map single parsed UPI txn → UPIInwardRecord
|
|
"""
|
|
try:
|
|
txn_amt = cls.convert_amount(parsed_txn.get("amount"))
|
|
txn_date_raw = parsed_txn.get("txn_date", "")
|
|
txn_date = cls.convert_date(txn_date_raw)
|
|
|
|
record = UPIInwardRecord(
|
|
rrn=(parsed_txn.get("rrn") or "").strip(),
|
|
txn_date=txn_date,
|
|
txn_amt=txn_amt,
|
|
credit_account=(parsed_txn.get("credit_account") or "").strip(),
|
|
status=cls.normalize_status(parsed_txn.get("status")),
|
|
note=(parsed_txn.get("note") or "").strip(),
|
|
bank_code=bankcode,
|
|
txn_type=(parsed_txn.get("type") or "").strip()
|
|
)
|
|
|
|
return record
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error mapping UPI transaction: {e}", exc_info=True)
|
|
raise
|
|
|
|
@classmethod
|
|
def map_transactions(
|
|
cls,
|
|
parsed_transactions: List[Dict[str, Any]],
|
|
bankcode: str
|
|
) -> List[UPIInwardRecord]:
|
|
"""
|
|
Map list of parsed UPI transactions
|
|
"""
|
|
records: List[UPIInwardRecord] = []
|
|
|
|
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)} UPI transactions for bank {bankcode}")
|
|
return records |