mirror of
https://github.com/searxng/searxng.git
synced 2026-07-17 21:41:24 +00:00
97 lines
2.8 KiB
Python
97 lines
2.8 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
"""SearchZee is a small, indie project, web and news results pulled from
|
|
independent search infrastructure."""
|
|
|
|
import typing as t
|
|
from urllib.parse import urlencode
|
|
|
|
from searx.exceptions import SearxEngineAPIException
|
|
from searx.extended_types import SXNG_Response
|
|
from searx.network import get
|
|
from searx.result_types import EngineResults
|
|
from searx.utils import extr, html_to_text
|
|
from searx.enginelib import EngineCache
|
|
|
|
if t.TYPE_CHECKING:
|
|
from searx.search.processors import OnlineParams
|
|
|
|
about = {
|
|
"website": "https://searchzee.com",
|
|
"official_api_documentation": None,
|
|
"use_official_api": False,
|
|
"require_api_key": False,
|
|
"results": "JSON",
|
|
"description": (
|
|
"SearchZee is a small, indie project, the web and news results"
|
|
" are pulled from an independent search infrastructure."
|
|
),
|
|
}
|
|
categories: list[str] = None # type: ignore[reportAssignmentType]
|
|
|
|
paging = True
|
|
|
|
SearchzeeCategType = t.Literal["web", "news"]
|
|
searchzee_categ: SearchzeeCategType = None # type: ignore[reportAssignmentType]
|
|
|
|
|
|
CACHE: EngineCache
|
|
"""Cache for storing the scraped API Token."""
|
|
|
|
base_url = "https://searchzee.com"
|
|
|
|
# only supports for news
|
|
time_range_map = {"day": "pd", "week": "pw", "month": "pm", "year": "py"}
|
|
|
|
|
|
def setup(engine_settings: dict[str, t.Any]) -> bool:
|
|
if searchzee_categ not in t.get_args(SearchzeeCategType):
|
|
raise ValueError("invalid category: %s" % searchzee_categ)
|
|
|
|
global CACHE # pylint: disable=global-statement
|
|
CACHE = EngineCache(engine_settings["name"]) # type: ignore[reportAny]
|
|
return True
|
|
|
|
|
|
def _obtain_api_token() -> str:
|
|
token: str | None = CACHE.get("token") # type: ignore[reportAny]
|
|
if token:
|
|
return token
|
|
|
|
token_resp = get(
|
|
f"{base_url}/app.js",
|
|
)
|
|
if not token_resp.ok:
|
|
raise SearxEngineAPIException("failed to obtain api key")
|
|
|
|
token = extr(token_resp.text, "const SEARCHZEE_API_TOKEN = \"", "\";")
|
|
CACHE.set("token", token, expire=3600)
|
|
|
|
return token
|
|
|
|
|
|
def request(query: str, params: "OnlineParams"):
|
|
params["headers"]["X-SearchZee-Token"] = _obtain_api_token()
|
|
|
|
args = {"q": query, "type": searchzee_categ, "offset": params["pageno"] - 1}
|
|
if params["time_range"]:
|
|
args["freshness"] = time_range_map[params["time_range"]]
|
|
params["url"] = f"{base_url}/api/search?{urlencode(args)}"
|
|
|
|
|
|
def response(resp: "SXNG_Response") -> EngineResults:
|
|
res = EngineResults()
|
|
|
|
results: list[dict[str, str]] = resp.json()["results"] # type: ignore[reportAny]
|
|
|
|
for result in results:
|
|
res.add(
|
|
res.types.MainResult(
|
|
url=result["url"],
|
|
title=html_to_text(result["title"]),
|
|
content=html_to_text(result["summary"]),
|
|
thumbnail=result.get("thumbnail") or "",
|
|
)
|
|
)
|
|
|
|
return res
|