[fix] heexy: circumvent botblocking

1 files changed, 41 insertions, 5 deletions

diff --git a/searx/engines/heexy.py b/searx/engines/heexy.py
index fd41dbd3d..f566a7953 100644
--- a/searx/engines/heexy.py
+++ b/searx/engines/heexy.py
@@ -13,9 +13,13 @@ It seems to use Bing internally, as the image thumbnails are loaded from Bing.
 from urllib.parse import urlencode
 
 import typing as t
+from lxml import html
 
-from searx.exceptions import SearxEngineAccessDeniedException
+from searx.enginelib import EngineCache
+from searx.network import get
+from searx.exceptions import SearxEngineAPIException, SearxEngineAccessDeniedException
 from searx.result_types import EngineResults
+from searx.utils import eval_xpath, extract_text, gen_useragent
 
 if t.TYPE_CHECKING:
     from searx.extended_types import SXNG_Response
@@ -38,14 +42,45 @@ heexy_categ = "web"
 """Category to search in. Can be either "web" or "image"."""
 
 
-base_url = "https://seapi.heexy.org"
+base_url = "https://heexy.org"
+api_url = "https://seapi.heexy.org"
 safe_search_map = {0: "off", 1: "on", 2: "on"}
 
+CACHE: EngineCache
+"""Cache for storing the ``X-Data-Cacheft`` token (acts like an API key)."""
+
+
+def setup(engine_settings: dict[str, t.Any]) -> bool:
+    global CACHE  # pylint: disable=global-statement
 
-def init(_):
     if heexy_categ not in ("web", "image"):
         raise ValueError("invalid search category: %s" % heexy_categ)
 
+    CACHE = EngineCache(engine_settings["name"])
+    return True
+
+
+def _get_api_token(query: str) -> str:
+    """The API token is independent of the search query. We just need any query
+    to obtain it initially, and don't hardcode it here to decrease chances of
+    getting blocked. The token must be passed as ``X-Data-Cacheft`` header."""
+
+    cached_token: str = CACHE.get("token")
+    if cached_token:
+        return cached_token
+
+    resp = get(f"{base_url}/search?q={query}", headers={"User-Agent": gen_useragent()})
+    if not resp.ok:
+        raise SearxEngineAPIException("failed to obtain request token: invalid response code")
+
+    doc = html.fromstring(resp.text)
+    token = extract_text(eval_xpath(doc, "//html/@data-cacheft"))
+    if not token:
+        raise SearxEngineAPIException("failed to obtain request token: no token found")
+
+    CACHE.set("token", token)
+    return token
+
 
 def request(query: str, params: "OnlineParams") -> None:
     args = {
@@ -56,8 +91,9 @@ def request(query: str, params: "OnlineParams") -> None:
     if params["searxng_locale"] != "all":
         args["lang"] = params["searxng_locale"].split("-")[0]
 
-    params["url"] = f"{base_url}/search/{heexy_categ}?{urlencode(args)}"
-    params["headers"]["Origin"] = base_url
+    params["url"] = f"{api_url}/search/{heexy_categ}?{urlencode(args)}"
+    params["headers"]["X-Data-Cacheft"] = _get_api_token(query)
+    params["headers"]["Origin"] = api_url
 
 
 def response(resp: "SXNG_Response"):