410 lines
14 KiB
Python
410 lines
14 KiB
Python
"""CV text cleanup, scoring-JD builder, and the ATS scorer singleton.
|
|
|
|
Pure module: no FastAPI imports and no HTTPException.
|
|
|
|
Designer-made resumes position every glyph individually, so pypdf hands back
|
|
"S K I L L S" instead of "SKILLS". In that layout a single space is glyph
|
|
padding and a run of two or more spaces is the real word gap, which is what
|
|
the `despace_line` decorator keys off to rebuild readable lines.
|
|
|
|
The scoring pieces reuse the bulk-ats engine, installed editable from the repo
|
|
root (`pip install -e ..` -> `import app.*`), and share llm_setup's process-wide
|
|
AsyncOpenAI client rather than opening a second connection pool.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from app.core.config import Settings, get_settings
|
|
from app.services.llm import OpenAIScorer
|
|
# One definition, two extractors. The bulk-ATS engine and this recruiting path
|
|
# both read CVs with pypdf and both broke the same way on glyph-fragmented
|
|
# files, so the repair lives in the package that owns PDF handling and is
|
|
# re-exported here for the callers that import it from this module.
|
|
from app.services.pdf import extract_pdf_text, is_glyph_fragmented # noqa: F401
|
|
from dotenv import load_dotenv
|
|
|
|
from job.candidate.decorators import despace_line, normalize_unicode
|
|
|
|
load_dotenv()
|
|
|
|
# Backend-local per-candidate error code for cases the bulk-ats engine never sees
|
|
# (its uploads always have bytes; inbox attachments can vanish from disk).
|
|
FILE_NOT_FOUND = "FILE_NOT_FOUND"
|
|
|
|
_scorer: OpenAIScorer | None = None
|
|
|
|
|
|
def get_scoring_settings() -> Settings:
|
|
"""Validated scoring knobs (model family, token floor, size limits).
|
|
|
|
Reads real env vars, which load_dotenv() above has populated from the nearest
|
|
.env (backend/.env, else the repo root). Shared names (OPENAI_MODEL,
|
|
OPENAI_MAX_OUTPUT_TOKENS) therefore match what llm_setup uses.
|
|
"""
|
|
return get_settings()
|
|
|
|
|
|
def get_scorer() -> OpenAIScorer:
|
|
"""Process-wide scorer over llm_setup's shared AsyncOpenAI client.
|
|
|
|
Lazy so that a missing/invalid OPENAI configuration surfaces on the first
|
|
scoring request, not at import; llm_setup.init_llm() in the app lifespan has
|
|
normally created and verified the client before this ever runs.
|
|
"""
|
|
global _scorer
|
|
if _scorer is None:
|
|
from llm_setup import get_client
|
|
|
|
settings = get_scoring_settings()
|
|
_scorer = OpenAIScorer(
|
|
get_client(),
|
|
model=settings.openai_model,
|
|
max_output_tokens=settings.openai_max_output_tokens,
|
|
effort=settings.openai_effort,
|
|
enable_cache=settings.openai_enable_prompt_cache,
|
|
)
|
|
return _scorer
|
|
|
|
|
|
def build_job_description(job) -> str:
|
|
"""Deterministic scoring JD from a JobPosts row.
|
|
|
|
Byte-stable per job: derived only from stored column values, in fixed order,
|
|
because OpenAI prompt caching works on exact prefix match — one volatile byte
|
|
(an id, a timestamp) would stop the whole batch from reusing the cache.
|
|
Deliberately excludes post_text (hashtags, salary, LinkedIn formatting) and
|
|
salary; list order is taken as stored.
|
|
"""
|
|
lines = [f"Job Title: {job.title}"]
|
|
if job.employment_type:
|
|
lines.append(f"Employment Type: {job.employment_type}")
|
|
if job.location:
|
|
lines.append(f"Location: {job.location}")
|
|
if job.experience_min is not None and job.experience_max is not None:
|
|
lines.append(f"Experience Required: {job.experience_min}-{job.experience_max} years")
|
|
elif job.experience_min is not None:
|
|
lines.append(f"Experience Required: {job.experience_min}+ years")
|
|
elif job.experience_max is not None:
|
|
lines.append(f"Experience Required: up to {job.experience_max} years")
|
|
if job.description:
|
|
lines += ["", "Description:", job.description.strip()]
|
|
if job.requirements:
|
|
lines += ["", "Mandatory Requirements:"]
|
|
lines += [f"- {item}" for item in job.requirements]
|
|
if job.optional_skills:
|
|
lines += ["", "Preferred (nice to have):"]
|
|
lines += [f"- {item}" for item in job.optional_skills]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def candidate_base_fields(source):
|
|
fields = {
|
|
"filename": source["safe_name"],
|
|
"file_path": source["file_path"],
|
|
"content_sha256": source["sha256"],
|
|
"candidate_email": source.get("candidate_email"),
|
|
}
|
|
# Omit empty FKs so an upsert cannot wipe a link that this source does not know.
|
|
inbox_mid = source.get("inbox_message_id")
|
|
if inbox_mid:
|
|
fields["inbox_message_id"] = inbox_mid
|
|
manual_id = source.get("manual_upload_candidate_id")
|
|
if manual_id:
|
|
fields["manual_upload_candidate_id"] = manual_id
|
|
return fields
|
|
|
|
|
|
def candidate_failed_fields(source, code, message):
|
|
return {
|
|
**candidate_base_fields(source),
|
|
"status": "failed",
|
|
"error_code": str(code),
|
|
"error_message": message,
|
|
"candidate_name": None,
|
|
"job_title": None,
|
|
"current_company": None,
|
|
"years_experience": None,
|
|
"match_score": None,
|
|
"matched_keywords": [],
|
|
"missing_keywords": [],
|
|
"summary_critique": None,
|
|
"linkedin_url": None,
|
|
}
|
|
|
|
|
|
def candidate_completed_fields(source, result):
|
|
return {
|
|
**candidate_base_fields(source),
|
|
"status": "completed",
|
|
"candidate_name": result.candidate_name,
|
|
"job_title": result.job_title,
|
|
"current_company": result.current_company,
|
|
"years_experience": result.years_experience,
|
|
"match_score": result.match_score,
|
|
"matched_keywords": result.matched_keywords,
|
|
"missing_keywords": result.missing_keywords,
|
|
"summary_critique": result.summary_critique,
|
|
"professional_summary": result.professional_summary,
|
|
"linkedin_url": None,
|
|
"error_code": None,
|
|
"error_message": None,
|
|
}
|
|
|
|
|
|
def extract_pdf_link_uris(reader) -> list[str]:
|
|
"""Clickable /URI annotations that pypdf's extract_text() never returns.
|
|
|
|
Designer CVs put LinkedIn (and portfolio) behind an icon; the URL lives on
|
|
the annotation, not in the text layer. Appending these after page text is
|
|
what lets linkedin_utils see a profile the recruiter can open.
|
|
"""
|
|
found: list[str] = []
|
|
seen: set[str] = set()
|
|
try:
|
|
pages = reader.pages
|
|
except Exception:
|
|
return found
|
|
for page in pages:
|
|
try:
|
|
annots = page.get("/Annots")
|
|
if annots is None:
|
|
continue
|
|
if hasattr(annots, "get_object"):
|
|
annots = annots.get_object()
|
|
except Exception:
|
|
continue
|
|
if not annots:
|
|
continue
|
|
for annot in annots:
|
|
try:
|
|
obj = annot.get_object() if hasattr(annot, "get_object") else annot
|
|
action = obj.get("/A") if obj is not None else None
|
|
if action is not None and hasattr(action, "get_object"):
|
|
action = action.get_object()
|
|
uri = None
|
|
if action is not None:
|
|
uri = action.get("/URI")
|
|
if uri is None and obj is not None:
|
|
uri = obj.get("/URI")
|
|
if uri is None:
|
|
continue
|
|
value = str(uri).strip()
|
|
if value and value not in seen:
|
|
seen.add(value)
|
|
found.append(value)
|
|
except Exception:
|
|
continue
|
|
return found
|
|
|
|
|
|
@normalize_unicode
|
|
@despace_line
|
|
def normalize_spaced_text(text) -> str:
|
|
"""Turn raw pypdf output into readable text, leaving normal lines untouched.
|
|
|
|
The decorators have already folded the unicode and rebuilt the glyph-padded
|
|
lines; what is left is the whitespace tidy-up that every line wants.
|
|
"""
|
|
if not text:
|
|
return ""
|
|
lines = [re.sub(r" {2,}", " ", line).strip() for line in text.splitlines()]
|
|
return re.sub(r"\n{3,}", "\n\n", "\n".join(lines)).strip()
|
|
|
|
|
|
# Mirrors frontend Inbox.jsx sourceFrom — board name is tagged in the To address.
|
|
INBOX_SOURCES = (
|
|
"Microsoft Outlook",
|
|
"Career Portal",
|
|
"Manual CV Upload",
|
|
"LinkedIn",
|
|
"Indeed",
|
|
"Rozee",
|
|
"Mustakbil",
|
|
"Employee Referral",
|
|
"Recruitment Agency",
|
|
"Campus Hiring",
|
|
"Walk-in",
|
|
)
|
|
|
|
|
|
def _letters_only(value: str) -> str:
|
|
return re.sub(r"[^a-z]", "", (value or "").lower())
|
|
|
|
|
|
def source_from_message_to(message_to: str | None) -> str:
|
|
raw = (message_to or "").strip()
|
|
if not raw:
|
|
return "Unknown"
|
|
flat = _letters_only(raw)
|
|
for name in INBOX_SOURCES:
|
|
if _letters_only(name) and _letters_only(name) in flat:
|
|
return name
|
|
return raw.split(",")[0].strip()
|
|
|
|
|
|
def documents_from_message(file_name: str | None, file_path: str | None) -> list[dict]:
|
|
names = [n.strip() for n in (file_name or "").split(",") if n.strip()]
|
|
paths = [p.strip() for p in (file_path or "").split(",") if p.strip()]
|
|
out = []
|
|
for i, name in enumerate(names):
|
|
out.append({"name": name, "path": paths[i] if i < len(paths) else None})
|
|
if not out and paths:
|
|
for path in paths:
|
|
out.append({"name": path.rsplit("/", 1)[-1].rsplit("\\", 1)[-1], "path": path})
|
|
return out
|
|
|
|
|
|
_ATTACHMENTS_ROOT = Path(__file__).resolve().parents[2] / "inbox" / "decoded_attachments"
|
|
|
|
|
|
def contained_download_path(stored_path: str | None) -> Path | None:
|
|
"""Resolve `stored_path` only if it sits inside decoded_attachments.
|
|
|
|
Never follows a client-supplied path. Returns None on any failure so the
|
|
caller can 404 rather than 403 (a 403 would confirm the file exists).
|
|
|
|
Stored paths may be host-absolute Windows paths (see resolve_attachment_path).
|
|
Those fail the containment check against this process's attachments dir; fall
|
|
back to the basename under decoded_attachments, then contain that too.
|
|
"""
|
|
if not stored_path or not str(stored_path).strip():
|
|
return None
|
|
root = _ATTACHMENTS_ROOT.resolve()
|
|
raw = str(stored_path).strip()
|
|
basename = Path(raw.replace("\\", "/")).name
|
|
|
|
def _contained(path: Path) -> Path | None:
|
|
try:
|
|
resolved = path.resolve()
|
|
resolved.relative_to(root)
|
|
except (OSError, RuntimeError, ValueError):
|
|
return None
|
|
if not resolved.is_file():
|
|
return None
|
|
return resolved
|
|
|
|
try:
|
|
candidate = Path(raw)
|
|
if not candidate.is_absolute():
|
|
candidate = root / basename
|
|
hit = _contained(candidate)
|
|
if hit is not None:
|
|
return hit
|
|
except (OSError, RuntimeError, ValueError):
|
|
pass
|
|
if basename:
|
|
return _contained(root / basename)
|
|
return None
|
|
|
|
|
|
# Same system prefixes inbox/models._is_linkable_sender rejects — anything we
|
|
# accept here must remain linkable when insert_email creates the users row.
|
|
_SKIP_SENDER_PREFIXES = (
|
|
"noreply", "no-reply", "donotreply", "do-not-reply",
|
|
"mailer-daemon", "postmaster", "bounce",
|
|
)
|
|
_ROLE_LOCAL_PARTS = frozenset({
|
|
"info", "hr", "careers", "jobs", "admin", "support", "contact", "sales",
|
|
"recruitment", "office", "team", "hello", "enquiry", "inquiry", "recruit",
|
|
"talent", "hiring", "apply", "applications", "webmaster", "helpdesk",
|
|
})
|
|
_EMAIL_RE = re.compile(
|
|
r"(?i)\b([a-z0-9][a-z0-9._%+\-]{0,63})@([a-z0-9](?:[a-z0-9\-]{0,61}[a-z0-9])?"
|
|
r"(?:\.[a-z0-9](?:[a-z0-9\-]{0,61}[a-z0-9])?)+)\b"
|
|
)
|
|
_PHONE_RE = re.compile(r"(?:\+?\d[\d\s\-().]{7,}\d)")
|
|
_LABEL_RE = re.compile(r"(?i)\b(?:e[\-\s]?mail|mail[\s\-]?id|contact)\b")
|
|
_REF_HEADING_RE = re.compile(r"(?i)^\s*(?:references?|referees?)\b")
|
|
_REF_MENTION_RE = re.compile(
|
|
r"(?i)\b(?:reference|referee|manager|supervisor|contact\s+person)\b"
|
|
)
|
|
_HEADER_LINE_COUNT = 12
|
|
_MIN_ACCEPT_SCORE = 3
|
|
|
|
|
|
def _presumed_name_tokens(lines: list[str]) -> list[str]:
|
|
"""First non-empty line with 2+ alpha tokens and no digits/@ — CV name header."""
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
continue
|
|
if any(ch.isdigit() for ch in stripped) or "@" in stripped:
|
|
continue
|
|
tokens = [_letters_only(t) for t in re.split(r"\s+", stripped) if _letters_only(t)]
|
|
if len(tokens) >= 2:
|
|
return tokens
|
|
return []
|
|
|
|
|
|
def _email_local_ok(local: str) -> bool:
|
|
lowered = (local or "").lower()
|
|
if lowered in _ROLE_LOCAL_PARTS:
|
|
return False
|
|
return not lowered.startswith(_SKIP_SENDER_PREFIXES)
|
|
|
|
|
|
def extract_candidate_email(text: str) -> tuple[str | None, list[str]]:
|
|
"""Pick the candidate's own email from CV text, or None when ambiguous/absent.
|
|
|
|
Returns ``(best, all_plausible)``. Ambiguity is intentional — a wrong guess
|
|
would create a user under a stranger's address and mail them a confirm link.
|
|
"""
|
|
if not text or not text.strip():
|
|
return None, []
|
|
|
|
lines = text.splitlines()
|
|
name_tokens = _presumed_name_tokens(lines)
|
|
in_references = False
|
|
scored: list[tuple[int, int, str]] = [] # (score, first_line_idx, email)
|
|
seen: dict[str, int] = {} # lower email -> index in scored
|
|
|
|
for idx, line in enumerate(lines):
|
|
if _REF_HEADING_RE.search(line):
|
|
in_references = True
|
|
for match in _EMAIL_RE.finditer(line):
|
|
local, domain = match.group(1), match.group(2)
|
|
if not _email_local_ok(local):
|
|
continue
|
|
email = f"{local}@{domain}".lower()
|
|
score = 0
|
|
if idx < _HEADER_LINE_COUNT:
|
|
score += 3
|
|
local_letters = _letters_only(local)
|
|
if local_letters and any(
|
|
tok and (tok in local_letters or local_letters in tok)
|
|
for tok in name_tokens
|
|
):
|
|
score += 3
|
|
if _LABEL_RE.search(line) or _PHONE_RE.search(line):
|
|
score += 1
|
|
if in_references:
|
|
score -= 5
|
|
if _REF_MENTION_RE.search(line):
|
|
score -= 3
|
|
|
|
if email in seen:
|
|
prev_i = seen[email]
|
|
prev_score, _, _ = scored[prev_i]
|
|
if score > prev_score:
|
|
scored[prev_i] = (score, idx, email)
|
|
continue
|
|
seen[email] = len(scored)
|
|
scored.append((score, idx, email))
|
|
|
|
if not scored:
|
|
return None, []
|
|
|
|
scored.sort(key=lambda t: (-t[0], t[1]))
|
|
plausible = [email for _, _, email in scored]
|
|
top_score, _, top_email = scored[0]
|
|
runner_up = scored[1][0] if len(scored) > 1 else None
|
|
if top_score < _MIN_ACCEPT_SCORE:
|
|
return None, plausible
|
|
if runner_up is not None and top_score <= runner_up:
|
|
return None, plausible
|
|
return top_email, plausible
|
|
|