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
from urllib.parse import urlsplit

import httpx
from dotenv import load_dotenv

from matching.ranking import rank_profile

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


# Moved to matching/ranking.py so the CV Bank ranks stored resumes with the
# same arithmetic instead of growing a second copy that drifts. Re-exported
# under the original name: every call site here and in talent/views.py is
# unchanged, and the numbers this produces are identical.
relevance_score = rank_profile


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,
    }