# SPDX-License-Identifier: AGPL-3.0-or-later
"""Yandex Search API (the official **paid** `Yandex Search API v2`_).

Unlike the :origin:`yandex <searx/engines/yandex.py>` engine (which scrapes the
public HTML interface and is prone to CAPTCHA blocking), this engine talks to
the official, paid Yandex Cloud Search API.  It requires a Yandex Cloud account,
a *folder id* and an *API key*.

The API answers with a Base64-encoded XML document, which is decoded and
parsed here.

Configuration
=============

The engine is inactive by default because it needs credentials.  To enable it,
set ``inactive: false`` and add your ``api_key`` and ``yandex_folder_id`` to
:origin:`searx/settings.yml`:

.. code:: yaml

  - name: yandex api
    engine: yandex_api
    shortcut: yda
    categories: [general, web]
    inactive: false
    api_key: ""           # Yandex Cloud API key (``Api-Key``)
    yandex_folder_id: ""  # Yandex Cloud folder id
    # optional, see below:
    yandex_default_language: en

.. _Yandex Search API v2:
   https://aistudio.yandex.ru/docs/en/search-api/api-ref/WebSearch/search.html
"""

import math
import typing as t
from base64 import b64decode

from lxml import etree

from searx.exceptions import SearxEngineAPIException
from searx.result_types import EngineResults
from searx.utils import extract_text

if t.TYPE_CHECKING:
    from searx.extended_types import SXNG_Response
    from searx.search.processors import OnlineParams


about = {
    "website": "https://yandex.cloud/en/services/search-api",
    "wikidata_id": "Q5281",
    "official_api_documentation": "https://aistudio.yandex.ru/docs/en/search-api/api-ref/WebSearch/search.html",
    "use_official_api": True,
    "require_api_key": True,
    "results": "XML",
}

# Engine configuration
categories = ["general", "web"]
paging = True
safesearch = True

# Credentials, overwritten via settings.yml
api_key: str = ""
"""Yandex Cloud API key, passed as ``Authorization: Api-Key <api_key>``."""

yandex_folder_id: str = ""
"""Yandex Cloud folder id the API key belongs to."""

# Search tuning, overwritten via settings.yml
yandex_default_language: str = "en"
"""Default query language.  It selects the Yandex search domain (e.g. yandex.ru
for ``ru``, yandex.com for ``en``) and the language of the search-result
notifications, but only as a fallback -- a request whose own locale matches
:py:obj:`language_map` overrides it.  Must be one of its keys: ``ru``, ``be``,
``kk``, ``uk``, ``tr`` or ``en``."""

region: str = ""
"""Optional Yandex `region id`.
Only meaningful together with ``SEARCH_TYPE_RU``.

__ https://aistudio.yandex.ru/docs/en/search-api/reference/regions.html
"""

page_size: int = 10
"""Number of results requested per page."""

base_url = "https://searchapi.api.cloud.yandex.net/v2/web/search"

# searxng safesearch level -> Yandex familyMode
safesearch_map = {
    0: "FAMILY_MODE_NONE",
    1: "FAMILY_MODE_MODERATE",
    2: "FAMILY_MODE_STRICT",
}

# Map a query language to a (search_type, l10n) pair.  It drives both the
# per-request override (when the query's locale matches) and the
# ``yandex_default_language`` default.
language_map = {
    "ru": ("SEARCH_TYPE_RU", "LOCALIZATION_RU"),
    "be": ("SEARCH_TYPE_BE", "LOCALIZATION_BE"),
    "kk": ("SEARCH_TYPE_KK", "LOCALIZATION_KK"),
    "uk": ("SEARCH_TYPE_RU", "LOCALIZATION_UK"),
    "tr": ("SEARCH_TYPE_TR", "LOCALIZATION_TR"),
    "en": ("SEARCH_TYPE_COM", "LOCALIZATION_EN"),
    # Uzbek ('uz') is intentionally omitted: Yandex offers SEARCH_TYPE_UZ but no
    # matching LOCALIZATION_UZ, so such queries fall back to the defaults above.
}


def setup(_):
    """Validate credentials and paging limits when the engine is loaded."""
    if not api_key or not yandex_folder_id:
        raise SearxEngineAPIException("missing 'api_key' and/or 'yandex_folder_id' in engine settings")
    if not 1 <= page_size <= 100:
        raise SearxEngineAPIException("'page_size' must be in the range 1..100 (Yandex 'groupsOnPage')")
    if yandex_default_language not in language_map:
        raise SearxEngineAPIException(f"'yandex_default_language' must be one of {sorted(language_map)}")


def request(query: str, params: "OnlineParams"):

    # Yandex returns at most 250 results for a query.
    max_page = math.ceil(250 / page_size)
    if params["pageno"] > max_page:
        params["url"] = None
        return

    if len(query) > 400:
        # Yandex rejects a 'queryText' longer than 400 characters; decline the
        # request gracefully instead of provoking an API error.
        params["url"] = None
        return

    lang = params["searxng_locale"].split("-")[0].lower()
    req_search_type, req_l10n = language_map.get(lang, language_map[yandex_default_language])

    body: dict[str, t.Any] = {
        "query": {
            "searchType": req_search_type,
            "queryText": query,
            "familyMode": safesearch_map[params["safesearch"]],
            # the API uses a 0-based page index
            "page": str(params["pageno"] - 1),
        },
        "groupSpec": {
            "groupMode": "GROUP_MODE_FLAT",
            "groupsOnPage": str(page_size),
            "docsInGroup": "1",
        },
        "l10n": req_l10n,
        "folderId": yandex_folder_id,
        "responseFormat": "FORMAT_XML",
    }
    # Yandex accepts a 'region' only together with the Russian search type.
    if region and req_search_type == "SEARCH_TYPE_RU":
        body["region"] = region

    params["method"] = "POST"
    params["url"] = base_url
    params["headers"]["Authorization"] = f"Api-Key {api_key}"
    params["headers"]["Content-Type"] = "application/json"
    params["json"] = body


def _raw_xml(resp: "SXNG_Response") -> bytes:
    """Extract and Base64-decode the XML payload out of the JSON envelope.

    The synchronous ``/v2/web/search`` endpoint returns ``{"rawData":
    "<base64>"}`` on success; HTTP errors are raised upstream via
    ``raise_for_httperror``.
    """
    data: dict[str, t.Any] = resp.json()

    raw_data = data.get("rawData")
    if raw_data is None:
        raise SearxEngineAPIException("Yandex Search API: no 'rawData' in response")

    return b64decode(raw_data)


def response(resp: "SXNG_Response") -> EngineResults:
    res = EngineResults()

    dom = etree.fromstring(_raw_xml(resp))  # pylint: disable=c-extension-no-member

    # An <error> inside <response> signals an application error.  Code 15 simply
    # means "nothing was found" and must not raise.
    error = dom.find(".//response/error")
    if error is not None:
        if error.get("code") == "15":
            return res
        raise SearxEngineAPIException(f"Yandex Search API error {error.get('code')}: {error.text}")

    for doc in dom.iterfind(".//doc"):
        url = extract_text(doc.find("url"), allow_none=True)
        title = extract_text(doc.find("title"), allow_none=True)
        if not url or not title:
            continue

        content = extract_text(doc.find("headline"), allow_none=True)
        if not content:
            passages = doc.find("passages")
            if passages is not None:
                content = " ".join(extract_text(p) or "" for p in passages.iterfind("passage")).strip()

        res.add(
            res.types.MainResult(
                url=url,
                title=title,
                content=content or "",
            )
        )

    return res