"""Unified, MANDATORY server-side job-description preprocessing. Every JD entering the résumé pipeline — from the Chrome extension, Telegram, a server-side URL fetch, pasted text, or any future client — MUST pass through `preprocess_jd()` before keyword extraction. The pipeline never trusts a client to deliver clean text. Design contract: * Input is treated as UNTRUSTED data (may be a whole scraped page, may contain injected instructions, recruiter cards, related jobs, hashtags, UI chrome). * Output isolates the actual role requirements and reports a confidence score plus diagnostics. * FAIL-SAFE: when JD content cannot be isolated with confidence, `ok=False` and the caller must NOT proceed to modify a résumé (return manual-review status). This module reuses the HTML noise selectors / line-noise list already proven in `jd_from_url.py` (single source of truth for those constants) and adds text-mode cleaning, section isolation, person/hashtag/handle stripping, and the confidence gate on top. """ from __future__ import annotations import re from dataclasses import dataclass, field, asdict from typing import Dict, List # Reuse the proven constants/helpers rather than re-deriving them (single source). from .jd_from_url import ( _JD_SIGNALS, _LINE_NOISE, _scrub_lines, _has_jd_signal, _extract_from_html, ) # ── Section headings that DELIMIT genuine role content vs page/company noise ── # Headings whose CONTENT we keep (role requirements). _KEEP_HEADINGS = ( "responsibilit", "requirement", "qualification", "what you'll do", "what you will do", "what you'll bring", "who you are", "your role", "in this role", "day to day", "day-to-day", "key skills", "must have", "must-have", "nice to have", "nice-to-have", "preferred", "the role", "about the role", "about the job", "role overview", "job summary", "what we're looking for", "what we are looking for", "your impact", "skills", "experience", "we're looking for", "we are looking for", ) # Headings whose CONTENT is company/marketing/page noise (drop the section body). _DROP_HEADINGS = ( "about us", "about the company", "who we are", "our story", "our mission", "our values", "our culture", "life at", "why join", "benefits", "perks", "what we offer", "equal opportunity", "eeo", "diversity", "compensation", "salary", "related jobs", "similar jobs", "people also viewed", "recommended for you", "more jobs", "recruiter", "hiring manager", "meet the team", "follow", "followers", "connect with", "share this job", ) # Lines that are engagement / subscription / notification / social chrome. _ENGAGEMENT_NOISE = ( "like", "comment", "share", "repost", "reactions", "followers", "following", "subscribe", "notification", "ll remind", "remind you", "trial ends", "days before", "renews", "cancel anytime", "see more", "see less", "show more", "show less", "load more", "view all", "connections", "who viewed", "people you may know", "add to your feed", "premium", "upgrade", "try free", "get started free", "start free trial", ) # Prompt-injection / manipulation phrases. A line matching any of these is # dropped BEFORE extraction, so neither the LLM nor the deterministic fallback # ever sees "add X as a required skill" style instructions embedded in the page. _INJECTION_RE = re.compile( r"(ignore\s+(all\s+)?(previous|prior|above)\s+instructions" r"|disregard\s+(the\s+)?(above|previous|prior|earlier)" r"|add\s+[\w,\s/&+-]+\s+as\s+(a\s+)?(required|mandatory|preferred|must[- ]have)" r"|you\s+(must|should|need to)\s+(add|include|output|treat|extract|ignore|append)" r"|(system|developer)\s+prompt" r"|as\s+an?\s+(ai|language\s+model|assistant)" r"|prompt\s+injection" r"|override\s+(the\s+)?(instructions|rules|system))", re.I, ) _HASHTAG_RE = re.compile(r"(?:^|\s)#\w[\w-]*", re.UNICODE) _HANDLE_RE = re.compile(r"(?:^|\s)@\w[\w.\-]*", re.UNICODE) _MULTISPACE_RE = re.compile(r"[ \t ]+") # A camel/glued job-board hashtag with NO spaces (e.g. "warehousejobs", # "dubaicareers", "noonuae") — very high-signal scrape noise, never a real skill. _GLUED_JOBTAG_RE = re.compile( r"\b\w*(?:jobs?|careers?|hiring|vacan\w+|walkin\w*|recruit\w*|" r"opportunit\w+|openings?)\b", re.I, ) @dataclass class PreprocessResult: ok: bool clean_text: str = "" sections: Dict[str, str] = field(default_factory=dict) confidence: float = 0.0 diagnostics: Dict = field(default_factory=dict) dropped_samples: List[str] = field(default_factory=list) reason: str = "" def to_dict(self) -> dict: return asdict(self) def _looks_like_html(raw: str) -> bool: low = (raw or "")[:4000].lower() return ("" in low or " str: """Whitespace + encoding normalization.""" if not text: return "" # Common mojibake / smart punctuation → ASCII-ish. text = (text.replace("‘", "'").replace("’", "'") .replace("“", '"').replace("”", '"') .replace("–", "-").replace("—", "-") .replace(" ", " ").replace("", "")) text = _MULTISPACE_RE.sub(" ", text) text = re.sub(r"\n{3,}", "\n\n", text) return text.strip() def _strip_social(line: str) -> str: """Remove hashtags and @handles from a line.""" line = _HASHTAG_RE.sub(" ", line) line = _HANDLE_RE.sub(" ", line) return _MULTISPACE_RE.sub(" ", line).strip() def _is_engagement(low: str) -> bool: # Short lines that ARE an engagement/subscription token. if len(low) <= 40 and any(low == n or low.startswith(n + " ") or low == n + "s" for n in _ENGAGEMENT_NOISE): return True return any(n in low for n in ("ll remind", "trial ends", "days before", "cancel anytime", "start free trial", "be an early applicant", "easy apply")) def _looks_like_person_line(line: str) -> bool: """A standalone recruiter/employee card line: 1-4 Title-Case words, no verb, not tied to a reporting relationship. Conservative — only drops SHORT lines that are just a name (optionally with a title after a dash/comma).""" s = line.strip() if len(s) > 60 or not s: return False # "Reports to" / "reporting to" relationships are legitimate JD content — keep. if re.search(r"report(s|ing)?\s+to", s, re.I): return False # Never treat a known section heading as a person line. low_full = s.lower().rstrip(":").strip() if any(low_full == h or low_full.startswith(h) for h in _KEEP_HEADINGS + _DROP_HEADINGS): return False head = re.split(r"[-–—,|]", s, 1)[0].strip() words = head.split() if not (1 <= len(words) <= 4): return False # All words Title-Case alphabetic (a name), and the line has no lowercase verb. if not all(re.match(r"^[A-Z][a-z'.]+$", w) for w in words): return False # Reject if it contains a common role/skill word (that'd be a real heading). low = head.lower() if any(k in low for k in ("manager", "engineer", "product", "developer", "analyst", "designer", "lead", "director", "scientist", "specialist", "consultant")): return False return True def _clean_text_lines(text: str) -> tuple[str, list[str]]: """Line-level scrub: drop UI/legal/engagement/person/hashtag noise. Returns (clean_text, dropped_samples).""" dropped: list[str] = [] out: list[str] = [] for ln in text.splitlines(): raw = ln.strip() if not raw: out.append("") continue low = raw.lower() if _INJECTION_RE.search(raw): dropped.append(raw) continue if any(n in low for n in _LINE_NOISE): dropped.append(raw) continue if _is_engagement(low): dropped.append(raw) continue if _looks_like_person_line(raw): dropped.append(raw) continue cleaned = _strip_social(raw) if not cleaned: dropped.append(raw) continue out.append(cleaned) return "\n".join(out).strip(), dropped def _isolate_sections(text: str) -> tuple[Dict[str, str], str]: """Split text into heading-delimited sections; keep role-requirement sections, drop company/marketing/related-jobs sections. Returns (kept_sections, kept_text). Heuristic heading = a short line (<80 chars) that matches a known heading and is not itself a sentence. When no headings are found, the whole (line-cleaned) text is treated as one 'body' section.""" lines = text.splitlines() sections: Dict[str, List[str]] = {} cur = "_preamble" sections[cur] = [] order: List[str] = [cur] def _heading_of(line: str) -> str | None: """A STANDALONE heading line only — not an inline-labelled content line. 'Requirements' / 'About Us' are headings; 'Requirements: 5+ years ...' is content (substantial text follows the label, so it stays in its section).""" s = line.strip() if not s or len(s) > 60: return None low = s.lower().rstrip(":").strip() for h in _KEEP_HEADINGS + _DROP_HEADINGS: if low == h: return h if low.startswith(h): residual = low[len(h):].strip(" :-–—").strip() # Heading only if ≤2 residual words (e.g. "about the role"). if len(residual.split()) <= 2: return h return None for ln in lines: h = _heading_of(ln) if h is not None: cur = h if cur not in sections: sections[cur] = [] order.append(cur) continue sections[cur].append(ln) # If the page has REAL role-content headings, anything before the first such # heading (_preamble) is page chrome (recruiter card, hashtags, related jobs, # "is hiring" lines) — drop it. Only trust the preamble when no headings exist. has_keep_heading = any( name != "_preamble" and any(name == k or name.startswith(k) for k in _KEEP_HEADINGS) and "\n".join(sections[name]).strip() for name in order ) kept: Dict[str, str] = {} kept_parts: List[str] = [] for name in order: body = "\n".join(sections[name]).strip() if not body: continue is_drop = any(name == d or name.startswith(d) for d in _DROP_HEADINGS) if is_drop: continue if name == "_preamble": if has_keep_heading: continue # pre-heading chrome — drop when real sections exist if not _has_jd_signal(body) and len(body) < 200: kept.setdefault("_preamble", body) continue kept[name] = body kept_parts.append(body) kept_text = "\n\n".join(kept_parts).strip() if not kept_text: # nothing matched keep-headings → fall back to whole body whole = "\n".join(l for n in order for l in sections[n]).strip() kept_text = whole kept = {"_body": whole} if whole else {} return kept, kept_text def _score_confidence(clean_text: str, sections: Dict[str, str]) -> float: """0..1 confidence that we isolated a real JD (not a contaminated page).""" if not clean_text: return 0.0 low = clean_text.lower() signal_hits = sum(1 for s in _JD_SIGNALS if s in low) has_reqs = any("requirement" in n or "responsibilit" in n or "qualification" in n or "what you" in n or "the role" in n for n in sections) length = len(clean_text) score = 0.0 score += min(signal_hits / 6.0, 1.0) * 0.5 # JD-signal density score += 0.25 if has_reqs else 0.0 # found a requirements-type section score += 0.25 if 250 <= length <= 20000 else (0.1 if length >= 120 else 0.0) return round(min(score, 1.0), 3) def preprocess_jd(raw: str, *, company: str = "", min_confidence: float = 0.4) -> PreprocessResult: """Isolate genuine job-description content from any (untrusted) input. Args: raw: the JD input — plain text OR raw HTML, from ANY source. company: hiring company name (used later; kept for diagnostics parity). min_confidence: below this, `ok=False` (fail-safe — do not modify résumé). Returns a PreprocessResult. Callers MUST check `.ok` before extraction. """ raw = raw or "" diag: Dict = {"input_chars": len(raw), "input_mode": None} if not raw.strip(): return PreprocessResult(ok=False, reason="empty_input", diagnostics=diag) # 1. HTML vs text. if _looks_like_html(raw): diag["input_mode"] = "html" _title, extracted = _extract_from_html(raw) base = extracted or "" else: diag["input_mode"] = "text" base = raw base = _normalize(base) # 2. Section isolation FIRST (while headings are intact) — keep role content, # drop company/marketing/related-jobs sections by heading. sections, section_text = _isolate_sections(base) # 3. Line-level noise scrub of the kept text (UI/legal/engagement/person/hashtag). line_clean, dropped = _clean_text_lines(section_text) # 4. Drop glued job-board tags token-wise (they survive line scrub inside prose). line_clean = _GLUED_JOBTAG_RE.sub(" ", line_clean) clean_text = _normalize(_MULTISPACE_RE.sub(" ", line_clean)) # 5. Confidence gate. confidence = _score_confidence(clean_text, sections) diag.update({ "output_chars": len(clean_text), "sections_kept": list(sections.keys()), "lines_dropped": len(dropped), "jd_signal": _has_jd_signal(clean_text), }) if not clean_text or len(clean_text) < 120 or not _has_jd_signal(clean_text): return PreprocessResult( ok=False, clean_text=clean_text, sections=sections, confidence=confidence, diagnostics=diag, dropped_samples=dropped[:20], reason="no_jd_content_isolated") if confidence < min_confidence: return PreprocessResult( ok=False, clean_text=clean_text, sections=sections, confidence=confidence, diagnostics=diag, dropped_samples=dropped[:20], reason="low_confidence") return PreprocessResult( ok=True, clean_text=clean_text, sections=sections, confidence=confidence, diagnostics=diag, dropped_samples=dropped[:20], reason="ok") if __name__ == "__main__": # ponytail: runnable self-check, no framework contaminated = """ Noon.com | 1,120+ followers Sivani Sanjana is hiring #dubaijobs #noonuae #warehousejobs Amit Virmani commented on this People also viewed Senior Analyst at Amazon · Dubai We'll remind you 7 days before your trial ends About the Role We are looking for a Product Manager to own the e-commerce roadmap. Responsibilities: stakeholder management, A/B testing, SQL, product analytics. Requirements: 5+ years product management experience. Agile delivery. About Us Noon is the region's homegrown marketplace founded by Mohamed Alabbar. """ r = preprocess_jd(contaminated, company="Noon") assert r.ok, f"expected ok, got {r.reason} (conf={r.confidence})" low = r.clean_text.lower() for bad in ("sivani sanjana", "amit virmani", "dubaijobs", "noonuae", "trial ends", "people also viewed", "mohamed alabbar", "1,120+ followers"): assert bad not in low, f"contamination survived: {bad!r}\n{r.clean_text}" for good in ("stakeholder management", "a/b testing", "product management"): assert good in low, f"genuine JD content dropped: {good!r}" # Garbage-only input must FAIL safe. g = preprocess_jd("#jobs #hiring follow us • 1,120 followers like comment share") assert not g.ok, "garbage page should fail-safe" print("jd_preprocess self-check PASSED (conf=%.2f, sections=%s)" % (r.confidence, list(r.sections.keys())))