266 lines
8.9 KiB
Python
266 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Data access layer for ACH 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 TransactionRecord, ProcessedFile
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class Repository:
|
|
"""Data access layer for ACH processing."""
|
|
|
|
def __init__(self):
|
|
"""Initialize repository with connector."""
|
|
self.connector = get_connector()
|
|
|
|
def validate_account_exists(self, account_number: str) -> bool:
|
|
"""
|
|
Validate if account number exists in dep_account table.
|
|
|
|
Args:
|
|
account_number: Account number to validate (cbs_acct)
|
|
|
|
Returns:
|
|
True if account exists in dep_account.link_accno, False otherwise
|
|
"""
|
|
conn = self.connector.get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT COUNT(*) FROM dep_account WHERE link_accno = :accno",
|
|
{'accno': account_number}
|
|
)
|
|
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()
|
|
|
|
def bulk_insert_transactions(self, transactions: List[TransactionRecord]) -> tuple:
|
|
"""
|
|
Bulk insert transaction records into ach_api_log.
|
|
Records with invalid account numbers are silently skipped.
|
|
|
|
Args:
|
|
transactions: List of TransactionRecord objects
|
|
|
|
Returns:
|
|
Tuple of (inserted_count, skipped_count)
|
|
"""
|
|
if not transactions:
|
|
logger.warning("No transactions to insert")
|
|
return 0, 0
|
|
|
|
# Validate accounts and filter out invalid ones
|
|
valid_transactions = []
|
|
skipped_count = 0
|
|
|
|
for txn in transactions:
|
|
if self.validate_account_exists(txn.cbs_acct):
|
|
valid_transactions.append(txn)
|
|
else:
|
|
skipped_count += 1
|
|
|
|
if not valid_transactions:
|
|
logger.info(f"All {skipped_count} transactions skipped (invalid accounts)")
|
|
return 0, skipped_count
|
|
|
|
conn = self.connector.get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
|
|
# Prepare batch data
|
|
batch_data = [txn.to_dict() for txn in valid_transactions]
|
|
|
|
# Execute batch insert
|
|
insert_sql = """
|
|
INSERT INTO ach_api_log_temp (
|
|
narration, status, bankcode, jrnl_id,
|
|
tran_date, cbs_acct, tran_amt, TXNIND
|
|
) VALUES (
|
|
:narration, :status, :bankcode, :jrnl_id,
|
|
:tran_date, :cbs_acct, :tran_amt, :TXNIND
|
|
)
|
|
"""
|
|
|
|
cursor.executemany(insert_sql, batch_data)
|
|
conn.commit()
|
|
|
|
inserted_count = len(valid_transactions)
|
|
logger.info(f"Inserted {inserted_count} transactions, skipped {skipped_count} (invalid accounts)")
|
|
return inserted_count, skipped_count
|
|
|
|
except Exception as e:
|
|
conn.rollback()
|
|
logger.error(f"Error inserting transactions: {e}", exc_info=True)
|
|
raise
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
def is_file_processed(self, filename: str, bankcode: str) -> bool:
|
|
"""
|
|
Check if file has already been processed for a specific bank.
|
|
|
|
Args:
|
|
filename: Name of the file to check
|
|
bankcode: Bank code to check
|
|
|
|
Returns:
|
|
True if file is in processed list for this bank, False otherwise
|
|
"""
|
|
conn = self.connector.get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT COUNT(*) FROM ach_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}")
|
|
return False
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
def mark_file_processed(self, processed_file: ProcessedFile) -> bool:
|
|
"""
|
|
Insert record into ach_processed_files to mark file as processed.
|
|
|
|
Args:
|
|
processed_file: ProcessedFile object with file metadata
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
conn = self.connector.get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
|
|
file_data = processed_file.to_dict()
|
|
insert_sql = """
|
|
INSERT INTO ach_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:
|
|
conn.rollback()
|
|
logger.error(f"Error marking file as processed: {e}", exc_info=True)
|
|
return False
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
def get_processed_files(self, bankcode: Optional[str] = None) -> List[str]:
|
|
"""
|
|
Get list of processed filenames.
|
|
|
|
Args:
|
|
bankcode: Optional bankcode filter
|
|
|
|
Returns:
|
|
List of filenames that have been processed
|
|
"""
|
|
conn = self.connector.get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
|
|
if bankcode:
|
|
cursor.execute(
|
|
"SELECT filename FROM ach_processed_files WHERE bankcode = :bankcode ORDER BY processed_at DESC",
|
|
{'bankcode': bankcode}
|
|
)
|
|
else:
|
|
cursor.execute("SELECT filename FROM ach_processed_files ORDER BY processed_at DESC")
|
|
|
|
filenames = [row[0] for row in cursor.fetchall()]
|
|
return filenames
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error retrieving processed files: {e}")
|
|
return []
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
def call_ach_api_txn_post(self) -> bool:
|
|
"""
|
|
Call the ach_api_txn_post procedure to process inserted transactions.
|
|
Should be called once per processing cycle after all files are inserted.
|
|
|
|
Returns:
|
|
True if procedure executed successfully, False otherwise
|
|
"""
|
|
conn = self.connector.get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
logger.info("Calling ach_api_txn_post procedure to process all inserted transactions...")
|
|
cursor.execute("CALL ach_api_txn_post()")
|
|
conn.commit()
|
|
logger.info("ach_api_txn_post procedure executed successfully")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Error calling ach_api_txn_post procedure: {e}", exc_info=True)
|
|
return False
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
def verify_tables_exist(self):
|
|
"""
|
|
Verify that required database tables exist.
|
|
If tables are missing, terminate the program.
|
|
"""
|
|
conn = self.connector.get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
|
|
# Check if ach_api_log table exists
|
|
try:
|
|
cursor.execute("SELECT COUNT(*) FROM ach_api_log_temp WHERE ROWNUM = 1")
|
|
logger.info("✓ ach_api_log_temp table exists")
|
|
except Exception as e:
|
|
logger.error(f"✗ ach_api_log_temp table not found: {e}")
|
|
raise SystemExit("FATAL: ach_api_log_temp table must be created manually before running this application")
|
|
|
|
# Check if ach_processed_files table exists
|
|
try:
|
|
cursor.execute("SELECT COUNT(*) FROM ach_processed_files WHERE ROWNUM = 1")
|
|
logger.info("✓ ach_processed_files table exists")
|
|
except Exception as e:
|
|
logger.error(f"✗ ach_processed_files table not found: {e}")
|
|
raise SystemExit("FATAL: ach_processed_files table must be created manually before running this application")
|
|
|
|
logger.info("Database tables verified successfully")
|
|
|
|
except SystemExit:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error verifying tables: {e}", exc_info=True)
|
|
raise SystemExit(f"FATAL: Error verifying database tables: {e}")
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|