push_data.py

import os
import sys
import json

import pandas as pd
import pymongo
import certifi
from dotenv import load_dotenv

from networksecurity.exception.exception import NetworkSecurityException

load_dotenv()

MONGO_DB_URL = os.getenv("MONGODB_URL_KEY")
print(MONGO_DB_URL)

ca = certifi.where()

class NetworkDataExtract():
    def __init__(self):
        try:
            pass
        except Exception as e:
            raise NetworkSecurityException(e,sys)
        
    def csv_to_json_convertor(self,file_path):
        try:
            data=pd.read_csv(file_path)
            data.reset_index(drop=True,inplace=True)
            records=list(json.loads(data.T.to_json()).values())
            return records
        except Exception as e:
            raise NetworkSecurityException(e,sys)
        
    def insert_data_mongodb(self,records,database,collection):
        try:
            self.database=database
            self.collection=collection
            self.records=records
            self.mongo_client = pymongo.MongoClient(MONGO_DB_URL, tlsCAFile=ca)
            self.database = self.mongo_client[self.database]
            self.collection = self.database[self.collection]
            self.collection.insert_many(self.records)
            return(len(self.records))
        except Exception as e:
            raise NetworkSecurityException(e,sys)
        

if __name__ == '__main__':  # execution of the etl pipeline
    FILE_PATH = os.path.join("Network_Data", "phisingData.csv")
    DATABASE = "PRINCEAI"
    Collection = "NetworkData"

    networkobj = NetworkDataExtract()

    print("Step 1: Converting CSV to JSON...")
    records = networkobj.csv_to_json_convertor(file_path=FILE_PATH)

    print("Step 2: Conversion done")
    print("Total records:", records)

    print("Step 3: Inserting into MongoDB...")
    no_of_records = networkobj.insert_data_mongodb(records, DATABASE, Collection)

    print("Inserted records:", no_of_records)