commit 6421a07f2be1c18206d13c0332a38816f20d52be Author: Bishwajeet Kumar Rajak Date: Thu Jun 11 00:43:44 2026 +0530 first commit diff --git a/__pycache__/config.cpython-314.pyc b/__pycache__/config.cpython-314.pyc new file mode 100644 index 0000000..9fdca69 Binary files /dev/null and b/__pycache__/config.cpython-314.pyc differ diff --git a/__pycache__/logging_config.cpython-314.pyc b/__pycache__/logging_config.cpython-314.pyc new file mode 100644 index 0000000..e4e4f52 Binary files /dev/null and b/__pycache__/logging_config.cpython-314.pyc differ diff --git a/api/api_client.py b/api/api_client.py new file mode 100644 index 0000000..3cb77c4 --- /dev/null +++ b/api/api_client.py @@ -0,0 +1,38 @@ +import requests +from logging_config import get_logger + +logger = get_logger(__name__) + +class FetchQueueClient: + + def __init__(self, config): + self.url = config.FETCH_QUEUE_URL + + def fetch(self, payload): + logger.info("API request payload: %s", payload) + + try: + response = requests.post( + self.url, + json=payload, + verify=False, + timeout=20 + ) + + logger.info("API status: %s", response.status_code) + + if not response.ok: + logger.error("API error response: %s", response.text) + + response.raise_for_status() + + result = response.json() + + logger.debug("API full response: %s", result) + + return result + + except Exception: + logger.exception("API call failed") + raise +`` \ No newline at end of file diff --git a/config.py b/config.py new file mode 100644 index 0000000..5c31879 --- /dev/null +++ b/config.py @@ -0,0 +1,32 @@ +import os + +class Config: + + BANK_CODE_MAPPING = { + "9": "0015", "12": "0014", "13": "0016", "14": "0009", + "15": "0005", "16": "0001", "7": "0013", "6": "0002", + "5": "0007", "17": "0020", "1": "0012", "10": "0017", + "11": "0018", "2": "0003", "4": "0004", + "8": "0006", "3": "0021" + } + + FETCH_QUEUE_URL = "https://43.225.3.224:443/IPKS_Queue_Generation/fetchAllQueue" + + def __init__(self): + self.db_user = os.getenv("DB_USER", "") + self.db_password = os.getenv("DB_PASSWORD", "") + self.db_dsn = os.getenv("DB_DSN", "") + + self.poll_interval_minutes = int(os.getenv("POLL_INTERVAL_MINUTES", "60")) + self.bank_codes = self._parse_bank_codes() + + def _parse_bank_codes(self): + codes_str = os.getenv( + "BANK_CODES", + "0001,0002,0003,0004,0005,0006,0007,0009,0012,0013,0014,0015,0016,0017,0018,0020,0021" + ) + return [c.strip() for c in codes_str.split(",") if c.strip()] + + def validate(self): + if not self.db_user or not self.db_password or not self.db_dsn: + raise RuntimeError("Database configuration missing") \ No newline at end of file diff --git a/db/__pycache__/oracle_connector.cpython-314.pyc b/db/__pycache__/oracle_connector.cpython-314.pyc new file mode 100644 index 0000000..415bbde Binary files /dev/null and b/db/__pycache__/oracle_connector.cpython-314.pyc differ diff --git a/db/db.py b/db/db.py new file mode 100644 index 0000000..72608a3 --- /dev/null +++ b/db/db.py @@ -0,0 +1,87 @@ +from logging_config import get_logger + +logger = get_logger(__name__) + +class OracleDB: + + def __init__(self, connector): + self.connector = connector + + def fetch_all(self, sql, params=None): + conn = None + cur = None + try: + conn = self.connector.get_connection() + cur = conn.cursor() + + cur.execute(sql, params or {}) + + cols = [c[0] for c in cur.description] + rows = [dict(zip(cols, r)) for r in cur.fetchall()] + + logger.info("Fetched %s rows from database", len(rows)) + + return rows + + except Exception: + logger.exception("Error in fetch_all") + raise + + finally: + if cur: + cur.close() + if conn: + conn.close() + + def execute(self, sql, params): + conn = None + cur = None + try: + conn = self.connector.get_connection() + cur = conn.cursor() + + cur.execute(sql, params) + conn.commit() + + logger.info("Insert/Update executed successfully") + + except Exception: + if conn: + conn.rollback() + logger.exception("Error executing SQL") + raise + + finally: + if cur: + cur.close() + if conn: + conn.close() + + def call_procedure(self, name): + conn = None + cur = None + try: + conn = self.connector.get_connection() + cur = conn.cursor() + + logger.info("Calling procedure %s", name) + + try: + cur.callproc(name) + except Exception: + cur.execute(f"BEGIN {name}; END;") + + conn.commit() + + logger.info("Procedure executed successfully") + return True + + except Exception: + logger.exception("Procedure call failed") + return False + + finally: + if cur: + cur.close() + if conn: + conn.close() \ No newline at end of file diff --git a/db/oracle_connector.py b/db/oracle_connector.py new file mode 100644 index 0000000..9ef0856 --- /dev/null +++ b/db/oracle_connector.py @@ -0,0 +1,23 @@ +import oracledb +from logging_config import get_logger + +logger = get_logger(__name__) + +class OracleConnector: + + def __init__(self, config): + self.config = config + self.pool = None + + def get_connection(self): + if not self.pool: + logger.info("Initializing Oracle connection pool") + self.pool = oracledb.create_pool( + user=self.config.db_user, + password=self.config.db_password, + dsn=self.config.db_dsn, + min=1, + max=5, + increment=1 + ) + return self.pool.acquire() \ No newline at end of file diff --git a/logging_config.py b/logging_config.py new file mode 100644 index 0000000..8c76ad9 --- /dev/null +++ b/logging_config.py @@ -0,0 +1,29 @@ +import logging +import logging.handlers +from pathlib import Path + +def setup_logging(): + Path("logs").mkdir(exist_ok=True) + + root_logger = logging.getLogger() + root_logger.setLevel(logging.INFO) + root_logger.handlers.clear() + + formatter = logging.Formatter( + "%(asctime)s %(levelname)s %(name)s - %(message)s" + ) + + console = logging.StreamHandler() + console.setFormatter(formatter) + root_logger.addHandler(console) + + file_handler = logging.handlers.RotatingFileHandler( + "logs/application.log", + maxBytes=10 * 1024 * 1024, + backupCount=5 + ) + file_handler.setFormatter(formatter) + root_logger.addHandler(file_handler) + +def get_logger(name): + return logging.getLogger(name) \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..17e3fd6 --- /dev/null +++ b/main.py @@ -0,0 +1,27 @@ +from logging_config import setup_logging +from config import Config +from db.oracle_connector import OracleConnector +from db.db import OracleDB +from repository.repository import Repository +from api.api_client import FetchQueueClient +from processor.processor import Processor +from scheduler.scheduler import Scheduler + +def main(): + setup_logging() + + config = Config() + config.validate() + + connector = OracleConnector(config) + db = OracleDB(connector) + repo = Repository(db) + api = FetchQueueClient(config) + + processor = Processor(config, repo, api) + scheduler = Scheduler(processor, config) + + scheduler.run() + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/models/bank_mapping.py b/models/bank_mapping.py new file mode 100644 index 0000000..ce1f3a6 --- /dev/null +++ b/models/bank_mapping.py @@ -0,0 +1,7 @@ +class BankMapping: + + def __init__(self, mapping): + self.reverse = {v: k for k, v in mapping.items()} + + def get_dccb(self, bank_code): + return self.reverse.get(bank_code) \ No newline at end of file diff --git a/models/outward_log_model.py b/models/outward_log_model.py new file mode 100644 index 0000000..9b867b2 --- /dev/null +++ b/models/outward_log_model.py @@ -0,0 +1,22 @@ +class OutwardLogModel: + + def __init__(self, bank, r, api): + self.bankcode = bank + self.jrnl_id = api.get("journalId") + self.ref_no = api.get("remarks") + + self.tran_date = r["TXN_DATE"].strftime("%d%m%Y") + self.txn_amt = float(r["TXN_AMT"]) + + self.recv_ac = r["DEST_AC_NO"] + self.send_ac = api.get("accountNum") + + self.ifsc = r["IFSC_CODE"] + self.sender_name = r["REMITTER_NAME"] + self.receiver_name = r["BENEFICIARY_NAME"] + self.beneficiary_address = r["BENEFICIARY_ADD"] + + self.sender_info = api.get("sendeReceiverInfo") + + def to_dict(self): + return self.__dict__ \ No newline at end of file diff --git a/processor/processor.py b/processor/processor.py new file mode 100644 index 0000000..fd9178d --- /dev/null +++ b/processor/processor.py @@ -0,0 +1,91 @@ +from logging_config import get_logger +from models.bank_mapping import BankMapping +from models.outward_log_model import OutwardLogModel + +logger = get_logger(__name__) + +class Processor: + + def __init__(self, config, repo, api): + self.config = config + self.repo = repo + self.api = api + self.mapping = BankMapping(config.BANK_CODE_MAPPING) + + def process(self): + total_inserted = 0 + + logger.info("Starting processing...") + + for bank_code in self.config.bank_codes: + + try: + logger.info("Processing bank %s", bank_code) + + dccb = self.mapping.get_dccb(bank_code) + + if not dccb: + logger.error("No mapping for bank %s", bank_code) + continue + + records = self.repo.get_records(dccb) + + if not records: + logger.info("No records found for bank %s", bank_code) + continue + + payload = { + "bankCode": bank_code, + "rrn": "090686552", + "branchCode": str(records[0]["BR_CODE"]).zfill(3), + "itemType": 1, + "queueType": "02", + "reterieveTop": "20" + } + + response = self.api.fetch(payload) + + if response.get("status") != "SUCCESS": + logger.error("API error for bank %s", bank_code) + continue + + items = response.get("response", {}).get("VwQueueItems", []) + + api_map = { + (int(x["queueItemId"]), float(x["txnAmt"])): x + for x in items + } + + for r in records: + + key = (int(r["CBS_QUEUE_NO2"]), float(r["TXN_AMT"])) + item = api_map.get(key) + + if not item: + continue + + if item.get("status") != "Processed": + continue + + ref = item.get("remarks") + + if not ref: + continue + + if self.repo.exists_by_ref(ref): + logger.info("Duplicate ref %s", ref) + continue + + model = OutwardLogModel(bank_code, r, item) + + self.repo.insert_outward_log(model.to_dict()) + total_inserted += 1 + + logger.info("Inserted ref %s", ref) + + except Exception: + logger.exception("Error processing bank %s", bank_code) + + logger.info("Processing completed total inserted %s", total_inserted) + + return total_inserted diff --git a/repository/repository.py b/repository/repository.py new file mode 100644 index 0000000..cc81719 --- /dev/null +++ b/repository/repository.py @@ -0,0 +1,50 @@ +from logging_config import get_logger + +logger = get_logger(__name__) + +class Repository: + + def __init__(self, db): + self.db = db + + def get_records(self, dccb): + logger.info("Fetching records for DCCB=%s", dccb) + + return self.db.fetch_all(""" + SELECT TXN_DATE,DCCB_CODE,BR_CODE,SRC_AC_NO,DEST_AC_NO, + IFSC_CODE,TXN_AMT,BENEFICIARY_NAME, + BENEFICIARY_ADD,REMITTER_NAME,CBS_QUEUE_NO2 + FROM NEFT_RTGS_TXN_QUEUE + WHERE DCCB_CODE = :d + AND TRUNC(TXN_DATE)=TRUNC(SYSDATE) + """, {"d": dccb}) + + def exists_by_ref(self, ref): + logger.debug("Checking duplicate REF=%s", ref) + + res = self.db.fetch_all( + "SELECT 1 FROM outward_neft_api_log WHERE REF_NO=:r", + {"r": ref} + ) + + return len(res) > 0 + + def insert_outward_log(self, d): + logger.debug("Insert data: %s", d) + + self.db.execute(""" + INSERT INTO outward_neft_api_log( + BANKCODE,TXNCODE,TXNIND,JRNL_ID,REF_NO,TRAN_DATE, + TXN_AMT,RECVR_ACCT_NO,SENDER_ACCT_NO, + RECIEVER_IFSC,SENDER_NAME,RECIEVER_NAME, + BENEFICIARY_ADDRESS,SENDER_TO_RECIVER_INFO,MSG_TYPE + ) + VALUES ( + :bankcode,'OUTWARD NEFT','DR',:jrnl_id,:ref_no,:tran_date, + :txn_amt,:recv_ac,:send_ac, + :ifsc,:sender_name,:receiver_name, + :beneficiary_address,:sender_info,'N06') + """, d) + + def call_proc(self): + return self.db.call_procedure("neft_api_txn_post") \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..0e3667f --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +oracledb==2.0.0 +requests \ No newline at end of file diff --git a/scheduler/scheduler.py b/scheduler/scheduler.py new file mode 100644 index 0000000..e7320c5 --- /dev/null +++ b/scheduler/scheduler.py @@ -0,0 +1,29 @@ +import time +from logging_config import get_logger + +logger = get_logger(__name__) + +class Scheduler: + + def __init__(self, processor, config): + self.processor = processor + self.config = config + + def run(self): + logger.info("Scheduler started") + + while True: + try: + count = self.processor.process() + + logger.info("Inserted count %s", count) + + if count > 0: + logger.info("Calling procedure") + self.processor.repo.call_proc() + + except Exception: + logger.exception("Scheduler error") + + logger.info("Sleeping for next cycle") + time.sleep(self.config.poll_interval_minutes * 60)