added some new endpoint

backend/app/api/endpoints.py

@@ -1,4 +1,5 @@
 
+from pathlib import Path
 from fastapi import APIRouter, HTTPException, UploadFile, File, Depends
 from typing import List, Optional
 from pydantic import BaseModel
@@ -14,12 +15,20 @@ from app.services.email_reader import parse_email_bytes, read_emails_from_direct
 from app.services.duplicate_checker import check_duplicate
 from app.models.email_model import EmailData
 from app.models.request_type_model import RequestTypeModel
-from app.services.gemeni_classification import analyze_intent, classify_email_gemeni, extract_text_from_attachment, get_primary_intent
 from app.services.retrieve_email_process import process_single_email
 from config import settings
 
 router = APIRouter()
 
+# Define the directory where attachments will be saved
+SAVE_DIR = Path("data/attachments")
+SAVE_FILE_PATH = SAVE_DIR / settings.settings.ALLOWED_PRIORITY_RULES_FILENAME
+
+if not SAVE_DIR.exists():
+    SAVE_DIR.mkdir(parents=True)
+    print(f"Directory created: {SAVE_DIR}")
+else:
+    print(f"Directory already exists: {SAVE_DIR}")
 
 @router.post("/process-emails-upload/", response_model=List[EmailData])
 async def process_email_files(files: List[UploadFile] = File(...)):
@@ -37,6 +46,7 @@ async def process_email_files(files: List[UploadFile] = File(...)):
                     sub_request_type=email_result["sub_request_type"],
                     confidence_score=email_result["confidence_score"],
                     duplicate_flag=email_result["duplicate_flag"],
+                    all_extracted_numbers=email_result["extracted_numbers_list"]
                 )
                 results.append(email_resp)
         except Exception as e:
@@ -80,3 +90,54 @@ async def process_email_directory():
         except Exception as e:
             print(f"Error processing {file_path}: {e}")
     return results
+
+
+@router.post("/upload-priority-rules/",
+             description="""
+Upload a JSON file containing the priority rules for request type identification and numerical extraction.
+**Sample JSON file structure:**
+
+```json
+{
+  "priority_rules": {
+    "is_prioritization_extraction": true,
+    "request_type_identification": {
+      "order": ["email_content", "document_content"],
+      "fallback": "document_content"
+    },
+    "numerical_field_extraction": {
+      "preferred_source": ["attachments"],
+      "fallback": "email_body"
+    },
+    "banking_numeric_keys": [
+      "loan_amount",
+      "balance",
+      "interest_rate"
+      .......
+    ]
+  }
+}"
+"""
+)
+async def upload_rules(file: UploadFile = File(...)):
+
+    # Validate filename
+    if file.filename != settings.settings.ALLOWED_PRIORITY_RULES_FILENAME:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid file name. Please use the appropriate file name: {settings.settings.ALLOWED_PRIORITY_RULES_FILENAME}"
+        )
+    
+   # Check if file is a JSON
+    if file.content_type != "application/json":
+        raise HTTPException(status_code=400, detail="Only JSON files are allowed.")
+
+    # Save file to the specified location
+    try:
+        file_content = await file.read()
+        with open(SAVE_FILE_PATH, "wb") as f:
+            f.write(file_content)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error saving file: {e}")
+    
+    return {"message": "Rules JSON uploaded successfully."}

backend/app/models/email_model.py

@@ -1,5 +1,5 @@
 from pydantic import BaseModel
-from typing import List, Optional
+from typing import Any, Dict, List, Optional
 
 class EmailData(BaseModel):
     sender: str
@@ -8,3 +8,4 @@ class EmailData(BaseModel):
     sub_request_type: Optional[str] = None
     confidence_score: Optional[float] = None
     duplicate_flag: bool = False
+    all_extracted_numbers: Optional[List[Dict[str, Any]]] = None

backend/app/services/gemeni_classification.py

@@ -1,4 +1,5 @@
-from config.settings import Settings
+import re
+from typing import List, Optional
 import google.generativeai as genai
 import json
 import email
@@ -6,6 +7,8 @@ import io
 import PyPDF2
 import docx
 import mimetypes
+
+from config.settings import Settings
    
 
 genai.configure(api_key=Settings.GEMENI_API_KEY_TOKEN)
@@ -97,7 +100,6 @@ def analyze_intent(text):
         print(f"Gemini API error (Intent): {e}")
         return ""
 
-
 def classify_email_gemeni(subject, body):
     """Classifies an email based on request type and sub-request type."""
     results = []
@@ -119,8 +121,6 @@ def classify_email_gemeni(subject, body):
     }}
     """
    
-
-    
     response = model.generate_content(PROMPT)
 
     if response and hasattr(response, "_result"):
@@ -158,4 +158,59 @@ def get_primary_intent(email_content, attach_content):
     return response.text  # Extracted primary intent
 
 
+def extract_key_number_with_llm(context: str, rules: Optional[dict] = None) -> Optional[str]:
+    """
+    Uses the Gemini LLM to extract the most appropriate numerical value(s)
+    from the provided context. The prompt instructs the model to return a 
+    JSON array of objects if more than one key is found. Each object should
+    have the banking key as the key and the corresponding numerical value as its value.
+    
+    :param context: The text context from which key numerical value(s) should be extracted.
+    :param rules: Optional rules dictionary. If not provided, it will be loaded.
+    :return: The extracted key numerical value(s) as a JSON array (list of dicts), or None if not found.
+    """
+        
+    banking_keys = rules.get("priority_rules", {}).get("banking_numeric_keys", [])
+    keys_str = ", ".join(banking_keys)
+    
+    # Construct the prompt instructing the model to return a JSON array
+    prompt = (
+        f"Analyze the following context and extract all relevant numerical values that represent key banking references. "
+        f"Consider the following banking numeric keys as potential candidates: {keys_str}. "
+        f"If you find more than one, return them in a JSON array. Each element in the array should be an object with "
+        f"the banking key as the key and the corresponding number as its value. If only one is found, still return a JSON array "
+        f"with a single object. Return only the JSON array with no additional text.\n\n"
+        f"Context: {context}"
+    )
+    
+    try:
+        response = model.generate_content(prompt)
+        if response and hasattr(response, "_result"):
+            text_response = response._result.candidates[0].content.parts[0].text.strip()
+            # Clean the JSON response by removing code fences if they exist.
+            cleaned_response = clean_json_response(text_response)
+            try:
+                # Validate that the cleaned response is valid JSON.
+                parsed = json.loads(cleaned_response)
+                # Re-serialize the parsed object to get a standardized JSON string.
+                return json.dumps(parsed)
+            except Exception as json_err:
+                print(f"JSON parsing error: {json_err}")
+                print("Cleaned Response received:", cleaned_response)
+                return None
+    except Exception as e:
+        print(f"Error during LLM extraction: {e}")
+    
+    return None
+    
 
+def clean_json_response(text: str) -> str:
+    """
+    Removes markdown code fences and any extra backticks from the response.
+    """
+    # Remove leading and trailing backticks or code fence markers
+    # Remove a leading "```json" if present
+    text = re.sub(r"^```json", "", text).strip()
+    # Remove trailing "```" if present
+    text = re.sub(r"```$", "", text).strip()
+    return text

backend/app/services/retrieve_email_process.py

@@ -1,41 +1,96 @@
 
-from typing import Optional
+import json
+import os
+from pathlib import Path
+import re
+from typing import List, Optional
 from app.services.duplicate_checker import check_duplicate
 from app.services.email_reader import parse_email_bytes
-from app.services.gemeni_classification import classify_email_gemeni, extract_text_from_attachment
+from app.services.gemeni_classification import classify_email_gemeni, extract_key_number_with_llm, extract_text_from_attachment
+from config import settings
 
 
 async def process_single_email(file_content: bytes, filename: str) -> Optional[dict]:
     """Processes a single email content."""
+    # Load customizable priority rules
+    rules = load_priority_rules()  # Expected to return a dict with "priority_rules" key
+    priority_config = rules.get("priority_rules", {})
+    use_priority = priority_config.get("is_prioritization_extraction", False)
+    all_extracted_numbers = []  # This will hold the combined results from all attachments
+
+    # Parse the email into its components
     email_data = parse_email_bytes(file_content, filename)
     if email_data:
-        attachment_text = ""
-        for attachment in email_data["attachments"]:
-            attachment_text += extract_text_from_attachment(attachment["content"], attachment["filename"])
+        # attachment_text = ""
+        # for attachment in email_data["attachments"]:
+        #     attachment_text += extract_text_from_attachment(attachment["content"], attachment["filename"])
 
         email_chain_text = email_data["email_chain_text"]
         email_body_text = email_data["body"]
 
-        # 1. Separate Classification:
-        document_result = classify_email_gemeni(email_data["subject"], attachment_text) if attachment_text else ("Unknown", "Unknown", "0")
-        email_chain_result = classify_email_gemeni(email_data["subject"], email_chain_text) if email_chain_text else ("Unknown", "Unknown", "0")
-        primary_email_result = classify_email_gemeni(email_data["subject"], email_body_text)
+        # Process attachments: extract text and numerical fields
+        attachment_text = ""
+        extracted_numbers = []
+        for attachment in email_data.get("attachments", []):
+            text = extract_text_from_attachment(attachment["content"], attachment["filename"])
+            attachment_text += text + "\n"
+            # extracted_numbers.extend(extract_key_number_with_llm(text,rules))
+            extracted_numbers_json = extract_key_number_with_llm(text, rules)
+            if extracted_numbers_json:
+                try:
+                    parsed_result = json.loads(extracted_numbers_json)  # Convert JSON string to Python object
+                    # Check if the parsed result is a list; if so, merge it into our overall list.
+                    if isinstance(parsed_result, list):
+                        all_extracted_numbers.extend(parsed_result)
+                    else:
+                        all_extracted_numbers.append(parsed_result)
+                except Exception as e:
+                    print("Error parsing JSON result from LLM:", e)
+
+        # Choose classification logic based on priority configuration
+        if use_priority:
+            # -- Priority Based Extraction Logic --
+            identification_order = priority_config.get("request_type_identification", {}).get("order", [])
+            classification_source = ""
+            if "email_content" in identification_order and email_body_text.strip():
+                classification_source = email_body_text
+            elif "document_content" in identification_order and attachment_text.strip():
+                classification_source = attachment_text
+            else:
+                classification_source = email_body_text  # default fallback
+            
+            primary_result = classify_email_gemeni(email_data["subject"], classification_source)
+
+            # special condition to check if priority is email content and email has multi thread then we 
+            # have to compare confidence score with primary confidence score 
+            if "email_content" in identification_order:
+                email_chain_result = classify_email_gemeni(email_data["subject"], email_chain_text) if email_chain_text else ("Unknown", "Unknown", "0")
+                email_chain_confidence = float(email_chain_result[2])
+                primary_email_confidence = float(primary_result[2])
+                if email_chain_confidence > primary_email_confidence and email_chain_confidence:
+                    primary_result  = email_chain_result
 
-        # 2. Confidence Score Comparison:
-        document_confidence = float(document_result[2])
-        email_chain_confidence = float(email_chain_result[2])
-        primary_email_confidence = float(primary_email_result[2])
+        else:
+            # 1. Separate Classification:
+            document_result = classify_email_gemeni(email_data["subject"], attachment_text) if attachment_text else ("Unknown", "Unknown", "0")
+            email_chain_result = classify_email_gemeni(email_data["subject"], email_chain_text) if email_chain_text else ("Unknown", "Unknown", "0")
+            primary_email_result = classify_email_gemeni(email_data["subject"], email_body_text)
 
-        best_result = primary_email_result  # Default to email body
-        if document_confidence > primary_email_confidence and document_confidence > email_chain_confidence:
-            best_result = document_result
-        elif email_chain_confidence > primary_email_confidence and email_chain_confidence > document_confidence:
-            best_result = email_chain_result
+            # 2. Confidence Score Comparison:
+            document_confidence = float(document_result[2])
+            email_chain_confidence = float(email_chain_result[2])
+            primary_email_confidence = float(primary_email_result[2])
 
-        # 3. Refined Classification:
-        request_type = best_result[0]
-        sub_request_type = best_result[1]
-        confidence_score = best_result[2]
+            primary_result  = primary_email_result  # Default to email body
+            if document_confidence > primary_email_confidence and document_confidence > email_chain_confidence:
+                primary_result  = document_result
+            elif email_chain_confidence > primary_email_confidence and email_chain_confidence > document_confidence:
+                primary_result  = email_chain_result
+
+        # Extract classification results
+        request_type = primary_result[0]
+        sub_request_type = primary_result[1]
+        confidence_score = primary_result[2]
 
         duplicate_flag, duplicate_reason = check_duplicate(email_data["body"])
         email_obj = {
@@ -45,8 +100,56 @@ async def process_single_email(file_content: bytes, filename: str) -> Optional[d
             "sub_request_type": sub_request_type,
             "confidence_score": confidence_score,
             "duplicate_flag": duplicate_flag,
+            "extracted_numbers_list": all_extracted_numbers
         }
         return email_obj
     else:
         print(f"Parsing failed for file: {filename}")
-        return None
+        return None
+    
+
+def load_priority_rules() -> dict:
+    """
+    Loads the priority rules from a JSON file.
+    The JSON file should be located at 'config/rules.json'.
+    If the file does not exist, a default rules dictionary is returned.
+    
+    Expected JSON structure example:
+    {
+      "priority_rules": {
+        "is_prioritization_extraction": true,
+        "request_type_identification": {
+          "order": ["email_content", "document_content"],
+          "fallback": "document_content"
+        },
+        "numerical_field_extraction": {
+          "preferred_source": ["attachments"],
+          "fallback": "email_body"
+        }
+      }
+    }
+    """
+    # Get the rules directory and filename from environment variables
+    RULES_FILENAME =  settings.settings.ALLOWED_PRIORITY_RULES_FILENAME
+    RULES_DIR = Path("data/attachments")
+    # Build the full file path using pathlib
+    RULES_FILE_PATH = Path(RULES_DIR) / RULES_FILENAME
+    
+    if os.path.exists(RULES_FILE_PATH):
+        with open(RULES_FILE_PATH, "r") as file:
+            return json.load(file)
+    
+    # Return default rules if file doesn't exist
+    return {
+        "priority_rules": {
+            "is_prioritization_extraction": False,
+            "request_type_identification": {
+                "order": ["email_content", "document_content"],
+                "fallback": "document_content"
+            },
+            "numerical_field_extraction": {
+                "preferred_source": ["attachments"],
+                "fallback": "email_body"
+            }
+        }
+    }

backend/config/settings.py

@@ -9,15 +9,14 @@ class Settings:
     ENV = os.getenv("ENV")  # Default to "local" if ENV is not set
     # API Token
     HUGGINGFACE_API_TOKEN = os.getenv("HUGGINGFACE_API_TOKEN") or os.environ.get("HUGGINGFACE_API_TOKEN")
-    
-    GEMENI_API_KEY_TOKEN = os.getenv("GEMENI_API_KEY") or os.environ.get("GEMENI_API_KEY")
 
+    GEMENI_API_KEY_TOKEN = os.getenv("GEMENI_API_KEY") or os.environ.get("GEMENI_API_KEY")
 
     # Model Path
     MODEL_NAME = os.getenv("MODEL_NAME") or os.environ.get("MODEL_NAME") # Default if not set
     OCR_LANGUAGE = os.getenv("OCR_LANGUAGE", "eng")
     directory_path = os.getenv("EMAIL_DIRECTORY_PATH") or os.environ.get("EMAIL_DIRECTORY_PATH")  # Make configurable
-    
+    ALLOWED_PRIORITY_RULES_FILENAME=os.getenv("ALLOWED_PRIORITY_RULES_FILENAME") or os.environ.get("ALLOWED_PRIORITY_RULES_FILENAME")  # Make configurable
     # Debugging info
     print(f"Running in {ENV} mode with model path: {MODEL_NAME}")
     print(f"Running in {ENV} mode with TOKEN: {HUGGINGFACE_API_TOKEN}")