from __future__ import annotations import hashlib import math import re from datetime import datetime from .narrative import career_text, current_role, full_text, profile, summary_text PROF_WEIGHT = {"beginner": 0.30, "intermediate": 0.60, "advanced": 0.85, "expert": 1.00} TIER_ORDER = {"trap": 0, "weak": 1, "possible": 2, "strong": 3} AI_SKILL_TERMS = ( "machine learning", "deep learning", "nlp", "pytorch", "tensorflow", "llm", "rag", "vector", "embedding", "fine-tun", "lora", "qlora", "faiss", "ranking", "recommendation", "search", "retrieval", "python", "evaluation", ) ASSESSMENT_TERMS = ( "machine learning", "deep learning", "nlp", "pytorch", "tensorflow", "llm", "rag", "vector", "embedding", "ranking", "retrieval", "python", "evaluation", "search", ) CV_SPEECH_ROBOTICS = ("computer vision", "image classification", "speech", "tts", "robotics", "autonomous") DEFAULT_CONSULTING = ( "tcs", "tata consultancy", "infosys", "wipro", "accenture", "cognizant", "capgemini", "hcl", "tech mahindra", "mindtree", "ltimindtree", "mphasis", "hexaware", ) STARTUP_SIZES = {"1-10", "11-50", "51-200", "201-500"} def _lower(x: object) -> str: return str(x or "").lower() def _year(date_text: object) -> int | None: m = re.search(r"\d{4}", str(date_text or "")) return int(m.group(0)) if m else None def _title_tier(title: str, spec: dict) -> tuple[str, float]: tl = _lower(title) tiers = spec.get("title_tiers") or {} for trap in tiers.get("trap", []): if _lower(trap) in tl: return "trap", 0.0 for strong in tiers.get("strong", []): if _lower(strong) in tl: return "strong", 1.0 for possible in tiers.get("possible", []): if _lower(possible) in tl: return "possible", 0.70 if any(k in tl for k in ("machine learning", " ml", " ai ", "ai engineer", "nlp", "search", "recommend", "data scien")): return "possible", 0.65 return "weak", 0.35 def _best_title_tier(candidate: dict, spec: dict) -> tuple[str, float, str]: best_tier, best_score, best_title = "weak", 0.35, str(profile(candidate).get("current_title") or "") for role in candidate.get("career_history") or []: title = str(role.get("title") or "") tier, base = _title_tier(title, spec) months = int(role.get("duration_months") or 0) weighted = base * min(1.0, 0.35 + months / 30.0) if TIER_ORDER[tier] > TIER_ORDER[best_tier] or ( TIER_ORDER[tier] == TIER_ORDER[best_tier] and weighted > best_score ): best_tier, best_score, best_title = tier, weighted, title cur_title = str(profile(candidate).get("current_title") or "") cur_tier, cur_score = _title_tier(cur_title, spec) if TIER_ORDER[cur_tier] >= TIER_ORDER[best_tier]: return cur_tier, max(best_score, cur_score), cur_title or best_title return best_tier, best_score, best_title def _family_score(text: str, terms: list[str], hits_out: list[str], cap_hits: int = 3) -> float: hits = [term for term in terms if _lower(term) in text] hits_out.extend(hits[:3]) return min(1.0, len(hits) / cap_hits) def career_evidence(text: str, spec: dict) -> tuple[float, list[str]]: tl = _lower(text) families = spec.get("career_evidence_terms") or {} weights = {"retrieval": 0.28, "ranking": 0.28, "production": 0.24, "llm": 0.12, "plain_language": 0.08} hits: list[str] = [] score = 0.0 for family, weight in weights.items(): score += weight * _family_score(tl, [str(x) for x in families.get(family, [])], hits) return min(1.0, score), list(dict.fromkeys(hits))[:6] def _blended_evidence(candidate: dict, spec: dict) -> tuple[float, list[str], float, float]: ctext = career_text(candidate) stext = summary_text(candidate) career_score, career_hits = career_evidence(ctext, spec) summary_score, summary_hits = career_evidence(stext, spec) combined = min(1.0, 0.82 * career_score + 0.18 * summary_score) hits = list(dict.fromkeys([*career_hits, *summary_hits]))[:6] return combined, hits, career_score, summary_score def skill_trust(candidate: dict) -> tuple[float, int, list[str]]: values: list[float] = [] ai_values: list[float] = [] ai_names: list[str] = [] for skill in candidate.get("skills") or []: name = str(skill.get("name") or "") nl = name.lower() prof = PROF_WEIGHT.get(_lower(skill.get("proficiency")), 0.40) endorsements = int(skill.get("endorsements") or 0) months = float(skill.get("duration_months") or 0) trust = prof * math.log1p(endorsements) * min(months, 48) / 48.0 trust = min(1.0, trust / 3.0) values.append(trust) if any(term in nl for term in AI_SKILL_TERMS): ai_values.append(trust) ai_names.append(name) if ai_values: return sum(ai_values) / len(ai_values), len(ai_values), ai_names[:6] return (sum(values) / len(values) if values else 0.0), 0, [] def assessment_score(candidate: dict) -> float: scores = (candidate.get("redrob_signals") or {}).get("skill_assessment_scores") or {} vals = [float(v) for k, v in scores.items() if any(term in _lower(k) for term in ASSESSMENT_TERMS)] if not vals: return 0.0 return max(0.0, min(1.0, sum(vals) / len(vals) / 100.0)) def _education_score(candidate: dict) -> float: edu = candidate.get("education") or [] if not edu: return 0.25 score = 0.0 for item in edu: field = _lower(item.get("field_of_study")) degree = _lower(item.get("degree")) tier = _lower(item.get("tier")) if any(x in field for x in ("computer", "machine learning", "artificial", "data science", "software", "electrical")): score += 0.35 if any(x in degree for x in ("m.tech", "m.s", "ms", "phd", "master")): score += 0.20 score += {"tier_1": 0.25, "tier_2": 0.18, "tier_3": 0.10, "tier_4": 0.05}.get(tier, 0.05) return min(1.0, score) def _company_scale_score(candidate: dict) -> float: p = profile(candidate) role = current_role(candidate) score = 0.0 for size in (p.get("current_company_size"), role.get("company_size")): if str(size or "") in STARTUP_SIZES: score += 0.35 elif str(size or "") in {"501-1000", "1001-5000"}: score += 0.20 elif str(size or "") == "10001+": score += 0.05 return min(1.0, score) def _work_mode_fit(candidate: dict) -> float: mode = _lower((candidate.get("redrob_signals") or {}).get("preferred_work_mode")) if mode in ("hybrid", "flexible"): return 1.0 if mode == "remote": return 0.82 if mode == "onsite": return 0.90 return 0.70 def _platform_activity_score(candidate: dict) -> float: sig = candidate.get("redrob_signals") or {} views = int(sig.get("profile_views_received_30d") or 0) apps = int(sig.get("applications_submitted_30d") or 0) search = int(sig.get("search_appearance_30d") or 0) saved = int(sig.get("saved_by_recruiters_30d") or 0) score = 0.0 if views >= 20: score += 0.25 elif views >= 5: score += 0.12 if apps >= 3: score += 0.15 if search >= 10: score += 0.25 elif search >= 3: score += 0.12 if saved >= 3: score += 0.20 elif saved >= 1: score += 0.10 return min(1.0, score) def _verified_trust(candidate: dict) -> float: sig = candidate.get("redrob_signals") or {} flags = [bool(sig.get("verified_email")), bool(sig.get("verified_phone")), bool(sig.get("linkedin_connected"))] return sum(flags) / 3.0 def _yoe_location_fit(candidate: dict, spec: dict) -> float: p = profile(candidate) sig = candidate.get("redrob_signals") or {} yoe = float(p.get("years_of_experience") or 0) yspec = spec.get("yoe", {}) ymin, ymax = float(yspec.get("min", 5.0)), float(yspec.get("max", 9.0)) if ymin <= yoe <= ymax: yoe_score = 1.0 elif yoe < ymin: yoe_score = max(0.25, 1.0 - (ymin - yoe) / 4.0) else: yoe_score = max(0.35, 1.0 - (yoe - ymax) / 7.0) loc_text = f"{p.get('location', '')} {p.get('country', '')}".lower() boosts = [_lower(x) for x in spec.get("location_boost", [])] loc_score = 1.0 if any(x and x in loc_text for x in boosts) else 0.35 if sig.get("willing_to_relocate"): loc_score = max(loc_score, 0.70) return 0.60 * yoe_score + 0.40 * loc_score def _consulting_only(candidate: dict, spec: dict) -> bool: roles = candidate.get("career_history") or [] if len(roles) < 2: return False firms = tuple(_lower(x) for x in spec.get("consulting_companies", DEFAULT_CONSULTING)) consulting_roles = 0 for role in roles: blob = f"{role.get('company','')} {role.get('industry','')} {role.get('description','')}".lower() if any(firm in blob for firm in firms) or any(term in blob for term in ("consulting", "outsourc", "staffing")): consulting_roles += 1 return consulting_roles == len(roles) def _product_company_score(candidate: dict, spec: dict) -> float: roles = candidate.get("career_history") or [] if not roles: return 0.0 terms = tuple(_lower(x) for x in spec.get("product_company_terms", [])) score = 0.0 for role in roles: blob = f"{role.get('company','')} {role.get('industry','')} {role.get('description','')}".lower() if any(term in blob for term in terms): score += 0.35 if not any(firm in blob for firm in DEFAULT_CONSULTING): score += 0.10 return min(1.0, score) def _research_only(candidate: dict, career_ev: float) -> bool: blob = full_text(candidate).lower() research = any(term in blob for term in ("research scientist", "academic lab", "phd", "paper", "publication", "university lab")) production = any(term in blob for term in ("production", "shipped", "deployed", "real users", "a/b test", "serving")) return research and not production and career_ev < 0.35 def _langchain_only(candidate: dict, career_ev: float) -> bool: blob = full_text(candidate).lower() has_framework = "langchain" in blob or "llamaindex" in blob real_ir = any(term in blob for term in ("retrieval", "ranking", "faiss", "bm25", "elasticsearch", "recommendation", "ndcg", "mrr")) return has_framework and not real_ir and career_ev < 0.35 def _title_chaser(candidate: dict) -> bool: roles = sorted(candidate.get("career_history") or [], key=lambda r: str(r.get("start_date") or ""), reverse=True) recent = roles[:5] short_roles = sum(1 for role in recent if int(role.get("duration_months") or 0) < 18) title_words = " ".join(_lower(r.get("title")) for r in recent) senior_titles = sum(1 for word in ("senior", "staff", "principal", "lead") if word in title_words) return len(recent) >= 4 and short_roles >= 4 and senior_titles >= 2 def _summary_title_mismatch(candidate: dict) -> bool: p = profile(candidate) summary = _lower(p.get("summary")) title = _lower(p.get("current_title")) if "marketing manager" in summary and "marketing" not in title: return True if "my professional background is in marketing manager" in summary and "marketing" not in title: return True return False def _cv_speech_without_ir(candidate: dict) -> bool: blob = full_text(candidate).lower() cv_hits = sum(1 for term in CV_SPEECH_ROBOTICS if term in blob) ir_hits = sum(1 for term in ("retrieval", "ranking", "recommendation", "nlp", "search", "semantic") if term in blob) return cv_hits >= 3 and ir_hits == 0 def honeypot_flag(candidate: dict, title_tier: str, career_ev: float, ai_count: int, trust: float) -> tuple[bool, list[str]]: p = profile(candidate) yoe = float(p.get("years_of_experience") or 0) skills = candidate.get("skills") or [] roles = candidate.get("career_history") or [] reasons: list[str] = [] expert_tiny = [s for s in skills if _lower(s.get("proficiency")) == "expert" and float(s.get("duration_months") or 0) < 3] if len(expert_tiny) >= 3: reasons.append("expert skills with near-zero duration") expert_zero = [s for s in skills if _lower(s.get("proficiency")) == "expert" and float(s.get("duration_months") or 0) == 0] if expert_zero: reasons.append("expert skill with zero duration") if len(skills) >= 10 and sum(int(s.get("endorsements") or 0) for s in skills) == 0: reasons.append("many skills with zero endorsements") years = [_year(r.get("start_date")) for r in roles if _year(r.get("start_date"))] if years and yoe >= 8 and min(years) > datetime.now().year - 4: reasons.append("claimed YoE contradicts role start dates") total_role_months = sum(int(r.get("duration_months") or 0) for r in roles) if yoe > 0 and total_role_months > int(yoe * 12 + 30): reasons.append("role durations exceed total experience") if yoe > 0 and any(int(r.get("duration_months") or 0) > int(yoe * 12 + 6) for r in roles): reasons.append("single role longer than total experience") edu_end_years = [int(e.get("end_year")) for e in candidate.get("education") or [] if e.get("end_year")] if edu_end_years and years: earliest_job = min(years) latest_edu = max(edu_end_years) if latest_edu > earliest_job + 2 and yoe >= 6: reasons.append("education timeline conflicts with claimed experience") assess = assessment_score(candidate) if assess >= 0.85 and title_tier == "trap" and career_ev < 0.20: reasons.append("high assessments but no matching career proof") if title_tier in ("trap", "weak") and ai_count >= 8 and trust < 0.20 and career_ev < 0.30: reasons.append("keyword-stuffed profile without career proof") return bool(reasons), reasons[:4] def extract_features(candidate: dict, spec: dict) -> dict: p = profile(candidate) sig = candidate.get("redrob_signals") or {} role = current_role(candidate) evidence, evidence_terms, career_only_ev, summary_ev = _blended_evidence(candidate, spec) title = str(p.get("current_title") or "") tier, tier_score, best_title = _best_title_tier(candidate, spec) trust, ai_count, ai_skill_names = skill_trust(candidate) assess = assessment_score(candidate) consulting = _consulting_only(candidate, spec) stuffer = tier in ("weak", "trap") and ai_count >= 6 and trust < 0.35 and evidence < 0.45 research_only = _research_only(candidate, evidence) langchain_only = _langchain_only(candidate, evidence) title_chaser = _title_chaser(candidate) mismatch = _summary_title_mismatch(candidate) cv_only = _cv_speech_without_ir(candidate) hp, hp_reasons = honeypot_flag(candidate, tier, evidence, ai_count, trust) anti = 0.0 if stuffer: anti += 0.50 if consulting: anti += 0.30 if research_only: anti += 0.40 if langchain_only: anti += 0.25 if title_chaser: anti += 0.20 if mismatch: anti += 0.35 if cv_only: anti += 0.20 if tier == "trap" and evidence < 0.60: anti += 0.80 ctext = career_text(candidate) edu_top = "" if candidate.get("education"): e0 = candidate["education"][0] edu_top = f"{e0.get('degree','')} {e0.get('field_of_study','')}".strip() return { "candidate_id": candidate.get("candidate_id"), "current_title": title, "best_career_title": best_title, "current_company": str(p.get("current_company") or ""), "current_industry": str(p.get("current_industry") or ""), "years_of_experience": float(p.get("years_of_experience") or 0), "location": str(p.get("location") or ""), "country": str(p.get("country") or ""), "career_evidence": evidence, "career_only_evidence": career_only_ev, "summary_evidence": summary_ev, "career_terms": "|".join(evidence_terms), "title_tier": tier, "title_tier_score": tier_score, "skill_trust": min(1.0, trust), "ai_skill_count": ai_count, "ai_skill_names": "|".join(ai_skill_names[:5]), "assessment_score": assess, "education_score": _education_score(candidate), "education_top": edu_top, "company_scale_score": _company_scale_score(candidate), "work_mode_fit": _work_mode_fit(candidate), "platform_activity_score": _platform_activity_score(candidate), "verified_trust": _verified_trust(candidate), "yoe_location_fit": _yoe_location_fit(candidate, spec), "product_company_score": _product_company_score(candidate, spec), "anti_pattern_penalty": min(1.0, anti), "honeypot_flag": hp, "honeypot_reasons": "|".join(hp_reasons), "stuffer_flag": stuffer, "consulting_only": consulting, "research_only": research_only, "langchain_only": langchain_only, "title_chaser": title_chaser, "summary_title_mismatch": mismatch, "cv_speech_without_ir": cv_only, "preferred_work_mode": str(sig.get("preferred_work_mode") or ""), "open_to_work_flag": bool(sig.get("open_to_work_flag", False)), "willing_to_relocate": bool(sig.get("willing_to_relocate", False)), "recruiter_response_rate": float(sig.get("recruiter_response_rate") or 0), "avg_response_time_hours": float(sig.get("avg_response_time_hours") or 0), "notice_period_days": int(sig.get("notice_period_days") or 90), "profile_completeness_score": float(sig.get("profile_completeness_score") or 0), "last_active_date": sig.get("last_active_date") or "", "github_activity_score": float(sig.get("github_activity_score") if sig.get("github_activity_score") is not None else -1), "saved_by_recruiters_30d": int(sig.get("saved_by_recruiters_30d") or 0), "profile_views_received_30d": int(sig.get("profile_views_received_30d") or 0), "search_appearance_30d": int(sig.get("search_appearance_30d") or 0), "interview_completion_rate": float(sig.get("interview_completion_rate") or 0), "offer_acceptance_rate": float(sig.get("offer_acceptance_rate") if sig.get("offer_acceptance_rate") is not None else -1), "verified_email": bool(sig.get("verified_email", False)), "verified_phone": bool(sig.get("verified_phone", False)), "linkedin_connected": bool(sig.get("linkedin_connected", False)), "career_hash": hashlib.sha1(ctext.encode("utf-8", errors="ignore")).hexdigest(), "career_hash_count": 1, "current_role_excerpt": str(role.get("description") or "")[:180].replace("\n", " "), } def attach_hash_counts(features_df): counts = features_df["career_hash"].value_counts().to_dict() features_df["career_hash_count"] = features_df["career_hash"].map(counts).astype(int) return features_df def features_dataframe(candidates: list[dict], spec: dict): import pandas as pd df = pd.DataFrame(extract_features(candidate, spec) for candidate in candidates) return attach_hash_counts(df)