Upload 2 files

app.py

@@ -0,0 +1,285 @@
+import streamlit as st
+import pandas as pd
+import os
+import logging
+import re
+import uuid
+from chromadb import PersistentClient
+from sentence_transformers import SentenceTransformer
+from langchain_groq import ChatGroq
+from rag_utils_updated import extract_text, preprocess_text, get_embeddings, is_image_pdf, assess_cv, extract_job_requirements
+import plotly.graph_objects as go
+from dotenv import load_dotenv
+
+# Logging setup
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+logger = logging.getLogger(__name__)
+
+load_dotenv()
+
+if os.environ.get("LLM_PROMPT") is None:
+    st.error("LLM_PROMPT is missing. Check your .env file!")
+if os.environ.get("ADMIN_PASSWORD") is None:
+    st.error("ADMIN_PASSWORD is missing. Check your .env file!")
+
+st.title("CV Assessment and Ranking App")
+
+# Generate a unique session ID for temporary sessions
+if "session_id" not in st.session_state:
+    st.session_state.session_id = str(uuid.uuid4())[:8]  # Short unique session ID
+
+# Initialize session state variables
+for key in ["job_description", "requirements", "detailed_assessments", "cvs", "job_description_embedding"]:
+    if key not in st.session_state:
+        st.session_state[key] = None if key in ["job_description", "requirements", "job_description_embedding"] else {}
+if "assessment_completed" not in st.session_state:
+    st.session_state.assessment_completed = False
+if "admin_logged_in" not in st.session_state:
+    st.session_state.admin_logged_in = False
+
+# Persistent Storage for Embeddings
+PERMANENT_DB_PATH = "./cv_db"
+db_client = PersistentClient(path=PERMANENT_DB_PATH)
+st.session_state.collection = db_client.get_or_create_collection(f"cv_embeddings_{st.session_state.session_id}")
+
+if "embedding_model" not in st.session_state:
+    st.session_state.embedding_model = SentenceTransformer('all-mpnet-base-v2')
+if "groq_client" not in st.session_state:
+    st.session_state.groq_client = ChatGroq(api_key=os.environ.get("GROQ_API_KEY"))
+
+def clear_chroma_db():
+    """Clears only the current session's embeddings."""
+    try:
+        st.session_state.collection.delete(where={"session_id": st.session_state.session_id})  # Delete only this session's embeddings
+        st.info("Session-specific embeddings cleared. Starting fresh!")
+    except Exception as e:
+        st.error(f"Error clearing session embeddings: {e}")
+        st.stop()
+
+# Ensure the session clears its own embeddings on startup
+clear_chroma_db()
+
+import shutil
+
+def clear_all_sessions_data():
+    """Admin function to delete old session embeddings and reclaim disk space while keeping active sessions."""
+    try:
+        global db_client
+        existing_collections = db_client.list_collections()
+        
+        # Identify active sessions (all currently running session IDs)
+        active_sessions = [f"cv_embeddings_{st.session_state.session_id}"]
+
+        # Delete all collections except currently active ones
+        for collection_name in existing_collections:
+            if collection_name not in active_sessions:
+                db_client.delete_collection(collection_name)  # Delete only old session data
+        
+        # Force database compaction to free up space
+        db_client = None  # Close database connection
+        shutil.rmtree(PERMANENT_DB_PATH)  # Delete database folder
+        os.makedirs(PERMANENT_DB_PATH, exist_ok=True)  # Recreate empty database
+        
+        db_client = PersistentClient(path=PERMANENT_DB_PATH)  # Reinitialize database
+
+        st.success("Old session embeddings deleted. Active sessions retained. Database size optimized.")
+    except Exception as e:
+        st.error(f"Error deleting old session data: {e}")
+
+
+# Admin Panel for Clearing Old Data
+with st.sidebar:
+    st.subheader("Admin Login")
+    admin_user = st.text_input("Username", key="admin_user")
+    admin_pass = st.text_input("Password", type="password", key="admin_pass")
+    if st.button("Login as Admin"):
+        if admin_user == "admin" and admin_pass == os.environ.get("ADMIN_PASSWORD"):
+            st.session_state.admin_logged_in = True
+            st.success("Admin login successful!")
+        else:
+            st.error("Invalid credentials. Access denied.")
+    
+    if st.session_state.admin_logged_in:
+        st.subheader("Admin Actions")
+        if st.button("Clear All Stored Embeddings"):
+            clear_all_sessions_data()
+
+def process_cv(uploaded_file):
+    """Processes a single CV file: extracts text, preprocesses, and stores embeddings with a session ID."""
+    filename = uploaded_file.name
+    session_filename = f"{st.session_state.session_id}_{filename}"  # Unique per session
+
+    try:
+        if is_image_pdf(uploaded_file):
+            st.warning(f"{filename} appears to be an image-based PDF and cannot be processed.")
+            return None
+
+        text = extract_text(uploaded_file)
+        preprocessed_text = preprocess_text(text)
+        embedding = get_embeddings(preprocessed_text, st.session_state.embedding_model)
+
+        st.session_state.collection.add(
+            embeddings=[embedding],
+            documents=[preprocessed_text],
+            ids=[session_filename],  # Store session-unique ID
+            metadatas=[{"session_id": st.session_state.session_id, "filename": filename}]
+        )
+        return {"text": preprocessed_text, "embedding": embedding, "session_filename": session_filename}
+    except Exception as e:
+        st.error(f"Failed to process {filename}: {e}")
+        return None
+
+def parse_assessment(raw_response, requirements):
+    """Parses the LLM's assessment with robust error handling."""
+    matches = {
+        "technical_lead": "Not Found", 
+        "hr_specialist": "Not Found",
+        "project_manager": "Not Found",
+        "final_assessment": "Not Found",
+        "recommendation": "Not Found",
+        "technical_lead_score": "Not Found",
+        "hr_specialist_score": "Not Found",
+        "project_manager_score": "Not Found",
+        "final_assessment_score": "Not Found",
+    }
+    
+    try:
+        technical_lead_match = re.search(r"Technical Lead Assessment:\s*(.*?)\s*Technical Lead Score:\s*(\d+)", raw_response, re.IGNORECASE | re.DOTALL)
+        if technical_lead_match:
+            matches["technical_lead"] = technical_lead_match.group(1).strip()
+            matches["technical_lead_score"] = technical_lead_match.group(2)
+
+        hr_specialist_match = re.search(r"HR Specialist Assessment:\s*(.*?)\s*HR Specialist Score:\s*(\d+)", raw_response, re.IGNORECASE | re.DOTALL)
+        if hr_specialist_match:
+            matches["hr_specialist"] = hr_specialist_match.group(1).strip()
+            matches["hr_specialist_score"] = hr_specialist_match.group(2)
+
+        project_manager_match = re.search(r"Project Manager Assessment:\s*(.*?)\s*Project Manager Score:\s*(\d+)", raw_response, re.IGNORECASE | re.DOTALL)
+        if project_manager_match:
+            matches["project_manager"] = project_manager_match.group(1).strip()
+            matches["project_manager_score"] = project_manager_match.group(2)
+
+        final_assessment_match = re.search(r"Final Assessment:\s*(.*?)\s*Final Assessment Score:\s*(\d+)", raw_response, re.IGNORECASE | re.DOTALL)
+        if final_assessment_match:
+            matches["final_assessment"] = final_assessment_match.group(1).strip()
+            matches["final_assessment_score"] = final_assessment_match.group(2)
+
+        recommendation_match = re.search(r"Recommendation:\s*(.*?)$", raw_response, re.IGNORECASE | re.DOTALL)
+        if recommendation_match:
+            matches["recommendation"] = recommendation_match.group(1).strip()
+    except Exception as e:
+        print(f"Error parsing assessment: {e}")
+    
+    return matches
+
+# 1. Input Job Description
+st.subheader("Enter Job Description")
+requirements_source = st.radio("Source:", ("File Upload", "Web Page Link", "Text Input"))
+
+if requirements_source == "File Upload":
+    uploaded_file = st.file_uploader("Upload Job Requirements (PDF/DOCX)", type=["pdf", "docx"])
+    if uploaded_file:
+        st.session_state.job_description = extract_text(uploaded_file)
+elif requirements_source == "Text Input":
+    st.session_state.job_description = st.text_area("Enter Job Requirements", height=200)
+
+if st.session_state.job_description:
+    st.success("Job description uploaded successfully!")
+    if st.session_state.job_description_embedding is None:
+        st.session_state.job_description_embedding = get_embeddings(st.session_state.job_description, st.session_state.embedding_model)
+    if not st.session_state.requirements:
+        st.session_state.requirements = extract_job_requirements(st.session_state.job_description, st.session_state.groq_client)
+        if st.session_state.requirements:
+            with st.expander("Extracted Job Requirements:"):
+                for req in st.session_state.requirements:
+                    st.write(f"- {req}")
+
+# 2. Upload CVs
+st.subheader("Upload CVs (Folder)")
+uploaded_files = st.file_uploader("Choose CV files", accept_multiple_files=True)
+
+if uploaded_files and not st.session_state.assessment_completed:
+    with st.spinner("Processing uploaded CVs, please wait..."):
+        st.write(f"{len(uploaded_files)} CV(s) uploaded.")
+        st.session_state.cvs = {}
+
+        for uploaded_file in uploaded_files:
+            result = process_cv(uploaded_file)
+            if result:
+                st.session_state.cvs[result["session_filename"]] = result
+
+        st.success("CV embeddings created successfully!")
+    st.session_state.assessment_completed = True
+
+# Perform detailed assessments automatically
+if st.session_state.assessment_completed:
+    st.write("Performing detailed assessments...")
+    detailed_assessments = st.session_state.detailed_assessments  # Store reference for efficiency
+    if not detailed_assessments:
+        with st.spinner("Assessing CVs..."):
+            for filename, cv_data in st.session_state.cvs.items():
+                try:
+                    assessment = assess_cv(cv_data["text"], st.session_state.requirements, filename, st.session_state.groq_client)
+                    detailed_assessments[filename] = assessment
+                except Exception as e:
+                    st.error(f"Error assessing {filename}: {e}")
+    st.success("Detailed assessments complete!")
+
+
+
+    st.subheader("Candidates Assessment and Ranking")
+    assessments_df = pd.DataFrame([{**parse_assessment(a["raw_response"], st.session_state.requirements), "filename": f} for f, a in st.session_state.detailed_assessments.items()])
+    assessments_df = assessments_df.sort_values(by='final_assessment_score', ascending=False)
+    st.dataframe(assessments_df)
+
+    st.subheader("Detailed Assessment Results")
+    
+    # Iterate through the DataFrame rows to display the UI for each assessment
+    for index, row in assessments_df.iterrows():
+        st.write(f"**Filename:** {row['filename']}")
+        scores = {
+            "Technical Lead": int(row["technical_lead_score"]),
+            "HR Specialist": int(row["hr_specialist_score"]),
+            "Project Manager": int(row["project_manager_score"]),
+            "Final Assessment": int(row["final_assessment_score"]),
+        }
+        scores_df = pd.DataFrame(list(scores.items()), columns=["Expert", "Score"])
+
+        # Create Plotly bar chart with annotations
+        fig = go.Figure(data=[go.Bar(
+            x=scores_df["Expert"],
+            y=scores_df["Score"],
+            text=scores_df["Score"],
+            textposition='auto',
+        )])
+        fig.update_layout(yaxis_range=[0, 100])
+
+        # Create columns layout
+        col1, col2 = st.columns([1, 3])
+
+        # Display bar chart in the first column with a unique key
+        with col1:
+            st.plotly_chart(fig, use_container_width=True, key=f"chart_{index}")
+
+        # Display collapsed panels in the second column
+        with col2:
+            with st.expander("Technical Lead Assessment"):
+                st.write(f"{row['technical_lead']}")
+                st.write(f"**Technical Lead Score:** {row['technical_lead_score']}")
+
+            with st.expander("HR Specialist Assessment"):
+                st.write(f"{row['hr_specialist']}")
+                st.write(f"**HR Specialist Score:** {row['hr_specialist_score']}")
+
+            with st.expander("Project Manager Assessment"):
+                st.write(f"{row['project_manager']}")
+                st.write(f"**Project Manager Score:** {row['project_manager_score']}")
+
+            with st.expander("Final Assessment"):
+                st.write(f"{row['final_assessment']}")
+                st.write(f"**Final Assessment Score:** {row['final_assessment_score']}")
+
+            with st.expander("Recommendation"):
+                st.write(f"{row['recommendation']}")
+
+        st.write("---")

rag_utils_updated.py

@@ -0,0 +1,185 @@
+import os
+import logging
+import requests
+import json
+import PyPDF2
+import docx
+from bs4 import BeautifulSoup
+from chromadb import PersistentClient
+from langchain_groq import ChatGroq
+from langchain.prompts import ChatPromptTemplate
+from langchain.output_parsers import PydanticOutputParser
+from pydantic import BaseModel, Field, ValidationError
+from typing import List
+from sentence_transformers import SentenceTransformer  # Import SentenceTransformer
+from dotenv import load_dotenv
+
+# Setup logging
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+logger = logging.getLogger(__name__)
+
+# --- Text Extraction ---
+def extract_text(uploaded_file):
+    try:
+        if isinstance(uploaded_file, str):
+            return extract_text_from_webpage(uploaded_file)
+        elif hasattr(uploaded_file, 'type') and uploaded_file.type == "application/pdf":
+            if is_image_pdf(uploaded_file):
+                logger.warning(f"Image-based PDF detected: {uploaded_file.name}")
+                return ""  # Skip processing
+            return extract_text_from_pdf(uploaded_file)
+        elif hasattr(uploaded_file, 'type') and uploaded_file.type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
+            return extract_text_from_docx(uploaded_file)
+        return ""
+    except Exception as e:
+        logger.error(f"Error extracting text: {e}")
+        return ""
+
+def is_image_pdf(uploaded_file):
+    """Check if a PDF is image-based (contains no selectable text)."""
+    try:
+        reader = PyPDF2.PdfReader(uploaded_file)
+        for page in reader.pages:
+            if page.extract_text():
+                return False  # Text is present, so not an image PDF
+        return True  # No text detected, likely an image-based PDF
+    except Exception as e:
+        logger.error(f"Error checking if PDF is image-based: {e}")
+        return True  # Assume image PDF if error occurs
+
+def extract_text_from_pdf(uploaded_file):
+    try:
+        reader = PyPDF2.PdfReader(uploaded_file)
+        return "\n".join([page.extract_text() or "" for page in reader.pages])
+    except Exception as e:
+        logger.error(f"Error reading PDF {uploaded_file.name}: {e}")
+        return ""
+
+def extract_text_from_docx(uploaded_file):
+    try:
+        doc = docx.Document(uploaded_file)
+        return "\n".join([para.text for para in doc.paragraphs])
+    except Exception as e:
+        logger.error(f"Error reading DOCX: {e}")
+        return ""
+
+def extract_text_from_webpage(url):
+    try:
+        response = requests.get(url)
+        response.raise_for_status()
+        soup = BeautifulSoup(response.content, 'html.parser')
+        return soup.get_text(separator='\n')
+    except requests.exceptions.RequestException as e:
+        logger.error(f"Error fetching webpage: {e}")
+        return ""
+
+def preprocess_text(text):
+    return text.lower()
+
+def get_embeddings(text, model):
+    return model.encode(text)
+
+def get_similar_cvs(cvs, job_description_embedding, collection):
+    results = collection.query(
+        query_embeddings=[job_description_embedding],
+        n_results=len(cvs),
+        include=["distances", "metadatas"]
+    )
+
+    similar_cvs = []
+    for i in range(len(results['metadatas'][0])): # Corrected loop
+        metadata = results['metadatas'][0][i]
+        if metadata: #Check if metadata exists
+            filename = metadata.get('filename') # Use .get to handle missing keys
+            if filename: # Check if filename exists in metadata
+                similarity_score = 1 - results['distances'][0][i]
+                similar_cvs.append({
+                    "filename": filename,
+                    "initial_score": similarity_score
+                })
+            else:
+                logger.warning(f"Metadata for CV at index {i} is missing 'filename'.")
+        else:
+            logger.warning(f"No metadata found for CV at index {i}.")
+
+
+    similar_cvs.sort(key=lambda x: x['initial_score'], reverse=True)
+    return similar_cvs
+
+# ... (CV Assessment & Ranking functions)
+
+class RequirementAssessment(BaseModel):
+    requirement: str
+    match: str = Field(pattern="^(Yes|No|Partial|Not Applicable)$")
+    evidence: str
+    justification: str
+
+class CandidateAssessment(BaseModel):
+    filename: str
+    requirements: List[RequirementAssessment]
+    overall_assessment: str = Field(pattern="^(Excellent|Good|Fair|Poor)$")
+    recommendation: str = Field(pattern="^(Interview|Reject|Maybe)$")
+    justification: str
+
+
+import time
+import requests
+import json
+from pydantic import ValidationError
+
+
+def assess_cv(cv_text, requirements, filename, groq_client, max_retries=3, retry_delay=2):
+    """Assess CV against specific job requirements with Tree-of-Thoughts."""
+
+    requirements_str = "\n".join([f"- {req}" for req in requirements])
+    prompt_template = ChatPromptTemplate.from_template(
+
+        template = os.environ.get("LLM_PROMPT")
+
+    )
+    
+    prompt = prompt_template.format_messages(requirements=requirements_str, cv_text=cv_text)
+
+    # ... (rest of the assess_cv function remains the same)
+    for attempt in range(max_retries):
+        try:
+            response = groq_client.invoke(prompt, timeout=30)
+            response_content = response.content
+
+            return {"filename": filename, "raw_response": response_content}
+
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Network error during Groq API call: {e}")
+            if attempt == max_retries - 1:
+                return {"filename": filename, "error": "Network error during LLM call"}
+            else:
+                logger.warning(f"Network error, retrying in {retry_delay} seconds (attempt {attempt+1}/{max_retries}).")
+                time.sleep(retry_delay)
+                retry_delay *= 2
+
+        except Exception as e:
+            logger.error(f"Groq API error (attempt {attempt + 1}/{max_retries}): {e}")
+            if attempt == max_retries - 1:
+                return {"filename": filename, "error": "General LLM failure"}
+            else:
+                logger.warning(f"General LLM error, retrying in {retry_delay} seconds (attempt {attempt+1}/{max_retries}).")
+                time.sleep(retry_delay)
+                retry_delay *= 2
+
+    return {"filename": filename, "error": "LLM call failed after multiple retries."}
+
+def extract_job_requirements(job_description, groq_client):
+        """Extracts job requirements from the job description using the LLM."""
+        prompt_template = ChatPromptTemplate.from_template(
+            template="Extract the key job requirements from the following job description:\n\n{job_description}\n\nRequirements:"
+        )
+        prompt = prompt_template.format_messages(job_description=job_description)
+
+        try:
+            response = groq_client.invoke(prompt, timeout=30)
+            requirements_text = response.content.strip()
+            requirements = [req.strip() for req in requirements_text.split('\n') if req.strip()]
+            return requirements
+        except Exception as e:
+            logger.error(f"Error extracting job requirements: {e}")
+            return []