"""Networking tools for the agent — web_search + web_fetch.

These give the Code-as-Action agent the same online reach opencode's `websearch`
and `webfetch` tools provide, implemented over the stdlib + (optionally) requests
and BeautifulSoup, which the kernel ships preinstalled. No API key is required:
search walks a chain of keyless engines (DuckDuckGo → Bing → DuckDuckGo lite →
Mojeek) with scholarly fast paths (a DOI resolves via Crossref, an arXiv id via
the arXiv API); fetch downloads a URL and converts HTML to readable markdown/text.

Networking can be globally gated by ``OPENAI4S_ALLOW_NETWORK`` (default on);
the daemon's Customize → Network panel flips this.
"""

from __future__ import annotations

import base64
import contextlib
import errno
import hashlib
import html as _html
import ipaddress
import json
import os
import pathlib
import re
import socket
import stat
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, BinaryIO, Iterator

_UA = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/124.0 Safari/537.36"
)
_HARDLINK_UNSUPPORTED = frozenset(
    {
        errno.EPERM,
        errno.EXDEV,
        errno.ENOSYS,
        getattr(errno, "ENOTSUP", errno.EPERM),
        getattr(errno, "EOPNOTSUPP", errno.EPERM),
    }
)
_FAKE_IP_NETWORK = ipaddress.ip_network("198.18.0.0/15")
_TRUE_VALUES = frozenset({"1", "true", "yes", "on"})


def network_allowed() -> bool:
    return os.environ.get("OPENAI4S_ALLOW_NETWORK", "1") not in ("0", "false", "no")


class NetworkDisabled(RuntimeError):
    pass


class SSRFBlocked(RuntimeError):
    pass


def _require_network() -> None:
    if not network_allowed():
        raise NetworkDisabled(
            "networking is disabled (enable it in Customize → Network / set "
            "OPENAI4S_ALLOW_NETWORK=1)"
        )


def _fake_ip_host_allowed(host: str) -> bool:
    """Accept a proxy-synthetic address only across a narrow trust boundary.

    Clash-style Fake-IP DNS maps public names into RFC 2544's
    ``198.18.0.0/15`` benchmarking range and a TUN adapter translates the
    subsequent connection back to the original hostname.  Treating that range
    as generally public would create an SSRF hole, so compatibility requires
    all three conditions: an explicit process opt-in, a hostname rather than
    an IP literal, and a built-in or user-approved egress domain.

    Only the last two are per-*host*, which is why they live here rather than
    beside the address test: ``getaddrinfo`` returns one entry per socktype,
    so a per-address form re-read the environment, re-parsed the host and
    rebuilt the whole egress catalog two or three times for one lookup.
    """

    enabled = (
        os.environ.get("OPENAI4S_ALLOW_FAKE_IP_DNS", "").strip().lower() in _TRUE_VALUES
    )
    if not enabled:
        return False
    try:
        ipaddress.ip_address(host.rstrip("."))
    except ValueError:
        pass
    else:
        return False
    from openai4s import egress

    return egress.domain_in_allowlist(host)


def _host_is_private(host: str) -> bool:
    """True if `host` resolves to a loopback / private / link-local (incl. cloud
    metadata 169.254.169.254) / reserved address — anything an agent-controlled
    URL should not be able to reach (SSRF guard)."""
    if not host:
        return False
    try:
        infos = socket.getaddrinfo(host, None)
    except (socket.gaierror, UnicodeError):
        return False  # let the request itself fail normally
    fake_ip_host: bool | None = None
    for info in infos:
        try:
            addr = ipaddress.ip_address(info[4][0])
        except ValueError:
            continue
        if (
            addr.is_private
            or addr.is_loopback
            or addr.is_link_local
            or addr.is_reserved
            or addr.is_multicast
            or addr.is_unspecified
        ):
            if addr.version == 4 and addr in _FAKE_IP_NETWORK:
                if fake_ip_host is None:
                    fake_ip_host = _fake_ip_host_allowed(host)
                if fake_ip_host:
                    continue
            return True
    return False


def guard_url(url: str) -> None:
    """Refuse a URL that resolves to a private, loopback or metadata address.

    Public because more than one module needs it: `_http_get` applies it per
    redirect hop, and `host/endpoints.py` applies it to an agent-supplied
    endpoint URL before probing. Reaching for `_guard_url` across a package
    boundary is what `test_backend_import_contract` refuses, and rightly — a
    guard two subsystems depend on is surface, not an internal.
    """
    return _guard_url(url)


def _guard_url(url: str) -> None:
    if os.environ.get("OPENAI4S_ALLOW_PRIVATE_FETCH", "") in ("1", "true", "yes"):
        return  # explicit opt-in (e.g. fetching a local model endpoint)
    # Percent-decoded, because both clients that actually connect decode the
    # authority first: `urllib.request.Request._parse` does `unquote(self.host)`
    # and `requests` normalizes the same way. Guarding the encoded spelling
    # guards a host nobody dials -- `http://169%2e254%2e169%2e254/` yields a
    # `gaierror` here (fail-open) and the metadata address in the client.
    host = urllib.parse.unquote(urllib.parse.urlparse(url).hostname or "")
    if _host_is_private(host):
        raise SSRFBlocked(
            f"refusing to fetch a private/loopback/metadata address: {host!r} "
            "(the Windows launcher auto-detects trusted Fake-IP DNS; set "
            "OPENAI4S_ALLOW_PRIVATE_FETCH=1 only for a trusted local target)"
        )


# --------------------------------------------------------------------------- #
#  low-level fetch
# --------------------------------------------------------------------------- #
#: Ceiling on a single response body held in memory. Enforced while reading,
#: not after: a cap applied to an already-allocated body is a description of
#: how big the allocation was, not a bound on it.
MAX_FETCH_BYTES = 32 * 1024 * 1024


class ResponseTooLarge(RuntimeError):
    """A body exceeded the byte ceiling and was abandoned mid-read."""


def _read_capped(reader: Any, limit: int) -> bytes:
    """Read at most ``limit`` bytes, then stop and say so.

    `resp.read()` with no argument is what this replaces. It allocates whatever
    the server chooses to send, which for a capability an agent can point at an
    arbitrary URL is the server deciding how much memory this process uses.
    """
    chunks: list[bytes] = []
    total = 0
    while True:
        chunk = reader.read(64 * 1024)
        if not chunk:
            return b"".join(chunks)
        total += len(chunk)
        if total > limit:
            raise ResponseTooLarge(
                f"response exceeds {limit} bytes; aborted after {total}"
            )
        chunks.append(chunk)


def _copy_capped(reader: Any, writer: BinaryIO, limit: int) -> tuple[int, str]:
    """Stream at most ``limit`` bytes to ``writer`` while hashing them."""

    total = 0
    digest = hashlib.sha256()
    while True:
        chunk = reader.read(64 * 1024)
        if not chunk:
            return total, digest.hexdigest()
        total += len(chunk)
        if total > limit:
            raise ResponseTooLarge(
                f"response exceeds {limit} bytes; aborted after {total}"
            )
        writer.write(chunk)
        digest.update(chunk)


def _hash_capped(reader: Any, limit: int) -> tuple[int, str]:
    """Hash at most ``limit`` bytes without accumulating them in memory."""

    total = 0
    digest = hashlib.sha256()
    while True:
        chunk = reader.read(64 * 1024)
        if not chunk:
            return total, digest.hexdigest()
        total += len(chunk)
        if total > limit:
            raise ResponseTooLarge(
                f"response exceeds {limit} bytes; aborted after {total}"
            )
        digest.update(chunk)


def _path_matches_regular_inode(path: pathlib.Path, expected: os.stat_result) -> bool:
    try:
        current = path.lstat()
    except OSError:
        return False
    return stat.S_ISREG(current.st_mode) and os.path.samestat(expected, current)


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    """Surface a 3xx as an HTTPError instead of quietly following it.

    `urllib.request.urlopen` follows redirects inside the stdlib, so a caller
    that means to inspect every hop never sees the intermediate ones. That is
    exactly what `_http_get` and `web_probe` both need to prevent, so the
    handler lives here rather than being defined twice.
    """

    def redirect_request(self, *_args, **_kwargs):  # noqa: ANN002, ANN003
        return None


def _no_redirect_opener() -> urllib.request.OpenerDirector:
    """Build the opener per call rather than once at import.

    Reusing a module-level opener looks like the obvious optimisation and it is
    the wrong trade twice over: assembling a handler chain is nothing next to
    an HTTP request, and an opener created at import time cannot be replaced by
    a test that patches `urllib.request.build_opener` -- which is exactly how
    `web_probe`'s own test drives this code.
    """
    return urllib.request.build_opener(_NoRedirect)


#: The 3xx codes a redirect-following client would act on. 304 is deliberately
#: absent -- it is a cache answer, not a redirect, and has no Location.
_REDIRECT_CODES = frozenset({301, 302, 303, 307, 308})


@contextlib.contextmanager
def _open_http_response(
    url: str,
    *,
    timeout: float = 30.0,
    headers: dict | None = None,
    method: str = "GET",
    _max_redirects: int = 5,
) -> Iterator[tuple[Any, str, str]]:
    """Open one guarded response and keep it live for the caller to consume.

    Redirects are followed manually so the SSRF and egress guards apply to every
    hop. The response is always closed when the caller finishes or raises.
    """

    _require_network()
    method = str(method or "GET").upper()
    if method not in ("GET", "HEAD"):
        raise ValueError(f"unsupported method {method!r}; expected GET or HEAD")
    hdrs = {"User-Agent": _UA, "Accept": "*/*"}
    if headers:
        hdrs.update(headers)
    try:
        import requests  # type: ignore
    except ImportError:
        requests = None  # type: ignore

    cur = url
    from openai4s import egress

    for _hop in range(_max_redirects + 1):
        egress.check_url(cur)
        _guard_url(cur)
        if requests is not None:
            response = requests.request(
                method,
                cur,
                headers=hdrs,
                timeout=timeout,
                allow_redirects=False,
                stream=True,
            )
            try:
                if response.is_redirect and response.headers.get("Location"):
                    cur = urllib.parse.urljoin(cur, response.headers["Location"])
                    continue
                status_code = int(response.status_code)
                if not 200 <= status_code < 300:
                    response.raise_for_status()
                    raise RuntimeError(f"HTTP request failed with status {status_code}")
                yield (
                    response.raw,
                    response.url,
                    response.headers.get("Content-Type", ""),
                )
                return
            finally:
                response.close()
        req = urllib.request.Request(cur, headers=hdrs, method=method)
        try:
            # NOT `urlopen`. The stdlib opener follows redirects internally, so
            # every hop would not pass through the guards above.
            response = _no_redirect_opener().open(req, timeout=timeout)  # noqa: S310
        except urllib.error.HTTPError as error:
            location = (error.headers or {}).get("Location") if error.headers else None
            if error.code in _REDIRECT_CODES and location:
                error.close()
                cur = urllib.parse.urljoin(cur, location)
                continue
            raise
        with response:
            yield (
                response,
                response.geturl(),
                response.headers.get("Content-Type", ""),
            )
            return
    raise RuntimeError("too many redirects")


def _http_get(
    url: str,
    *,
    timeout: float = 30.0,
    headers: dict | None = None,
    method: str = "GET",
    max_bytes: int | None = None,
    _max_redirects: int = 5,
) -> tuple[bytes, str, str]:
    """Fetch a URL, following redirects MANUALLY so the SSRF guard is applied to
    every hop (a public URL can 30x-redirect to a metadata/loopback target).

    ``method`` may be GET or HEAD. HEAD exists so a caller can ask whether a
    resource is there without downloading it -- a DOI existence probe was the
    reason three bundled skills reached for raw ``urllib`` instead of this
    function, and so bypassed the egress allowlist and this guard entirely.
    It goes through exactly the same per-hop checks; a HEAD is a request.

    Returns (body_bytes, final_url, content_type). For HEAD the body is empty.
    """
    method = str(method or "GET").upper()
    limit = MAX_FETCH_BYTES if max_bytes is None else int(max_bytes)
    with _open_http_response(
        url,
        timeout=timeout,
        headers=headers,
        method=method,
        _max_redirects=_max_redirects,
    ) as (reader, final_url, content_type):
        body = b"" if method == "HEAD" else _read_capped(reader, limit)
        return body, final_url, content_type


# --------------------------------------------------------------------------- #
#  HTML -> text / markdown
# --------------------------------------------------------------------------- #
def _html_to_markdown(html_text: str) -> str:
    """Best-effort HTML → markdown. Uses BeautifulSoup when available; otherwise
    a compact regex stripper."""
    try:
        from bs4 import BeautifulSoup, NavigableString  # type: ignore
    except ImportError:
        return _strip_tags(html_text)

    soup = BeautifulSoup(html_text, "html.parser")
    for tag in soup(
        [
            "script",
            "style",
            "noscript",
            "svg",
            "head",
            "nav",
            "footer",
            "form",
            "iframe",
        ]
    ):
        tag.decompose()

    parts: list[str] = []

    # Tags whose presence among a container's descendants means the container is
    # structural (recurse so those blocks format themselves) rather than a leaf
    # holding inline prose (emit its whole text).
    structural = [
        "h1",
        "h2",
        "h3",
        "h4",
        "h5",
        "h6",
        "p",
        "li",
        "pre",
        "blockquote",
        "ul",
        "ol",
        "table",
        "section",
        "article",
    ]

    def _walk(node) -> None:
        name = getattr(node, "name", None)
        if name is None:
            return
        if name in ("h1", "h2", "h3", "h4", "h5", "h6"):
            level = int(name[1])
            parts.append("\n" + "#" * level + " " + node.get_text(" ", strip=True))
        elif name in ("p", "blockquote"):
            txt = node.get_text(" ", strip=True)
            if txt:
                parts.append("\n" + txt)
        elif name == "li":
            txt = node.get_text(" ", strip=True)
            if txt:
                parts.append("- " + txt)
        elif name in ("pre", "code"):
            txt = node.get_text("\n", strip=True)
            if txt:
                parts.append("\n```\n" + txt + "\n```")
        elif name in ("section", "article", "div", "span", "a", "td", "th", "dd", "dt"):
            # Containers/inline wrappers. If they enclose block-level content,
            # recurse so those blocks format themselves; otherwise emit the whole
            # node as one paragraph so bare text and <a> children aren't dropped
            # (e.g. the arXiv abstract <blockquote> text and <div class=authors>
            # author links, which the old code silently discarded).
            if node.find(structural) is not None:
                for child in node.children:
                    if isinstance(child, NavigableString):
                        stray = str(child).strip()
                        if stray:
                            parts.append("\n" + stray)
                    else:
                        _walk(child)
            else:
                txt = node.get_text(" ", strip=True)
                if txt:
                    parts.append("\n" + txt)
        else:
            for child in getattr(node, "children", []):
                _walk(child)

    body = soup.body or soup
    for child in getattr(body, "children", []):
        _walk(child)
    text = "\n".join(p for p in parts if p.strip())
    text = re.sub(r"\n{3,}", "\n\n", text).strip()
    if not text:  # some pages nest oddly — fall back to a flat get_text
        text = soup.get_text("\n", strip=True)
        text = re.sub(r"\n{3,}", "\n\n", text).strip()
    return text


def _strip_tags(html_text: str) -> str:
    text = re.sub(r"(?is)<(script|style|head|nav|footer|form).*?</\1>", " ", html_text)
    text = re.sub(r"(?s)<[^>]+>", " ", text)
    text = _html.unescape(text)
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n\s*\n\s*\n+", "\n\n", text)
    return text.strip()


def web_fetch(
    url: str,
    fmt: str = "markdown",
    timeout: float = 30.0,
    max_chars: int = 20000,
    *,
    method: str = "GET",
    user_agent: str | None = None,
) -> dict:
    """Fetch a URL and return its content. fmt ∈ {markdown, text, html, json}.

    ``method="HEAD"`` asks only whether the resource is there, and returns no
    ``content``. ``user_agent`` overrides the default one for services that
    require a contactable identity -- Crossref and OpenAlex serve their "polite
    pool" only to callers who send one, and without these two options three
    bundled skills used raw ``urllib`` instead, which meant their requests were
    subject to neither the egress allowlist nor the SSRF guard.
    """
    if not re.match(r"^https?://", url, re.I):
        url = "https://" + url
    if str(method or "GET").upper() == "HEAD":
        # Delegated, and deliberately does not follow redirects -- see
        # `web_probe`. Two spellings of "does this exist" that disagree about
        # redirects is one more than anybody can keep straight, and the
        # non-following answer is both the more informative one (doi.org's own
        # 302/404 rather than the publisher's) and the narrower one (a single
        # guarded request instead of a chain).
        #
        # No `content` key at all rather than an empty one: "" reads as "the
        # resource is empty", when what happened is that we did not ask.
        return {
            **web_probe(url, timeout=timeout, user_agent=user_agent),
            "method": "HEAD",
        }
    headers = {"User-Agent": user_agent} if user_agent else None
    body, final_url, ctype = _http_get(
        url, timeout=timeout, headers=headers, method=method
    )
    raw = body.decode("utf-8", errors="replace")
    is_html = ("html" in ctype.lower()) or bool(re.search(r"(?i)<html", raw[:2000]))
    if fmt == "html":
        content = raw
    elif fmt == "json" or "json" in ctype.lower():
        try:
            content = json.dumps(json.loads(raw), ensure_ascii=False, indent=2)
        except Exception:  # noqa: BLE001
            content = raw
    elif is_html:
        content = _html_to_markdown(raw) if fmt == "markdown" else _strip_tags(raw)
    else:
        content = raw
    truncated = len(content) > max_chars
    return {
        "url": final_url,
        "content_type": ctype,
        "truncated": truncated,
        "content": content[:max_chars],
        # The response exactly as it came off the wire, before decoding and
        # before any reformatting. `content` above has been through
        # `decode(errors="replace")` — which maps every invalid byte sequence
        # onto the same U+FFFD — and, for JSON, through a load/dump round trip
        # that discards the original whitespace entirely. Two materially
        # different responses can produce identical `content`, so a hash taken
        # over `content` cannot answer "are these the same bytes we received".
        # These describe the complete body even when `content` is truncated.
        "raw_sha256": hashlib.sha256(body).hexdigest(),
        "raw_bytes": len(body),
    }


def web_probe(
    url: str,
    *,
    timeout: float = 15.0,
    user_agent: str | None = None,
) -> dict:
    """Ask whether a URL is there, and report the *origin server's own* answer.

    A HEAD that follows redirects cannot answer the question a probe is usually
    asked. ``https://doi.org/<doi>`` returns 302 for a registered DOI and 404
    for an unregistered one; follow the 302 and you get the publisher's status
    instead, which may be a 403 paywall for a DOI that certainly exists. So
    this makes exactly one hop and returns what came back.

    Not following is also the narrower behaviour: one guarded request rather
    than a chain of them. The SSRF and egress checks still apply to that hop --
    a probe is a request, and exempting it would turn this into an existence
    oracle for the host's private network.

    Returns ``{url, status, location, exists}``. ``exists`` is the 2xx/3xx
    judgement callers actually want; ``status`` is there for the ones that need
    to tell 401 from 404.
    """
    if not re.match(r"^https?://", url, re.I):
        url = "https://" + url
    _require_network()
    from openai4s import egress

    egress.check_url(url)
    _guard_url(url)

    hdrs = {"User-Agent": user_agent or _UA, "Accept": "*/*"}

    request = urllib.request.Request(url, headers=hdrs, method="HEAD")
    try:
        with _no_redirect_opener().open(
            request, timeout=timeout
        ) as response:  # noqa: S310
            status = int(getattr(response, "status", 0) or 200)
            location = response.headers.get("Location") or ""
    except urllib.error.HTTPError as error:
        status = int(error.code)
        location = (error.headers or {}).get("Location") or ""
    except urllib.error.URLError as error:
        # No status could be obtained at all. Reported as a status of 0 rather
        # than as an exception, because "unreachable" is an answer to the
        # question asked and a caller probing a list of DOIs should not have
        # one connection failure end the batch.
        return {
            "url": url,
            "status": 0,
            "location": "",
            "exists": False,
            "error": str(error.reason),
        }
    return {
        "url": url,
        "status": status,
        "location": location,
        "exists": 200 <= status < 400,
    }


def web_download(
    url: str,
    destination: "os.PathLike[str] | str",
    *,
    timeout: float = 60.0,
    max_bytes: int = 64 * 1024 * 1024,
    user_agent: str | None = None,
) -> dict:
    """Fetch a URL straight to a file, bounded, through the same guards.

    `web_fetch` decodes to text, so it is the wrong shape for a ZIP or a
    coordinate file -- which is why a bundled skill downloaded the RRUFF
    spectra archive with raw ``urllib`` and, in doing so, skipped the egress
    allowlist and the SSRF guard that every hop of `_http_get` applies.

    The byte ceiling is the caller's, defaulting well above a real dataset and
    well below "whatever the server feels like sending", and it is enforced
    while streaming into a temporary sibling. Only a complete response is
    atomically published, so a failed or oversized request cannot corrupt an
    existing destination. Confinement of ``destination`` is deliberately NOT
    done here: this module knows nothing about sessions or workspaces. The Host
    service that exposes this resolves the path against the session workspace
    first, because a capability that writes wherever it is told is a capability
    that writes outside the session.
    """
    if not re.match(r"^https?://", url, re.I):
        url = "https://" + url
    path = pathlib.Path(destination)
    path.parent.mkdir(parents=True, exist_ok=True)
    headers = {"User-Agent": user_agent} if user_agent else None
    limit = int(max_bytes)
    if limit < 0:
        raise ValueError("max_bytes must be non-negative")
    stage = tempfile.TemporaryDirectory(
        prefix=f".{path.name}.download-",
        dir=path.parent,
        ignore_cleanup_errors=True,
    )
    stage_path = pathlib.Path(stage.name)
    temporary = stage_path / "response.part"
    publish_link = stage_path / "publish.link"
    verified_identity: os.stat_result | None = None
    try:
        with temporary.open("x+b") as target:
            with _open_http_response(url, timeout=timeout, headers=headers) as (
                reader,
                final_url,
                ctype,
            ):
                first_digest = _copy_capped(reader, target, limit)
            target.flush()
            verified_identity = os.fstat(target.fileno())
            publication_source = publish_link
            try:
                os.link(temporary, publish_link, follow_symlinks=False)
            except OSError as exc:
                if exc.errno not in _HARDLINK_UNSUPPORTED:
                    raise
                # FAT-family and some network filesystems cannot create hard
                # links. The private 0700 staging directory still lets them
                # use atomic replace, with the same held-fd checks before and
                # after publication. Writers to one destination must be
                # serialized on this compatibility path.
                publication_source = temporary
            if not _path_matches_regular_inode(publication_source, verified_identity):
                raise RuntimeError("download staging path changed before publication")

            # Re-read the held inode before replacing an existing destination.
            # A same-UID watcher can modify a named staging inode in place
            # without changing its identity; catching that here preserves the
            # previous destination instead of publishing bytes we did not
            # receive from the guarded response.
            target.seek(0)
            if _hash_capped(target, limit) != first_digest:
                raise RuntimeError("download bytes changed before publication")
            if not _path_matches_regular_inode(publication_source, verified_identity):
                raise RuntimeError("download staging path changed before publication")

            os.replace(publication_source, path)
            if not _path_matches_regular_inode(path, verified_identity):
                raise RuntimeError("download destination changed during publication")
            target.seek(0)
            if _hash_capped(target, limit) != first_digest:
                raise RuntimeError("download bytes changed during publication")
            if not _path_matches_regular_inode(path, verified_identity):
                raise RuntimeError("download destination changed during verification")
            size, sha256 = first_digest
    finally:
        stage.cleanup()
    return {
        "url": final_url,
        "path": str(path),
        "bytes": size,
        "content_type": ctype,
        "sha256": sha256,
    }


# --------------------------------------------------------------------------- #
#  web search (keyless, multi-engine + scholarly fast paths)
# --------------------------------------------------------------------------- #
_RETRY_PAUSE = 1.5  # seconds before the one retry pass (rate-limits are bursty)

_DOI_RE = re.compile(r"\b(10\.\d{4,9}/[^\s\"'<>]+)", re.I)
_ARXIV_ID_RE = re.compile(r"\b(\d{4}\.\d{4,5})(v\d+)?\b")

# Redirector/ad links no engine result should surface.
_AD_URL_BITS = (
    "duckduckgo.com/y.js",
    "bing.com/aclick",
    "doubleclick.net",
    "googleadservices.com",
)
_TRACKING_PARAMS = {"fbclid", "gclid", "msclkid", "mc_cid", "mc_eid", "igshid"}


def _tavily_key() -> str:
    """Tavily API key from the environment. A key saved from the UI (global
    Search settings) is written to this env var — live by the Search settings
    endpoint and at daemon startup — so a UI-entered key works without .env."""
    return os.environ.get("OPENAI4S_TAVILY_API_KEY", "").strip()


def _tavily_search(query: str, num_results: int, timeout: float) -> list[dict]:
    """Authenticated Tavily search (https://tavily.com). Tried FIRST when a key
    is configured (env ``OPENAI4S_TAVILY_API_KEY`` or the UI Search setting),
    because the keyless scrapers below are increasingly bot-blocked /
    rate-limited. Returns [] (silent fall-through to the keyless chain) when the
    key is unset or anything errors, so search never hard-depends on the key.
    stdlib-only POST — no extra deps."""
    key = _tavily_key()
    if not key or timeout <= 0:
        return []
    body = json.dumps(
        {
            "query": query,
            "max_results": max(1, min(int(num_results), 20)),
            "search_depth": "basic",
        }
    ).encode("utf-8")
    req = urllib.request.Request(
        "https://api.tavily.com/search",
        data=body,
        method="POST",
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {key}"},
    )
    try:
        with urllib.request.urlopen(req, timeout=min(timeout, 15.0)) as resp:
            payload = json.loads(resp.read().decode("utf-8", "replace"))
    except Exception:  # noqa: BLE001 - any failure → fall back to keyless engines
        return []
    out: list[dict] = []
    for r in payload.get("results") or []:
        url = (r.get("url") or "").strip()
        if not url:
            continue
        out.append(
            {
                "title": (r.get("title") or "").strip() or url,
                "url": url,
                "snippet": (r.get("content") or "").strip()[:500],
            }
        )
        if len(out) >= num_results:
            break
    return out


def web_search(query: str, num_results: int = 8, timeout: float = 20.0) -> dict:
    """Multi-engine web search. When ``OPENAI4S_TAVILY_API_KEY`` is set, an
    authenticated Tavily query is tried first (the keyless scrapers get
    bot-blocked). A DOI in the query is answered straight
    from Crossref and an arXiv id from the arXiv API (structured, reliable
    metadata); otherwise the engine chain DuckDuckGo → Bing → DuckDuckGo lite →
    Mojeek is walked until one returns hits, retrying once (and once more with a
    simplified query) if everything comes back empty. Results are deduplicated
    by normalized URL. `timeout` is the budget for the WHOLE call (every engine,
    the retry pass, and the fallback combined), so hanging endpoints can't blow
    past it. Returns {query, count, results:[{title,url,snippet}], source}."""
    _require_network()
    query = (query or "").strip()
    if not query:
        return {"query": query, "count": 0, "results": [], "note": "empty query"}
    # One wall-clock deadline shared across the entire call so a set of stalling
    # engines can never run the caller ~10x past `timeout` (each engine and the
    # retry/simplified passes draw down the same remaining budget).
    deadline = time.monotonic() + max(timeout, 1.0)
    routed = _identifier_route(query, deadline)
    if routed:
        source, results = routed
        return {
            "query": query,
            "count": len(results),
            "results": results[:num_results],
            "source": source,
        }
    # Authenticated engine first — the keyless scrapers below get bot-blocked;
    # silently falls through when the key is unset or the call errors.
    if _time_left(deadline) > 0:
        tav = _tavily_search(query, num_results, _req_timeout(deadline))
        if tav:
            return {
                "query": query,
                "count": len(tav),
                "results": tav[:num_results],
                "source": "tavily",
            }
    results, source = _engine_sweep(query, num_results, deadline)
    note = None
    if not results and _time_left(deadline) > 0:
        simplified = _simplify_query(query)
        if simplified and simplified.lower() != query.lower():
            results, source = _engine_sweep(simplified, num_results, deadline)
            if results:
                note = (
                    f"no hits for the full query — showing results for the "
                    f"simplified query {simplified!r}"
                )
    out: dict = {"query": query, "count": len(results), "results": results}
    if source:
        out["source"] = source
    if note:
        out["note"] = note
    if not results:
        out["note"] = (
            "no results from any engine (DuckDuckGo/Bing/Mojeek) — "
            "they may be rate-limiting; retry shortly with different "
            "terms, or use host.web_fetch on a known URL / a specific "
            "database API instead."
        )
    return out


# ---- budget helpers -------------------------------------------------------- #
def _time_left(deadline: float) -> float:
    return deadline - time.monotonic()


def _req_timeout(deadline: float) -> float:
    """Per-request cap: the smaller of the remaining call budget and 12s (so one
    stalled engine can't eat the whole budget on its own)."""
    return max(0.0, min(_time_left(deadline), 12.0))


# ---- scholarly identifier routing ----------------------------------------- #
def _identifier_route(query: str, deadline: float) -> tuple[str, list[dict]] | None:
    """If the query carries a scholarly identifier, resolve it via the matching
    structured API (far more reliable than scraping an engine for it):
    DOI → Crossref, arXiv id (when 'arxiv' appears in the query) → arXiv API.
    Returns (source, results) or None to fall through to the engines."""
    m = _DOI_RE.search(query)
    if m and _req_timeout(deadline) > 0:
        results = _crossref_lookup(m.group(1).rstrip(".,;)"), _req_timeout(deadline))
        if results:
            return "crossref", results
    if "arxiv" in query.lower():
        m = _ARXIV_ID_RE.search(query)
        if m and _req_timeout(deadline) > 0:
            results = _arxiv_lookup(
                m.group(1) + (m.group(2) or ""), _req_timeout(deadline)
            )
            if results:
                return "arxiv", results
    return None


def _crossref_lookup(doi: str, timeout: float) -> list[dict]:
    url = "https://api.crossref.org/works/" + urllib.parse.quote(doi)
    try:
        body, _f, _c = _http_get(url, timeout=timeout)
        msg = json.loads(body.decode("utf-8", errors="replace")).get("message") or {}
    except Exception:  # noqa: BLE001
        return []
    title = " ".join(msg.get("title") or []).strip()
    if not title:
        return []
    authors = ", ".join(
        f"{a.get('given', '')} {a.get('family', '')}".strip()
        for a in (msg.get("author") or [])[:6]
        if isinstance(a, dict)
    )
    container = " ".join(msg.get("container-title") or []).strip()
    year = ""
    try:
        year = str((msg.get("issued") or {}).get("date-parts", [[""]])[0][0] or "")
    except Exception:  # noqa: BLE001
        pass
    abstract = _strip_tags(msg.get("abstract") or "")
    head = " — ".join(b for b in (authors, container, year) if b)
    snippet = (head + (". " if head and abstract else "") + abstract)[:500]
    return [
        {
            "title": title,
            "url": msg.get("URL") or f"https://doi.org/{doi}",
            "snippet": snippet,
        }
    ]


def _arxiv_lookup(arxiv_id: str, timeout: float) -> list[dict]:
    url = "https://export.arxiv.org/api/query?id_list=" + urllib.parse.quote(arxiv_id)
    try:
        body, _f, _c = _http_get(url, timeout=timeout)
    except Exception:  # noqa: BLE001
        return []
    raw = body.decode("utf-8", errors="replace")
    out: list[dict] = []
    for entry in re.finditer(r"<entry>(.*?)</entry>", raw, re.S):
        e = entry.group(1)

        def _tag(name: str, e: str = e) -> str:
            m = re.search(rf"<{name}[^>]*>(.*?)</{name}>", e, re.S)
            if not m:
                return ""
            return _html.unescape(re.sub(r"\s+", " ", m.group(1)).strip())

        title = _tag("title")
        # the API reports an unknown id as an <entry> titled "Error"
        if not title or title.lower() == "error":
            continue
        link_m = re.search(r"<id>\s*(https?://\S+?)\s*</id>", e)
        out.append(
            {
                "title": title,
                "url": (
                    link_m.group(1) if link_m else f"https://arxiv.org/abs/{arxiv_id}"
                ),
                "snippet": _tag("summary")[:500],
            }
        )
    return out


# ---- engine chain ---------------------------------------------------------- #
def _engine_sweep(
    query: str, num_results: int, deadline: float
) -> tuple[list[dict], str | None]:
    """Walk the engine chain until one returns hits; one whole-chain retry after
    a short pause (the keyless endpoints rate-limit in bursts). Every engine and
    the retry pause draw from the shared `deadline`, so the sweep stops as soon
    as the caller's overall budget is spent rather than plowing on."""
    for attempt in (0, 1):
        for name, fn in _ENGINES:
            eng_timeout = _req_timeout(deadline)
            if eng_timeout <= 0:
                return [], None
            try:
                results = _dedupe(fn(query, num_results, eng_timeout))
            except Exception:  # noqa: BLE001
                results = []
            if results:
                return results[:num_results], name
        # retry pass only when the pause itself still fits in the budget
        if attempt == 0:
            if _time_left(deadline) <= _RETRY_PAUSE:
                break
            time.sleep(_RETRY_PAUSE)
    return [], None


def _simplify_query(query: str) -> str:
    """Zero-hit fallback: drop exact-phrase quotes and site: filters, keep the
    first 8 tokens — over-constrained queries are the usual cause of no hits."""
    q = re.sub(r"[\"“”]", " ", query)
    q = re.sub(r"\bsite:\S+", " ", q)
    toks = [t for t in re.split(r"\s+", q) if t]
    return " ".join(toks[:8]).strip()


def _norm_url(url: str) -> str:
    """Canonical form for dedup: lowercase scheme/host, no fragment, no
    utm_*/click-id tracking params, no trailing slash."""
    try:
        p = urllib.parse.urlsplit(url)
    except ValueError:
        return url
    keep = [
        (k, v)
        for k, v in urllib.parse.parse_qsl(p.query, keep_blank_values=True)
        if not k.lower().startswith("utm_") and k.lower() not in _TRACKING_PARAMS
    ]
    return urllib.parse.urlunsplit(
        (
            (p.scheme or "https").lower(),
            p.netloc.lower(),
            p.path.rstrip("/") or "/",
            urllib.parse.urlencode(keep),
            "",
        )
    )


def _dedupe(results: list[dict]) -> list[dict]:
    seen: set[str] = set()
    out: list[dict] = []
    for r in results or []:
        url = (r.get("url") or "").strip()
        if not url or any(bit in url for bit in _AD_URL_BITS):
            continue
        key = _norm_url(url)
        if key in seen:
            continue
        seen.add(key)
        out.append(r)
    return out


def _ddg_unwrap(href: str) -> str:
    # DuckDuckGo wraps hits as /l/?uddg=<encoded target>
    if "uddg=" in href:
        try:
            q = urllib.parse.urlparse(href).query
            target = urllib.parse.parse_qs(q).get("uddg", [None])[0]
            if target:
                return urllib.parse.unquote(target)
        except Exception:  # noqa: BLE001
            pass
    if href.startswith("//"):
        return "https:" + href
    return href


def _bing_unwrap(href: str) -> str:
    # Bing sometimes wraps organic hits as /ck/a?...&u=a1<base64url target>
    if "bing.com/ck/" not in href:
        return href
    try:
        q = urllib.parse.parse_qs(urllib.parse.urlsplit(href).query)
        u = (q.get("u") or [""])[0]
        if u.startswith("a1"):
            pad = "=" * (-len(u[2:]) % 4)
            return base64.urlsafe_b64decode(u[2:] + pad).decode("utf-8", "replace")
    except Exception:  # noqa: BLE001
        pass
    return href


def _ddg_html(query: str, num_results: int, timeout: float) -> list[dict]:
    url = "https://html.duckduckgo.com/html/?q=" + urllib.parse.quote(query)
    try:
        body, _final, _ct = _http_get(
            url, timeout=timeout, headers={"Referer": "https://duckduckgo.com/"}
        )
    except Exception:  # noqa: BLE001
        return []
    raw = body.decode("utf-8", errors="replace")
    out: list[dict] = []
    try:
        from bs4 import BeautifulSoup  # type: ignore

        soup = BeautifulSoup(raw, "html.parser")
        for res in soup.select(".result, .web-result"):
            a = res.select_one("a.result__a")
            if not a:
                continue
            snip_el = res.select_one(".result__snippet")
            out.append(
                {
                    "title": a.get_text(" ", strip=True),
                    "url": _ddg_unwrap(a.get("href", "")),
                    "snippet": snip_el.get_text(" ", strip=True) if snip_el else "",
                }
            )
            if len(out) >= num_results:
                break
    except ImportError:
        for m in re.finditer(r'result__a[^>]*href="([^"]+)"[^>]*>(.*?)</a>', raw, re.S):
            href, title = m.group(1), _strip_tags(m.group(2))
            out.append({"title": title, "url": _ddg_unwrap(href), "snippet": ""})
            if len(out) >= num_results:
                break
    return out


def _bing_html(query: str, num_results: int, timeout: float) -> list[dict]:
    url = (
        "https://www.bing.com/search?q="
        + urllib.parse.quote(query)
        + "&count="
        + str(min(max(num_results, 1), 30))
    )
    try:
        body, _f, _c = _http_get(
            url, timeout=timeout, headers={"Accept-Language": "en-US,en;q=0.8"}
        )
    except Exception:  # noqa: BLE001
        return []
    raw = body.decode("utf-8", errors="replace")
    out: list[dict] = []
    try:
        from bs4 import BeautifulSoup  # type: ignore

        soup = BeautifulSoup(raw, "html.parser")
        for li in soup.select("li.b_algo"):
            a = li.select_one("h2 a")
            if not a or not a.get("href"):
                continue
            snip = li.select_one(".b_caption p") or li.select_one("p")
            out.append(
                {
                    "title": a.get_text(" ", strip=True),
                    "url": _bing_unwrap(a["href"]),
                    "snippet": snip.get_text(" ", strip=True) if snip else "",
                }
            )
            if len(out) >= num_results:
                break
    except ImportError:
        for m in re.finditer(
            r'<li class="b_algo".*?<h2[^>]*><a[^>]+href="([^"]+)"[^>]*>'
            r"(.*?)</a></h2>(.*?)</li>",
            raw,
            re.S,
        ):
            snip_m = re.search(r"<p[^>]*>(.*?)</p>", m.group(3), re.S)
            out.append(
                {
                    "title": _strip_tags(m.group(2)).strip(),
                    "url": _bing_unwrap(_html.unescape(m.group(1))),
                    "snippet": _strip_tags(snip_m.group(1)).strip() if snip_m else "",
                }
            )
            if len(out) >= num_results:
                break
    return out


def _ddg_lite(query: str, num_results: int, timeout: float) -> list[dict]:
    url = "https://lite.duckduckgo.com/lite/?q=" + urllib.parse.quote(query)
    try:
        body, _f, _c = _http_get(
            url, timeout=timeout, headers={"Referer": "https://duckduckgo.com/"}
        )
    except Exception:  # noqa: BLE001
        return []
    raw = body.decode("utf-8", errors="replace")
    out: list[dict] = []
    anchors = list(
        re.finditer(
            r"<a[^>]+class=['\"]result-link['\"][^>]+href=['\"]([^'\"]+)['\"][^>]*>"
            r"(.*?)</a>",
            raw,
            re.S,
        )
    )
    for i, m in enumerate(anchors):
        # the snippet <td class="result-snippet"> sits between this link row
        # and the next result's link row
        seg_end = anchors[i + 1].start() if i + 1 < len(anchors) else len(raw)
        seg = raw[m.end() : seg_end]
        snip_m = re.search(
            r"<td[^>]*class=['\"]result-snippet['\"][^>]*>(.*?)</td>", seg, re.S
        )
        out.append(
            {
                "title": _strip_tags(m.group(2)).strip(),
                "url": _ddg_unwrap(_html.unescape(m.group(1))),
                "snippet": _strip_tags(snip_m.group(1)).strip() if snip_m else "",
            }
        )
        if len(out) >= num_results:
            break
    return out


def _mojeek_html(query: str, num_results: int, timeout: float) -> list[dict]:
    url = "https://www.mojeek.com/search?q=" + urllib.parse.quote(query)
    try:
        body, _f, _c = _http_get(url, timeout=timeout)
    except Exception:  # noqa: BLE001
        return []
    raw = body.decode("utf-8", errors="replace")
    out: list[dict] = []
    try:
        from bs4 import BeautifulSoup  # type: ignore

        soup = BeautifulSoup(raw, "html.parser")
        for li in soup.select("ul.results-standard li"):
            a = li.select_one("h2 a") or li.select_one("a.title")
            if not a or not a.get("href"):
                continue
            snip = li.select_one("p.s")
            out.append(
                {
                    "title": a.get_text(" ", strip=True),
                    "url": a["href"],
                    "snippet": snip.get_text(" ", strip=True) if snip else "",
                }
            )
            if len(out) >= num_results:
                break
    except ImportError:
        for m in re.finditer(
            r'<h2><a[^>]+href="([^"]+)"[^>]*>(.*?)</a></h2>\s*'
            r'(?:<p class="s">(.*?)</p>)?',
            raw,
            re.S,
        ):
            out.append(
                {
                    "title": _strip_tags(m.group(2)).strip(),
                    "url": _html.unescape(m.group(1)),
                    "snippet": _strip_tags(m.group(3) or "").strip(),
                }
            )
            if len(out) >= num_results:
                break
    return out


# Ordered by result quality (snippet richness) and rate-limit tolerance.
_ENGINES: tuple[tuple[str, object], ...] = (
    ("duckduckgo", _ddg_html),
    ("bing", _bing_html),
    ("duckduckgo-lite", _ddg_lite),
    ("mojeek", _mojeek_html),
)