first commit
This commit is contained in:
@@ -0,0 +1,6 @@
|
||||
"""Processors module for ACH file processing."""
|
||||
|
||||
from .data_mapper import UPIDataMapper
|
||||
from .file_processor import FileProcessor
|
||||
|
||||
__all__ = ['UPIDataMapper', 'FileProcessor']
|
||||
@@ -0,0 +1,132 @@
|
||||
#!/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
|
||||
@@ -0,0 +1,166 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Main file processor for end-to-end ACH file processing.
|
||||
Orchestrates download, parsing, mapping, and database insertion.
|
||||
"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import gzip
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from logging_config import get_logger
|
||||
from upi_parser import UPIParser
|
||||
from db.repository import Repository
|
||||
from db.models import ProcessedFile
|
||||
from sftp.sftp_client import SFTPClient
|
||||
from .data_mapper import UPIDataMapper
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class FileProcessor:
|
||||
|
||||
|
||||
def __init__(self, repository: Repository = None, sftp_client: SFTPClient = None):
|
||||
self.repository = repository or Repository()
|
||||
self.sftp_client = sftp_client or SFTPClient()
|
||||
self.temp_dir = tempfile.gettempdir()
|
||||
|
||||
def process_file(
|
||||
self,
|
||||
filename: str,
|
||||
bankcode: str,
|
||||
remote_path: str
|
||||
) -> bool:
|
||||
|
||||
local_path = os.path.join(self.temp_dir, filename)
|
||||
|
||||
try:
|
||||
logger.info(f"Starting processing: {filename} (bank: {bankcode})")
|
||||
|
||||
# Step 1: Check if already processed
|
||||
if self.repository.is_file_processed(filename, bankcode):
|
||||
logger.info(f"File already processed for {bankcode}: {filename}")
|
||||
return True
|
||||
|
||||
# Step 2: Download file
|
||||
if not self.sftp_client.download_file(remote_path, local_path):
|
||||
raise Exception(f"Failed to download file: {remote_path}")
|
||||
|
||||
# NEW: Handle .gz file
|
||||
file_to_parse = local_path
|
||||
|
||||
if filename.lower().endswith(".gz"):
|
||||
logger.info(f"Extracting GZ file: {filename}")
|
||||
|
||||
extracted_path = local_path[:-3] # remove .gz
|
||||
|
||||
with gzip.open(local_path, 'rb') as f_in:
|
||||
with open(extracted_path, 'wb') as f_out:
|
||||
shutil.copyfileobj(f_in, f_out)
|
||||
|
||||
file_to_parse = extracted_path
|
||||
|
||||
# Step 3: Parse file
|
||||
parser = UPIParser(file_to_parse)
|
||||
transactions, metadata, summary = parser.parse()
|
||||
|
||||
if not transactions:
|
||||
logger.warning(f"No transactions found in {filename}")
|
||||
|
||||
processed_file = ProcessedFile(
|
||||
filename=filename, # ORIGINAL .gz NAME
|
||||
bankcode=bankcode,
|
||||
file_path=remote_path,
|
||||
transaction_count=0,
|
||||
status='SUCCESS'
|
||||
)
|
||||
self.repository.mark_file_processed(processed_file)
|
||||
return True
|
||||
|
||||
# Step 4: Map transactions
|
||||
mapped_records = UPIDataMapper.map_transactions(transactions, bankcode)
|
||||
|
||||
# Step 5: Insert to database
|
||||
inserted_count, skipped_count = self.repository.bulk_insert_transactions(mapped_records)
|
||||
|
||||
# Step 6: Mark processed
|
||||
processed_file = ProcessedFile(
|
||||
filename=filename, # ORIGINAL .gz NAME
|
||||
bankcode=bankcode,
|
||||
file_path=remote_path,
|
||||
transaction_count=inserted_count,
|
||||
status='SUCCESS'
|
||||
)
|
||||
self.repository.mark_file_processed(processed_file)
|
||||
|
||||
logger.info(
|
||||
f"Successfully processed {filename}: "
|
||||
f"{inserted_count} inserted, {skipped_count} skipped (non-ipks accounts)"
|
||||
)
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing {filename}: {e}", exc_info=True)
|
||||
|
||||
try:
|
||||
processed_file = ProcessedFile(
|
||||
filename=filename,
|
||||
bankcode=bankcode,
|
||||
file_path=remote_path,
|
||||
transaction_count=0,
|
||||
status='FAILED',
|
||||
error_message=str(e)[:2000]
|
||||
)
|
||||
self.repository.mark_file_processed(processed_file)
|
||||
except Exception as mark_error:
|
||||
logger.error(f"Failed to mark file as failed: {mark_error}")
|
||||
|
||||
return False
|
||||
|
||||
finally:
|
||||
# UPDATED CLEANUP (handles both .gz + extracted file)
|
||||
try:
|
||||
if os.path.exists(local_path):
|
||||
os.remove(local_path)
|
||||
|
||||
if 'file_to_parse' in locals() and file_to_parse != local_path:
|
||||
if os.path.exists(file_to_parse):
|
||||
os.remove(file_to_parse)
|
||||
|
||||
logger.debug(f"Cleanup done for {filename}")
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Error cleaning up local file {local_path}: {e}")
|
||||
|
||||
def process_files(self, files_to_process: list) -> dict:
|
||||
|
||||
stats = {
|
||||
'total': len(files_to_process),
|
||||
'successful': 0,
|
||||
'failed': 0,
|
||||
'files': []
|
||||
}
|
||||
|
||||
for filename, bankcode, remote_path in files_to_process:
|
||||
success = self.process_file(filename, bankcode, remote_path)
|
||||
stats['successful'] += 1 if success else 0
|
||||
stats['failed'] += 0 if success else 1
|
||||
stats['files'].append({
|
||||
'filename': filename,
|
||||
'bankcode': bankcode,
|
||||
'success': success
|
||||
})
|
||||
|
||||
logger.info(f"Processing complete: {stats['successful']}/{stats['total']} successful")
|
||||
return stats
|
||||
|
||||
def __enter__(self):
|
||||
if self.sftp_client and not self.sftp_client.sftp:
|
||||
self.sftp_client.connect()
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
if self.sftp_client:
|
||||
self.sftp_client.disconnect()
|
||||
Reference in New Issue
Block a user