HR-ATS-Portal/backend/talent/plugins.py

"""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 _csv_env(name: str, default: str) -> list[str]:
    return [s.strip() for s in os.getenv(name, default).split(",") if s.strip()]


# The user's own companies: their CURRENT employees must never appear in sourced
# results. Names drive the always-on server-side filter (case-insensitive
# substring, so "Utopia Brands Pakistan" matches too). URLs drive the actor's
# excludeCurrentCompanies filter, which wants full LinkedIn company URLs and
# stops those profiles from being scraped (and paid for) at all.
APIFY_EXCLUDE_COMPANIES = _csv_env("APIFY_EXCLUDE_COMPANIES", "Utopia Brands,Utopia Deals")
APIFY_EXCLUDE_COMPANY_URLS = _csv_env(
    "APIFY_EXCLUDE_COMPANY_URLS",
    "https://www.linkedin.com/company/utopiadeals,"
    "https://www.linkedin.com/company/utopia-brands-usa,"
    "https://www.linkedin.com/company/utopiabrands",
)


def _matches_excluded(text) -> bool:
    haystack = " ".join(str(text or "").lower().split())
    return bool(haystack) and any(
        name.lower() in haystack for name in APIFY_EXCLUDE_COMPANIES
    )


def is_excluded_profile(profile: dict) -> bool:
    """True when the person currently works at one of the excluded companies.

    The headline is only consulted when no current company was extracted, so an
    "ex-Utopia" headline on someone now elsewhere does not exclude them.
    """
    company = (profile or {}).get("current_company")
    if _matches_excluded(company):
        return True
    return not company and _matches_excluded((profile or {}).get("headline"))

# 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
) -> 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
    if APIFY_EXCLUDE_COMPANY_URLS:
        actor_input["excludeCurrentCompanies"] = APIFY_EXCLUDE_COMPANY_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,
    }