Spaces:
Sleeping
Sleeping
| """ | |
| src/text_builder.py — Builds a candidate's text representation for embedding/BM25. | |
| Integrates the Statistical CDF and NLI Categorical Hypotheses into the candidate text blob. | |
| """ | |
| import numpy as np | |
| from config import TIER1_SKILLS, TIER2_SKILLS | |
| def get_cdf_category(percentile: float) -> str: | |
| if percentile <= 10: | |
| return "extremely low" | |
| elif percentile <= 25: | |
| return "below average" | |
| elif percentile <= 75: | |
| return "average" | |
| elif percentile <= 90: | |
| return "high" | |
| else: | |
| return "exceptional" | |
| def build_cdf_sentences(candidate: dict, cdfs: dict) -> str: | |
| """ | |
| Generate semantically rich sentences from raw platform signals using CDF percentiles. | |
| """ | |
| sig = candidate.get('redrob_signals', {}) | |
| sentences = [] | |
| # 1. Recruiter Response Rate | |
| val = sig.get('recruiter_response_rate', 0.0) | |
| pct = (np.searchsorted(cdfs['recruiter_response_rate'], val, side='right') / len(cdfs['recruiter_response_rate'])) * 100 | |
| cat = get_cdf_category(pct) | |
| sentences.append(f"Candidate shows {cat} engagement with recruiters.") | |
| # 2. Avg Response Time (lower is better) | |
| val = sig.get('avg_response_time_hours', 0.0) | |
| pct = (np.searchsorted(cdfs['avg_response_time_hours'], val, side='right') / len(cdfs['avg_response_time_hours'])) * 100 | |
| pct = 100.0 - pct # invert so lower is better | |
| cat = get_cdf_category(pct) | |
| sentences.append(f"Candidate average response time is {cat}.") | |
| # 3. Interview Completion Rate | |
| val = sig.get('interview_completion_rate', 0.0) | |
| pct = (np.searchsorted(cdfs['interview_completion_rate'], val, side='right') / len(cdfs['interview_completion_rate'])) * 100 | |
| cat = get_cdf_category(pct) | |
| sentences.append(f"Candidate interview completion rate is {cat}.") | |
| # 4. Offer Acceptance Rate | |
| val = sig.get('offer_acceptance_rate', -1) | |
| if val == -1: | |
| sentences.append("Candidate has no offer acceptance history.") | |
| else: | |
| pct = (np.searchsorted(cdfs['offer_acceptance_rate'], val, side='right') / len(cdfs['offer_acceptance_rate'])) * 100 | |
| cat = get_cdf_category(pct) | |
| sentences.append(f"Candidate offer acceptance history is {cat}.") | |
| # 5. Github Activity Score | |
| val = sig.get('github_activity_score', -1) | |
| if val == -1: | |
| sentences.append("Candidate has no GitHub activity history.") | |
| else: | |
| pct = (np.searchsorted(cdfs['github_activity_score'], val, side='right') / len(cdfs['github_activity_score'])) * 100 | |
| cat = get_cdf_category(pct) | |
| sentences.append(f"Candidate GitHub activity level is {cat}.") | |
| # 6. Profile Completeness Score | |
| val = sig.get('profile_completeness_score', 0.0) | |
| pct = (np.searchsorted(cdfs['profile_completeness_score'], val, side='right') / len(cdfs['profile_completeness_score'])) * 100 | |
| cat = get_cdf_category(pct) | |
| sentences.append(f"Candidate profile completeness score is {cat}.") | |
| # 7. Categorical preferences | |
| willing = sig.get('willing_to_relocate', False) | |
| sentences.append("Candidate is willing to relocate to another city." if willing else "Candidate is not willing to relocate.") | |
| work_mode = sig.get('preferred_work_mode', 'flexible') | |
| sentences.append(f"Candidate preferred work mode is {work_mode}.") | |
| notice = sig.get('notice_period_days', 30) | |
| sentences.append(f"Candidate notice period is {notice} days.") | |
| # 8. Location | |
| loc = candidate.get('profile', {}).get('location', '') | |
| if loc: | |
| sentences.append(f"Candidate current location is {loc}.") | |
| return " ".join(sentences) | |
| def build_candidate_text(candidate: dict, max_chars: int = 1000, cdfs: dict | None = None) -> str: | |
| """ | |
| Build a dense, signal-rich text blob for a candidate, combining resume details with CDF behavior sentences. | |
| """ | |
| parts = [] | |
| # 1. Career descriptions (most signal-dense — lead with them) | |
| for job in candidate.get('career_history', [])[:5]: | |
| desc = (job.get('description') or '').strip() | |
| if desc: | |
| title = job.get('title', '') | |
| company = job.get('company', '') | |
| parts.append(f"{title} at {company}: {desc[:600]}") | |
| # 2. Profile headline + summary (intent narrative) | |
| headline = (candidate['profile'].get('headline') or '').strip() | |
| summary = (candidate['profile'].get('summary') or '').strip() | |
| if headline: | |
| parts.append(headline) | |
| if summary: | |
| parts.append(summary[:500]) | |
| # 3. Current title (role signal) | |
| current_title = (candidate['profile'].get('current_title') or '').strip() | |
| if current_title: | |
| parts.append(current_title) | |
| # 4. Only advanced/expert skills with meaningful duration (reduce noise) | |
| expert_skills = [ | |
| s['name'] for s in candidate.get('skills', []) | |
| if s.get('proficiency') in ('advanced', 'expert') | |
| and s.get('duration_months', 0) > 6 | |
| ] | |
| if expert_skills: | |
| parts.append(' '.join(expert_skills)) | |
| # 5. Education field of study (light signal) | |
| for edu in candidate.get('education', [])[:2]: | |
| fos = (edu.get('field_of_study') or '').strip() | |
| if fos: | |
| parts.append(fos) | |
| # 6. Append CDF & NLI behavioral profile sentences if CDF data is available | |
| if cdfs is not None: | |
| parts.append(build_cdf_sentences(candidate, cdfs)) | |
| full_text = ' '.join(filter(None, parts)) | |
| return full_text[:max_chars] | |
| def build_jd_text(jd_text: str, max_chars: int = 1500) -> str: | |
| """ | |
| Clean and truncate the job description text for embedding. | |
| """ | |
| import re | |
| text = re.sub(r'\s+', ' ', jd_text).strip() | |
| return text[:max_chars] | |
| def build_bm25_tokens(candidate: dict) -> list[str]: | |
| """ | |
| Build a tokenized list for the BM25 index. | |
| """ | |
| import re | |
| def tokenize(text: str) -> list[str]: | |
| text = text.lower() | |
| text = re.sub(r'[^a-z0-9\s]', ' ', text) | |
| return [t for t in text.split() if len(t) > 1] | |
| tokens = [] | |
| # Skills — all of them (BM25 is used for exact-term recall) | |
| for skill in candidate.get('skills', []): | |
| tokens.extend(tokenize(skill.get('name', ''))) | |
| # Career descriptions | |
| for job in candidate.get('career_history', [])[:5]: | |
| tokens.extend(tokenize(job.get('description', '')[:800])) | |
| tokens.extend(tokenize(job.get('title', ''))) | |
| tokens.extend(tokenize(job.get('industry', ''))) | |
| # Profile | |
| tokens.extend(tokenize(candidate['profile'].get('headline', ''))) | |
| tokens.extend(tokenize(candidate['profile'].get('summary', '')[:500])) | |
| tokens.extend(tokenize(candidate['profile'].get('current_title', ''))) | |
| # Education | |
| for edu in candidate.get('education', []): | |
| tokens.extend(tokenize(edu.get('field_of_study', ''))) | |
| tokens.extend(tokenize(edu.get('degree', ''))) | |
| # Certifications | |
| for cert in candidate.get('certifications', []): | |
| tokens.extend(tokenize(cert.get('name', ''))) | |
| tokens.extend(tokenize(cert.get('issuer', ''))) | |
| return tokens | |
| def build_jd_tokens(jd_text: str) -> list[str]: | |
| """ | |
| Tokenize the JD for BM25 querying. | |
| """ | |
| import re | |
| text = jd_text.lower() | |
| text = re.sub(r'[^a-z0-9\s]', ' ', text) | |
| return [t for t in text.split() if len(t) > 1] | |