# app.py
# Modern Dark Mode Streamlit Application for AI Talent Screening (FINAL - Remove Coloring, Increase Max Profiles to 7)
import streamlit as st
# Import necessary libraries
from transformers import BertTokenizer, BertForSequenceClassification, T5Tokenizer, T5ForConditionalGeneration
import torch
import numpy as np
import re
import io
import time
import pandas as pd
import PyPDF2
from docx import Document
import plotly.express as px # Import Plotly for stable charting
# Set page config with modern dark theme and wide layout
st.set_page_config(
page_title="AI Data/Tech Talent Screening Tool",
page_icon="🚀",
layout="wide",
initial_sidebar_state="expanded",
)
# --- CUSTOM MODERN DARK MODE CSS OVERHAUL (UNCHANGED) ---
st.markdown("""
""", unsafe_allow_html=True)
# --- CONFIGURATION AND CORE UTILITIES (UNCHANGED) ---
skills_list = [
'python', 'sql', 'c++', 'java', 'tableau', 'machine learning', 'data analysis',
'business intelligence', 'r', 'tensorflow', 'pandas', 'spark', 'scikit-learn', 'aws',
'javascript', 'scala', 'go', 'ruby', 'pytorch', 'keras', 'deep learning', 'nlp',
'computer vision', 'azure', 'gcp', 'docker', 'kubernetes', 'hadoop', 'kafka',
'airflow', 'power bi', 'matplotlib', 'seaborn', 'plotly', 'ggplot', 'mysql',
'postgresql', 'mongodb', 'redis', 'git', 'linux', 'api', 'rest',
'rust', 'kotlin', 'typescript', 'julia', 'snowflake', 'bigquery', 'cassandra',
'neo4j', 'hugging face', 'langchain', 'onnx', 'xgboost', 'terraform', 'ansible',
'jenkins', 'gitlab ci', 'qlik', 'looker', 'd3 js', 'blockchain', 'quantum computing',
'cybersecurity', 'project management', 'technical writing', 'business analysis',
'agile methodologies', 'communication', 'team leadership',
'databricks', 'synapse', 'delta lake', 'streamlit', 'fastapi', 'graphql', 'mlflow', 'kedro'
]
skills_pattern = re.compile(r'\b(' + '|'.join(re.escape(skill) for skill in skills_list) + r')\b', re.IGNORECASE)
# Helper functions for CV parsing
def extract_text_from_pdf(file):
try:
pdf_reader = PyPDF2.PdfReader(file)
text = ""
for page in pdf_reader.pages:
page_text = page.extract_text()
if page_text:
text += page_text + "\n"
return text.strip()
except Exception as e:
st.error(f"Error extracting text from PDF: {str(e)}")
return ""
def extract_text_from_docx(file):
try:
doc = Document(file)
text = ""
for paragraph in doc.paragraphs:
text += paragraph.text + "\n"
return text.strip()
except Exception as e:
st.error(f"Error extracting text from Word document: {str(e)}")
return ""
def extract_text_from_file(uploaded_file):
if uploaded_file.name.endswith('.pdf'): return extract_text_from_pdf(uploaded_file)
elif uploaded_file.name.endswith('.docx'): return extract_text_from_docx(uploaded_file)
return ""
def normalize_text(text):
text = text.lower()
# Clean up text for better NLP processing
text = re.sub(r'_|-|,\s*collaborated in agile teams|,\s*developed solutions for|,\s*led projects involving|,\s*designed applications with|,\s*built machine learning models for|,\s*implemented data pipelines for|,\s*deployed cloud-based solutions|,\s*optimized workflows for|,\s*contributed to data-driven projects', '', text)
return text
def check_experience_mismatch(resume, job_description):
resume_match = re.search(r'(\d+)\s*years?|senior', resume.lower())
job_match = re.search(r'(\d+)\s*years?(?:\s+\w+)*\+|senior\+', job_description.lower())
if resume_match and job_match:
resume_years = resume_match.group(0)
job_years = job_match.group(0)
if 'senior' in resume_years: resume_num = 10
else: resume_num = int(re.search(r'\d+', resume_years).group(0)) if re.search(r'\d+', resume_years) else 0
if 'senior+' in job_years: job_num = 10
else: job_num = int(re.search(r'\d+', job_years).group(0)) if re.search(r'\d+', job_years) else 0
if resume_num < job_num: return f"Experience mismatch: Resume has {resume_years.strip()}, job requires {job_years.strip()}"
return None
def validate_input(text, is_resume=True):
if not text.strip() or len(text.strip()) < 10: return "Input is too short (minimum 10 characters)."
text_normalized = normalize_text(text)
if is_resume and not skills_pattern.search(text_normalized): return "Please include at least one data/tech skill (e.g., python, sql, databricks)."
if is_resume and not re.search(r'\d+\s*year(s)?|senior', text.lower()): return "Please include experience (e.g., '3 years experience' or 'senior')."
return None
@st.cache_resource
def load_models():
# Placeholder for model loading
bert_model_path = 'scmlewis/bert-finetuned-isom5240'
bert_tokenizer = BertTokenizer.from_pretrained(bert_model_path)
bert_model = BertForSequenceClassification.from_pretrained(bert_model_path, num_labels=2)
t5_tokenizer = T5Tokenizer.from_pretrained('t5-small')
t5_model = T5ForConditionalGeneration.from_pretrained('t5-small')
device = torch.device('cpu')
bert_model.to(device)
t5_model.to(device)
bert_model.eval()
t5_model.eval()
return bert_tokenizer, bert_model, t5_tokenizer, t5_model, device
@st.cache_data
def tokenize_inputs(resumes, job_description, _bert_tokenizer, _t5_tokenizer):
job_description_norm = normalize_text(job_description)
bert_inputs = [f"resume: {normalize_text(resume)} [sep] job: {job_description_norm}" for resume in resumes]
bert_tokenized = _bert_tokenizer(bert_inputs, return_tensors='pt', padding=True, truncation=True, max_length=64)
t5_inputs = []
for resume in resumes:
prompt = re.sub(r'\b[Cc]\+\+\b', 'c++', resume)
prompt_normalized = normalize_text(prompt)
t5_inputs.append(f"summarize: {prompt_normalized}")
t5_tokenized = _t5_tokenizer(t5_inputs, return_tensors='pt', padding=True, truncation=True, max_length=64)
return bert_tokenized, t5_inputs, t5_tokenized
@st.cache_data
def extract_skills(text):
text_normalized = normalize_text(text)
text_normalized = re.sub(r'[,_-]', ' ', text_normalized)
found_skills = skills_pattern.findall(text_normalized)
return set(s.lower() for s in found_skills)
@st.cache_data
def classify_and_summarize_batch(resume, job_description, _bert_tokenized, _t5_input, _t5_tokenized, _job_skills_set):
_, bert_model, t5_tokenizer, t5_model, device = st.session_state.models
try:
# BERT Classification
bert_tokenized = {k: v.to(device) for k, v in _bert_tokenized.items()}
with torch.no_grad():
outputs = bert_model(**bert_tokenized)
logits = outputs.logits
probabilities = torch.softmax(logits, dim=1).cpu().numpy()
predictions = np.argmax(probabilities, axis=1)
confidence_threshold = 0.85
prob, pred = probabilities[0], predictions[0]
# T5 Summarization
t5_tokenized = {k: v.to(device) for k, v in _t5_tokenized.items()}
with torch.no_grad():
t5_outputs = t5_model.generate(
t5_tokenized['input_ids'],
attention_mask=t5_tokenized['attention_mask'],
max_length=30,
min_length=8,
num_beams=2,
no_repeat_ngram_size=3,
length_penalty=2.0,
early_stopping=True
)
summaries = [t5_tokenizer.decode(output, skip_special_tokens=True, clean_up_tokenization_spaces=True) for output in t5_outputs]
# Skill and Experience Check
resume_skills_set = extract_skills(resume)
skill_overlap = len(_job_skills_set.intersection(resume_skills_set)) / len(_job_skills_set) if _job_skills_set else 0
suitability = "Relevant"
warning = "None"
exp_warning = check_experience_mismatch(resume, job_description)
if skill_overlap < 0.4:
suitability = "Irrelevant"
warning = "Low skill match (<40%) with job requirements"
elif exp_warning:
suitability = "Uncertain"
warning = exp_warning
elif prob[pred] < confidence_threshold:
suitability = "Uncertain"
warning = f"Lower AI confidence: {prob[pred]:.2f}"
elif skill_overlap < 0.5:
suitability = "Irrelevant"
warning = "Skill overlap is present but not a strong match (<50%)"
# Final Summary Formatting (HR-friendly)
detected_skills = list(set(skills_pattern.findall(normalize_text(resume))))
exp_match = re.search(r'\d+\s*years?|senior', resume.lower())
if detected_skills and exp_match: final_summary = f"Key Skills: {', '.join(detected_skills)}. Experience: {exp_match.group(0).capitalize()}"
elif detected_skills: final_summary = f"Key Skills: {', '.join(detected_skills)}"
else: final_summary = f"Experience: {exp_match.group(0).capitalize() if exp_match else 'Unknown'}"
# Color codes based on theme (STILL NEEDED for the scorecards/tiles and CSV)
if suitability == "Relevant": color = "#4CAF50"
elif suitability == "Irrelevant": color = "#F44336"
else: color = "#FFC107"
return {"Suitability": suitability, "Data/Tech Related Skills Summary": final_summary, "Warning": warning or "None", "Suitability_Color": color}
except Exception as e:
return {"Suitability": "Error", "Data/Tech Related Skills Summary": "Failed to process profile", "Warning": str(e), "Suitability_Color": "#F44336"}
@st.cache_data
def generate_skill_pie_chart(resumes):
# Skill chart logic
skill_counts = {}
total_resumes = len([r for r in resumes if r.strip()])
if total_resumes == 0: return None
for resume in resumes:
if resume.strip():
resume_lower = normalize_text(resume)
found_skills = skills_pattern.findall(resume_lower)
for skill in found_skills:
skill_counts[skill.lower()] = skill_counts.get(skill.lower(), 0) + 1
if not skill_counts: return None
sorted_skills = sorted(skill_counts.items(), key=lambda item: item[1], reverse=True)
top_n = 8
if len(sorted_skills) > top_n:
top_skills = dict(sorted_skills[:top_n-1])
other_count = sum(count for _, count in sorted_skills[top_n-1:])
top_skills["Other Skills"] = other_count
else:
top_skills = dict(sorted_skills)
chart_df = pd.DataFrame(list(top_skills.items()), columns=['Skill', 'Count'])
# PLOTLY IMPLEMENTATION
fig = px.pie(
chart_df,
values='Count',
names='Skill',
title='Top Candidate Skill Frequency',
hole=0.3, # Donut chart style
color_discrete_sequence=px.colors.qualitative.Plotly
)
# Update layout for dark theme
fig.update_layout(
paper_bgcolor='rgba(0,0,0,0)',
plot_bgcolor='rgba(0,0,0,0)',
font_color='#F8F8F8',
title_font_color='#42A5F5',
title_font_size=20,
legend_title_font_color='#B0B0B0',
)
fig.update_traces(textinfo='percent+label', marker=dict(line=dict(color='#3A3A3A', width=1.5)))
return fig
def render_sidebar():
"""Render sidebar content with professional HR language."""
SUCCESS_COLOR = "#4CAF50"
WARNING_COLOR = "#FFC107"
DANGER_COLOR = "#F44336"
PRIMARY_COLOR = "#42A5F5"
with st.sidebar:
st.markdown(f"""
TALENT SCREENING ASSISTANT
Powered by Advanced NLP (BERT + T5)
""", unsafe_allow_html=True)
with st.expander("📝 Quick Guide for HR", expanded=True):
st.markdown("""
**1. Set Requirements (Tab 1)**:
- Enter the **Job Description** (JD). Be clear about required skills and experience (e.g., '5 years+').
**2. Upload Candidates (Tab 2)**:
- Upload or paste up to **7 Candidate Profiles** (PDF/DOCX/Text). <-- **UPDATED TO 7**
- Profiles must contain key technical skills and explicit experience.
**3. Run Screening**:
- Click the **Run Candidate Screening** button.
**4. Review Report (Tab 3)**:
- View the summary scorecard and detailed table for swift assessment.
""")
with st.expander("🎯 Screening Outcomes Explained", expanded=False):
st.markdown(f"""
- **Relevant**: Strong match across all criteria. Proceed to interview.
- **Irrelevant**: Low skill overlap or poor fit. Pass on candidate.
- **Uncertain**: Flagged due to Experience Mismatch or Lower AI confidence. Requires manual review.
""", unsafe_allow_html=True)
# --- MAIN APPLICATION LOGIC ---
def main():
"""Main function to run the Streamlit app for resume screening."""
render_sidebar()
# Initialize session state
if 'resumes' not in st.session_state: st.session_state.resumes = ["Expert in python, machine learning, tableau, 4 years experience", "", ""]
if 'input_job_description' not in st.session_state: st.session_state.input_job_description = "Data scientist requires python, machine learning, 3 years+"
if 'results' not in st.session_state: st.session_state.results = []
if 'valid_resumes' not in st.session_state: st.session_state.valid_resumes = []
if 'models' not in st.session_state: st.session_state.models = None
st.markdown("🚀 AI DATA/TECH TALENT SCREENING TOOL
", unsafe_allow_html=True)
tab_setup, tab_resumes, tab_results = st.tabs(["1. Job Requirement Setup", "2. Candidate Profile Upload", "3. Screening Report & Analytics"])
# --- TAB 1: Setup & Job Description ---
with tab_setup:
st.markdown("## 📋 Define Job Requirements")
st.info("Please enter the **Job Description** below. This is essential for the AI to accurately match skills and experience levels.")
job_description = st.text_area(
"Job Description Text",
value=st.session_state.input_job_description,
height=150,
key="job_description_tab",
placeholder="e.g., Data engineer role requires 5 years+ experience with Python, AWS, and Databricks. Must have leadership experience."
)
st.session_state.input_job_description = job_description
validation_error = validate_input(job_description, is_resume=False)
if validation_error and job_description.strip():
st.warning(f"Input Check: Job Description missing key details. {validation_error}")
# --- TAB 2: Manage Resumes ---
with tab_resumes:
st.markdown(f"## 📁 Upload Candidate Profiles ({len(st.session_state.resumes)}/7)") # <-- **UPDATED TO 7**
st.info("Upload or paste candidate text below. The AI requires **key technical skills and experience statements** to function.")
# Manage resume inputs
for i in range(len(st.session_state.resumes)):
status_icon = "⚪" # Default: Pending
validation_error = validate_input(st.session_state.resumes[i], is_resume=True)
if not st.session_state.resumes[i].strip():
status_icon = "📂" # Empty/Needs Input
is_expanded = False
elif validation_error:
status_icon = "⚠️" # Warning/Error
is_expanded = True
else:
status_icon = "✅" # Valid
is_expanded = False
with st.expander(f"**{status_icon} CANDIDATE PROFILE {i+1}**", expanded=is_expanded):
uploaded_file = st.file_uploader(
f"Upload Profile (PDF or DOCX) for Candidate {i+1}",
type=['pdf', 'docx'],
key=f"file_upload_{i}"
)
if uploaded_file is not None:
extracted_text = extract_text_from_file(uploaded_file)
if extracted_text: st.session_state.resumes[i] = extracted_text
else: st.session_state.resumes[i] = ""
st.session_state.resumes[i] = st.text_area(
f"Candidate Profile Text",
value=st.session_state.resumes[i],
height=100,
key=f"resume_{i}_tab",
placeholder="e.g., Expert in Python, SQL, and 3 years experience in data science."
)
if validation_error and st.session_state.resumes[i].strip():
st.warning(f"Profile Check: Candidate {i+1} flagged. {validation_error}")
st.markdown("
", unsafe_allow_html=True)
col_add, col_remove, _ = st.columns([1, 1, 3])
with col_add:
if st.button("➕ Add Candidate Slot", use_container_width=True) and len(st.session_state.resumes) < 7: # <-- **UPDATED TO 7**
st.session_state.resumes.append("")
st.rerun()
with col_remove:
if st.button("➖ Remove Candidate Slot", use_container_width=True) and len(st.session_state.resumes) > 1:
st.session_state.resumes.pop()
st.rerun()
# --- ACTION BUTTONS ---
st.markdown("---")
col_btn1, col_btn2, _ = st.columns([1, 1, 3])
with col_btn1:
analyze_clicked = st.button("✅ Run Candidate Screening", type="primary", use_container_width=True)
with col_btn2:
reset_clicked = st.button("♻️ Reset All Inputs", use_container_width=True)
st.markdown("---")
# Handle reset and analysis logic
if reset_clicked:
st.session_state.resumes = ["", "", ""]
st.session_state.input_job_description = ""
st.session_state.results = []
st.session_state.valid_resumes = []
st.session_state.models = None
st.rerun()
if analyze_clicked:
valid_resumes = []
all_inputs_valid = True
for i, resume in enumerate(st.session_state.resumes):
validation_error = validate_input(resume, is_resume=True)
if not validation_error and resume.strip(): valid_resumes.append(resume)
elif validation_error and resume.strip():
st.error(f"Screening Blocked: Candidate {i+1} failed pre-screening validation. Fix input.")
all_inputs_valid = False
job_validation_error = validate_input(job_description, is_resume=False)
if job_validation_error and job_description.strip(): st.error(f"Screening Blocked: Job Description failed validation. Fix input."); all_inputs_valid = False
if valid_resumes and job_description.strip() and all_inputs_valid:
if st.session_state.models is None:
with st.spinner("Initializing AI Model, please wait..."): st.session_state.models = load_models()
st.session_state.results = []
st.session_state.valid_resumes = valid_resumes
total_steps = len(valid_resumes)
with st.spinner("Processing Candidate Profiles..."):
progress_bar = st.progress(0); status_text = st.empty()
bert_tokenizer, _, t5_tokenizer, _, _ = st.session_state.models
status_text.text("Status: Preparing inputs and extracting job skills...")
bert_tokenized, t5_inputs, t5_tokenized = tokenize_inputs(valid_resumes, job_description, bert_tokenizer, t5_tokenizer)
job_skills_set = extract_skills(job_description)
results = []
for i, resume in enumerate(valid_resumes):
status_text.text(f"Status: Analyzing Profile {i+1} of {total_steps}...")
bert_tok_single = {
'input_ids': bert_tokenized['input_ids'][i].unsqueeze(0),
'attention_mask': bert_tokenized['attention_mask'][i].unsqueeze(0)
}
t5_tok_single = {
'input_ids': t5_tokenized['input_ids'][i].unsqueeze(0),
'attention_mask': t5_tokenized['attention_mask'][i].unsqueeze(0)
}
result = classify_and_summarize_batch(
resume,
job_description,
bert_tok_single,
t5_inputs[i],
t5_tok_single,
job_skills_set
)
result["Resume"] = f"Candidate {i+1}"
results.append(result)
progress_bar.progress((i + 1) / total_steps)
st.session_state.results = results
status_text.empty(); progress_bar.empty()
st.success("Screening Complete. Results are available in the 'Screening Report & Analytics' tab. 🎉")
else:
st.error("Screening cannot run. Ensure at least one valid candidate profile and a job description are provided.")
# --- TAB 3: Results (The Professional Report) ---
with tab_results:
st.markdown("## 📊 Screening Results Summary")
if st.session_state.results:
# --- Scorecard Metrics (Professional Tiles) ---
results_df = pd.DataFrame(st.session_state.results)
total = len(results_df)
relevant_count = len(results_df[results_df['Suitability'] == 'Relevant'])
review_count = len(results_df[results_df['Suitability'] == 'Uncertain'])
irrelevant_count = len(results_df[results_df['Suitability'].isin(['Irrelevant', 'Error'])])
st.markdown(f"#### Overview: {total} Candidate Profiles Processed")
PRIMARY_COLOR = "#42A5F5"
SUCCESS_COLOR = "#4CAF50"
WARNING_COLOR = "#FFC107"
DANGER_COLOR = "#F44336"
col1, col2, col3, col4 = st.columns(4)
with col1:
st.markdown(f"""
""", unsafe_allow_html=True)
with col2:
st.markdown(f"""
RELEVANT MATCHES
{relevant_count}
""", unsafe_allow_html=True)
with col3:
st.markdown(f"""
REQUIRES REVIEW
{review_count}
""", unsafe_allow_html=True)
with col4:
st.markdown(f"""
IRRELEVANT / ERROR
{irrelevant_count}
""", unsafe_allow_html=True)
st.markdown("---")
# --- Detailed Report Table ---
st.markdown("### 📋 Detailed Screening Results")
# **FIX**: Explicitly drop the 'Suitability_Color' column to ensure it is not displayed,
# and remove all cell coloring logic.
display_df = results_df.drop(columns=['Suitability_Color'])
# Rename columns for professional display
display_df = display_df.rename(
columns={
'Data/Tech Related Skills Summary': '**PROFILE SUMMARY**',
'Warning': '**FLAGGING REASON**',
'Resume': '**PROFILE ID**',
'Suitability': '**SUITABILITY**'
}
)
# Display the simple, unstyled DataFrame
st.dataframe(
display_df,
use_container_width=True
)
# --- Download and Chart Section ---
st.markdown("
", unsafe_allow_html=True)
col_dl, col_chart_expander = st.columns([1, 3])
with col_dl:
csv_buffer = io.StringIO()
# Drop the internal column before download
results_df.drop(columns=['Suitability_Color']).to_csv(csv_buffer, index=False)
st.download_button(
"💾 Download Full Report (CSV)",
csv_buffer.getvalue(),
file_name="Talent_Screening_Report.csv",
mime="text/csv",
use_container_width=True
)
with col_chart_expander:
with st.expander("📈 Skill Distribution Analytics (Plotly)", expanded=False):
if st.session_state.valid_resumes:
fig = generate_skill_pie_chart(st.session_state.valid_resumes)
if fig:
st.plotly_chart(fig, use_container_width=True)
else:
st.info("No recognized technical skills found in the profiles for charting.")
else:
st.info("No valid candidate profiles to analyze.")
else:
st.info("Please complete the setup and upload tabs, then run the screening to generate the report.")
if __name__ == "__main__":
main()