import aiohttp import asyncio import nest_asyncio import streamlit as st import time from pdfminer.high_level import extract_text from docx import Document import os from pdf2image import convert_from_bytes import base64 import io import certifi import ssl # from dotenv import load_dotenv import json from prompts import ( ANALYSIS_PROMPT, ATS_PROMPT, OBJECTIVE_ANALYSIS_PROMPT, SCORING_PROMPT, DEFUALT_PREP_GUIDE, LOW_PREP_GUIDE ) # load_dotenv() api_key = os.getenv('OPENAI_API_KEY') nest_asyncio.apply() def generate_prompts(num_images): return { "SCORING": SCORING_PROMPT, "ANALYSIS": ANALYSIS_PROMPT, "OBJECTIVE_ANALYSIS": OBJECTIVE_ANALYSIS_PROMPT, "ATS": ATS_PROMPT.format(pages=num_images), } def encode_image(image): buffered = io.BytesIO() image.save(buffered, format="PNG") return base64.b64encode(buffered.getvalue()).decode('utf-8') def extract_text_from_pdf(file): try: # Reset file pointer to the start of the file file.seek(0) text = extract_text(file) return text except Exception as e: print(f"An error occurred: {e}") return None def extract_text_from_file(file): file_extension = os.path.splitext(file.name)[1] if file_extension == '.pdf': return extract_text_from_pdf(file) elif file_extension == '.docx': doc = Document(file) return "\n".join([paragraph.text for paragraph in doc.paragraphs]) elif file_extension == '.txt': return file.read().decode('utf-8') else: return "Unsupported file type" async def generate_completion(message, api_key): url = "https://api.openai.com/v1/chat/completions" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" } payload = { "model": "gpt-4o", "messages": message, "temperature": 0.1, "response_format": {"type": "json_object"} } ssl_context = ssl.create_default_context(cafile=certifi.where()) # Use certifi to specify the CA bundle for SSL certificate verification async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=ssl_context)) as session: async with session.post(url, headers=headers, json=payload) as response: if response.status != 200: print(f"Error: {response.status}") return None response_json = await response.json() return response_json async def main(resume_text, job_description): analysis_types_and_prompts = { "SCORING": SCORING_PROMPT, "ANALYSIS": ANALYSIS_PROMPT, "OBJECTIVE_ANALYSIS": OBJECTIVE_ANALYSIS_PROMPT, "ATS": ATS_PROMPT.format(pages=pdf_length) } tasks = [] for analysis, prompt in analysis_types_and_prompts.items(): messages = construct_message( analysis, prompt, resume_text, job_description, base_image if analysis == "ATS" else None, ) tasks.append(generate_completion(messages,api_key)) scoring_task = asyncio.create_task(tasks[0]) other_tasks = tasks[1:] other_tasks_futures = [asyncio.create_task(task) for task in other_tasks] result_scoring = await scoring_task scoring_json = json.loads(result_scoring['choices'][0]['message']['content']) if len(scoring_json) == 1: return scoring_json, None # Error dictionary other_results = await asyncio.gather(*other_tasks_futures) responses_async = [result_scoring] + other_results if scoring_json['Overall']['Score'] <=50: guide_prompt = LOW_PREP_GUIDE else: guide_prompt = DEFUALT_PREP_GUIDE return responses_async, guide_prompt def construct_message(analysis, prompt, resume_text, job_description=None, base_image=None): if analysis == "ATS": messages = [ {"role": "system", "content": prompt}, {"role": "user", "content": f"Here is the Job Description: {job_description}"}, {"role": "user", "content": []} ] for image in base_image: messages[2]["content"].append({ "type": "image_url", "image_url": { "url": f"data:image/png;base64,{image}" }, }) return messages elif analysis == "OBJECTIVE_ANALYSIS": return ([ {"role": "system", "content": prompt}, {"role": "user", "content": f"Here is the resume text: {resume_text}"}]) else: return ([ {"role": "system", "content": prompt}, {"role": "user", "content": f"Here is the resume text: {resume_text} \n Here is the job description: {job_description}"}]) st.set_page_config(page_title="Resume Analyzer", layout="centered") st.title("Resume Analyzer") job_description_source = st.radio("Job Description Source", ("Upload File", "Paste Text")) if job_description_source == "Upload File": job_description_file = st.file_uploader("Upload Job Description", type=['pdf', 'docx', 'txt']) job_description = extract_text_from_file(job_description_file) if job_description_file else None else: job_description = st.text_area("Paste Job Description Here") resume_file = st.file_uploader("Upload Resume", type=['pdf', 'docx', 'txt']) if job_description: st.subheader("Job Description Preview") st.text_area("Job Description", job_description, height=200) if resume_file: resume_text = extract_text_from_file(resume_file) st.subheader("Resume Preview") st.text_area("Resume", resume_text, height=200) resume_file.seek(0) file_bytes = resume_file.read() images = convert_from_bytes(file_bytes) pdf_length = len(images) base_image = [] for image in images: base_image.append(encode_image(image)) if st.button("Analyze Resume"): if job_description and resume_file: start_time = time.time() responses_async, guide_prompt = asyncio.run(main(resume_text, job_description)) end_time = time.time() total_time = end_time - start_time guide_message = construct_message("GUIDE", guide_prompt, resume_text, job_description) guide_response = asyncio.run(generate_completion(guide_message, api_key)) if isinstance(responses_async, dict) and "Error" in responses_async: error_message = f"Improper file entered: {responses_async['Error']}" st.write(error_message) else: st.subheader("Analysis Result") st.text_area("Scoring Analysis", responses_async[0]['choices'][0]['message']['content'], height=400) st.text_area("Gap Analysis", responses_async[1]['choices'][0]['message']['content'], height=400) st.text_area("Objective Analysis", responses_async[2]['choices'][0]['message']['content'], height=400) st.text_area("ATS Analysis", responses_async[3]['choices'][0]['message']['content'], height=400) st.text_area("Interview Prep Guide", guide_response['choices'][0]['message']['content'], height=400) st.write(f"Image prompt input tokens: {responses_async[2]['usage']['prompt_tokens']}") # st.write(f"Time taken for OpenAI response: {response_time:.2f} seconds") st.write(f"Total time taken for execution: {total_time:.2f} seconds") #Testing JSON Respones: # testing.run_tests(responses_async, guide_response) else: st.error("Please provide both the job description and resume.")