added validation to consider only ipks mirror account numbers
This commit is contained in:
@@ -19,26 +19,67 @@ class Repository:
|
|||||||
"""Initialize repository with connector."""
|
"""Initialize repository with connector."""
|
||||||
self.connector = get_connector()
|
self.connector = get_connector()
|
||||||
|
|
||||||
def bulk_insert_transactions(self, transactions: List[TransactionRecord]) -> int:
|
def validate_account_exists(self, account_number: str) -> bool:
|
||||||
"""
|
"""
|
||||||
Bulk insert transaction records into ach_api_log_temp git.
|
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:
|
Args:
|
||||||
transactions: List of TransactionRecord objects
|
transactions: List of TransactionRecord objects
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Number of inserted records
|
Tuple of (inserted_count, skipped_count)
|
||||||
"""
|
"""
|
||||||
if not transactions:
|
if not transactions:
|
||||||
logger.warning("No transactions to insert")
|
logger.warning("No transactions to insert")
|
||||||
return 0
|
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()
|
conn = self.connector.get_connection()
|
||||||
try:
|
try:
|
||||||
cursor = conn.cursor()
|
cursor = conn.cursor()
|
||||||
|
|
||||||
# Prepare batch data
|
# Prepare batch data
|
||||||
batch_data = [txn.to_dict() for txn in transactions]
|
batch_data = [txn.to_dict() for txn in valid_transactions]
|
||||||
|
|
||||||
# Execute batch insert
|
# Execute batch insert
|
||||||
insert_sql = """
|
insert_sql = """
|
||||||
@@ -54,9 +95,9 @@ class Repository:
|
|||||||
cursor.executemany(insert_sql, batch_data)
|
cursor.executemany(insert_sql, batch_data)
|
||||||
conn.commit()
|
conn.commit()
|
||||||
|
|
||||||
count = len(transactions)
|
inserted_count = len(valid_transactions)
|
||||||
logger.info(f"Successfully inserted {count} transactions into ach_api_log_temp")
|
logger.info(f"Inserted {inserted_count} transactions, skipped {skipped_count} (invalid accounts)")
|
||||||
return count
|
return inserted_count, skipped_count
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
conn.rollback()
|
conn.rollback()
|
||||||
|
|||||||
@@ -93,8 +93,8 @@ class FileProcessor:
|
|||||||
mapped_records = DataMapper.map_transactions(transactions, bankcode)
|
mapped_records = DataMapper.map_transactions(transactions, bankcode)
|
||||||
logger.info(f"Mapped {len(mapped_records)} transactions")
|
logger.info(f"Mapped {len(mapped_records)} transactions")
|
||||||
|
|
||||||
# Step 5: Insert to database
|
# Step 5: Insert to database (with account validation)
|
||||||
inserted_count = self.repository.bulk_insert_transactions(mapped_records)
|
inserted_count, skipped_count = self.repository.bulk_insert_transactions(mapped_records)
|
||||||
|
|
||||||
# Step 6: Mark file as processed
|
# Step 6: Mark file as processed
|
||||||
processed_file = ProcessedFile(
|
processed_file = ProcessedFile(
|
||||||
@@ -106,7 +106,7 @@ class FileProcessor:
|
|||||||
)
|
)
|
||||||
self.repository.mark_file_processed(processed_file)
|
self.repository.mark_file_processed(processed_file)
|
||||||
|
|
||||||
logger.info(f"Successfully processed {filename}: {inserted_count} transactions inserted")
|
logger.info(f"Successfully processed {filename}: {inserted_count} inserted, {skipped_count} skipped (invalid accounts)")
|
||||||
return True
|
return True
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
|||||||
Reference in New Issue
Block a user