commit 6ec74f1c6457837d469dfc144f13696c21672539 Author: Bishwajeet Kumar Rajak Date: Sat Jun 13 14:58:21 2026 +0530 first commit diff --git a/.claude/settings.local.json b/.claude/settings.local.json new file mode 100644 index 0000000..54559c9 --- /dev/null +++ b/.claude/settings.local.json @@ -0,0 +1,10 @@ +{ + "permissions": { + "allow": [ + "Bash(python3 -m venv:*)", + "Bash(source venv/bin/activate)", + "Bash(python:*)", + "Bash(pip install:*)" + ] + } +} diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..ed5239f --- /dev/null +++ b/.env.example @@ -0,0 +1,23 @@ +# Database Configuration +DB_USER=pacs_db +DB_PASSWORD=pacs_db +DB_HOST=testipksdb.c7q7defafeea.ap-south-1.rds.amazonaws.com +DB_PORT=1521 +DB_SERVICE_NAME=IPKSDB +DB_POOL_MIN=2 +DB_POOL_MAX=10 + +# SFTP Configuration +SFTP_HOST=192.168.1.100 +SFTP_PORT=22 +SFTP_USERNAME=ipks +SFTP_PASSWORD=secure_password +SFTP_BASE_PATH=/home/ipks/IPKS_FILES/REPORTS + +# Processing Configuration +POLL_INTERVAL_MINUTES=30 +BATCH_SIZE=100 +BANK_CODES=HDFC,ICICI,SBI,AXIS,PNB + +# Logging Configuration +LOG_LEVEL=INFO diff --git a/__pycache__/config.cpython-39.pyc b/__pycache__/config.cpython-39.pyc new file mode 100644 index 0000000..b8cc4c2 Binary files /dev/null and b/__pycache__/config.cpython-39.pyc differ diff --git a/__pycache__/logging_config.cpython-39.pyc b/__pycache__/logging_config.cpython-39.pyc new file mode 100644 index 0000000..1baf02d Binary files /dev/null and b/__pycache__/logging_config.cpython-39.pyc differ diff --git a/__pycache__/neft_inward_parser.cpython-39.pyc b/__pycache__/neft_inward_parser.cpython-39.pyc new file mode 100644 index 0000000..0843f1f Binary files /dev/null and b/__pycache__/neft_inward_parser.cpython-39.pyc differ diff --git a/__pycache__/scheduler.cpython-39.pyc b/__pycache__/scheduler.cpython-39.pyc new file mode 100644 index 0000000..e610bc5 Binary files /dev/null and b/__pycache__/scheduler.cpython-39.pyc differ diff --git a/config.py b/config.py new file mode 100644 index 0000000..de3f895 --- /dev/null +++ b/config.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +""" +Configuration management for ACH file processing pipeline. +Loads and validates environment variables. +""" + +import os +from pathlib import Path +from logging_config import get_logger + +logger = get_logger(__name__) + + +class Config: + """Application configuration from environment variables.""" + + def __init__(self): + """Initialize configuration from environment.""" + self._validate_env_file() + self._load_database_config() + self._load_sftp_config() + self._load_processing_config() + + def _validate_env_file(self): + """Check if .env file exists.""" + if not Path('.env').exists(): + logger.warning(".env file not found. Using environment variables or defaults.") + + def _load_database_config(self): + """Load database configuration.""" + self.db_user = os.getenv('DB_USER', 'pacs_db') + self.db_password = os.getenv('DB_PASSWORD', 'pacs_db') + self.db_host = os.getenv('DB_HOST', 'testipksdb.c7q7defafeea.ap-south-1.rds.amazonaws.com') + self.db_port = int(os.getenv('DB_PORT', '1521')) + self.db_service_name = os.getenv('DB_SERVICE_NAME', 'IPKSDB') + self.db_pool_min = int(os.getenv('DB_POOL_MIN', '2')) + self.db_pool_max = int(os.getenv('DB_POOL_MAX', '10')) + + def _load_sftp_config(self): + """Load SFTP configuration.""" + self.sftp_host = os.getenv('SFTP_HOST', '43.225.3.224') + self.sftp_port = int(os.getenv('SFTP_PORT', '4650')) + self.sftp_username = os.getenv('SFTP_USERNAME', 'ipkssftp') + self.sftp_password = os.getenv('SFTP_PASSWORD', 'CqBx@j6G9vqNW#1') + self.sftp_base_path = os.getenv('SFTP_BASE_PATH', '/home/ipks/IPKS_FILES/REPORTS') + + def _load_processing_config(self): + """Load processing configuration.""" + self.poll_interval_minutes = int(os.getenv('POLL_INTERVAL_MINUTES', '30')) + self.batch_size = int(os.getenv('BATCH_SIZE', '100')) + self.bank_codes = self._parse_bank_codes() + self.log_level = os.getenv('LOG_LEVEL', 'INFO') + + def _parse_bank_codes(self): + """Parse bank codes from comma-separated environment variable.""" + codes_str = os.getenv('BANK_CODES', '0001,0002,0003,0004,0005,0006,0007,0009,0012,0013,0014,0015,0016,0017,0018,0020,0021') + return [code.strip() for code in codes_str.split(',') if code.strip()] + + def get_db_connection_string(self): + """Generate Oracle connection string.""" + return f"{self.db_user}/{self.db_password}@{self.db_host}:{self.db_port}/{self.db_service_name}" + + def validate(self): + """Validate critical configuration.""" + if not self.db_user or not self.db_password: + raise ValueError("Database credentials not configured") + if not self.sftp_username: + logger.warning("SFTP username not configured") + if not self.bank_codes: + raise ValueError("No bank codes configured") + logger.info(f"Configuration validated. Bank codes: {', '.join(self.bank_codes)}") + + +# Global config instance +config = None + + +def get_config(): + """Get or create global config instance.""" + global config + if config is None: + config = Config() + return config + + +if __name__ == '__main__': + cfg = get_config() + cfg.validate() + print(f"Bank Codes: {cfg.bank_codes}") + print(f"SFTP Host: {cfg.sftp_host}:{cfg.sftp_port}") + print(f"Database: {cfg.db_host}:{cfg.db_port}/{cfg.db_service_name}") + print(f"Poll Interval: {cfg.poll_interval_minutes} minutes") diff --git a/db/__init__.py b/db/__init__.py new file mode 100644 index 0000000..701b306 --- /dev/null +++ b/db/__init__.py @@ -0,0 +1,6 @@ +"""Database module for ACH file processing.""" + +from .oracle_connector import OracleConnector +from .repository import Repository + +__all__ = ['OracleConnector', 'Repository'] diff --git a/db/models.py b/db/models.py new file mode 100644 index 0000000..475fe21 --- /dev/null +++ b/db/models.py @@ -0,0 +1,50 @@ +from dataclasses import dataclass +from datetime import datetime +from decimal import Decimal +from typing import Optional + + +@dataclass +class UPIInwardRecord: + rrn: str + txn_date: str + txn_amt: Optional[Decimal] + credit_account: str + status: str + note: str + bank_code: str + txn_type: str # will be CREDIT / DEBIT from parser + + def to_dict(self): + return { + "BANKCODE": self.bank_code, + "RRN": self.rrn, + "TXN_DATE": self.txn_date, + "TXN_AMT": self.txn_amt, + "TXN_TYPE": self.txn_type, # converted to CR before insert + "BENEFICIARY_ACCOUNT": self.credit_account, + "STATUS": self.status, + "NOTE": self.note, + } + + +@dataclass +class ProcessedFile: + filename: str + bankcode: str + file_path: str + transaction_count: int + status: str = 'SUCCESS' + error_message: Optional[str] = None + processed_at: Optional[datetime] = None + + def to_dict(self): + return { + 'filename': self.filename, + 'bankcode': self.bankcode, + 'file_path': self.file_path, + 'transaction_count': self.transaction_count, + 'status': self.status, + 'error_message': self.error_message, + 'processed_at': self.processed_at or datetime.now(), + } \ No newline at end of file diff --git a/db/oracle_connector.py b/db/oracle_connector.py new file mode 100644 index 0000000..15ab91b --- /dev/null +++ b/db/oracle_connector.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +""" +Oracle database connection pool manager using oracledb. +Manages connections with pooling and health checks. + +oracledb is the modern, simpler replacement for cx_Oracle. +No Oracle Instant Client required - uses Thick or Thin mode. +""" + +import oracledb +from logging_config import get_logger +from config import get_config + +logger = get_logger(__name__) + + +class OracleConnector: + """Manages Oracle database connections with pooling.""" + + def __init__(self): + """Initialize connection pool.""" + self.pool = None + self.config = get_config() + + def initialize_pool(self): + """Create connection pool.""" + try: + # Build connection string for oracledb + # Format: user/password@host:port/service_name + connection_string = ( + f"{self.config.db_user}/{self.config.db_password}@" + f"{self.config.db_host}:{self.config.db_port}/{self.config.db_service_name}" + ) + + # Create connection pool using oracledb API + # Note: oracledb uses 'min' and 'max' for pool sizing + self.pool = oracledb.create_pool( + dsn=connection_string, + min=self.config.db_pool_min, + max=self.config.db_pool_max, + increment=1, + ) + + logger.debug(f"Oracle connection pool initialized: min={self.config.db_pool_min}, max={self.config.db_pool_max}") + return True + except oracledb.DatabaseError as e: + logger.error(f"Failed to initialize connection pool: {e}", exc_info=True) + return False + except Exception as e: + logger.error(f"Unexpected error initializing pool: {e}", exc_info=True) + return False + + def get_connection(self): + """Get connection from pool.""" + if not self.pool: + self.initialize_pool() + + try: + conn = self.pool.acquire() + logger.debug("Connection acquired from pool") + return conn + except oracledb.DatabaseError as e: + logger.error(f"Failed to acquire connection: {e}", exc_info=True) + raise + except Exception as e: + logger.error(f"Unexpected error acquiring connection: {e}", exc_info=True) + raise + + def close_pool(self): + """Close connection pool.""" + if self.pool: + try: + self.pool.close() + logger.info("Connection pool closed") + except Exception as e: + logger.error(f"Error closing pool: {e}") + + def test_connection(self): + """Test database connectivity.""" + try: + conn = self.get_connection() + cursor = conn.cursor() + cursor.execute("SELECT 1 FROM dual") + result = cursor.fetchone() + cursor.close() + conn.close() + logger.debug("Database connection test successful") + return True + except Exception as e: + logger.error(f"Database connection test failed: {e}") + return False + + def __enter__(self): + """Context manager entry.""" + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Context manager exit.""" + self.close_pool() + + +# Global connector instance +_connector = None + + +def get_connector(): + """Get or create global connector instance.""" + global _connector + if _connector is None: + _connector = OracleConnector() + return _connector diff --git a/db/repository.py b/db/repository.py new file mode 100644 index 0000000..cb0a48a --- /dev/null +++ b/db/repository.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +""" +Data access layer for UPI inward file processing. +Handles CRUD operations and transaction management. +""" + +from typing import List, Optional +from logging_config import get_logger +from .oracle_connector import get_connector +from .models import UPIInwardRecord, ProcessedFile + +logger = get_logger(__name__) + + +class Repository: + """Data access layer for UPI inward processing.""" + + def __init__(self): + self.connector = get_connector() + + # --------------------------------------------------------- + # Account validation (same as NEFT) + # --------------------------------------------------------- + def validate_account_exists(self, account_number: str) -> bool: + + if not account_number: + return False + + last12 = str(account_number)[-12:] + + conn = self.connector.get_connection() + try: + cursor = conn.cursor() + cursor.execute( + "SELECT COUNT(*) FROM dep_account WHERE link_accno = :accno", + {'accno': last12} + ) + count = cursor.fetchone()[0] + return count > 0 + + except Exception as e: + logger.warning(f"Error validating account {account_number}: {e}") + return False + + finally: + cursor.close() + conn.close() + + # --------------------------------------------------------- + # UPDATED: UPI BULK INSERT WITH VALIDATION + # --------------------------------------------------------- + def bulk_insert_transactions(self, transactions: List[UPIInwardRecord]) -> tuple: + + if not transactions: + logger.warning("No transactions to insert") + return 0, 0 + + valid_transactions = [] + skipped_count = 0 + + for txn in transactions: + acct = txn.credit_account + + # Condition 1: account exists + account_valid = self.validate_account_exists(acct) + + # Condition 2: only CREDIT transactions + is_credit = (txn.txn_type == "CREDIT") + + if account_valid and is_credit: + # Force DB value to CR + txn.txn_type = "CR" + valid_transactions.append(txn) + else: + skipped_count += 1 + + if not valid_transactions: + logger.info(f"All {skipped_count} transactions skipped") + return 0, skipped_count + + conn = self.connector.get_connection() + cursor = None + + try: + cursor = conn.cursor() + + batch_data = [txn.to_dict() for txn in valid_transactions] + logger.info(batch_data) + + insert_sql = """ + INSERT INTO pacs_db.upi_inward_api_log ( + BANKCODE, + RRN, + TXN_DATE, + TXN_AMT, + TXN_TYPE, + CREDIT_ACCOUNT, + STATUS, + NOTE + ) VALUES ( + :BANKCODE, + :RRN, + :TXN_DATE, + :TXN_AMT, + :TXN_TYPE, + :CREDIT_ACCOUNT, + :STATUS, + :NOTE + ) + """ + + cursor.executemany(insert_sql, batch_data) + logger.info(f"executemany rowcount: {cursor.rowcount}") + + conn.commit() + logger.info("UPI transactions committed successfully") + + inserted_count = len(valid_transactions) + + logger.info(f"Inserted {inserted_count} UPI transactions") + return inserted_count, skipped_count + + except Exception as e: + if conn: + conn.rollback() + logger.error(f"Error inserting UPI transactions: {e}", exc_info=True) + raise + + finally: + if cursor: + cursor.close() + conn.close() + + # --------------------------------------------------------- + # FILE TRACKING (UNCHANGED) + # --------------------------------------------------------- + def is_file_processed(self, filename: str, bankcode: str) -> bool: + + conn = self.connector.get_connection() + cursor = None + + try: + cursor = conn.cursor() + cursor.execute( + """ + SELECT COUNT(*) + FROM neft_processed_files + WHERE filename = :filename + AND bankcode = :bankcode + """, + {'filename': filename, 'bankcode': bankcode} + ) + count = cursor.fetchone()[0] + return count > 0 + + except Exception as e: + logger.error(f"Error checking processed file: {e}", exc_info=True) + return False + + finally: + if cursor: + cursor.close() + conn.close() + + def mark_file_processed(self, processed_file: ProcessedFile) -> bool: + + conn = self.connector.get_connection() + cursor = None + + try: + cursor = conn.cursor() + + file_data = processed_file.to_dict() + + insert_sql = """ + INSERT INTO neft_processed_files ( + filename, bankcode, file_path, transaction_count, + status, error_message, processed_at + ) VALUES ( + :filename, :bankcode, :file_path, :transaction_count, + :status, :error_message, :processed_at + ) + """ + + cursor.execute(insert_sql, file_data) + conn.commit() + + logger.info(f"Marked file as processed: {processed_file.filename}") + return True + + except Exception as e: + if conn: + conn.rollback() + logger.error(f"Error marking file as processed: {e}", exc_info=True) + return False + + finally: + if cursor: + cursor.close() + conn.close() diff --git a/logging_config.py b/logging_config.py new file mode 100644 index 0000000..1b22df3 --- /dev/null +++ b/logging_config.py @@ -0,0 +1,51 @@ +import logging +import logging.handlers +import os +from pathlib import Path + +def setup_logging(log_level=logging.INFO, log_dir="logs"): + """ + Configure logging with both console and file handlers. + + Args: + log_level: logging level (default: logging.INFO) + log_dir: directory to store log files + """ + # Create logs directory if it doesn't exist + Path(log_dir).mkdir(exist_ok=True) + + # Get root logger + logger = logging.getLogger() + logger.setLevel(log_level) + + # Clear existing handlers + logger.handlers.clear() + + # Create formatter + formatter = logging.Formatter( + fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + datefmt='%Y-%m-%d %H:%M:%S' + ) + + # Console handler + console_handler = logging.StreamHandler() + console_handler.setLevel(log_level) + console_handler.setFormatter(formatter) + logger.addHandler(console_handler) + + # File handler (rotating) + log_file = os.path.join(log_dir, 'app.log') + file_handler = logging.handlers.RotatingFileHandler( + log_file, + maxBytes=10 * 1024 * 1024, # 10MB + backupCount=5 + ) + file_handler.setLevel(log_level) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + return logger + +def get_logger(name): + """Get a logger instance for a specific module.""" + return logging.getLogger(name) diff --git a/processors/__init__.py b/processors/__init__.py new file mode 100644 index 0000000..4d08ef9 --- /dev/null +++ b/processors/__init__.py @@ -0,0 +1,6 @@ +"""Processors module for ACH file processing.""" + +from .data_mapper import UPIDataMapper +from .file_processor import FileProcessor + +__all__ = ['UPIDataMapper', 'FileProcessor'] diff --git a/processors/data_mapper.py b/processors/data_mapper.py new file mode 100644 index 0000000..26b04c8 --- /dev/null +++ b/processors/data_mapper.py @@ -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 \ No newline at end of file diff --git a/processors/file_processor.py b/processors/file_processor.py new file mode 100644 index 0000000..65bde88 --- /dev/null +++ b/processors/file_processor.py @@ -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() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f55780d --- /dev/null +++ b/requirements.txt @@ -0,0 +1,23 @@ +# Core dependencies +python-dotenv==1.0.0 + +# Database (modern Oracle driver - simpler than cx_Oracle) +oracledb==2.0.0 + +# SFTP +paramiko==3.4.0 +cryptography==41.0.7 + +# Scheduling +schedule==1.2.0 + +# Configuration +python-decouple==3.8 + +# Timezone support +pytz==2023.3 + +# Development dependencies +pytest==7.4.0 +black==23.7.0 +flake8==6.0.0 diff --git a/scheduler.py b/scheduler.py new file mode 100644 index 0000000..5544d72 --- /dev/null +++ b/scheduler.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +""" +NEFT file processing scheduler. +Runs polling loop every 30 minutes to process new files. +""" + +import signal +import time +import sys +from datetime import datetime +from logging_config import get_logger, setup_logging +from config import get_config +from db import OracleConnector, Repository +from sftp import SFTPClient, FileMonitor +from processors import FileProcessor + +logger = get_logger(__name__) + + +class Scheduler: + """Main scheduler for NEFT file processing.""" + + def __init__(self): + """Initialize scheduler.""" + self.config = get_config() + self.config.validate() + self.running = True + self.cycle_count = 0 + + # Setup signal handlers for graceful shutdown + signal.signal(signal.SIGTERM, self._signal_handler) + signal.signal(signal.SIGINT, self._signal_handler) + + def _signal_handler(self, signum, frame): + """Handle shutdown signals gracefully.""" + logger.info(f"Received signal {signum}, shutting down gracefully...") + self.running = False + + def initialize_database(self): + """Initialize database connection and verify tables exist.""" + try: + connector = OracleConnector() + if connector.test_connection(): + logger.info("Database connection test passed") + repository = Repository() + # repository.verify_tables_exist() + return True + else: + logger.error("Database connection test failed") + return False + except SystemExit as e: + logger.error(f"Database initialization failed: {e}") + raise + except Exception as e: + logger.error(f"Error initializing database: {e}", exc_info=True) + return False + + def run_processing_cycle(self): + """Run single file processing cycle.""" + self.cycle_count += 1 + logger.info(f"=== Starting processing cycle {self.cycle_count} ===") + + sftp_client = SFTPClient() + repository = Repository() + + try: + # Connect to SFTP + if not sftp_client.connect(): + logger.error("Failed to connect to SFTP server") + return + + # Scan for new files across all banks + monitor = FileMonitor(sftp_client) + new_files = [] + today_str = datetime.now().strftime("%d%m%Y") + logger.info(f'listing file for {today_str}') + + for bank_code in self.config.bank_codes: + # Get list of files already processed for this specific bank + bank_processed = repository.get_processed_files(bank_code) + remote_path = f"{self.config.sftp_base_path}/{bank_code}/UPI" + + + pattern = f"*_{today_str}_*_UPI.txt.gz" + files = sftp_client.list_files(remote_path, pattern=pattern) + + + for filename in files: + if filename not in bank_processed: + full_path = f"{remote_path}/{filename}" + new_files.append((filename, bank_code, full_path)) + logger.info(f"Found new file: {filename} (bank: {bank_code})") + else: + logger.debug(f"Skipping already processed file for {bank_code}: {filename}") + + if not new_files: + logger.info("No new files to process") + return + + logger.info(f"Found {len(new_files)} new files to process") + + # Process files + processor = FileProcessor(repository, sftp_client) + stats = processor.process_files(new_files) + + # Log summary + logger.info(f"Cycle {self.cycle_count} complete:") + logger.info(f" Total files: {stats['total']}") + logger.info(f" Successful: {stats['successful']}") + logger.info(f" Failed: {stats['failed']}") + + # Call neft_api_txn_post procedure once per cycle to process all inserted transactions + # if stats['successful'] > 0: + # logger.info("Calling neft_api_txn_post procedure for all inserted transactions...") + # if repository.call_neft_api_txn_post(): + # logger.info("Transaction post-processing completed successfully") + # else: + # logger.error("Transaction post-processing failed") + + except Exception as e: + logger.error(f"Error in processing cycle: {e}", exc_info=True) + + finally: + sftp_client.disconnect() + + def run(self): + """Run scheduler main loop.""" + logger.info("="*80) + logger.info("NEFT_INWARD File Processing Scheduler Started") + logger.info(f"Poll Interval: {self.config.poll_interval_minutes} minutes") + logger.info(f"Bank Codes: {', '.join(self.config.bank_codes)}") + logger.info("="*80) + + # Initialize database + try: + if not self.initialize_database(): + logger.error("Failed to initialize database. Exiting.") + return + except SystemExit as e: + logger.error(f"Fatal error: {e}") + raise + + # Run processing loop + poll_interval_seconds = self.config.poll_interval_minutes * 60 + + while self.running: + try: + self.run_processing_cycle() + except Exception as e: + logger.error(f"Unexpected error in processing cycle: {e}", exc_info=True) + + # Wait for next cycle + if self.running: + logger.info(f"Waiting {self.config.poll_interval_minutes} minutes until next cycle...") + time.sleep(poll_interval_seconds) + + logger.info("Scheduler shutdown complete") + + +def main(): + """Main entry point.""" + # Setup logging + setup_logging() + + # Create and run scheduler + scheduler = Scheduler() + scheduler.run() + + +if __name__ == '__main__': + main() diff --git a/sftp/__init__.py b/sftp/__init__.py new file mode 100644 index 0000000..588ec13 --- /dev/null +++ b/sftp/__init__.py @@ -0,0 +1,6 @@ +"""SFTP module for ACH file processing.""" + +from .sftp_client import SFTPClient +from .file_monitor import FileMonitor + +__all__ = ['SFTPClient', 'FileMonitor'] diff --git a/sftp/file_monitor.py b/sftp/file_monitor.py new file mode 100644 index 0000000..163eeda --- /dev/null +++ b/sftp/file_monitor.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +""" +File monitoring and discovery for UPI files. +Scans SFTP directories for new files across multiple banks. + +Filename convention handled: + _DDMMYYYY__UPI.txt.gz + +Example: + TUMLUK_31052026_8C_UPI.txt.gz +""" + +import re +from typing import List, Tuple, Dict +from logging_config import get_logger +from config import get_config +from .sftp_client import SFTPClient + +logger = get_logger(__name__) + + +class FileMonitor: + """Monitors SFTP for new UPI files.""" + + def __init__(self, sftp_client: SFTPClient = None): + self.config = get_config() + self.sftp_client = sftp_client or SFTPClient() + + def scan_for_new_files(self, processed_filenames: List[str]) -> List[Tuple[str, str, str]]: + """ + Scan all bank directories for new UPI files. + + Returns: + List of (filename, bankcode, full_remote_path) + """ + new_files: List[Tuple[str, str, str]] = [] + + for bank_code in self.config.bank_codes: + remote_path = f"{self.config.sftp_base_path}/{bank_code}/UPI" + + # ✅ Pattern for UPI files + files = self.sftp_client.list_files( + remote_path, + pattern='*_UPI.txt.gz' + ) + + for filename in files: + if filename not in processed_filenames: + full_path = f"{remote_path}/{filename}" + new_files.append((filename, bank_code, full_path)) + logger.info(f"Found new UPI file: {filename} (bank: {bank_code})") + else: + logger.debug(f"Skipping already processed UPI file: {filename}") + + logger.info(f"UPI scan complete: Found {len(new_files)} new files") + return new_files + + @staticmethod + def parse_filename(filename: str) -> Dict[str, str]: + """ + Parse UPI filename. + + Expected format: + _DDMMYYYY__UPI.txt.gz + + Example: + TUMLUK_31052026_8C_UPI.txt.gz + """ + + pattern = r'^([A-Z]+)_(\d{2})(\d{2})(\d{4})_([A-Z0-9]+)_UPI\.txt\.gz$' + match = re.match(pattern, filename, flags=re.IGNORECASE) + + if not match: + logger.warning(f"Could not parse UPI filename: {filename}") + return {} + + location, day, month, year, code = match.groups() + + return { + 'filename': filename, + 'location': location, + 'day': day, + 'month': month, + 'year': year, + 'code': code, + 'timestamp': f"{day}/{month}/{year}" + } + + def __enter__(self): + if not self.sftp_client.sftp: + self.sftp_client.connect() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.sftp_client.disconnect() \ No newline at end of file diff --git a/sftp/sftp_client.py b/sftp/sftp_client.py new file mode 100644 index 0000000..4399915 --- /dev/null +++ b/sftp/sftp_client.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 +""" +SFTP client for file operations. +Handles connection, file discovery, and download operations. +""" + +import paramiko +import os +from pathlib import Path +from logging_config import get_logger +from config import get_config + +logger = get_logger(__name__) + + +class SFTPClient: + """SFTP operations for ACH file processing.""" + + def __init__(self): + """Initialize SFTP client.""" + self.config = get_config() + self.sftp = None + self.ssh = None + + def connect(self) -> bool: + """ + Establish SFTP connection. + + Returns: + True if successful, False otherwise + """ + try: + # Create SSH client + self.ssh = paramiko.SSHClient() + self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + + # Connect + self.ssh.connect( + self.config.sftp_host, + port=self.config.sftp_port, + username=self.config.sftp_username, + password=self.config.sftp_password, + timeout=10 + ) + + # Get SFTP channel + self.sftp = self.ssh.open_sftp() + logger.info(f"Connected to SFTP server: {self.config.sftp_host}:{self.config.sftp_port}") + return True + + except Exception as e: + logger.error(f"Failed to connect to SFTP server: {e}", exc_info=True) + return False + + def disconnect(self): + """Close SFTP connection.""" + try: + if self.sftp: + self.sftp.close() + if self.ssh: + self.ssh.close() + logger.info("SFTP connection closed") + except Exception as e: + logger.error(f"Error closing SFTP connection: {e}") + + def list_files(self, remote_path: str, pattern: str) -> list: + """ + List files matching pattern in remote directory. + + Args: + remote_path: Path on SFTP server + pattern: File pattern to match (e.g., 'ACH_*.txt') + + Returns: + List of matching filenames + """ + if not self.sftp: + logger.error("SFTP not connected") + return [] + + try: + files = [] + try: + items = self.sftp.listdir_attr(remote_path) + except FileNotFoundError: + logger.warning(f"Directory not found: {remote_path}") + return [] + + import fnmatch + for item in items: + if fnmatch.fnmatch(item.filename, pattern): + files.append(item.filename) + + logger.debug(f"Found {len(files)} files matching {pattern} in {remote_path}") + return sorted(files) + + except Exception as e: + logger.error(f"Error listing files in {remote_path}: {e}", exc_info=True) + return [] + + def download_file(self, remote_path: str, local_path: str) -> bool: + """ + Download file from SFTP server. + + Args: + remote_path: Full path on SFTP server + local_path: Local destination path + + Returns: + True if successful, False otherwise + """ + if not self.sftp: + logger.error("SFTP not connected") + return False + + try: + # Create local directory if needed + Path(local_path).parent.mkdir(parents=True, exist_ok=True) + + # Download file + self.sftp.get(remote_path, local_path) + logger.info(f"Downloaded file: {remote_path} -> {local_path}") + return True + + except Exception as e: + logger.error(f"Error downloading file {remote_path}: {e}", exc_info=True) + return False + + def get_file_size(self, remote_path: str) -> int: + """ + Get size of remote file. + + Args: + remote_path: Full path on SFTP server + + Returns: + File size in bytes, or -1 if error + """ + if not self.sftp: + logger.error("SFTP not connected") + return -1 + + try: + stat = self.sftp.stat(remote_path) + return stat.st_size + except Exception as e: + logger.error(f"Error getting file size {remote_path}: {e}") + return -1 + + def __enter__(self): + """Context manager entry.""" + self.connect() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Context manager exit.""" + self.disconnect() diff --git a/upi.py b/upi.py new file mode 100644 index 0000000..c282ee3 --- /dev/null +++ b/upi.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +""" +Main application entry point. +Runs ACH file processing scheduler. +""" + +import logging +from logging_config import setup_logging, get_logger +from scheduler import Scheduler + +# Initialize logging +logging.getLogger("paramiko").setLevel(logging.WARNING) +logger = setup_logging(log_level=logging.INFO) +app_logger = get_logger(__name__) + + +def main(): + """Main application function.""" + app_logger.info("Application started") + + try: + # Run the scheduler + scheduler = Scheduler() + scheduler.run() + app_logger.info("Application completed successfully") + except KeyboardInterrupt: + app_logger.info("Application interrupted by user") + except Exception as e: + app_logger.error(f"An error occurred: {e}", exc_info=True) + raise + + +if __name__ == "__main__": + main() diff --git a/upi_parser.py b/upi_parser.py new file mode 100644 index 0000000..baeb3a8 --- /dev/null +++ b/upi_parser.py @@ -0,0 +1,270 @@ +import os +import logging +from typing import List, Dict, Tuple, Optional +from decimal import Decimal, InvalidOperation + +# ------------------------- +# Logger +# ------------------------- +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +# ------------------------- +# Helpers +# ------------------------- +def normalize_text(val: str) -> str: + return str(val or "").strip() + + +def to_decimal(val: str) -> Optional[Decimal]: + try: + return Decimal(val.strip()) + except Exception: + return None + + +# ------------------------- +# Parser Class +# ------------------------- +class UPIParser: + """ + Parser for UPI Bank Switch Report (NPCI Cycle) + """ + + EXPECTED_HEADER = [ + "rrn", + "type", + "txn_datetime", + "credit_account", + "status", + "amount", + "note", + ] + + def __init__(self, file_path: str): + self.file_path = file_path + self.transactions: List[Dict] = [] + self.file_metadata: Dict = {} + self.summary_data: Dict = {} + + # ------------------------- + # MAIN PARSE + # ------------------------- + def parse(self) -> Tuple[List[Dict], Dict, Dict]: + try: + rows = self._read_rows() + + self.file_metadata = { + "source_file": os.path.basename(self.file_path), + "row_count": len(rows), + "columns": self.EXPECTED_HEADER, + } + + for idx, row in enumerate(rows, start=1): + txn = self._row_to_transaction(row, idx) + if txn: + self.transactions.append(txn) + + self.summary_data = self._build_summary(self.transactions) + + logger.info(f"Parsed {len(self.transactions)} rows from {self.file_path}") + return self.transactions, self.file_metadata, self.summary_data + + except Exception as e: + logger.error(f"Error parsing file: {e}", exc_info=True) + raise + + # ------------------------- + # READ FILE + # ------------------------- + def _read_rows(self) -> List[List[str]]: + rows = [] + + with open(self.file_path, 'r', encoding='utf-8', errors='replace') as f: + for line in f: + line = line.strip() + + # Skip unwanted lines + if ( + not line + or "UPI BANK SWITCH REPORT" in line + or line.startswith("===") + or line.startswith("---") + or line.startswith("RRN") + ): + continue + + parts = line.split('|') + + if len(parts) < 7: + logger.debug(f"Skipping malformed row: {line}") + continue + + rows.append(parts[:7]) + + return rows + + # ------------------------- + # ROW -> TRANSACTION + # ------------------------- + def _row_to_transaction(self, row: List[str], row_num: int) -> Optional[Dict]: + + txn = { + "rrn": normalize_text(row[0]), + "type": normalize_text(row[1]), + "txn_datetime": normalize_text(row[2]), + "credit_account": normalize_text(row[3]), + "status": normalize_text(row[4]), + "amount": "", + "note": normalize_text(row[6]), + } + + # Amount conversion + amt = to_decimal(row[5]) + txn["amount"] = amt if amt is not None else "" + + # Split datetime + if txn["txn_datetime"]: + try: + date, time = txn["txn_datetime"].split(" ") + txn["txn_date"] = date + txn["txn_time"] = time + except Exception: + txn["txn_date"] = txn["txn_datetime"] + txn["txn_time"] = "" + + if not txn["rrn"]: + logger.debug(f"Skipping row {row_num}: Missing RRN") + return None + + return txn + + # ------------------------- + # SUMMARY + # ------------------------- + def _build_summary(self, txns: List[Dict]) -> Dict: + total_count = len(txns) + total_amount = Decimal("0") + + by_status: Dict[str, Dict] = {} + by_type: Dict[str, Dict] = {} + + for t in txns: + amt = t.get("amount") or Decimal("0") + + if isinstance(amt, str): + try: + amt = Decimal(amt) + except: + amt = Decimal("0") + + total_amount += amt + + status = t.get("status", "").upper() + txn_type = t.get("type", "").upper() + + # Status grouping + if status not in by_status: + by_status[status] = {"count": 0, "amount": Decimal("0")} + by_status[status]["count"] += 1 + by_status[status]["amount"] += amt + + # Type grouping + if txn_type not in by_type: + by_type[txn_type] = {"count": 0, "amount": Decimal("0")} + by_type[txn_type]["count"] += 1 + by_type[txn_type]["amount"] += amt + + # Convert Decimal → string + by_status_final = { + k: {"count": v["count"], "amount": f"{v['amount']:.2f}"} + for k, v in by_status.items() + } + + by_type_final = { + k: {"count": v["count"], "amount": f"{v['amount']:.2f}"} + for k, v in by_type.items() + } + + return { + "total_count": total_count, + "total_amount": f"{total_amount:.2f}", + "by_status": by_status_final, + "by_type": by_type_final, + } + + +# ------------------------- +# PRINT FUNCTIONS +# ------------------------- +def print_transactions(transactions: List[Dict], limit: Optional[int] = 50): + + headers = [ + ("RRN", 15), + ("TYPE", 8), + ("DATE", 10), + ("TIME", 10), + ("ACCOUNT", 20), + ("STATUS", 10), + ("AMOUNT", 12), + ("NOTE", 25), + ] + + header_line = " ".join([f"{h:<{w}}" for h, w in headers]) + print("\n" + "=" * len(header_line)) + print(header_line) + print("=" * len(header_line)) + + for i, txn in enumerate(transactions): + row = [ + txn.get("rrn", ""), + txn.get("type", ""), + txn.get("txn_date", ""), + txn.get("txn_time", ""), + txn.get("credit_account", ""), + txn.get("status", ""), + f"{txn.get('amount')}" if txn.get("amount") else "", + txn.get("note", ""), + ] + + print(" ".join(f"{str(val)[:w]:<{w}}" for val, (h, w) in zip(row, headers))) + + if limit and i + 1 >= limit: + print(f"... ({len(transactions) - limit} more rows)") + break + + print("=" * len(header_line)) + print(f"Total: {len(transactions)} records\n") + + +def print_metadata(metadata: Dict): + print("\n===== FILE METADATA =====") + for k, v in metadata.items(): + print(f"{k.upper():20}: {v}") + print("=========================\n") + + +def print_summary(summary: Dict): + print("\n===== SUMMARY =====") + for k, v in summary.items(): + print(f"{k.upper()}: {v}") + print("===================\n") + + +# ------------------------- +# MAIN RUNNER +# ------------------------- +if __name__ == "__main__": + + file_path = r"C:\Users\2780475\Desktop\Test\TUMLUK_31052026_8C_UPI.txt.gz" + + parser = UPISwitchParser(file_path) + + transactions, metadata, summary = parser.parse() + + print_metadata(metadata) + print_transactions(transactions, limit=50) + print_summary(summary) + + logger.info(f"Parsing complete. Extracted {len(transactions)} transactions") \ No newline at end of file