"""FastAPI server for the Resume Analysis and Matching System.""" import os import re import subprocess import json import shutil import subprocess import json import tempfile import uvicorn import ollama from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.middleware.cors import CORSMiddleware from sentence_transformers import SentenceTransformer from typing import List # Adjust imports to use the existing project structure from CHROMA_DB.collections import ChromaDBManager from main import extract_job_description, index_directory # --- App Initialization & Global Objects --- os.environ["TOKENIZERS_PARALLEILLISM"] = "false" print("Loading embedding model...") model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2") print("Model loaded.") main_chroma_manager = ChromaDBManager() app = FastAPI( title="Resume Analysis and Matching System", description="An API for matching resumes to job descriptions using a RAG architecture.", version="0.1.0", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # --- Business Logic --- def extract_structured_data(resume_text: str) -> dict: """ Extracts name, skills, and years of experience from resume text using regex. """ # Attempt to extract name from the first few lines # This is a simple pattern and might need refinement for various resume formats. name_pattern = r"^[A-Z][a-z]+(?: [A-Z][a-z]+(?: [A-Z][a-z]+)?)?" name_match = re.search(name_pattern, resume_text.split('\n')[0]) name = name_match.group(0) if name_match else "Unknown Candidate" skills_list = [ "python", "java", "c++", "c#", "javascript", "typescript", "react", "angular", "vue", "nodejs", "express", "django", "flask", "fastapi", "ruby", "rails", "php", "laravel", "sql", "mysql", "postgresql", "mongodb", "redis", "docker", "kubernetes", "aws", "azure", "gcp", "terraform", "ansible", "jenkins", "git", "jira", "scrum", "agile", "machine learning", "deep learning", "tensorflow", "pytorch", "scikit-learn", "pandas", "numpy", "data analysis", "data science", "natural language processing", "computer vision", "html", "css", "tailwind", "bootstrap" ] extracted_skills = [] for skill in skills_list: if re.search(r'\b' + re.escape(skill) + r'\b', resume_text, re.IGNORECASE): extracted_skills.append(skill.capitalize()) experience_pattern = r'(\d+\+?)\s*years? of experience' match = re.search(experience_pattern, resume_text, re.IGNORECASE) experience = match.group(1) + "+ years" if match else "Not specified" return { "name": name, "skills": list(set(extracted_skills)), "experience": experience } def summarize_matches_with_llm_api(job_text: str, matches: dict) -> str: """ Uses a local LLM via Ollama to generate a summary and returns it. If it fails, it returns a user-friendly error message. """ print("\n\n🤖 Generating AI Summary for Top Matches...") context = "" for i, (fname, match) in enumerate(matches.items(), 1): context += f"--- Resume {i}: {fname} ---\n" context += f"Relevance: {match['match_percentage']}%\n" context += f"Matching Section ({match['section_name']}):\n{match['text']}\n\n" prompt = f""" You are an expert HR assistant. Your task is to analyze the following resumes and provide a summary of why they are a good fit for the given job description. **Job Description:** {job_text} **Top Matching Resumes:** {context} **Your Task:** Based on the job description and the provided resume snippets, write a concise summary for each of the top 2-3 candidates. Highlight their key qualifications, relevant experience, and skills that align with the job requirements. Keep it brief and to the point. """ # Skip LLM summarization for now return "LLM summarization temporarily disabled." # --- API Endpoints --- @app.get("/api/status", tags=["Monitoring"]) async def get_status(): """A simple endpoint to confirm the API is running.""" return {"status": "ok", "message": "API is running."} @app.post("/api/match-resumes", tags=["Matching"]) async def match_resumes( job_description: UploadFile = File(...), resumes: List[UploadFile] = File(...) ): """ Upload a job description and resumes, perform on-the-fly indexing and matching, and return results. """ temp_dir = tempfile.mkdtemp() try: jd_path = os.path.join(temp_dir, job_description.filename) with open(jd_path, "wb") as buffer: shutil.copyfileobj(job_description.file, buffer) resumes_dir = os.path.join(temp_dir, "resumes") os.makedirs(resumes_dir) resume_full_texts = {} # Dictionary to store full text of each resume for resume in resumes: resume_path = os.path.join(resumes_dir, resume.filename) with open(resume_path, "wb") as buffer: shutil.copyfileobj(resume.file, buffer) # Read full text of the resume using the universal parser from KNOWLEDGE_EXTRACTOR # This assumes the universal_parser can handle various document types. # I need to import universal_parser from KNOWLEDGE_EXTRACTOR.universal_parser from KNOWLEDGE_EXTRACTOR.universal_parser import UniversalParser parser = UniversalParser() try: parsed_data = parser.parse_file(resume_path) if parsed_data and parsed_data.get("text"): # Assuming 'text' key holds the full content resume_full_texts[resume.filename] = parsed_data["text"] else: print(f"Warning: Could not extract text from {resume.filename} using UniversalParser.") resume_full_texts[resume.filename] = "" except Exception as e: print(f"Error parsing {resume.filename} with UniversalParser: {e}") resume_full_texts[resume.filename] = "" temp_collection_name = f"temp_collection_{os.urandom(8).hex()}" temp_sections_collection_name = f"temp_sections_collection_{os.urandom(8).hex()}" temp_chroma_manager = ChromaDBManager( in_memory=True, collection_name=temp_collection_name, sections_collection_name=temp_sections_collection_name ) print(f"Starting on-the-fly indexing for {len(resumes)} resumes into collection '{temp_collection_name}'...") index_directory(resumes_dir, model, temp_chroma_manager) print("On-the-fly indexing complete.") job_text, job_embedding = extract_job_description(jd_path, model) results = temp_chroma_manager.query( query_text=job_text, query_embedding=job_embedding, top_k=20, # Increase top_k to get more matches min_similarity=0.1, ) if not results or not results.get("matches"): return { "job_text": job_text, "matches": [], "summary": "No matching resumes found in the uploaded files.", "overall_scores": {} } # Group matches by filename and keep the best section match best_matches = {} for match in results["matches"]: fname = match["filename"] if fname not in best_matches or match["match_percentage"] > best_matches[fname]["match_percentage"]: # Get full text for structured data extraction full_resume_text = resume_full_texts.get(fname, "") structured_data = extract_structured_data(full_resume_text) match['name'] = structured_data['name'] match['skills'] = structured_data['skills'] match['experience'] = structured_data['experience'] best_matches[fname] = match # Sort matches by relevance sorted_matches = sorted( best_matches.items(), key=lambda item: item[1]['match_percentage'], reverse=True ) # Get overall resume scores overall_scores = {} if results.get("resume_scores"): overall_scores = { match['filename']: results['resume_scores'].get(match['resume_id'], 0) for match in best_matches.values() } # Sort overall scores overall_scores = dict( sorted(overall_scores.items(), key=lambda x: x[1], reverse=True) ) summary = summarize_matches_with_llm_api(job_text, dict(sorted_matches)) return { "job_text": job_text, "matches": [ { "filename": filename, "name": match["name"], "relevance": match["match_percentage"], "best_section": match["section_name"], "section_text": match["text"], "skills": match["skills"], "experience": match["experience"] } for filename, match in sorted_matches ], "overall_scores": overall_scores, "summary": summary } except Exception as e: import traceback traceback.print_exc() raise HTTPException(status_code=500, detail=str(e)) finally: shutil.rmtree(temp_dir) @app.post("/api/index-resumes", tags=["Indexing"]) async def index_resumes_endpoint(resumes_path: str = "DATA_resume"): """ Triggers the indexing of resumes from the specified directory. """ if not os.path.isdir(resumes_path): raise HTTPException(status_code=404, detail=f"Directory not found: {resumes_path}") try: print(f"Starting indexing for directory: {resumes_path} into main database.") index_directory(resumes_path, model, main_chroma_manager) return {"status": "success", "message": f"Indexing complete for {resumes_path}."} except Exception as e: raise HTTPException(status_code=500, detail=f"Indexing failed: {e}") @app.get("/api/resume-embedding/{resume_id}", tags=["Resumes"]) async def get_resume_embedding(resume_id: str): """Retrieve the full resume text embedding given a resume ID.""" embedding = main_chroma_manager.get_resume_embedding(resume_id) if not embedding: raise HTTPException(status_code=404, detail=f"Resume with ID '{resume_id}' not found.") return {"embedding": embedding} @app.post("/api/summarize-resume", tags=["Resumes"]) async def summarize_resume(resume_embedding: dict, job_description: str): """Summarize the resume information using the LLM.""" try: # Create a temporary file to store the resume embedding with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".json") as f: json.dump(resume_embedding, f) temp_file_path = f.name # Run the SLM_manager/augemented_generation.py script with the temporary file command = [ "python", "/Users/deepandee/Desktop/RAG/SLM_manager/augemented_generation.py", "--resume_file", temp_file_path, "--job_description", job_description, ] process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = process.communicate() # Check for errors if stderr: print(f"Error summarizing resume: {stderr.decode()}") raise HTTPException(status_code=500, detail=f"Error summarizing resume: {stderr.decode()}") # Extract the summary from the output summary = stdout.decode().strip() except Exception as e: raise HTTPException(status_code=500, detail=str(e)) finally: # Clean up the temporary file os.remove(temp_file_path) return {"summary": summary} # --- Server Startup --- if __name__ == "__main__": uvicorn.run("api:app", host="0.0.0.0", port=8000, reload=True)