summaryrefslogtreecommitdiff
path: root/searx/engines/exaapi.py
blob: 81092f117d9306189fd6caabb565d044034077de (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Engine to search using the official `Exa Search API`_. Exa is a search engine for AI agents.

.. _Exa Search API: https://exa.ai/docs/reference/search

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

The engine has the following mandatory setting:

- :py:obj:`api_key`

You can obtain an API key from the `API Key section <https://dashboard.exa.ai/api-keys>`_ in the Exa dashboard.

Optional settings are:

- :py:obj:`results_per_page`
- :py:obj:`search_type`
- :py:obj:`content_mode`
- :py:obj:`content_max_characters`

.. code:: yaml

  - name: exaapi
    engine: exaapi
    api_key: "..."
    results_per_page: 10
    search_type: auto
    content_mode: highlights
    inactive: false

The API supports SafeSearch and region-aware results.
"""

import typing as t

from dateutil import parser

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

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


SearchType = t.Literal["fast", "auto", "instant", "deep", "deep-lite", "deep-reasoning"]
ContentMode = t.Literal["highlights", "text"]

about = {
    "website": "https://exa.ai",
    "wikidata_id": None,
    "official_api_documentation": "https://exa.ai/docs/reference/search",
    "use_official_api": True,
    "require_api_key": True,
    "results": "JSON",
}

api_key: str = ""
"""API key for Exa Search API (required)."""

categories = ["general", "web"]
safesearch = True

base_url = "https://api.exa.ai/search"
results_per_page: int = 10
"""Maximum number of results per request. Value must be between 1 and 100, default is 10."""

search_type: SearchType = "auto"
"""Search type. Default is auto, see documentation for more information."""

content_mode: ContentMode = "highlights"
"""Content to request from the API: ``highlights`` (excerpts) or ``text`` (page text)."""

content_max_characters: int = 500
"""Maximum characters for the requested content."""


def init(_):
    if not api_key:
        raise SearxEngineAPIException("No API key provided")
    if not 1 <= results_per_page <= 100:
        raise ValueError("results_per_page must be between 1 and 100")
    if search_type not in t.get_args(SearchType):
        raise ValueError(f"Unsupported search type: {search_type}")
    if content_mode not in t.get_args(ContentMode):
        raise ValueError(f"Unsupported content mode: {content_mode}")
    if content_max_characters < 1:
        raise ValueError("content_max_characters must be at least 1")


def _contents_payload() -> dict[str, t.Any]:
    if content_mode == "text":
        return {"text": {"maxCharacters": content_max_characters, "stripLinks": True}}
    return {"highlights": {"maxCharacters": content_max_characters}}


def _extract_content(result: dict[str, t.Any]) -> str:
    if content_mode == "text":
        return html_to_text(result.get("text") or "")
    return html_to_text(" ".join(result.get("highlights") or []))


def request(query: str, params: "OnlineParams") -> None:
    """Create the API request."""
    body: dict[str, t.Any] = {
        "query": query,
        "type": search_type,
        "numResults": results_per_page,
        "contents": _contents_payload(),
    }

    # Apply SafeSearch if enabled
    if params["safesearch"]:
        body["moderation"] = True

    # Apply region-aware results if specified
    locale_parts = params["searxng_locale"].split("-")
    region = locale_parts[-1]
    if len(locale_parts) > 1:
        body["userLocation"] = region.upper()

    params["url"] = base_url
    params["method"] = "POST"
    params["headers"]["x-api-key"] = api_key
    params["json"] = body


def _extract_published_date(value: str | None):
    """Extract and parse the published date from the API response.

    Args:
        value: Raw date string from the API

    Returns:
        Parsed datetime object or None if parsing fails
    """
    if not value:
        return None
    try:
        return parser.parse(value)
    except (parser.ParserError, TypeError, OverflowError):
        return None


def response(resp: "SXNG_Response") -> EngineResults:
    """Process the API response and return results."""
    res = EngineResults()

    for result in resp.json().get("results", []):
        url = result.get("url")
        if not url:
            continue

        res.add(
            res.types.MainResult(
                url=url,
                title=html_to_text(result.get("title") or url),
                content=_extract_content(result),
                thumbnail=result.get("image") or "",
                publishedDate=_extract_published_date(result.get("publishedDate")),
                author=result.get("author") or "",
            )
        )

    return res