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