changed download logs

.gitignore

@@ -1,3 +1,4 @@
 .venv
 .env
 __pycache__/
+audit_logs.json

.vscode/launch.json

@@ -0,0 +1,19 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Python Debugger: FastAPI",
+            "type": "debugpy",
+            "request": "launch",
+            "module": "uvicorn",
+            "args": [
+                "main:app",
+                "--reload"
+            ],
+            "jinja": true
+        }
+    ]
+}

ai_model/callAiModel.py

@@ -4,7 +4,7 @@ from langchain.prompts import PromptTemplate
 from langchain_huggingface import HuggingFaceEndpoint
 
 # HuggingFace Endpoint Initialization
-repo_id = "mistralai/Mixtral-8x7B-Instruct-v0.1"
+repo_id = ""
 llmModel = HuggingFaceEndpoint(
     repo_id=repo_id,
     max_new_tokens=512,

routes/route.py

@@ -160,7 +160,6 @@ async def add_or_update_chat(
         return {"message": "Chat added successfully."}
 
 
-
 @router.post(
     "/api/model/getAllTitleByUserId",
     tags=["model"],
@@ -183,6 +182,10 @@ async def get_all_title(
     return chatList
 
 
+from fastapi import APIRouter, Depends, HTTPException
+from datetime import datetime, timedelta, timezone
+import json
+
 
 @router.post(
     "/api/model/getAllLogsByDateRange",
@@ -196,24 +199,27 @@ async def get_logs_by_daterange(
     verify_token(token.credentials)
     current_date = datetime.now(timezone.utc)  # Ensure timezone awareness
 
-    # Determine start_date based on duration format
-    if " - " in data.duration:  # Handle custom date range
-        try:
+    # Determine start_date and end_date
+    try:
+        if " - " in data.duration:  # Custom date range
             start_date_str, end_date_str = map(str.strip, data.duration.split(" - "))
             start_date = datetime.fromisoformat(start_date_str)
             end_date = datetime.fromisoformat(end_date_str)
-        except ValueError:
-            raise ValueError("Invalid date format. Use ISO 8601 format: YYYY-MM-DDTHH:MM:SS.ssssss+00:00 - YYYY-MM-DDTHH:MM:SS.ssssss+00:00")
-    else:  # Handle predefined durations
-        if data.duration == "1 Day":
-            start_date = current_date - timedelta(days=1)
-        elif data.duration == "Current Week":
-            start_date = current_date - timedelta(weeks=1)
-        elif data.duration == "Last 2 Weeks":
-            start_date = current_date - timedelta(weeks=2)
-        else:
-            raise ValueError("Invalid duration selected")
-        end_date = current_date
+        else:  # Predefined durations
+            if data.duration == "1 Day":
+                start_date = current_date - timedelta(days=1)
+            elif data.duration == "Current Week":
+                start_date = current_date - timedelta(weeks=1)
+            elif data.duration == "Last 2 Weeks":
+                start_date = current_date - timedelta(weeks=2)
+            else:
+                raise HTTPException(status_code=400, detail="Invalid duration selected")
+            end_date = current_date
+    except ValueError:
+        raise HTTPException(
+            status_code=400,
+            detail="Invalid date format. Use ISO 8601 format: YYYY-MM-DDTHH:MM:SS.ssssss+00:00 - YYYY-MM-DDTHH:MM:SS.ssssss+00:00",
+        )
 
     # Ensure timestamps are in UTC
     start_date = start_date.astimezone(timezone.utc)
@@ -226,7 +232,14 @@ async def get_logs_by_daterange(
         encrypted_data = encrypted_record["encrypted_data"]
         decrypted_data = cipher.decrypt(encrypted_data).decode()
         log = json.loads(decrypted_data)
-        log_timestamp = datetime.fromisoformat(log["Timestamp"]).astimezone(timezone.utc)
+
+        # Ensure timestamp is parsed correctly
+        try:
+            log_timestamp = datetime.fromisoformat(log["timestamp"]).astimezone(
+                timezone.utc
+            )
+        except (ValueError, KeyError):
+            continue  # Skip logs with invalid timestamps
 
         if start_date <= log_timestamp <= end_date:
             logs_list.append(log)
@@ -252,7 +265,7 @@ async def delete_chat_history(
 
     # Attempt to delete the document
     delete_result = chat_collection.delete_one({"titleUId": titleUId, "userID": userID})
-    
+
     if delete_result.deleted_count == 0:
         raise HTTPException(status_code=404, detail="No chat history found to delete")
 

test.py

@@ -0,0 +1,11 @@
+from pymongo import MongoClient
+
+uri = "mongodb+srv://soham:Soham2001@potholedetection.gt3rb.mongodb.net/?retryWrites=true&w=majority&appName=potholeDetection"
+
+try:
+    client = MongoClient(uri)
+    db = client.admin  # Use the "admin" database for the ping command
+    result = db.command("ping")
+    print("Ping successful:", result)
+except Exception as e:
+    print("Ping failed:", e)

utils/download_mongodb_logs.py

@@ -1,51 +0,0 @@
-from pymongo import MongoClient
-from cryptography.fernet import Fernet
-import json
-from datetime import datetime, timedelta
-
-# MongoDB connection string
-connection_string = "mongodb+srv://soham2000:soham2000@cluster0.lbu4i.mongodb.net/?retryWrites=true&w=majority&appName=Cluster0"
-
-# Encryption key (generate this once and store securely)
-encryption_key = b"kFRvFTtpl6WDr0eceOteS5IAnv3ps0YtCavmUwllO0k="
-cipher = Fernet(encryption_key)
-
-client = MongoClient(connection_string)
-db = client["nodetest"] 
-collection = db["logs"] 
-logs_list = []
-# Calculate the date range based on the selected duration
-def download_logs(duration):
-    # Get the current date
-    current_date = datetime.now()
-
-    # Calculate the start and end date based on the selected duration
-    if duration == "1 day":
-        start_date = current_date - timedelta(days=1)
-        end_date = current_date
-    elif duration == "1 week":
-        start_date = current_date - timedelta(weeks=1)
-        end_date = current_date
-    elif duration == "2 weeks":
-        start_date = current_date - timedelta(weeks=2)
-        end_date = current_date
-    else:
-        raise ValueError("Invalid duration selected")
-
-    # Decrypt and filter logs based on the calculated date range
-    for encrypted_record in collection.find():
-        encrypted_data = encrypted_record["encrypted_data"]
-        decrypted_data = cipher.decrypt(encrypted_data).decode()
-        log = json.loads(decrypted_data)
-
-        # Parse the Timestamp field into a datetime object
-        log_timestamp = datetime.fromisoformat(log["Timestamp"])
-
-        # Check if the log falls within the specified date range
-        if start_date <= log_timestamp <= end_date:
-            logs_list.append(log)
-
-    return logs_list
-
-
-download_logs("1 week")

utils/email_validator.py

@@ -1,37 +0,0 @@
-from email.mime.text import MIMEText
-from email.mime.multipart import MIMEMultipart
-import smtplib
-
-# Function to send activation email with HTML content
-def send_activation_email(email: str, activation_otp: int):
-    sender_email = "chandratresoham@gmail.com"  # Update with your email address
-    sender_password = "wbuc okcv hzzn iwyx"  # Update with your email password
-  # Update with your website URL
-    
-    # HTML content for the email body
-    email_body = f"""
-    <html>
-        <body>
-            <p>
-                Hello,<br><br>
-                You have successfully registered to the system.<br><br>
-                Please enter below otp to verify your accout<br><br>
-                <p style="font-size:17px; font-weight:bold">{activation_otp}</p><br><br>
-                Thank you!<br>
-            </p>
-        </body>
-    </html>
-    """
-
-    # Create MIMEText object with HTML content
-    message = MIMEMultipart("alternative")
-    message['From'] = sender_email
-    message['To'] = email
-    message['Subject'] = "Activate your account"
-    message.attach(MIMEText(email_body, 'html'))
-
-    # Connect to SMTP server and send email
-    with smtplib.SMTP('smtp.gmail.com', 587) as server:  # Update with your SMTP server details
-        server.starttls()
-        server.login(sender_email, sender_password)
-        server.sendmail(sender_email, email, message.as_string())

utils/logs_generator.py

@@ -0,0 +1,80 @@
+import json
+from datetime import datetime, timedelta
+import random
+from faker import Faker
+
+# Initialize Faker instance
+fake = Faker()
+
+# Sample data
+actions = ["LOGIN", "LOGOUT", "SELECT", "UPDATE", "DELETE"]
+objects_accessed = [
+    "System Configurations",
+    "Employee Records",
+    "Order Details",
+    "User Profiles",
+    "Transaction Logs",
+    "Server Logs",
+    "Product Catalog",
+    "Payment Information",
+    "Customer Feedback",
+    "Internal Reports",
+    "Audit Trails",
+    "Access Logs",
+    "Security Policies",
+    "Employee Attendance",
+    "Network Configurations",
+    "HR Records",
+    "Financial Statements",
+    "Inventory Data",
+]
+status_options = ["Success", "Failure"]
+
+# Function to generate a random IP address
+def generate_ip():
+    return f"{random.randint(1, 255)}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(0, 255)}"
+
+# Function to generate a random timestamp within a specific day
+def generate_random_timestamp(date):
+    random_time = timedelta(seconds=random.randint(0, 86400))  # Random seconds in a day
+    return (date + random_time).isoformat()
+
+# Function to generate log entry
+def generate_log_entry(date):
+    timestamp = generate_random_timestamp(date)
+    user = fake.name()
+    action = random.choice(actions)
+    object_accessed = random.choice(objects_accessed)
+    ip_address = generate_ip()
+    status = random.choice(status_options)
+    
+    return {
+        "timestamp": timestamp,
+        "user": user,
+        "action": action,
+        "objectAccessed": object_accessed,
+        "ipAddress": ip_address,
+        "status": status,
+    }
+
+# Define your start and end date
+start_date = datetime(2025, 1, 1)
+end_date = datetime(2025, 8, 2)
+
+# Generate logs per day
+log_entries = []
+current_date = start_date
+while current_date <= end_date:
+    logs_per_day = random.randint(15, 20)  # Generate 15 to 20 logs per day
+    for _ in range(logs_per_day):
+        log_entries.append(generate_log_entry(current_date))
+    current_date += timedelta(days=1)  # Move to the next day
+
+# Define the path to the JSON file
+file_path = "audit_logs.json"
+
+# Write the generated logs to the file
+with open(file_path, "w") as json_file:
+    json.dump(log_entries, json_file, indent=4)
+
+print(f"Logs have been written to {file_path}")

utils/logstomongodb.py

@@ -0,0 +1,33 @@
+from pymongo import MongoClient
+from cryptography.fernet import Fernet
+import json
+
+# MongoDB connection string
+connection_string = "mongodb+srv://soham2000:soham2000@cluster0.lbu4i.mongodb.net/?retryWrites=true&w=majority&appName=Cluster0"
+
+# Encryption key (generate this once and store securely)
+# encryption_key = Fernet.generate_key()
+cipher = Fernet(b"kFRvFTtpl6WDr0eceOteS5IAnv3ps0YtCavmUwllO0k=")
+
+# Replace <db_password> with your actual password in the connection string
+client = MongoClient(connection_string)
+db = client['nodetest']  # Database name
+collection = db['logs']  # Collection name
+
+file_path = "audit_logs.json"  # Path to your logs file
+with open(file_path, 'r') as file:
+    logs = json.load(file)
+
+# Encrypt and insert logs
+for log in logs:
+    log_json = json.dumps(log)  # Convert the log object to JSON string
+    encrypted_log = cipher.encrypt(log_json.encode())  # Encrypt the JSON string
+    collection.insert_one({"encrypted_data": encrypted_log})  # Insert encrypted log
+
+print("Logs have been encrypted and inserted into the database.")
+
+# Decrypt a log (for testing)
+for encrypted_record in collection.find():
+    encrypted_data = encrypted_record['encrypted_data']
+    decrypted_data = cipher.decrypt(encrypted_data).decode()
+    print("Decrypted Log:", json.loads(decrypted_data))