Codingo / backend /services /resume_parser.py
husseinelsaadi
Richer CV experience, sharper STT, livelier LUNA voice
c5a47ef
Raw
History Blame Contribute Delete
17.7 kB
import os
import re
import json
import logging
from pathlib import Path
from typing import Dict, List, Tuple
from pdfminer.high_level import extract_text as pdf_extract_text
from docx import Document
# ---------------------------------------------------------------------------
# Parsing strategy
#
# The primary parser is an LLM (Groq) which reads the raw resume text and
# returns clean, structured fields. This is far more accurate than regex and
# handles any ATS-friendly CV. The legacy spaCy/transformer/regex path is kept
# only as a fallback for when no GROQ_API_KEY is configured or the LLM call
# fails. The heavy fallback dependencies (spaCy model, the manishiitg/resume-ner
# transformer, nltk) are imported lazily inside ``_regex_parse`` so the normal
# (LLM) path does not pay their startup cost.
# ---------------------------------------------------------------------------
# Cached lazily so we only build the Groq client once.
_llm = None
_llm_initialised = False
def _get_llm():
"""Return a cached ChatGroq client, or ``None`` when no key is configured."""
global _llm, _llm_initialised
if _llm_initialised:
return _llm
_llm_initialised = True
api_key = os.getenv("GROQ_API_KEY")
if not api_key:
logging.warning("GROQ_API_KEY not set; resume parsing will use the regex fallback.")
_llm = None
return _llm
try:
from langchain_groq import ChatGroq
_llm = ChatGroq(temperature=0, model_name="llama-3.3-70b-versatile", api_key=api_key)
except Exception as exc:
logging.error(f"Failed to initialise Groq for resume parsing: {exc}")
_llm = None
return _llm
# Expanded keyword lists
SKILL_KEYWORDS = {
# Programming Languages
"python", "java", "javascript", "typescript", "c++", "c#", "ruby", "go", "rust", "kotlin", "swift",
"php", "r", "matlab", "scala", "perl", "bash", "powershell", "sql", "html", "css",
# Frameworks & Libraries
"react", "angular", "vue", "node.js", "express", "django", "flask", "spring", "spring boot",
".net", "laravel", "rails", "fastapi", "pytorch", "tensorflow", "keras", "scikit-learn",
# Databases
"mysql", "postgresql", "mongodb", "redis", "elasticsearch", "cassandra", "oracle", "sql server",
# Cloud & DevOps
"aws", "azure", "gcp", "docker", "kubernetes", "jenkins", "terraform", "ansible", "ci/cd",
# Other Technical Skills
"machine learning", "deep learning", "data science", "nlp", "computer vision", "ai",
"rest api", "graphql", "microservices", "agile", "scrum", "git", "linux", "windows"
}
EDUCATION_PATTERNS = [
# Degrees
r"\b(bachelor|b\.?s\.?c?\.?|b\.?a\.?|b\.?tech|b\.?e\.?)\b",
r"\b(master|m\.?s\.?c?\.?|m\.?a\.?|m\.?tech|m\.?e\.?|mba)\b",
r"\b(ph\.?d\.?|doctorate|doctoral)\b",
r"\b(diploma|certificate|certification)\b",
# Fields of Study
r"\b(computer science|software engineering|information technology|it|cs)\b",
r"\b(electrical engineering|mechanical engineering|civil engineering)\b",
r"\b(data science|artificial intelligence|machine learning)\b",
r"\b(business administration|finance|accounting|marketing)\b",
# Institution indicators
r"\b(university|college|institute|school)\s+of\s+\w+",
r"\b\w+\s+(university|college|institute)\b"
]
JOB_TITLE_PATTERNS = [
r"\b(software|senior|junior|lead|principal|staff)\s*(engineer|developer|programmer)\b",
r"\b(data|business|system|security)\s*(analyst|scientist|engineer)\b",
r"\b(project|product|program|engineering)\s*manager\b",
r"\b(devops|cloud|ml|ai|backend|frontend|full[\s-]?stack)\s*(engineer|developer)\b",
r"\b(consultant|architect|specialist|coordinator|administrator)\b"
]
def extract_text(file_path: str) -> str:
"""Extract text from PDF or DOCX files"""
path = Path(file_path)
if path.suffix.lower() == ".pdf":
text = pdf_extract_text(file_path)
elif path.suffix.lower() == ".docx":
doc = Document(file_path)
text = "\n".join([p.text for p in doc.paragraphs])
else:
raise ValueError("Unsupported file format")
return text
def clean_text(text: str) -> str:
"""Clean and normalize text"""
# Remove multiple spaces and normalize
text = re.sub(r'\s+', ' ', text)
# Keep line breaks for section detection
text = re.sub(r'\n{3,}', '\n\n', text)
return text.strip()
def extract_sections(text: str) -> Dict[str, str]:
"""Extract different sections from resume"""
sections = {
'education': '',
'experience': '',
'skills': '',
'summary': ''
}
# Common section headers
section_patterns = {
'education': r'(education|academic|qualification|degree)',
'experience': r'(experience|employment|work\s*history|professional\s*experience|career)',
'skills': r'(skills|technical\s*skills|competencies|expertise)',
'summary': r'(summary|objective|profile|about)'
}
lines = text.split('\n')
current_section = None
for i, line in enumerate(lines):
line_lower = line.lower().strip()
# Check if this line is a section header
for section, pattern in section_patterns.items():
if re.search(pattern, line_lower) and len(line_lower) < 50:
current_section = section
break
# Add content to current section
if current_section and i > 0:
sections[current_section] += line + '\n'
return sections
def extract_name(text: str, entities: List) -> str:
"""Extract name using multiple methods"""
# Method 1: Use transformer model
name_parts = []
for ent in entities:
if ent["entity_group"].upper() in ["NAME", "PERSON", "PER"]:
name_parts.append(ent["word"].strip())
if name_parts:
# Clean and join name parts
full_name = " ".join(dict.fromkeys(name_parts))
full_name = re.sub(r'\s+', ' ', full_name).strip()
if len(full_name) > 3 and len(full_name.split()) <= 4:
return full_name
# Method 2: Use spaCy if available
nlp = _get_nlp()
if nlp:
doc = nlp(text[:500]) # Check first 500 chars
for ent in doc.ents:
if ent.label_ == "PERSON":
name = ent.text.strip()
if len(name) > 3 and len(name.split()) <= 4:
return name
# Method 3: Pattern matching for first few lines
first_lines = text.split('\n')[:5]
for line in first_lines:
line = line.strip()
# Look for name pattern (2-4 words, title case)
if re.match(r'^[A-Z][a-z]+(\s+[A-Z][a-z]+){1,3}$', line):
return line
return "Not Found"
def extract_skills(text: str, skill_section: str = "") -> List[str]:
"""Extract skills using multiple methods"""
skills_found = set()
# Prioritize skills section if available
search_text = skill_section + " " + text if skill_section else text
search_text = search_text.lower()
# Method 1: Direct keyword matching
for skill in SKILL_KEYWORDS:
if re.search(rf'\b{re.escape(skill.lower())}\b', search_text):
skills_found.add(skill)
# Method 2: Pattern-based extraction
# Look for skills in bullet points or comma-separated lists
skill_patterns = [
r'[•·▪▫◦‣⁃]\s*([A-Za-z\s\+\#\.]+)', # Bullet points
r'(?:skills?|technologies|tools?)[\s:]*([A-Za-z\s,\+\#\.]+)', # After keywords
]
for pattern in skill_patterns:
matches = re.findall(pattern, search_text, re.IGNORECASE)
for match in matches:
# Check each word/phrase in the match
potential_skills = re.split(r'[,;]', match)
for ps in potential_skills:
ps = ps.strip().lower()
if ps in SKILL_KEYWORDS:
skills_found.add(ps)
return list(skills_found)
def extract_education(text: str, edu_section: str = "") -> List[str]:
"""Extract education information"""
education_info = []
search_text = edu_section + " " + text if edu_section else text
# Extract degrees
for pattern in EDUCATION_PATTERNS:
matches = re.findall(pattern, search_text, re.IGNORECASE)
for match in matches:
if isinstance(match, tuple):
match = match[0]
education_info.append(match)
# Extract years (graduation years)
year_pattern = r'\b(19[0-9]{2}|20[0-9]{2})\b'
years = re.findall(year_pattern, search_text)
# Extract GPA if mentioned
gpa_pattern = r'(?:gpa|cgpa|grade)[\s:]*([0-9]\.[0-9]+)'
gpa_matches = re.findall(gpa_pattern, search_text, re.IGNORECASE)
return list(dict.fromkeys(education_info)) # Remove duplicates
def extract_experience(text: str, exp_section: str = "") -> List[str]:
"""Extract experience information"""
experience_info = []
search_text = exp_section + " " + text if exp_section else text
# Extract job titles
for pattern in JOB_TITLE_PATTERNS:
matches = re.findall(pattern, search_text, re.IGNORECASE)
for match in matches:
if isinstance(match, tuple):
match = ' '.join(match).strip()
experience_info.append(match)
# Extract years of experience
exp_patterns = [
r'(\d+)\+?\s*(?:years?|yrs?)(?:\s+of)?\s+experience',
r'experience\s*:?\s*(\d+)\+?\s*(?:years?|yrs?)',
r'(\d+)\+?\s*(?:years?|yrs?)\s+(?:as|in|of)',
]
for pattern in exp_patterns:
matches = re.findall(pattern, search_text, re.IGNORECASE)
if matches:
years = max(map(int, matches))
experience_info.append(f"{years}+ years experience")
break
# Extract company names (common patterns)
company_patterns = [
r'(?:at|@|company|employer)\s*:?\s*([A-Z][A-Za-z\s&\.\-]+)',
r'([A-Z][A-Za-z\s&\.\-]+)\s*(?:inc|llc|ltd|corp|company)',
]
for pattern in company_patterns:
matches = re.findall(pattern, search_text)
experience_info.extend(matches[:3]) # Limit to avoid false positives
return list(dict.fromkeys(experience_info))
# ---------------------------------------------------------------------------
# Lazy loaders for the fallback (regex/NER) path
# ---------------------------------------------------------------------------
_nlp = None
_nlp_loaded = False
_ner_pipeline = None
_ner_loaded = False
def _get_nlp():
"""Lazily load the spaCy model (used only by the regex fallback)."""
global _nlp, _nlp_loaded
if _nlp_loaded:
return _nlp
_nlp_loaded = True
try:
import spacy
_nlp = spacy.load("en_core_web_sm")
except Exception:
logging.warning("spaCy model en_core_web_sm unavailable; name detection degraded.")
_nlp = None
return _nlp
def _get_ner_pipeline():
"""Lazily load the resume-ner transformer (used only by the regex fallback)."""
global _ner_pipeline, _ner_loaded
if _ner_loaded:
return _ner_pipeline
_ner_loaded = True
try:
from transformers import AutoTokenizer, AutoModelForTokenClassification, pipeline
name = "manishiitg/resume-ner"
tok = AutoTokenizer.from_pretrained(name)
mdl = AutoModelForTokenClassification.from_pretrained(name)
_ner_pipeline = pipeline("ner", model=mdl, tokenizer=tok, aggregation_strategy="simple")
except Exception as exc:
logging.warning(f"resume-ner model unavailable ({exc}); using spaCy/regex for names.")
_ner_pipeline = None
return _ner_pipeline
# ---------------------------------------------------------------------------
# Primary parser: LLM (Groq)
# ---------------------------------------------------------------------------
def _coerce_list(value) -> List[str]:
"""Normalise the LLM's value (list or string) into a clean list of strings."""
if isinstance(value, list):
items = [str(v).strip() for v in value]
elif isinstance(value, str):
items = [p.strip() for p in re.split(r'[\n;]+', value)]
else:
items = []
# Drop empties and obvious "not found" placeholders, dedupe (case-insensitive)
seen, out = set(), []
for it in items:
if not it or it.lower() in ("not found", "n/a", "none", "-"):
continue
key = it.lower()
if key not in seen:
seen.add(key)
out.append(it)
return out
def _llm_extract(text: str) -> Dict[str, str]:
"""Extract structured fields from resume text with the Groq LLM.
Returns the same string-field shape as ``parse_resume``. Raises on any
failure so the caller can fall back to the regex parser.
"""
llm = _get_llm()
if llm is None:
raise RuntimeError("LLM unavailable")
# Keep the prompt bounded; resumes rarely need more than this many chars.
snippet = text[:6000]
prompt = f"""You are an expert resume parser. Read the resume text below and extract the candidate's details.
Return ONLY a JSON object (no markdown, no commentary) with exactly these keys:
- "name": the candidate's full name as a string (empty string if not found)
- "skills": an array of individual technical and professional skills (e.g. ["Python", "React", "AWS", "Project Management"])
- "education": an array of one-line entries, each "<Degree> in <Field> — <Institution>, <Year>" when available
- "experience": an array with ONE entry per role. Each entry is a single line of the form
"<Job Title> at <Company> (<dates or duration>): <1-2 sentence summary of the key responsibilities, projects, and technologies for that role>".
Base the summary strictly on that role's bullet points in the resume.
Rules:
- Use the actual content of the resume; never invent information.
- Keep each entry on a single line (no line breaks inside an entry).
- For experience, always include the role's responsibilities/description after the colon when the resume provides them.
- If a section is missing, return an empty array for it.
Resume text:
\"\"\"
{snippet}
\"\"\"
"""
response = llm.invoke(prompt)
raw = response.content if hasattr(response, "content") else str(response)
# Be robust to code fences or stray prose around the JSON.
cleaned = raw.strip()
if cleaned.startswith("```"):
cleaned = re.sub(r"^```[a-zA-Z]*\n?", "", cleaned).rstrip("`").strip()
start, end = cleaned.find("{"), cleaned.rfind("}")
if start == -1 or end == -1:
raise ValueError("LLM did not return JSON")
data = json.loads(cleaned[start:end + 1])
name = str(data.get("name", "")).strip()
skills = _coerce_list(data.get("skills"))
education = _coerce_list(data.get("education"))
experience = _coerce_list(data.get("experience"))
return {
"name": name if name else "Not Found",
"skills": ", ".join(skills[:20]) if skills else "Not Found",
"education": "\n".join(education[:6]) if education else "Not Found",
"experience": "\n".join(experience[:6]) if experience else "Not Found",
}
# ---------------------------------------------------------------------------
# Fallback parser: legacy spaCy/transformer/regex heuristics
# ---------------------------------------------------------------------------
def _regex_parse(text: str) -> Dict[str, str]:
"""Heuristic resume parsing used when the LLM is unavailable."""
sections = extract_sections(text)
entities = []
ner = _get_ner_pipeline()
if ner is not None:
try:
entities = ner(text[:1024]) # Limit for performance
except Exception:
entities = []
name = extract_name(text, entities)
skills = extract_skills(text, sections.get('skills', ''))
education = extract_education(text, sections.get('education', ''))
experience = extract_experience(text, sections.get('experience', ''))
return {
"name": name,
"skills": ", ".join(skills[:15]) if skills else "Not Found",
"education": ", ".join(education[:5]) if education else "Not Found",
"experience": ", ".join(experience[:5]) if experience else "Not Found",
}
def parse_resume(file_path: str, filename: str = None) -> Dict[str, str]:
"""Parse a resume into structured fields.
Tries the Groq LLM first (accurate, handles any CV layout) and falls back
to the legacy regex/NER heuristics if the LLM is unavailable or errors.
The return shape (string fields) is unchanged so existing callers work.
"""
raw_text = extract_text(file_path)
text = clean_text(raw_text)
try:
return _llm_extract(text)
except Exception as exc:
logging.warning(f"LLM resume parsing failed ({exc}); falling back to regex parser.")
return _regex_parse(text)
# Optional: Add confidence scores
def parse_resume_with_confidence(file_path: str) -> Dict[str, Tuple[str, float]]:
"""Parse resume with confidence scores for each field"""
result = parse_resume(file_path)
# Simple confidence calculation based on whether data was found
confidence_scores = {
"name": 0.9 if result["name"] != "Not Found" else 0.1,
"skills": min(0.9, len(result["skills"].split(",")) * 0.1) if result["skills"] != "Not Found" else 0.1,
"education": 0.8 if result["education"] != "Not Found" else 0.2,
"experience": 0.8 if result["experience"] != "Not Found" else 0.2
}
return {
key: (value, confidence_scores[key])
for key, value in result.items()
}