"""Apify REST helpers and LinkedIn profile normalization. Pure module: no FastAPI imports and no HTTPException. The default actor is HarvestAPI's no-cookie LinkedIn people search (harvestapi~linkedin-profile-search). Its input schema was verified live: `searchQuery` (fuzzy string), `maxItems` (int), `locations` (array of strings), `profileScraperMode` ("Short" | "Full" | "Full + email search"). Swapping actors later means changing APIFY_ACTOR_ID plus, at most, build_actor_input and normalize_profile. """ from __future__ import annotations import os import re from urllib.parse import urlsplit import httpx from dotenv import load_dotenv load_dotenv() # The user's console .env entry is APIFY_TOKEN; APIFY_API_TOKEN is the documented name. APIFY_API_TOKEN = os.getenv("APIFY_API_TOKEN") or os.getenv("APIFY_TOKEN") APIFY_API_BASE = os.getenv("APIFY_API_BASE", "https://api.apify.com/v2") APIFY_ACTOR_ID = os.getenv("APIFY_ACTOR_ID", "harvestapi~linkedin-profile-search") APIFY_MAX_RESULTS = int(os.getenv("APIFY_MAX_RESULTS", "25")) APIFY_PROFILE_MODE = os.getenv("APIFY_PROFILE_MODE", "Full") APIFY_TIMEOUT = float(os.getenv("APIFY_TIMEOUT", "30")) # Server-side spend ceiling per run (maxTotalChargeUsd). The HarvestAPI actor is # pay-per-EVENT ($0.10/search page + per-profile), so Apify's maxItems billing # param does not apply — it was rejected live with "Maximum cost per run is less # than the allowed minimum of $0.10". A 25-profile Full run costs ~$0.20. APIFY_MAX_COST_USD = float(os.getenv("APIFY_MAX_COST_USD", "1.0")) def _matches_excluded(text, names) -> bool: haystack = " ".join(str(text or "").lower().split()) if not haystack: return False return any(str(name).lower() in haystack for name in (names or []) if name) def is_excluded_profile(profile: dict, *, companies=None, universities=None) -> bool: """True when the person matches a configured company or university exclusion. Company: current employer, or headline only when no company was extracted so an "ex-…" headline on someone now elsewhere does not exclude them. University: any education school name on the sourced profile. Lists come from exclude_company / exclude_university — never hardcoded. """ company = (profile or {}).get("current_company") if _matches_excluded(company, companies): return True if not company and _matches_excluded((profile or {}).get("headline"), companies): return True if universities: raw = (profile or {}).get("raw") or {} for edu in extract_education(raw): if _matches_excluded(edu.get("school"), universities): return True return False # Apify run status -> talent_runs.status. Transitional states stay "running"; # unknown values also stay "running" so we never commit a terminal state we # don't understand (Buffer precedent). APIFY_STATUS_TO_LOCAL = { "READY": "running", "RUNNING": "running", "TIMING-OUT": "running", "ABORTING": "running", "SUCCEEDED": "succeeded", "FAILED": "failed", "TIMED-OUT": "timed_out", "ABORTED": "aborted", } TERMINAL_STATUSES = {"succeeded", "failed", "timed_out", "aborted"} def local_status(apify_status) -> str: return APIFY_STATUS_TO_LOCAL.get(str(apify_status or "").upper(), "running") class ApifyError(RuntimeError): def __init__(self, message: str, *, code: str | None = None): super().__init__(message) self.code = code def _headers() -> dict: if not APIFY_API_TOKEN: raise RuntimeError("APIFY_API_TOKEN is not configured") return { "Authorization": f"Bearer {APIFY_API_TOKEN}", "Content-Type": "application/json", } def _raise_for_response(response: httpx.Response) -> None: if response.status_code < 400: return try: error = (response.json() or {}).get("error") or {} except ValueError: error = {} if error.get("message"): raise ApifyError(error["message"], code=error.get("type")) raise httpx.HTTPStatusError( response.text, request=response.request, response=response ) # LinkedIn's years-of-experience facet, as the actor's yearsOfExperienceIds # enum defines it (verified from the actor's input schema): id -> (min, max) # whole years. A job's experience_min/experience_max selects every overlapping # bucket. EXPERIENCE_BUCKETS = { "1": (0, 0), # Less than 1 year "2": (1, 2), # 1 to 2 years "3": (3, 5), # 3 to 5 years "4": (6, 10), # 6 to 10 years "5": (11, 60), # More than 10 years } def years_of_experience_ids(experience_min, experience_max) -> list[str]: """Bucket ids overlapping [experience_min, experience_max]; [] = no filter.""" if experience_min is None and experience_max is None: return [] lo = int(experience_min) if experience_min is not None else 0 hi = int(experience_max) if experience_max is not None else 60 if hi < lo: lo, hi = hi, lo return [ bucket_id for bucket_id, (b_lo, b_hi) in EXPERIENCE_BUCKETS.items() if b_hi >= lo and b_lo <= hi ] # Job "locations" that are work arrangements, not places. Sending one as the # actor's locations filter returns an empty dataset — verified live: a run with # locations=["Remote"] found 0 profiles where the same query with a real # geography found 5. Filter them out instead of filtering by them. NON_GEOGRAPHIC_LOCATIONS = { "remote", "hybrid", "onsite", "on-site", "on site", "anywhere", "flexible", "wfh", "work from home", } def _geographic_location(value) -> str | None: text = str(value or "").strip() if not text or text.lower() in NON_GEOGRAPHIC_LOCATIONS: return None return text def _skill_terms(*entry_lists) -> list[str]: """Keyword-like entries only (max 3 words, 30 chars), deduped in order. Job requirements are sometimes skills ("Python", "Amazon Seller Central") and sometimes prose ("2-5 years of experience managing Amazon PPC campaigns..."). Prose in the fuzzy searchQuery strangles it — verified live: a sentence-stuffed query matched 2 people country-wide and 0 in Karachi, where the title alone finds plenty. """ terms: list[str] = [] seen: set[str] = set() for entries in entry_lists: for entry in entries or []: text = " ".join(str(entry).split()) if not text or text.lower() in seen: continue if len(text) <= 30 and len(text.split()) <= 3: terms.append(text) seen.add(text.lower()) return terms def build_actor_input( job: dict, *, max_results: int, overrides: dict | None = None, start_page: int = 1, exclude_company_urls: list[str] | None = None, ) -> dict: """Deterministic actor input from job fields. No LLM involved. The job title goes into LinkedIn's CURRENT-TITLE facet (currentJobTitles), not the keyword box: a keyword query matches words anywhere in a profile, so "AI Engineer Python..." returned a pool that was 52% generic software engineers (every full-stack profile mentions Python). Verified live: the facet alone returns full pages of genuinely AI-titled people. The keyword box carries only the skill terms. start_page > 1 continues a previous search deeper into the result pages (25 profiles per page), so a re-run surfaces new people instead of re-finding the first page. """ overrides = overrides or {} title = str(job.get("title") or "").strip()[:100] terms = _skill_terms(job.get("requirements"), job.get("optional_skills"))[:3] query = " ".join(terms).strip()[:200] if overrides.get("keywords"): query = str(overrides["keywords"]).strip()[:200] actor_input: dict = { "maxItems": max_results, "profileScraperMode": APIFY_PROFILE_MODE, } if title: actor_input["currentJobTitles"] = [title] if query: actor_input["searchQuery"] = query elif not title: # No facet and no terms: nothing left to search by. actor_input["searchQuery"] = "" experience_ids = years_of_experience_ids( job.get("experience_min"), job.get("experience_max") ) if experience_ids: actor_input["yearsOfExperienceIds"] = experience_ids urls = [u.strip() for u in (exclude_company_urls or []) if u and str(u).strip()] if urls: actor_input["excludeCurrentCompanies"] = urls if start_page and int(start_page) > 1: actor_input["startPage"] = min(int(start_page), 100) if "location" in overrides and overrides["location"] is not None: # An explicit override wins outright — "Remote" here means the caller # wants no geography constraint, not a fallback to the job's location. location = _geographic_location(overrides["location"]) else: location = _geographic_location(job.get("location")) if location: actor_input["locations"] = [location] return actor_input def broaden_actor_input(actor_input: dict) -> dict | None: """Next rung of the thin-results broadening ladder, or None when exhausted. A search ANDs facet + keywords + location + experience; in a single city that intersection can collapse to one person (seen live: Amazon PPC + Karachi returned 1). Rungs: (1) drop the keyword query, keeping the title facet; (2) drop the facet and search the title as keywords instead. Location, experience and company exclusions are never relaxed — they are user intent, not tuning. """ current = dict(actor_input) if current.get("currentJobTitles") and "searchQuery" in current: current.pop("searchQuery") return current if current.get("currentJobTitles"): title = current.pop("currentJobTitles")[0] current["searchQuery"] = title return current return None async def start_actor_run(actor_input: dict, *, actor_id: str | None = None) -> dict: """POST /acts/{id}/runs. maxTotalChargeUsd caps spend on Apify's side, independent of what the actor does with its input; the profile count itself is limited by the maxItems field inside the actor input.""" actor = actor_id or APIFY_ACTOR_ID async with httpx.AsyncClient(timeout=APIFY_TIMEOUT) as client: response = await client.post( f"{APIFY_API_BASE}/acts/{actor}/runs", params={"maxTotalChargeUsd": APIFY_MAX_COST_USD}, json=actor_input, headers=_headers(), ) _raise_for_response(response) data = (response.json() or {}).get("data") or {} if not data.get("id"): raise ApifyError("Apify did not return a run id") return data async def get_run(run_id: str) -> dict: async with httpx.AsyncClient(timeout=APIFY_TIMEOUT) as client: response = await client.get( f"{APIFY_API_BASE}/actor-runs/{run_id}", headers=_headers() ) _raise_for_response(response) return (response.json() or {}).get("data") or {} async def get_dataset_items(dataset_id: str, *, limit: int, offset: int = 0) -> list[dict]: async with httpx.AsyncClient(timeout=APIFY_TIMEOUT) as client: response = await client.get( f"{APIFY_API_BASE}/datasets/{dataset_id}/items", params={"format": "json", "clean": "true", "limit": limit, "offset": offset}, headers=_headers(), ) _raise_for_response(response) body = response.json() return body if isinstance(body, list) else [] async def get_me() -> dict: """Cheap auth sanity check; used by verification, not the request path.""" async with httpx.AsyncClient(timeout=APIFY_TIMEOUT) as client: response = await client.get(f"{APIFY_API_BASE}/users/me", headers=_headers()) _raise_for_response(response) return (response.json() or {}).get("data") or {} async def get_limits() -> dict: """GET /users/me/limits: monthly usage cycle, spend so far, and the cap. Prepaid credits surface through maxMonthlyUsageUsd, so cap minus spend is the account's remaining balance.""" async with httpx.AsyncClient(timeout=APIFY_TIMEOUT) as client: response = await client.get( f"{APIFY_API_BASE}/users/me/limits", headers=_headers() ) _raise_for_response(response) return (response.json() or {}).get("data") or {} def run_cost_usd(remote: dict) -> float: """What this Apify run actually charged, from the run object's usageTotalUsd ("Represents what you actually pay" — covers pay-per-event actors like HarvestAPI). 0.0 for anything absent or malformed.""" try: value = float((remote or {}).get("usageTotalUsd") or 0) except (TypeError, ValueError): return 0.0 return value if value > 0 else 0.0 def account_summary(limits_payload: dict) -> dict: """Map the raw /users/me/limits payload to the fields the UI shows. Every field is None when the payload lacks it; balance is clamped at 0 because a mid-cycle limit reduction can leave spend above the cap.""" payload = limits_payload or {} current = payload.get("current") or {} limits = payload.get("limits") or {} cycle = payload.get("monthlyUsageCycle") or {} def _number(value) -> float | None: try: return float(value) except (TypeError, ValueError): return None spent = _number(current.get("monthlyUsageUsd")) cap = _number(limits.get("maxMonthlyUsageUsd")) balance = max(cap - spent, 0.0) if spent is not None and cap is not None else None return { "spent_this_cycle_usd": spent, "monthly_limit_usd": cap, "balance_usd": balance, "cycle_ends_at": cycle.get("endAt"), } def normalize_linkedin_url(url) -> str | None: """Canonical dedupe key: https, lowercase host/path, no query or trailing slash.""" text = str(url or "").strip() if not text: return None if "//" not in text: text = f"https://{text}" parts = urlsplit(text) host = parts.netloc.lower() if "linkedin.com" not in host: return None path = parts.path.rstrip("/") return f"https://{host}{path}".lower() def _first_string(item: dict, *keys) -> str | None: for key in keys: value = item.get(key) if isinstance(value, str) and value.strip(): return value.strip() return None def _location_text(value) -> str | None: if isinstance(value, str) and value.strip(): return value.strip() if isinstance(value, dict): for key in ("linkedinText", "text", "name", "default"): nested = value.get(key) if isinstance(nested, str) and nested.strip(): return nested.strip() return None def _photo_url(item: dict) -> str | None: for key in ("photo", "profilePicture", "avatar", "photoUrl", "profilePic", "image"): value = item.get(key) if isinstance(value, str) and value.strip(): return value.strip() if isinstance(value, dict): nested = value.get("url") if isinstance(nested, str) and nested.strip(): return nested.strip() return None def _skills(item: dict) -> list[str]: """Up to 10 skill names; entries arrive as strings or {name: ...} dicts.""" names: list[str] = [] for key in ("topSkills", "skills"): for entry in item.get(key) or []: name = entry if isinstance(entry, str) else ( entry.get("name") if isinstance(entry, dict) else None ) if name and str(name).strip() and str(name).strip() not in names: names.append(str(name).strip()) if names: break return names[:10] def _current_position(item: dict) -> tuple[str | None, str | None]: """(title, company) from the most recent experience entry, however spelled.""" position = item.get("position") or item.get("currentPosition") if isinstance(position, dict): title = _first_string(position, "title", "role") company = _first_string(position, "companyName", "company") if title or company: return title, company experience = item.get("experience") or item.get("experiences") if isinstance(experience, list) and experience: entry = experience[0] if isinstance(entry, dict): company = _first_string(entry, "companyName", "company") if company is None: nested = entry.get("company") if isinstance(nested, dict): company = _first_string(nested, "name") return _first_string(entry, "title", "position", "role"), company return None, _first_string(item, "companyName", "currentCompany") _TOKEN_STOPWORDS = { "and", "or", "the", "of", "for", "with", "in", "a", "an", "to", # Requirement-prose filler that appears in almost every profile and would # inflate every score equally, flattening the ranking. "experience", "years", "year", "strong", "including", "ability", "knowledge", "skills", "understanding", "familiarity", "proficiency", "hands", "must", "have", "plus", "good", "excellent", "etc", } def _clean_phrase(text) -> str: cleaned = re.sub(r"[^a-z0-9+#]+", " ", str(text or "").lower()) return " ".join( t for t in cleaned.split() if len(t) > 1 and t not in _TOKEN_STOPWORDS ) def _match_tokens(*texts) -> set[str]: tokens: set[str] = set() for text in texts: tokens.update(_clean_phrase(text).split()) return tokens def relevance_score(job: dict, profile: dict) -> int: """0-100 job-fit rank for sorting, computed when a profile is persisted. Deterministic and free. Title component: a current title CONTAINING every job-title token scores 55 — containment, not exact phrase, because job titles rarely reappear verbatim ("Generative Engineer" vs the pool's "Generative AI Engineer"; seen live: the phrase rule dropped every real match to the scattered tier and compressed the whole pool into the 40s). The job title as an exact phrase in the headline scores 45; scattered token overlap caps at 35 — a keyword-stuffed headline ("AI/ML Engineer | Python | FastAPI | ...") must not outrank someone whose title IS the job title, which is exactly what token overlap alone did on live data. The headline tier stays phrase-only for the same reason: stuffed headlines contain every token of every hot title. Skills component (up to 45): GRADED token overlap between the content words of the job's requirements + optional skills and the person's title/headline/skills/summary. Graded, not per-term all-or-nothing: the title facet makes every sourced profile earn the same title points, so all differentiation lives here — an all-or-nothing single term put a whole live pool on exactly 60. """ job_title = _clean_phrase(job.get("title")) job_title_tokens = set(job_title.split()) title_text = _clean_phrase(profile.get("current_title")) headline_text = _clean_phrase(profile.get("headline")) if job_title and job_title_tokens <= set(title_text.split()): title_component = 55.0 elif job_title and job_title in headline_text: title_component = 45.0 else: role_tokens = set(title_text.split()) | set(headline_text.split()) ratio = ( len(job_title_tokens & role_tokens) / len(job_title_tokens) if job_title_tokens else 0.0 ) title_component = 35 * ratio job_tokens = _match_tokens( *(job.get("requirements") or []), *(job.get("optional_skills") or []) ) profile_tokens = _match_tokens( profile.get("current_title"), profile.get("headline"), " ".join(profile.get("skills") or []), profile.get("summary"), ) skills_ratio = ( len(job_tokens & profile_tokens) / len(job_tokens) if job_tokens else 0.0 ) return round(title_component + 45 * skills_ratio) def _date_text(value) -> str | None: """HarvestAPI dates arrive as {"month": "Jun", "year": 2025, "text": "Jun 2025"}.""" if isinstance(value, dict): text = value.get("text") if isinstance(text, str) and text.strip(): return text.strip() if isinstance(value, str) and value.strip(): return value.strip() return None def _entry_company(entry: dict) -> str | None: company = _first_string(entry, "companyName") if company is None and isinstance(entry.get("company"), dict): company = _first_string(entry["company"], "name") return company def extract_experience(raw: dict) -> list[dict]: """Employment history from a stored raw item, for the profile detail view.""" entries: list[dict] = [] for item in (raw or {}).get("experience") or []: if not isinstance(item, dict): continue start = _date_text(item.get("startDate")) end = _date_text(item.get("endDate")) description = str(item.get("description") or "").strip() entries.append({ "title": _first_string(item, "position", "title", "role"), "company": _entry_company(item), "employment_type": _first_string(item, "employmentType"), "location": _location_text(item.get("location")), "duration": _first_string(item, "duration"), "period": " – ".join(p for p in (start, end) if p) or None, "description": description[:400] or None, "skills": _skills(item)[:6], }) if len(entries) == 10: break return entries def extract_education(raw: dict) -> list[dict]: entries: list[dict] = [] for item in (raw or {}).get("education") or []: if not isinstance(item, dict): continue start = _date_text(item.get("startDate")) end = _date_text(item.get("endDate")) entries.append({ "school": _first_string(item, "schoolName", "school"), "degree": _first_string(item, "degree"), "field": _first_string(item, "fieldOfStudy", "field"), "period": _first_string(item, "period") or (" – ".join(p for p in (start, end) if p) or None), }) if len(entries) == 5: break return entries def normalize_profile(item: dict) -> dict | None: """Tolerant extraction of the card fields from one dataset item. Returns None (skip, not fail) when the item has no LinkedIn URL. The full item always rides along as `raw` so nothing is lost to key drift. """ if not isinstance(item, dict): return None url = normalize_linkedin_url( _first_string(item, "linkedinUrl", "url", "profileUrl", "publicProfileUrl", "link") ) if not url: return None name = _first_string(item, "fullName", "name") if not name: first = _first_string(item, "firstName") or "" last = _first_string(item, "lastName") or "" name = f"{first} {last}".strip() or None title, company = _current_position(item) return { "linkedin_url": url, "public_id": _first_string(item, "publicIdentifier", "publicId"), "full_name": name, "headline": _first_string(item, "headline", "subTitle", "occupation"), "location": _location_text(item.get("location")), "current_title": title, "current_company": company, "avatar_url": _photo_url(item), "summary": _first_string(item, "about", "summary"), "skills": _skills(item), "raw": item, }