| """FastAPI CV upload and job-match service. |
| |
| Run locally with: |
| uvicorn app:app --reload --host 127.0.0.1 --port 8000 |
| """ |
|
|
| import io |
| import json |
| import re |
| import threading |
| from pathlib import Path |
|
|
| import numpy as np |
| import torch |
| import uvicorn |
| from fastapi import FastAPI, File, HTTPException, UploadFile |
| from fastapi.responses import FileResponse |
| from fastapi.staticfiles import StaticFiles |
| from pydantic import BaseModel, Field |
| from sentence_transformers import CrossEncoder |
|
|
|
|
| ROOT = Path(__file__).resolve().parent |
| MODEL_DIR = ROOT / "model" |
| THRESHOLD_FILE = MODEL_DIR / "decision_threshold.json" |
| TEMPLATE_FILE = ROOT / "templates" / "index.html" |
| STATIC_DIR = ROOT / "static" |
| MAX_UPLOAD_BYTES = 15 * 1024 * 1024 |
|
|
| with THRESHOLD_FILE.open(encoding="utf-8") as handle: |
| DECISION_THRESHOLD = float(json.load(handle)["decision_threshold"]) |
|
|
| _model: CrossEncoder | None = None |
| _model_lock = threading.RLock() |
|
|
| SECTION_HEADERS = { |
| "summary": re.compile(r"^(?:(?:professional\s+)?summary|profile|about\s+me|objective|career\s+objective)$", re.I), |
| "experience": re.compile(r"^(?:(?:professional\s+)?(?:work\s+)?experience|employment\s+history|career\s+history)$", re.I), |
| "education": re.compile(r"^(?:education(?:al\s+background)?|academic\s+qualifications?)$", re.I), |
| "skills": re.compile(r"^(?:(?:technical\s+|core\s+|key\s+|professional\s+)?skills?|competenc(?:y|ies)|technologies)$", re.I), |
| } |
|
|
|
|
| class JobDetails(BaseModel): |
| job_title: str = Field(default="", max_length=300) |
| seniority: str = Field(default="", max_length=120) |
| industry: str = Field(default="", max_length=200) |
| must_have_skills: str = Field(default="", max_length=5000) |
| nice_to_have_skills: str = Field(default="", max_length=5000) |
| description: str = Field(default="", max_length=20000) |
| responsibilities: str = Field(default="", max_length=10000) |
| requirements: str = Field(default="", max_length=10000) |
|
|
|
|
| class CandidateDetails(BaseModel): |
| role: str = Field(default="", max_length=300) |
| seniority: str = Field(default="", max_length=120) |
| years_experience: str = Field(default="", max_length=80) |
| industry: str = Field(default="", max_length=200) |
| education: str = Field(default="", max_length=5000) |
| skills: str = Field(default="", max_length=8000) |
| summary: str = Field(default="", max_length=10000) |
| experience_bullets: str = Field(default="", max_length=15000) |
|
|
|
|
| class ScoreRequest(BaseModel): |
| job: JobDetails |
| candidate: CandidateDetails |
|
|
|
|
| def clean(value: object, fallback: str = "Not specified") -> str: |
| text = re.sub(r"\s+", " ", str(value or "")).strip() |
| return text or fallback |
|
|
|
|
| def clean_list(value: object) -> str: |
| text = clean(value) |
| if text == "Not specified": |
| return text |
| items = [item.strip() for item in re.split(r"[|,;\n]+", text) if item.strip()] |
| return ", ".join(items) if items else "Not specified" |
|
|
|
|
| def _is_section_header(line: str) -> bool: |
| candidate = re.sub(r"[#:|\-]+$", "", line.strip()) |
| return len(candidate) <= 45 and any(pattern.match(candidate) for pattern in SECTION_HEADERS.values()) |
|
|
|
|
| def _sections(text: str) -> dict[str, list[str]]: |
| sections: dict[str, list[str]] = {"other": []} |
| current = "other" |
| for raw_line in text.splitlines(): |
| line = raw_line.strip(" \t•*-\u2022") |
| if not line: |
| continue |
| header = re.sub(r"[#:|\-]+$", "", line.strip()) |
| matched = next((name for name, pattern in SECTION_HEADERS.items() if pattern.match(header)), None) |
| if matched: |
| current = matched |
| sections.setdefault(current, []) |
| continue |
| sections.setdefault(current, []).append(line) |
| return sections |
|
|
|
|
| def _first_non_contact_line(text: str) -> str: |
| for raw_line in text.splitlines()[:12]: |
| line = raw_line.strip(" \t•*-\u2022") |
| if not line or _is_section_header(line): |
| continue |
| if re.search(r"@|https?://|www\.|(?:\+?\d[\d ()-]{7,})", line, re.I): |
| continue |
| if len(line.split()) <= 10: |
| return line |
| return "" |
|
|
|
|
| def _section_text(section: list[str], limit: int = 800) -> str: |
| return " ".join(section).strip()[:limit] |
|
|
|
|
| def _section_items(section: list[str], limit: int = 1200) -> str: |
| return "; ".join(item for item in section if item.strip())[:limit] |
|
|
|
|
| def _inline_section(text: str, labels: str, next_labels: str, limit: int) -> str: |
| """Recover sections when a PDF extractor puts several headings on one line.""" |
| match = re.search( |
| rf"\b(?:{labels})\b\s*[:\-]\s*(.*?)(?=\s+\b(?:{next_labels})\b\s*[:\-]\s*|$)", |
| " ".join(text.split()), |
| re.I, |
| ) |
| return re.sub(r"\s+", " ", match.group(1)).strip()[:limit] if match else "" |
|
|
|
|
| def extract_candidate_fields(text: str) -> dict[str, str]: |
| sections = _sections(text) |
| all_text = " ".join(text.split()) |
| first_line = _first_non_contact_line(text) |
| role_keywords = ("developer", "engineer", "manager", "analyst", "designer", "specialist", "officer", "assistant", "consultant", "accountant", "teacher", "intern") |
|
|
| role = "" |
| for line in sections.get("other", [])[:8]: |
| if line == first_line: |
| continue |
| role_parts = re.split(r"\s*(?:\||•|·)\s*|\s{2,}", line) |
| role = next((part.strip() for part in role_parts if any(keyword in part.lower() for keyword in role_keywords)), "") |
| if role: |
| break |
|
|
| if not role: |
| role_match = re.search(r"\b(?:role|position|job\s+title)\s*[:\-]\s*([^|\n]+)", text, re.I) |
| role = role_match.group(1).strip() if role_match else first_line |
|
|
| seniority_match = re.search(r"\b(associate|intern|internship|entry[- ]?level|junior|mid[- ]?level|intermediate|senior|lead|principal|manager|director|executive)\b", all_text, re.I) |
| years_match = re.search(r"\b(\d+(?:\.\d+)?)\+?\s*(?:years?|yrs?)\b", all_text, re.I) |
| industry_terms = ( |
| "FinTech", "Banking", "Finance", "Software", "Information Technology", "Healthcare", "Education", |
| "Marketing", "Tourism", "Retail", "Logistics", "Construction", "Manufacturing", "Telecommunications", |
| "Insurance", "Hospitality", "Government", "E-commerce", |
| ) |
| industry = next((term for term in industry_terms if re.search(re.escape(term), all_text, re.I)), "") |
| if not industry: |
| industry_match = re.search(r"\bindustry\s*[:\-]\s*([^|\n]+)", text, re.I) |
| industry = industry_match.group(1).strip() if industry_match else "" |
|
|
| summary = _section_text(sections.get("summary", []), 900) or _inline_section( |
| text, r"professional\s+summary|summary|profile|objective", r"experience|education|skills|certifications", 900 |
| ) |
| if not summary: |
| summary = _section_text(sections.get("other", [])[1:5], 900) |
| education = _section_items(sections.get("education", []), 600) or _inline_section( |
| text, r"education|academic\s+qualifications?", r"experience|skills|summary|certifications", 600 |
| ) |
| skills = _section_items(sections.get("skills", [])) or _inline_section( |
| text, r"technical\s+skills|core\s+skills|skills|competenc(?:y|ies)|technologies", r"experience|education|summary|certifications", 1200 |
| ) |
| experience = _section_items(sections.get("experience", []), 1400) or _inline_section( |
| text, r"professional\s+experience|work\s+experience|experience|employment\s+history", r"education|skills|summary|certifications", 1400 |
| ) |
|
|
| return { |
| "role": role or first_line, |
| "seniority": seniority_match.group(1).replace("-", " ").title() if seniority_match else "", |
| "years_experience": years_match.group(1) if years_match else "", |
| "industry": industry, |
| "education": education, |
| "skills": skills, |
| "summary": summary, |
| "experience_bullets": experience, |
| } |
|
|
|
|
| def extract_cv_text(filename: str, contents: bytes) -> str: |
| suffix = Path(filename or "").suffix.lower() |
| if suffix == ".pdf": |
| text = "" |
| try: |
| import fitz |
|
|
| document = fitz.open(stream=contents, filetype="pdf") |
| text = "\n".join(block[4] for page in document for block in page.get_text("blocks", sort=True) if len(block) > 4) |
| except Exception: |
| text = "" |
| if len(text.strip()) < 40: |
| from pypdf import PdfReader |
|
|
| text = "\n".join(page.extract_text() or "" for page in PdfReader(io.BytesIO(contents)).pages) |
| return text |
| if suffix == ".docx": |
| from docx import Document |
|
|
| document = Document(io.BytesIO(contents)) |
| parts = [paragraph.text for paragraph in document.paragraphs] |
| parts.extend(cell.text for table in document.tables for row in table.rows for cell in row.cells) |
| return "\n".join(parts) |
| if suffix in {".txt", ".md"}: |
| for encoding in ("utf-8", "utf-16", "cp1252"): |
| try: |
| return contents.decode(encoding) |
| except UnicodeDecodeError: |
| continue |
| return contents.decode("utf-8", errors="ignore") |
| raise ValueError("Supported CV formats are PDF, DOCX and TXT.") |
|
|
|
|
| def build_resume_text(candidate: CandidateDetails) -> str: |
| return clean( |
| f""" |
| Candidate Role: {clean(candidate.role)}. |
| Seniority Level: {clean(candidate.seniority)}. |
| Years of Experience: {clean(candidate.years_experience)}. |
| Industry Experience: {clean(candidate.industry)}. |
| Education: {clean(candidate.education)}. |
| Skills: {clean_list(candidate.skills)}. |
| Professional Summary: {clean(candidate.summary)}. |
| Experience Highlights: {clean_list(candidate.experience_bullets)}. |
| """ |
| ) |
|
|
|
|
| def build_job_text(job: JobDetails) -> str: |
| return clean( |
| f""" |
| Job Title: {clean(job.job_title)}. |
| Required Seniority: {clean(job.seniority)}. |
| Industry: {clean(job.industry)}. |
| Must-Have Skills: {clean_list(job.must_have_skills)}. |
| Nice-to-Have Skills: {clean_list(job.nice_to_have_skills)}. |
| Job Description: {clean(job.description)}. |
| Responsibilities: {clean_list(job.responsibilities)}. |
| Requirements: {clean_list(job.requirements)}. |
| """ |
| ) |
|
|
|
|
| def get_model() -> CrossEncoder: |
| global _model |
| if _model is None: |
| with _model_lock: |
| if _model is None: |
| device = "cuda" if torch.cuda.is_available() else "cpu" |
| _model = CrossEncoder(str(MODEL_DIR), max_length=256, device=device) |
| return _model |
|
|
|
|
| def predict_match(job: JobDetails, candidate: CandidateDetails) -> float: |
| with _model_lock: |
| logits = np.asarray( |
| get_model().predict( |
| [(build_job_text(job), build_resume_text(candidate))], |
| activation_fn=torch.nn.Sigmoid(), |
| convert_to_numpy=True, |
| show_progress_bar=False, |
| ) |
| ).reshape(-1) |
| return float(np.clip(logits[0], 0.0, 1.0)) |
|
|
|
|
| app = FastAPI(title="Sri Lankan Job-Resume Matcher", version="2.0.0") |
| app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") |
|
|
|
|
| @app.get("/", response_class=FileResponse) |
| async def index() -> str: |
| return str(TEMPLATE_FILE) |
|
|
|
|
| @app.get("/health") |
| async def health() -> dict[str, object]: |
| return {"status": "ok", "model_loaded": _model is not None, "threshold": DECISION_THRESHOLD} |
|
|
|
|
| @app.post("/api/extract-cv") |
| async def extract_cv(file: UploadFile = File(...)) -> dict[str, object]: |
| filename = file.filename or "uploaded-cv" |
| contents = await file.read() |
| if len(contents) > MAX_UPLOAD_BYTES: |
| raise HTTPException(status_code=413, detail="CV file is too large. Maximum size is 15 MB.") |
| try: |
| text = re.sub(r"\r", "", extract_cv_text(filename, contents)) |
| text = re.sub(r"[ \t]+", " ", text) |
| text = re.sub(r"\n{3,}", "\n\n", text).strip() |
| if not text: |
| raise ValueError("No selectable text found. This may be a scanned PDF.") |
| except ValueError as exc: |
| raise HTTPException(status_code=400, detail=str(exc)) from exc |
| except Exception as exc: |
| raise HTTPException(status_code=400, detail=f"Could not read this CV: {exc}") from exc |
|
|
| candidate = extract_candidate_fields(text) |
| warnings = [] |
| if len(text) < 120: |
| warnings.append("Only a small amount of text was extracted. If this is a scanned PDF, OCR is needed.") |
| if not any(candidate.values()): |
| warnings.append("No candidate fields were detected. Review the text preview and fill the fields manually.") |
|
|
| return { |
| "filename": filename, |
| "preview": text[:2500] + ("…" if len(text) > 2500 else ""), |
| "candidate": candidate, |
| "warnings": warnings, |
| } |
|
|
|
|
| @app.post("/api/score") |
| async def score(request: ScoreRequest) -> dict[str, object]: |
| try: |
| probability = predict_match(request.job, request.candidate) |
| except Exception as exc: |
| raise HTTPException(status_code=500, detail=f"Model scoring failed: {exc}") from exc |
| return { |
| "verdict": "Suitable match" if probability >= DECISION_THRESHOLD else "Needs review", |
| "probability": probability, |
| "threshold": DECISION_THRESHOLD, |
| "disclaimer": "Use this score as a ranking aid only. Review the original CV and job requirements before making any hiring decision.", |
| } |
|
|
|
|
| if __name__ == "__main__": |
| uvicorn.run("app:app", host="127.0.0.1", port=8000, reload=True) |
|
|
