bishwajeet.rajak/upi_inward_file_based
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

updated

Browse Source
This commit is contained in:
bishwajeet.rajak
2026-06-16 15:37:55 +05:30
parent 78887017a0
commit 6f4ac190c1
1786 changed files with 316067 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/repository.py
+58 -61
View File
@@ -23,31 +23,33 @@ class Repository:
# ---------------------------------------------------------
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()
cursor = None
try:
cursor = conn.cursor()
acc = account_number[-12:]
if acc.startswith('0'):
acc = acc[1:]
cursor.execute(
"SELECT COUNT(*) FROM dep_account WHERE link_accno = :accno",
{'accno': last12}
{'accno': acc}
)
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()
if cursor:
cursor.close()
conn.close()
# ---------------------------------------------------------
# UPDATED: UPI BULK INSERT WITH VALIDATION
# BULK INSERT WITH VALIDATION
# ---------------------------------------------------------
def bulk_insert_transactions(self, transactions: List[UPIInwardRecord]) -> tuple:
@@ -61,14 +63,10 @@ class Repository:
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:
@@ -132,7 +130,7 @@ class Repository:
conn.close()
# ---------------------------------------------------------
# FILE TRACKING (UNCHANGED)
# FILE TRACKING
# ---------------------------------------------------------
def is_file_processed(self, filename: str, bankcode: str) -> bool:
@@ -200,8 +198,10 @@ class Repository:
conn.close()
def get_processed_files(self, bankcode: Optional[str] = None) -> List[str]:
conn = self.connector.get_connection()
cursor = None
try:
cursor = conn.cursor()
@@ -224,79 +224,76 @@ class Repository:
"""
)
filenames = [row[0] for row in cursor.fetchall()]
return filenames
return [row[0] for row in cursor.fetchall()]
except Exception as e:
logger.error(f"Error retrieving processed files: {e}", exc_info=True)
return []
finally:
if cursor:
cursor.close()
conn.close()
# def call_neft_api_txn_post(self) -> bool:
# conn = self.connector.get_connection()
# cursor = None
# try:
# cursor = conn.cursor()
# logger.info("Calling neft_api_txn_post procedure to process all inserted transactions...")
# ---------------------------------------------------------
# CALL STORED PROCEDURE
# ---------------------------------------------------------
def call_upi_api_txn_post(self) -> bool:
# try:
# cursor.callproc('neft_api_txn_post')
# except Exception:
# cursor.execute("BEGIN neft_api_txn_post; END;")
# conn.commit()
# logger.info("neft_api_txn_post procedure executed successfully")
# return True
# except Exception as e:
# logger.error(f"Error calling neft_api_txn_post procedure: {e}", exc_info=True)
# return False
# finally:
# if cursor:
# cursor.close()
# conn.close()
def verify_tables_exist(self):
conn = self.connector.get_connection()
cursor = None
try:
cursor = conn.cursor()
logger.info("Calling upi_api_txn_post procedure...")
try:
cursor.callproc('PACS_DB.UPI_API_TXN_POST')
except Exception:
cursor.execute("BEGIN PACS_DB.UPI_API_TXN_POST; END;")
conn.commit()
logger.info("upi_api_txn_post executed successfully")
return True
except Exception as e:
logger.error(f"Error calling procedure: {e}", exc_info=True)
return False
finally:
if cursor:
cursor.close()
conn.close()
# ---------------------------------------------------------
# VERIFY TABLES
# ---------------------------------------------------------
def verify_tables_exist(self):
conn = self.connector.get_connection()
cursor = None
try:
cursor = conn.cursor()
try:
cursor.execute("SELECT COUNT(*) FROM inward_upi_api_log WHERE ROWNUM = 1")
logger.info("✓ inward_upi_api_log table exists")
cursor.execute("SELECT 1 FROM inward_upi_api_log WHERE ROWNUM = 1")
logger.info("inward_upi_api_log exists")
except Exception as e:
logger.error(f"✗ inward_upi_api_log table not found: {e}")
raise SystemExit(
"FATAL: inward_upi_api_log table must be created manually before running this application"
"FATAL: inward_upi_api_log table missing"
)
try:
cursor.execute("SELECT COUNT(*) FROM upi_processed_files WHERE ROWNUM = 1")
logger.info("✓ upi_processed_files table exists")
cursor.execute("SELECT 1 FROM upi_processed_files WHERE ROWNUM = 1")
logger.info("upi_processed_files exists")
except Exception as e:
logger.error(f"✗ upi_processed_files table not found: {e}")
raise SystemExit(
"FATAL: upi_processed_files table must be created manually before running this application"
"FATAL: upi_processed_files table missing"
)
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:
if cursor:
cursor.close()
conn.close()