# SPDX-License-Identifier: Apache-2.0
# Copyright 2026 Aaron K. Clark
"""LinkedIn as a discovery source (#233) — the employer's posting, never Easy Apply.

LinkedIn's postings come in two kinds. **Easy Apply** keeps the application inside
LinkedIn; **offsite apply** sends the candidate to the employer's own careers site or
ATS. Only the second kind is wanted here: the poller finds the posting on LinkedIn,
reads where its Apply leads, and stages the lead with the **employer's posting** as
its link — the same #191 rule every aggregator gets. Easy Apply-only postings are
skipped and remembered, so they are never read twice.

Two ways in, same listings:

* **Signed in** — the candidate's own account. The .NET agent's browser signs in with
  the board account saved for ``linkedin.com`` (#221-style) and keeps the session
  cookies sealed on that row; this searches with them through the endpoints
  LinkedIn's own web app calls (``/voyager/api/…``). The job-detail call names the
  apply method outright: ``OffsiteApply`` with the employer's ``companyApplyUrl``,
  or an onsite (Easy Apply) method. LinkedIn's official API (OAuth) only opens job
  search to approved partners, which is why the site's own endpoints are used.
* **Guest** — no account. LinkedIn's public job search (``/jobs-guest/…``) renders
  cards for anyone and a posting page that carries the offsite apply link in a
  hidden ``<code id="applyUrl">`` element. Rate-limited hard (HTTP 429 after a few
  dozen calls), so it is the fallback while no session is kept. The approach — and
  the ``?url=`` extraction — follows JobSpy (MIT, https://github.com/speedyapply/JobSpy).

Every request is the tenant's own; nothing here is shared across tenants.
"""

from __future__ import annotations

import json
import logging
import re
import time
import urllib.parse
from collections.abc import Callable, Iterable
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from html import unescape
from typing import Any

import httpx

from applytrack.poll import Listing, classify

logger = logging.getLogger(__name__)

BASE = "https://www.linkedin.com"
SOURCE = "linkedin"
#: Board-account hosts that mean the candidate's LinkedIn sign-in.
ACCOUNT_HOSTS = ("linkedin.com", "www.linkedin.com")
#: How long a kept session is trusted before the agent renews it. The cookie itself
#: lasts a year; a renewal well inside that keeps a lapse from costing a poll.
SESSION_DAYS = 300
#: Keyword searches per poll (one page each). Every card that survives the keyword
#: check costs one job-detail call, so the cap is the account's request budget.
MAX_KEYWORD_QUERIES = 8
PAGE_SIZE = 25
#: Only postings listed within this window: the poller runs hourly and the ledger
#: remembers what it has read, so a day is plenty and keeps the detail calls few.
POSTED_WITHIN_SECONDS = 86_400
#: United States, LinkedIn's geo id; the search is remote-filtered when the profile is.
DEFAULT_GEO_ID = "103644278"
#: A pause between calls, the way a person's browser paces them.
PACE_SECONDS = 0.6

_OFFSITE = "com.linkedin.voyager.jobs.OffsiteApply"
_CARD_TYPE = "com.linkedin.voyager.dash.jobs.JobPostingCard"
_URN_ID_RE = re.compile(r"(\d+)\s*$")

# Guest HTML: the card list and the posting page (JobSpy's selectors, as regexes).
_GUEST_CARD_RE = re.compile(r'data-entity-urn="urn:li:jobPosting:(\d+)"(.*?)</li>', re.DOTALL)
_GUEST_TITLE_RE = re.compile(r'class="base-search-card__title[^"]*"[^>]*>(.*?)</h3>', re.DOTALL)
_GUEST_COMPANY_RE = re.compile(
    r'class="base-search-card__subtitle[^"]*"[^>]*>(.*?)</h4>', re.DOTALL
)
_GUEST_LOCATION_RE = re.compile(
    r'class="job-search-card__location[^"]*"[^>]*>(.*?)</span>', re.DOTALL
)
_GUEST_APPLY_RE = re.compile(r'<code\s+id="applyUrl"[^>]*>(.*?)</code>', re.DOTALL)
_GUEST_APPLY_URL_RE = re.compile(r"(?<=\?url=)[^\"&]+")
_TAG_RE = re.compile(r"<[^>]+>")
_WS_RE = re.compile(r"\s+")


class NeedsSignIn(Exception):
    """LinkedIn did not accept the kept session."""


class RateLimited(Exception):
    """LinkedIn answered 429: stop for this poll."""


@dataclass
class LinkedInAccount:
    """The tenant's LinkedIn sign-in and the session kept for it."""

    email: str
    session: str = ""
    session_expires_at: datetime | None = None

    @property
    def session_fresh(self) -> bool:
        return bool(self.cookies) and (
            self.session_expires_at is None or self.session_expires_at > datetime.now(timezone.utc)
        )

    @property
    def cookies(self) -> dict[str, str]:
        """The kept session as cookies: ``li_at`` and ``JSESSIONID``, or nothing."""
        return parse_session(self.session)


def parse_session(session: str) -> dict[str, str]:
    """The sealed session's plaintext — JSON ``{"li_at": …, "JSESSIONID": …}`` — as cookies.
    Anything without a ``li_at`` is no session."""
    if not session:
        return {}
    try:
        data = json.loads(session)
    except ValueError:
        return {}
    if not isinstance(data, dict) or not str(data.get("li_at") or "").strip():
        return {}
    out = {"li_at": str(data["li_at"]).strip()}
    js = str(data.get("JSESSIONID") or "").strip()
    out["JSESSIONID"] = js or 'ajax:0000000000000000000'
    return out


@dataclass
class Card:
    """One search result, before its posting is read."""

    job_id: str
    title: str
    company: str
    location: str

    @property
    def view_url(self) -> str:
        return f"{BASE}/jobs/view/{self.job_id}"


def _text(html: str) -> str:
    return _WS_RE.sub(" ", _TAG_RE.sub(" ", unescape(html or ""))).strip()


def _job_id(urn: object) -> str:
    m = _URN_ID_RE.search(str(urn or ""))
    return m.group(1) if m else ""


class Client:
    """One tenant's view of LinkedIn's job search over an ``httpx.Client``: signed in when
    the account has a session, the guest endpoints when it has none."""

    def __init__(self, client: httpx.Client, account: LinkedInAccount | None = None,
                 *, pace: Callable[[float], None] = time.sleep) -> None:
        self._client = client
        self._pace = pace
        self._cookies = account.cookies if account else {}
        for name, value in self._cookies.items():
            client.cookies.set(name, value, domain=".www.linkedin.com")

    @property
    def signed_in(self) -> bool:
        return bool(self._cookies)

    def _voyager_headers(self) -> dict[str, str]:
        # The web app sends its JSESSIONID back as the CSRF token, quotes stripped.
        return {
            "csrf-token": self._cookies.get("JSESSIONID", "").strip('"'),
            "x-restli-protocol-version": "2.0.0",
            "accept": "application/vnd.linkedin.normalized+json+2.1",
            "x-li-lang": "en_US",
            "referer": BASE + "/jobs/search/",
        }

    def _check(self, r: httpx.Response) -> None:
        if r.status_code == 429:
            raise RateLimited("LinkedIn answered 429")
        # A signed-in call that bounces to the sign-in page is the session's end.
        if r.status_code in (401, 403) or (
            r.status_code in (301, 302, 303, 307, 308)
            and any(s in r.headers.get("location", "") for s in ("/login", "/uas/", "/checkpoint"))
        ):
            raise NeedsSignIn(f"LinkedIn answered {r.status_code}")
        if r.status_code in (301, 302, 303, 307, 308):
            raise NeedsSignIn("LinkedIn redirected the request")
        r.raise_for_status()

    # -- signed in --------------------------------------------------------------

    def search(self, keywords: str, *, remote_only: bool, start: int = 0) -> list[Card]:
        """One page of the signed-in job search for ``keywords``."""
        filters = [f"timePostedRange:List(r{POSTED_WITHIN_SECONDS})"]
        if remote_only:
            filters.insert(0, "workplaceType:List(2)")
        query = (
            "(origin:JOB_SEARCH_PAGE_JOB_FILTER,"
            f"keywords:{urllib.parse.quote(keywords, safe='')},"
            f"locationUnion:(geoId:{DEFAULT_GEO_ID}),"
            f"selectedFilters:({','.join(filters)}),spellCorrectionEnabled:true)"
        )
        url = (
            f"{BASE}/voyager/api/voyagerJobsDashJobCards"
            "?decorationId=com.linkedin.voyager.dash.deco.jobs.search.JobSearchCardsCollection-220"
            f"&count={PAGE_SIZE}&q=jobSearch&query={query}&start={start}"
        )
        r = self._client.get(url, headers=self._voyager_headers())
        self._check(r)
        try:
            data = r.json()
        except ValueError as exc:
            raise NeedsSignIn("LinkedIn answered with a page, not the search") from exc
        return cards_from_search(data)

    def posting(self, job_id: str) -> dict[str, Any]:
        """The signed-in job posting: title, description, apply method."""
        r = self._client.get(
            f"{BASE}/voyager/api/jobs/jobPostings/{job_id}",
            headers={**self._voyager_headers(), "accept": "application/json"},
        )
        self._check(r)
        try:
            data = r.json()
        except ValueError as exc:
            raise NeedsSignIn("LinkedIn answered with a page, not the posting") from exc
        return data if isinstance(data, dict) else {}

    # -- guest ------------------------------------------------------------------

    def guest_search(self, keywords: str, *, remote_only: bool, start: int = 0) -> list[Card]:
        """One page of the public job search, rendered for anyone."""
        params: list[tuple[str, str | int | float | bool | None]] = [
            ("keywords", keywords), ("location", "United States"),
            ("f_TPR", f"r{POSTED_WITHIN_SECONDS}"), ("start", str(start)),
        ]
        if remote_only:
            params.append(("f_WT", "2"))
        r = self._client.get(
            f"{BASE}/jobs-guest/jobs/api/seeMoreJobPostings/search", params=params,
            headers={"accept": "text/html"},
        )
        if r.status_code == 429:
            raise RateLimited("LinkedIn answered 429")
        r.raise_for_status()
        return cards_from_guest_html(r.text)

    def guest_apply_link(self, job_id: str) -> str | None:
        """Where the public posting page's Apply leads: the employer's URL, or ``None`` for
        Easy Apply (no such link on the page)."""
        r = self._client.get(
            f"{BASE}/jobs-guest/jobs/api/jobPosting/{job_id}", headers={"accept": "text/html"}
        )
        if r.status_code == 429:
            raise RateLimited("LinkedIn answered 429")
        r.raise_for_status()
        return guest_apply_link_from_html(r.text)


# -- parsing ------------------------------------------------------------------


def cards_from_search(data: dict[str, Any]) -> list[Card]:
    """The job cards in a signed-in search response, in order, skipping cards with no
    title (ads and the state records that ride along)."""
    out: list[Card] = []
    included = data.get("included") if isinstance(data, dict) else None
    for item in included or []:
        if not isinstance(item, dict) or item.get("$type") != _CARD_TYPE:
            continue
        job_id = _job_id(item.get("jobPostingUrn"))
        title = (item.get("title") or {}).get("text") if isinstance(item.get("title"), dict) else ""
        title = " ".join(str(title or "").split()).strip()
        if not job_id or not title:
            continue
        company = str((item.get("primaryDescription") or {}).get("text") or "").strip()
        location = str((item.get("secondaryDescription") or {}).get("text") or "").strip()
        out.append(Card(job_id=job_id, title=title, company=company, location=location))
    return out


def apply_link(posting: dict[str, Any]) -> str | None:
    """The employer's apply URL from a signed-in posting, or ``None`` when the posting
    is Easy Apply (any onsite method) or says nothing."""
    method = posting.get("applyMethod")
    if not isinstance(method, dict):
        return None
    offsite = method.get(_OFFSITE)
    if not isinstance(offsite, dict):
        return None
    url = str(offsite.get("companyApplyUrl") or "").strip()
    return url if url.startswith(("http://", "https://")) else None


def cards_from_guest_html(html: str) -> list[Card]:
    out: list[Card] = []
    for job_id, body in _GUEST_CARD_RE.findall(html):
        title = _GUEST_TITLE_RE.search(body)
        company = _GUEST_COMPANY_RE.search(body)
        location = _GUEST_LOCATION_RE.search(body)
        if not title:
            continue
        out.append(Card(
            job_id=job_id,
            title=_text(title.group(1)),
            company=_text(company.group(1)) if company else "",
            location=_text(location.group(1)) if location else "",
        ))
    return out


def guest_apply_link_from_html(html: str) -> str | None:
    m = _GUEST_APPLY_RE.search(html)
    if not m:
        return None
    hit = _GUEST_APPLY_URL_RE.search(unescape(m.group(1)).strip().strip('"'))
    if not hit:
        return None
    url = urllib.parse.unquote(hit.group(0))
    return url if url.startswith(("http://", "https://")) else None


def _description(posting: dict[str, Any]) -> str:
    desc = posting.get("description")
    if isinstance(desc, dict):
        return " ".join(str(desc.get("text") or "").split())
    return " ".join(str(desc or "").split())


# An Easy Apply posting's source: "auto:linkedin:easy" once staged, which is what tells the
# agent's browser that this linkedin.com link is a dialog to drive and not a listing (#278).
# Keep in step with ``AtsProvider.LinkedInEasySource``.
EASY_SOURCE = f"{SOURCE}:easy"


def to_listing(
    card: Card,
    employer_link: str,
    description: str = "",
    *,
    remote: bool = False,
    easy: bool = False,
) -> Listing:
    """A card whose Apply leads to ``employer_link`` as the poller's :class:`Listing`: the
    LinkedIn posting stays the listing's link (the dedupe key, and an aggregator link the
    #191 rule resolves), the employer's URL rides as ``apply_link`` and is what gets stored.
    ``remote`` says the search was remote-filtered, which a card's location may not."""
    location = card.location
    if remote and "remote" not in location.lower():
        location = f"{location} (Remote)" if location else "Remote"
    return Listing(
        company=card.company,
        role=card.title,
        link=card.view_url,
        location=location,
        # Easy Apply has no employer link: the LinkedIn posting is where it is applied to.
        source=EASY_SOURCE if easy else SOURCE,
        description=description,
        apply_link="" if easy else employer_link,
    )


def queries_for(keywords: Iterable[str]) -> list[str]:
    """The keyword searches worth a page each: the tenant's keywords, de-duped and capped."""
    out: list[str] = []
    for k in keywords:
        q = " ".join(str(k).split()).strip()
        if q and q.lower() not in {x.lower() for x in out}:
            out.append(q)
        if len(out) >= MAX_KEYWORD_QUERIES:
            break
    return out


def fetch_listings(
    client: Client,
    keywords: Iterable[str],
    *,
    remote_only: bool,
    limit: int,
    already_seen: Callable[[str], bool] = lambda _url: False,
    remember: Callable[[str], None] = lambda _url: None,
    easy_apply: bool = False,
) -> list[Listing]:
    """One page per keyword, then the posting of every card that is new and names a
    keyword in its title (the fit score itself is the stager's, with the description);
    offsite-apply postings become listings with the employer's link, Easy Apply
    ones are remembered through ``remember`` and dropped. ``already_seen`` (by the
    LinkedIn posting URL) keeps a card the ledger knows from costing another call.
    A 429 ends the poll with what was gathered."""
    keyword_list = list(keywords)
    out: list[Listing] = []
    done: set[str] = set()
    try:
        for q in queries_for(keyword_list):
            cards = (client.search if client.signed_in else client.guest_search)(
                q, remote_only=remote_only
            )
            client._pace(PACE_SECONDS)
            for card in cards:
                if card.job_id in done or already_seen(card.view_url):
                    continue
                done.add(card.job_id)
                if not classify(card.title, "", keyword_list)[1]:
                    continue
                if client.signed_in:
                    posting = client.posting(card.job_id)
                    link = apply_link(posting)
                    description = _description(posting)
                else:
                    link = client.guest_apply_link(card.job_id)
                    description = ""
                client._pace(PACE_SECONDS)
                if not link:
                    # Easy Apply only. With the tenant's switch on it is staged for the agent's
                    # browser to drive LinkedIn's own dialog (#278); otherwise, as ever, skipped.
                    if easy_apply:
                        out.append(to_listing(card, "", description, remote=remote_only, easy=True))
                        if len(out) >= limit * 3:
                            return out
                        continue
                    logger.info(
                        "linkedin: %s — %s is Easy Apply only; skipped", card.company, card.title
                    )
                    remember(card.view_url)
                    continue
                if already_seen(link):
                    logger.info(
                        "linkedin: %s — %s destination link already seen; skipped",
                        card.company,
                        card.title,
                    )
                    remember(card.view_url)
                    continue
                out.append(to_listing(card, link, description, remote=remote_only))
                if len(out) >= limit * 3:
                    return out
    except RateLimited:
        logger.warning(
            "linkedin: rate-limited (429) after %d listings; the rest waits for the next poll",
            len(out),
        )
    return out


def session_expiry() -> datetime:
    return datetime.now(timezone.utc) + timedelta(days=SESSION_DAYS)