Blame for app.py

"""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)