"""Web tools for the assistant (Notion-style "research" capabilities).

Phase 1 of the documented web-toolset roadmap:

* ``web_search``  — query the self-hosted SearXNG instance (no API key) and,
  when it returns nothing, fall back to keyless HTML providers (DuckDuckGo,
  then Bing) so a dead meta-search instance never leaves the assistant
  answering « je n'ai pas accès à internet ».
* ``fetch_url``   — retrieve a public web page and return readable text.

Phase 2 (#92) additions:

* keyed providers — Tavily, Brave Search, SerpAPI and Exa are used first when
  their API key is configured (env, injected by Infisical in production);
* SQLite cache — search/fetch results are cached with a TTL
  (:mod:`backend.tools.webcache`);
* retry with backoff — transient network errors get one extra attempt;
* dynamic rendering — ``fetch_url(render=True)`` uses an isolated Playwright
  worker (optional dependency, graceful degradation).

All are READ-risk tools (no confirmation), rate-limited through the shared
registry, SSRF-guarded (scheme + private-address rejection), and size-capped.

Configuration (environment):
* ``OBSIGATE_SEARXNG_URL``   — defaults to https://search.dracodev.net
* ``OBSIGATE_WEB_TIMEOUT``   — seconds, default 10
* ``OBSIGATE_WEB_FALLBACK``  — ``0``/``false`` disables the keyless HTML
  fallbacks (SearXNG only), default enabled
* ``OBSIGATE_TAVILY_API_KEY`` / ``OBSIGATE_BRAVE_API_KEY`` /
  ``OBSIGATE_SERPAPI_API_KEY`` / ``OBSIGATE_EXA_API_KEY`` — optional keyed
  providers, tried before SearXNG when set
* ``OBSIGATE_WEB_PROVIDERS`` — optional comma-separated provider order
  (e.g. ``brave,searxng``); keyed providers without a key are skipped
* ``OBSIGATE_WEB_RETRY``     — extra attempts for transient network errors
  (default 1)
* ``OBSIGATE_WEB_CACHE_TTL`` — cache TTL seconds, ``0`` disables (default 900)
"""

from __future__ import annotations

import base64
import binascii
import html as html_lib
import ipaddress
import logging
import os
import re
import socket
import time
from collections.abc import Callable
from typing import Any
from urllib.parse import parse_qs, urlparse

import httpx

from backend.tools import webcache
from backend.tools.context import ToolError, ToolRisk, ToolScope
from backend.tools.registry import tool
from backend.tools.schemas import FetchUrlInput, WebSearchInput
from backend.tools.secrets import get_tool_key

logger = logging.getLogger("obsigate.tools.web")

SEARXNG_URL = os.environ.get("OBSIGATE_SEARXNG_URL", "https://search.dracodev.net")
WEB_TIMEOUT = float(os.environ.get("OBSIGATE_WEB_TIMEOUT", "10"))
WEB_FALLBACK_ENABLED = os.environ.get("OBSIGATE_WEB_FALLBACK", "1").strip().lower() not in {
    "0",
    "false",
    "no",
    "off",
}
WEB_RETRY_ATTEMPTS = int(os.environ.get("OBSIGATE_WEB_RETRY", "1"))
USER_AGENT = "ObsiGateAssistant/1.0 (+self-hosted vault AI)"
# Search engines reject non-browser agents on their public HTML endpoints.
BROWSER_UA = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)
# A minimal UA is not enough: Bing serves decoy SERPs (unrelated results) to
# requests missing the usual browser navigation headers.
BROWSER_HEADERS = {
    "User-Agent": BROWSER_UA,
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
    "Accept-Language": "fr-CA,fr;q=0.9,en-US;q=0.8,en;q=0.7",
    "Sec-Fetch-Dest": "document",
    "Sec-Fetch-Mode": "navigate",
    "Sec-Fetch-Site": "none",
    "Sec-Fetch-User": "?1",
    "Upgrade-Insecure-Requests": "1",
}
MAX_FETCH_BYTES = 1_500_000
MAX_TEXT_CHARS = 20_000

_BLOCKED_TAGS_RE = re.compile(
    r"<(script|style|noscript|template|svg)\b.*?</\1>", re.IGNORECASE | re.DOTALL
)
_TAG_RE = re.compile(r"<[^>]+>")
_BLOCK_SPLIT_RE = re.compile(
    r"</?(?:p|div|br|li|h[1-6]|tr|table|ul|ol|section|article|header|footer)\b[^>]*>",
    re.IGNORECASE,
)
_DDG_RESULT_RE = re.compile(
    r'<a[^>]*class="result__a"[^>]*href="([^"]+)"[^>]*>(.*?)</a>', re.IGNORECASE | re.DOTALL
)
_DDG_SNIPPET_RE = re.compile(
    r'<a[^>]*class="result__snippet"[^>]*>(.*?)</a>', re.IGNORECASE | re.DOTALL
)
_BING_RESULT_RE = re.compile(
    r'<h2[^>]*>\s*<a[^>]*href="([^"]+)"[^>]*>(.*?)</a>', re.IGNORECASE | re.DOTALL
)
_BING_SNIPPET_RE = re.compile(
    r'<p class="b_lineclamp[^"]*">(.*?)</p>', re.IGNORECASE | re.DOTALL
)


class SSRFError(ToolError):
    """Raised for a URL whose host is private/loopback or scheme unsupported."""


def _assert_public_http_url(url: str) -> str:
    """Reject non-http(s) schemes and private/loopback/link-local targets."""
    try:
        parsed = urlparse(url)
    except ValueError as e:
        raise SSRFError("URL invalide", code="invalid_url") from e
    if parsed.scheme not in ("http", "https"):
        raise SSRFError("Seuls les schémas http/https sont autorisés", code="invalid_scheme")
    host = parsed.hostname
    if not host:
        raise SSRFError("URL sans hôte", code="invalid_url")
    # Resolve the host so DNS-rebinding to internal IPs is also caught.
    try:
        infos = socket.getaddrinfo(host, None)
    except socket.gaierror as e:
        raise SSRFError(f"Hôte introuvable: {host}", code="dns_error") from e
    for info in infos:
        ip = ipaddress.ip_address(info[4][0])
        if (
            ip.is_private
            or ip.is_loopback
            or ip.is_link_local
            or ip.is_reserved
            or ip.is_multicast
            or ip.is_unspecified
        ):
            raise SSRFError("Accès aux adresses internes interdit", code="ssrf_blocked")
    return url


def _html_to_text(raw: str) -> str:
    """Cheap HTML → readable text: strip scripts/styles, tags, then compress.

    Comments (which may carry script-like payloads) are removed first.
    """
    text = re.sub(r"<!--.*?-->", " ", raw, flags=re.DOTALL)
    text = _BLOCKED_TAGS_RE.sub(" ", text)
    # Keep block boundaries as newlines before dropping the remaining tags.
    text = _BLOCK_SPLIT_RE.sub("\n", text)
    text = _TAG_RE.sub("", text)
    text = html_lib.unescape(text)
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r" ?\n ?", "\n", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()


def _response_text(resp: httpx.Response) -> str:
    """Decode a response body without relying on ``resp.text`` (easier to mock)."""
    return resp.content.decode(resp.encoding or "utf-8", errors="replace")


def _clean_fragment(fragment: str) -> str:
    return html_lib.unescape(_TAG_RE.sub("", fragment)).strip()


def _result(
    title: str, url: str, snippet: str, published: Any = None, score: Any = None
) -> dict[str, Any]:
    return {
        "title": (title or "")[:300],
        "url": url or "",
        "snippet": (snippet or "")[:600],
        "published": published,
        "score": score,
    }


def _with_retry(call: Callable[[], Any]) -> Any:
    """Run *call* with one extra attempt on transient network errors.

    House-made backoff (the roadmap's « tenacity ou boucle maison »): DNS
    blips and rate-limit hiccups are the common failure mode, and a single
    retry keeps the fallback chain from being consumed too early.
    """
    for attempt in range(1 + max(0, WEB_RETRY_ATTEMPTS)):
        try:
            return call()
        except httpx.TransportError:
            if attempt >= max(0, WEB_RETRY_ATTEMPTS):
                raise
            time.sleep(0.2 * (attempt + 1))
    raise RuntimeError("unreachable")  # pragma: no cover


def _env_key(name: str) -> str:
    """Read an API key: configuration-page store first, then environment."""
    return get_tool_key(name)


def _search_tavily(query: str, params: WebSearchInput) -> tuple[list[dict[str, Any]], list[str]]:
    """Tavily Search API (agent-oriented results, key required)."""
    resp = httpx.post(
        "https://api.tavily.com/search",
        json={
            "api_key": _env_key("OBSIGATE_TAVILY_API_KEY"),
            "query": query,
            "max_results": params.max_results,
            "search_depth": "basic",
            "include_answer": False,
        },
        headers={"User-Agent": USER_AGENT},
        timeout=WEB_TIMEOUT,
    )
    resp.raise_for_status()
    data = resp.json()
    return [
        _result(item.get("title") or "", item.get("url") or "", item.get("content") or "")
        for item in (data.get("results") or [])
    ], []


def _search_brave(query: str, params: WebSearchInput) -> tuple[list[dict[str, Any]], list[str]]:
    """Brave Search API (key required)."""
    resp = httpx.get(
        "https://api.search.brave.com/res/v1/web/search",
        params={"q": query, "count": params.max_results, "safesearch": "moderate"},
        headers={
            "X-Subscription-Id": _env_key("OBSIGATE_BRAVE_API_KEY"),
            "Accept": "application/json",
            "User-Agent": USER_AGENT,
        },
        timeout=WEB_TIMEOUT,
    )
    resp.raise_for_status()
    data = resp.json()
    return [
        _result(item.get("title") or "", item.get("url") or "", item.get("description") or "")
        for item in ((data.get("web") or {}).get("results") or [])
    ], []


def _search_serpapi(query: str, params: WebSearchInput) -> tuple[list[dict[str, Any]], list[str]]:
    """SerpAPI (Google SERP, key required)."""
    resp = httpx.get(
        "https://serpapi.com/search",
        params={"q": query, "api_key": _env_key("OBSIGATE_SERPAPI_API_KEY"),
                "num": params.max_results},
        headers={"User-Agent": USER_AGENT},
        timeout=WEB_TIMEOUT,
    )
    resp.raise_for_status()
    data = resp.json()
    return [
        _result(item.get("title") or "", item.get("link") or "", item.get("snippet") or "")
        for item in (data.get("organic_results") or [])
    ], []


def _search_exa(query: str, params: WebSearchInput) -> tuple[list[dict[str, Any]], list[str]]:
    """Exa neural search (key required)."""
    resp = httpx.post(
        "https://api.exa.ai/search",
        json={"query": query, "numResults": params.max_results},
        headers={
            "x-api-key": _env_key("OBSIGATE_EXA_API_KEY"),
            "User-Agent": USER_AGENT,
        },
        timeout=WEB_TIMEOUT,
    )
    resp.raise_for_status()
    data = resp.json()
    return [
        _result(item.get("title") or "", item.get("url") or "", (item.get("text") or "")[:600])
        for item in (data.get("results") or [])
    ], []


# Keyed providers: name -> (implementation, API key env var)
_KEYED_PROVIDERS: dict[str, tuple[_Provider, str]] = {
    "tavily": (_search_tavily, "OBSIGATE_TAVILY_API_KEY"),
    "brave": (_search_brave, "OBSIGATE_BRAVE_API_KEY"),
    "serpapi": (_search_serpapi, "OBSIGATE_SERPAPI_API_KEY"),
    "exa": (_search_exa, "OBSIGATE_EXA_API_KEY"),
}


def _search_searxng(
    query: str, params: WebSearchInput
) -> tuple[list[dict[str, Any]], list[str]]:
    """Query the self-hosted SearXNG instance (JSON API)."""
    url = SEARXNG_URL.rstrip("/") + "/search"
    resp = httpx.get(
        url,
        params={
            "q": query,
            "format": "json",
            "categories": params.category or "general",
            "pageno": max(1, params.page),
            **({"language": params.language} if params.language else {}),
            "safesearch": "1",
        },
        headers={"User-Agent": USER_AGENT},
        timeout=WEB_TIMEOUT,
        follow_redirects=False,
    )
    resp.raise_for_status()
    data = resp.json()
    results = [
        _result(
            item.get("title") or "",
            item.get("url") or "",
            item.get("content") or "",
            item.get("publishedDate"),
            item.get("score"),
        )
        for item in (data.get("results") or [])[: params.max_results]
    ]
    unresponsive = [
        name for entry in (data.get("unresponsive_engines") or [])
        for name in ([entry[0]] if isinstance(entry, (list, tuple)) and entry else [entry])
        if isinstance(name, str)
    ]
    return results, unresponsive


def _unwrap_duckduckgo_url(href: str) -> str:
    """DuckDuckGo HTML wraps hits in ``/l/?uddg=<urlencoded target>``."""
    href = html_lib.unescape(href)
    if href.startswith("//"):
        href = "https:" + href
    if "uddg=" in href:
        values = parse_qs(urlparse(href).query).get("uddg")
        if values:
            return values[0]
    return href


def _search_duckduckgo(
    query: str, params: WebSearchInput
) -> tuple[list[dict[str, Any]], list[str]]:
    """Keyless fallback: scrape the DuckDuckGo no-JS HTML endpoint."""
    resp = httpx.get(
        "https://html.duckduckgo.com/html/",
        params={"q": query, **({"kl": params.language} if params.language else {})},
        headers=BROWSER_HEADERS,
        timeout=WEB_TIMEOUT,
        follow_redirects=False,
    )
    resp.raise_for_status()
    body = _response_text(resp)
    snippets = [_clean_fragment(m.group(1)) for m in _DDG_SNIPPET_RE.finditer(body)]
    results: list[dict[str, Any]] = []
    for index, match in enumerate(_DDG_RESULT_RE.finditer(body)):
        results.append(
            _result(
                _clean_fragment(match.group(2)),
                _unwrap_duckduckgo_url(match.group(1)),
                snippets[index] if index < len(snippets) else "",
            )
        )
        if len(results) >= params.max_results:
            break
    return results, []


def _unwrap_bing_url(href: str) -> str:
    """Bing wraps hits in ``/ck/a?...&u=a1<base64url target>``."""
    href = html_lib.unescape(href)
    match = re.search(r"[?&]u=a1([A-Za-z0-9_\-]+)", href)
    if not match:
        return href
    token = match.group(1).replace("-", "+").replace("_", "/")
    token += "=" * (-len(token) % 4)
    try:
        return base64.b64decode(token).decode("utf-8", errors="replace")
    except (ValueError, binascii.Error):
        return href


def _search_bing(
    query: str, params: WebSearchInput
) -> tuple[list[dict[str, Any]], list[str]]:
    """Last-resort keyless fallback: scrape Bing's result page."""
    resp = httpx.get(
        "https://www.bing.com/search",
        params={"q": query, **({"setlang": params.language} if params.language else {})},
        headers=BROWSER_HEADERS,
        timeout=WEB_TIMEOUT,
        follow_redirects=False,
    )
    resp.raise_for_status()
    body = _response_text(resp)
    snippets = [_clean_fragment(m.group(1)) for m in _BING_SNIPPET_RE.finditer(body)]
    results: list[dict[str, Any]] = []
    for index, match in enumerate(_BING_RESULT_RE.finditer(body)):
        results.append(
            _result(
                _clean_fragment(match.group(2)),
                _unwrap_bing_url(match.group(1)),
                snippets[index] if index < len(snippets) else "",
            )
        )
        if len(results) >= params.max_results:
            break
    return results, []


_Provider = Callable[[str, WebSearchInput], "tuple[list[dict[str, Any]], list[str]]"]


def _provider_chain() -> list[tuple[str, _Provider]]:
    """Ordered providers: keyed APIs first, then self-hosted, then keyless.

    ``OBSIGATE_WEB_PROVIDERS`` (comma-separated) overrides the default order;
    unknown names are ignored and keyed providers without their key are skipped.
    """
    chain: list[tuple[str, _Provider]] = []
    configured = [
        name.strip().lower()
        for name in os.environ.get("OBSIGATE_WEB_PROVIDERS", "").split(",")
        if name.strip()
    ]
    for name in configured or list(_KEYED_PROVIDERS):
        entry = _KEYED_PROVIDERS.get(name)
        if entry and _env_key(entry[1]):
            chain.append((name, entry[0]))
    chain.append(("searxng", _search_searxng))
    if WEB_FALLBACK_ENABLED:
        chain.append(("duckduckgo", _search_duckduckgo))
        chain.append(("bing", _search_bing))
    return chain


@tool(
    name="web_search",
    description=(
        "Search the public web for current information and return ranked results "
        "(title, url, snippet). Use for facts outside the vault: weather, news, "
        "documentation, versions, prices, anything that needs live sources."
    ),
    input_model=WebSearchInput,
    risk=ToolRisk.READ,
    scopes=(ToolScope.IN_APP,),
)
def web_search(ctx, params: WebSearchInput) -> dict[str, Any]:
    """Try each configured provider and return the first non-empty result set."""
    query = params.query.strip()
    if not query:
        raise ToolError("Requête vide", code="invalid_arguments")

    key = webcache.cache_key("search", {
        "q": query,
        "max_results": params.max_results,
        "category": params.category,
        "language": params.language,
        "page": params.page,
    })
    cached = webcache.cache_get(key)
    if cached is not None:
        return {**cached, "cached": True}

    attempts: list[str] = []
    unresponsive: list[str] = []
    reachable = False
    last_error: Exception | None = None

    for name, provider in _provider_chain():
        attempts.append(name)

        def _attempt(p: _Provider = provider) -> tuple[list[dict[str, Any]], list[str]]:
            return p(query, params)

        try:
            results, engines = _with_retry(_attempt)
        except (httpx.HTTPError, ValueError, AttributeError) as e:
            logger.warning("web_search provider %s failed: %s", name, e)
            last_error = e
            continue
        reachable = True
        if engines:
            unresponsive = engines
        if results:
            payload: dict[str, Any] = {
                "query": query,
                "provider": name,
                "results": results,
                "count": len(results),
            }
            if unresponsive:
                payload["unresponsive_engines"] = unresponsive[:8]
            webcache.cache_set(key, payload)
            return payload

    if not reachable:
        raise ToolError(
            "Le moteur de recherche web est momentanément indisponible.",
            code="web_search_unavailable",
        ) from last_error

    # Every provider answered but returned nothing: tell the model explicitly
    # so it stops retrying the same query until its tool quota burns out.
    payload = {
        "query": query,
        "provider": attempts[-1],
        "results": [],
        "count": 0,
        "warning": (
            "Aucun résultat : les fournisseurs de recherche web sont "
            f"indisponibles ({', '.join(attempts)}). "
            "Ne relance pas la même recherche — dis-le à l'utilisateur."
        ),
    }
    if unresponsive:
        payload["unresponsive_engines"] = unresponsive[:8]
    return payload


@tool(
    name="fetch_url",
    description=(
        "Fetch a public web page (http/https) and return its readable text. "
        "Use after web_search to read a promising result in detail. HTML is "
        "converted to plain text; binary pages are rejected."
    ),
    input_model=FetchUrlInput,
    risk=ToolRisk.READ,
    scopes=(ToolScope.IN_APP,),
)
def fetch_url(ctx, params: FetchUrlInput) -> dict[str, Any]:
    """Retrieve one page, guard against SSRF, and extract its text."""
    url = _assert_public_http_url(params.url.strip())
    key = webcache.cache_key("fetch", {"url": url, "render": params.render})
    cached = webcache.cache_get(key)
    if cached is not None:
        return {**cached, "cached": True}

    if params.render:
        # Dynamic pages (SPA/React): delegated to the isolated Playwright
        # worker; the browser dependency stays optional (graceful error).
        from backend.tools.webrender import render_page

        payload = render_page(url)
        webcache.cache_set(key, payload)
        return payload

    try:
        # Follow redirects manually so every hop is re-checked against the
        # private-address SSRF guard (a public page can redirect to 127.0.0.1).
        resp = None
        for _hop in range(5):
            resp = httpx.get(
                url,
                headers={"User-Agent": USER_AGENT, "Accept": "text/html,application/xhtml+xml,*/*"},
                timeout=WEB_TIMEOUT,
                follow_redirects=False,
            )
            if resp.status_code in (301, 302, 303, 307, 308):
                location = resp.headers.get("location") or ""
                if not location:
                    break
                url = str(httpx.URL(url).join(location))
                url = _assert_public_http_url(url)
                continue
            break
        assert resp is not None
        resp.raise_for_status()
    except SSRFError:
        raise
    except httpx.HTTPError as e:
        logger.warning("fetch_url failed for %s: %s", url, e)
        raise ToolError("Impossible de récupérer la page.", code="fetch_unavailable") from e
    ctype = (resp.headers.get("content-type") or "").lower()
    if not any(t in ctype for t in ("html", "xml", "text", "json", "markdown")):
        raise ToolError(
            f"Type de contenu non pris en charge: {ctype.split(';')[0] or 'inconnu'}",
            code="unsupported_content_type",
        )
    raw = (resp.content[:MAX_FETCH_BYTES]).decode(resp.encoding or "utf-8", errors="replace")
    title_match = re.search(r"<title[^>]*>(.*?)</title>", raw, re.IGNORECASE | re.DOTALL)
    title = html_lib.unescape(title_match.group(1)).strip()[:300] if title_match else ""
    text = _html_to_text(raw)[:MAX_TEXT_CHARS]
    payload = {
        "url": str(resp.url),
        "status": resp.status_code,
        "title": title,
        "text": text,
        "truncated": len(raw) > MAX_TEXT_CHARS,
    }
    webcache.cache_set(key, payload)
    return payload