first commit

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
+``

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")

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()

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()

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)

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()

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)

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__

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

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")

requirements.txt

@@ -0,0 +1,2 @@
+oracledb==2.0.0
+requests

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)