import json import re from datetime import date, datetime from pathlib import Path from typing import Any import pandas as pd from sklearn.feature_extraction.text import HashingVectorizer ID_KEYS = ["candidate_id", "id", "user_id", "profile_id"] NAME_KEYS = ["candidate_name", "name", "full_name"] def load_candidates(path: str | Path) -> pd.DataFrame: path = Path(path) suffix = path.suffix.lower() if suffix == ".csv": return pd.read_csv(path) if suffix == ".jsonl": rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()] return pd.DataFrame(rows) if suffix == ".json": data = json.loads(path.read_text(encoding="utf-8")) if isinstance(data, dict): for key in ["candidates", "data", "profiles", "users"]: if key in data and isinstance(data[key], list): data = data[key] break return pd.DataFrame(data) if suffix == "": text = path.read_text(encoding="utf-8").strip() if not text: return pd.DataFrame() if text[0] == "[" or text[0] == "{": try: data = json.loads(text) if isinstance(data, dict): for key in ["candidates", "data", "profiles", "users"]: if key in data and isinstance(data[key], list): data = data[key] break return pd.DataFrame(data) except json.JSONDecodeError: pass rows = [json.loads(line) for line in text.splitlines() if line.strip()] return pd.DataFrame(rows) raise ValueError(f"Unsupported candidate file format: {path.suffix}") def clean_text(value: Any) -> str: if value is None or (isinstance(value, float) and pd.isna(value)): return "" if isinstance(value, (dict, list)): value = json.dumps(value, ensure_ascii=False) return re.sub(r"\s+", " ", str(value)).strip() def first_present(row: pd.Series, keys: list[str], default: str = "") -> str: for key in keys: if key in row and clean_text(row[key]): return clean_text(row[key]) return default def get_nested(row: pd.Series, key: str, default: Any = "") -> Any: if "." not in key: return row.get(key, default) value: Any = row for part in key.split("."): if isinstance(value, pd.Series): value = value.get(part, default) elif isinstance(value, dict): value = value.get(part, default) else: return default return value def parse_jsonish(value: Any) -> Any: if isinstance(value, (list, dict)): return value if not isinstance(value, str): return value value = value.strip() if not value: return [] try: return json.loads(value) except json.JSONDecodeError: return value def format_skills(skills: Any) -> tuple[str, float]: skills = parse_jsonish(skills) if not skills: return "", 0.0 proficiency_weight = {"beginner": 0.4, "intermediate": 0.7, "advanced": 1.0, "expert": 1.0} lines: list[str] = [] scores: list[float] = [] if isinstance(skills, list): for item in skills: if isinstance(item, dict): name = clean_text(item.get("name", "")) proficiency = clean_text(item.get("proficiency", "")).lower() endorsements = float(item.get("endorsements") or 0) duration = float(item.get("duration_months") or 0) if not name: continue lines.append( f"{name} - {proficiency or 'unknown'}, {int(duration)} months, {int(endorsements)} endorsements" ) p_score = proficiency_weight.get(proficiency, 0.5) d_score = min(duration / 48.0, 1.0) e_score = min(endorsements / 20.0, 1.0) scores.append((0.5 * p_score) + (0.3 * d_score) + (0.2 * e_score)) else: text = clean_text(item) if text: lines.append(text) scores.append(0.4) else: text = clean_text(skills) return text, 0.4 if text else 0.0 return "; ".join(lines), round(sum(scores) / max(len(scores), 1), 4) def days_since(date_value: Any) -> int | None: text = clean_text(date_value) if not text: return None try: parsed = datetime.fromisoformat(text).date() except ValueError: return None return max((date.today() - parsed).days, 0) def score_redrob_signals(signals: Any) -> tuple[str, float]: signals = parse_jsonish(signals) if not isinstance(signals, dict): return "", 0.0 profile_complete = float(signals.get("profile_completeness_score") or 0) / 100 response_rate = float(signals.get("recruiter_response_rate") or 0) interview_rate = float(signals.get("interview_completion_rate") or 0) offer_rate = signals.get("offer_acceptance_rate") offer_rate = 0.5 if offer_rate in [None, -1] else float(offer_rate) open_to_work = 1.0 if signals.get("open_to_work_flag") else 0.0 active_days = days_since(signals.get("last_active_date")) recency_score = 0.0 if active_days is not None: if active_days <= 7: recency_score = 1.0 elif active_days <= 30: recency_score = 0.8 elif active_days <= 90: recency_score = 0.45 else: recency_score = 0.1 response_hours = float(signals.get("avg_response_time_hours") or 999) response_speed = 1.0 if response_hours <= 12 else 0.8 if response_hours <= 24 else 0.5 if response_hours <= 72 else 0.2 recruiter_interest = min(float(signals.get("saved_by_recruiters_30d") or 0) / 10, 1.0) search_interest = min(float(signals.get("search_appearance_30d") or 0) / 100, 1.0) applications = min(float(signals.get("applications_submitted_30d") or 0) / 20, 1.0) verified = ( float(bool(signals.get("verified_email"))) + float(bool(signals.get("verified_phone"))) + float(bool(signals.get("linkedin_connected"))) ) / 3 github = float(signals.get("github_activity_score") or 0) github = 0.0 if github < 0 else min(github / 100, 1.0) score = ( 0.15 * profile_complete + 0.15 * recency_score + 0.15 * response_rate + 0.10 * response_speed + 0.10 * open_to_work + 0.10 * interview_rate + 0.05 * offer_rate + 0.07 * recruiter_interest + 0.05 * search_interest + 0.03 * applications + 0.03 * verified + 0.02 * github ) summary = ( f"profile completeness {profile_complete * 100:.0f}%, " f"last active {active_days if active_days is not None else 'unknown'} days ago, " f"open to work {bool(signals.get('open_to_work_flag'))}, " f"recruiter response rate {response_rate:.0%}, " f"avg response time {response_hours:.0f} hours, " f"interview completion {interview_rate:.0%}, " f"saved by recruiters 30d {int(signals.get('saved_by_recruiters_30d') or 0)}" ) return summary, round(score, 4) def build_profile_text(row: pd.Series) -> tuple[str, float]: skills_text, skill_score = format_skills(row.get("skills", "")) redrob_summary, activity_score = score_redrob_signals(row.get("redrob_signals", "")) parts = [ f"Candidate Name: {first_present(row, NAME_KEYS, clean_text(get_nested(row, 'profile.anonymized_name', 'Unknown')))}", f"Headline: {clean_text(row.get('headline', get_nested(row, 'profile.headline', '')))}", f"Current Role: {clean_text(row.get('current_role', row.get('title', get_nested(row, 'profile.current_title', ''))))}", f"Experience Years: {clean_text(row.get('experience_years', row.get('years_experience', get_nested(row, 'profile.years_of_experience', ''))))}", f"Current Industry: {clean_text(get_nested(row, 'profile.current_industry', ''))}", f"Location: {clean_text(get_nested(row, 'profile.location', ''))}, {clean_text(get_nested(row, 'profile.country', ''))}", f"Skills: {skills_text}", f"Work History: {clean_text(row.get('work_history', row.get('experience', row.get('career_history', ''))))}", f"Projects: {clean_text(row.get('projects', ''))}", f"Education: {clean_text(row.get('education', ''))}", f"Certifications: {clean_text(row.get('certifications', ''))}", f"Redrob Signals: {redrob_summary or clean_text(row.get('platform_activity', row.get('activity', '')))}", f"Achievements: {clean_text(row.get('achievements', ''))}", f"Summary: {clean_text(row.get('summary', row.get('bio', get_nested(row, 'profile.summary', ''))))}", ] return "\n".join(part for part in parts if not part.endswith(": ")), skill_score, activity_score def normalize_candidates(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df.columns = [str(col).strip().lower().replace(" ", "_") for col in df.columns] records = [] for pos, (_, row) in enumerate(df.iterrows()): profile_text, skill_score, activity_score = build_profile_text(row) candidate_id = first_present(row, ID_KEYS, str(pos)) candidate_name = first_present(row, NAME_KEYS, clean_text(get_nested(row, "profile.anonymized_name", "Unknown"))) work_history_text = clean_text(row.get("career_history", row.get("work_history", row.get("experience", "")))) records.append( { "candidate_id": candidate_id, "candidate_name": candidate_name, "profile_text": profile_text, "work_history_text": work_history_text, "structured_score": skill_score, "activity_score": activity_score, } ) out = pd.DataFrame(records) out = out.drop_duplicates(subset=["candidate_id"], keep="first") out = out[out["profile_text"].str.len() > 20].reset_index(drop=True) return out def prefilter_raw_active_candidates(df: pd.DataFrame) -> pd.DataFrame: """Cheap raw-schema filter to remove clearly inactive candidates before embedding.""" if "redrob_signals" not in df.columns: return df.copy() keep_mask = [] for _, row in df.iterrows(): signals = parse_jsonish(row.get("redrob_signals", {})) if not isinstance(signals, dict): keep_mask.append(True) continue active_days = days_since(signals.get("last_active_date")) is_recent = active_days is not None and active_days <= 60 is_moderately_recent = active_days is not None and active_days <= 180 is_open = bool(signals.get("open_to_work_flag")) response_rate = float(signals.get("recruiter_response_rate") or 0) saved_by_recruiters = int(signals.get("saved_by_recruiters_30d") or 0) applications = int(signals.get("applications_submitted_30d") or 0) keep_mask.append( is_recent or response_rate >= 0.50 or saved_by_recruiters >= 3 or (is_open and is_moderately_recent) or (is_open and response_rate >= 0.20) or (is_open and saved_by_recruiters > 0) or (is_open and applications > 0) ) return df[pd.Series(keep_mask, index=df.index)].reset_index(drop=True) def prefilter_raw_role_candidates( df: pd.DataFrame, job_description: str, top_n: int = 10000, ) -> pd.DataFrame: """Fast job-specific lexical prefilter before expensive normalization.""" if top_n <= 0 or len(df) <= top_n: out = df.copy() out["raw_role_score"] = 1.0 return out.reset_index(drop=True) texts = [build_raw_prefilter_text(row) for _, row in df.iterrows()] vectorizer = HashingVectorizer( lowercase=True, stop_words="english", ngram_range=(1, 2), n_features=2**18, alternate_sign=False, norm="l2", ) matrix = vectorizer.transform([job_description] + texts) scores = (matrix[1:] @ matrix[0].T).toarray().ravel() keep = scores.argsort()[-top_n:][::-1] out = df.iloc[keep].copy().reset_index(drop=True) out["raw_role_score"] = scores[keep] return out def build_raw_prefilter_text(row: pd.Series) -> str: profile = parse_jsonish(row.get("profile", {})) skills = parse_jsonish(row.get("skills", [])) career = parse_jsonish(row.get("career_history", [])) education = parse_jsonish(row.get("education", [])) parts: list[str] = [] if isinstance(profile, dict): for key in ["headline", "summary", "current_title", "current_industry"]: parts.append(clean_text(profile.get(key, ""))) if isinstance(skills, list): for skill in skills: if isinstance(skill, dict): parts.append(clean_text(skill.get("name", ""))) parts.append(clean_text(skill.get("proficiency", ""))) if isinstance(career, list): for item in career[:3]: if isinstance(item, dict): parts.append(clean_text(item.get("title", ""))) parts.append(clean_text(item.get("industry", ""))) parts.append(clean_text(item.get("description", ""))) if isinstance(education, list): for item in education[:2]: if isinstance(item, dict): parts.append(clean_text(item.get("degree", ""))) parts.append(clean_text(item.get("field_of_study", ""))) return " ".join(part for part in parts if part) def filter_active_candidates(candidates: pd.DataFrame, min_activity_score: float = 0.15) -> pd.DataFrame: """Remove clearly unavailable profiles while keeping borderline strong candidates.""" if "activity_score" not in candidates.columns: return candidates.copy() filtered = candidates[candidates["activity_score"].astype(float) >= min_activity_score].copy() return filtered.reset_index(drop=True)