mirror of
https://github.com/searxng/searxng.git
synced 2026-09-23 14:46:04 +00:00
[mod] brave-based engines: reuse result parsing methods (#6688)
Many engines use the JSON results from Brave (e.g. Brave, BraveAPI, Tusksearch and more), so the result parsing logic is always the same. This PR deduplicates this code.
This commit is contained in:
@@ -261,18 +261,18 @@ def response(resp: "SXNG_Response") -> EngineResults:
|
||||
|
||||
match brave_category:
|
||||
case "search" | "goggles":
|
||||
return _parse_results(_parse_search_result, resp)
|
||||
return _parse_results(parse_search_result, resp)
|
||||
case "news":
|
||||
return _parse_results(_parse_news_result, resp)
|
||||
return _parse_results(parse_news_result, resp)
|
||||
case "images":
|
||||
return _parse_results(_parse_image_result, resp)
|
||||
return _parse_results(parse_image_result, resp)
|
||||
case "videos":
|
||||
return _parse_results(_parse_video_result, resp)
|
||||
return _parse_results(parse_video_result, resp)
|
||||
case _:
|
||||
raise ValueError(f"Unsupported brave category: {brave_category}") # pyright: ignore[reportUnreachable]
|
||||
|
||||
|
||||
def _parse_search_result(result: dict[str, t.Any]) -> MainResult:
|
||||
def parse_search_result(result: dict[str, t.Any]) -> MainResult:
|
||||
thumbnail: dict[str, t.Any] = result.get("thumbnail", {})
|
||||
return MainResult(
|
||||
template="default.html",
|
||||
@@ -291,7 +291,7 @@ def _parse_secondary_items(json_data: dict[str, t.Any], results: EngineResults):
|
||||
videos_resp: dict[str, t.Any] = body_resp.get("videos", {})
|
||||
if videos_resp and "results" in videos_resp:
|
||||
for result in videos_resp.get("results", []):
|
||||
results.add(_parse_video_result(result))
|
||||
results.add(parse_video_result(result))
|
||||
# related queries -> suggestion
|
||||
query: dict[str, t.Any] = body_resp.get("query", {})
|
||||
if query and "related_queries" in query:
|
||||
@@ -300,7 +300,7 @@ def _parse_secondary_items(json_data: dict[str, t.Any], results: EngineResults):
|
||||
results.add(results.types.LegacyResult(suggestion=suggestion))
|
||||
|
||||
|
||||
def _parse_news_result(result: dict[str, t.Any]) -> MainResult:
|
||||
def parse_news_result(result: dict[str, t.Any]) -> MainResult:
|
||||
thumbnail: dict[str, t.Any] = result.get("thumbnail", {})
|
||||
return MainResult(
|
||||
title=result.get("title", ""),
|
||||
@@ -312,7 +312,7 @@ def _parse_news_result(result: dict[str, t.Any]) -> MainResult:
|
||||
)
|
||||
|
||||
|
||||
def _parse_image_result(result: dict[str, t.Any]) -> Image:
|
||||
def parse_image_result(result: dict[str, t.Any]) -> Image:
|
||||
properties: dict[str, t.Any] = result.get("properties", {})
|
||||
thumbnail: dict[str, t.Any] = result.get("thumbnail", {})
|
||||
width, height = properties.get("width"), properties.get("height")
|
||||
@@ -327,7 +327,7 @@ def _parse_image_result(result: dict[str, t.Any]) -> Image:
|
||||
)
|
||||
|
||||
|
||||
def _parse_video_result(result: dict[str, t.Any]) -> MainResult:
|
||||
def parse_video_result(result: dict[str, t.Any]) -> Video:
|
||||
video: dict[str, t.Any] = result.get("video", {})
|
||||
thumbnail: dict[str, t.Any] = result.get("thumbnail", {})
|
||||
|
||||
|
||||
@@ -27,11 +27,10 @@ The API supports paging and time filters.
|
||||
import typing as t
|
||||
|
||||
from urllib.parse import urlencode
|
||||
from dateutil import parser
|
||||
|
||||
from searx.engines.brave import parse_video_result
|
||||
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
|
||||
@@ -94,43 +93,13 @@ def request(query: str, params: "OnlineParams") -> None:
|
||||
params["headers"]["Accept"] = "application/json"
|
||||
|
||||
|
||||
def _extract_published_date(published_date_raw: str):
|
||||
"""Extract and parse the published date from the API response.
|
||||
|
||||
Args:
|
||||
published_date_raw: Raw date string from the API
|
||||
|
||||
Returns:
|
||||
Parsed datetime object or None if parsing fails
|
||||
"""
|
||||
if not published_date_raw:
|
||||
return None
|
||||
|
||||
try:
|
||||
return parser.parse(published_date_raw)
|
||||
except parser.ParserError:
|
||||
return None
|
||||
|
||||
|
||||
def response(resp: "SXNG_Response") -> EngineResults:
|
||||
"""Process the API response and return results."""
|
||||
res = EngineResults()
|
||||
data = resp.json()
|
||||
|
||||
for result in (data.get("web") or {}).get("results", []):
|
||||
thumbnail_obj = result.get("thumbnail")
|
||||
thumbnail = ""
|
||||
if thumbnail_obj and not thumbnail_obj.get("logo", False):
|
||||
thumbnail = thumbnail_obj.get("src") or ""
|
||||
|
||||
res.add(
|
||||
res.types.MainResult(
|
||||
url=result["url"],
|
||||
title=html_to_text(result["title"]),
|
||||
content=html_to_text(result.get("description", "")),
|
||||
publishedDate=_extract_published_date(result.get("age")),
|
||||
thumbnail=thumbnail,
|
||||
),
|
||||
)
|
||||
res = EngineResults()
|
||||
results_json = (data.get("web") or {}).get("results", [])
|
||||
for result in results_json:
|
||||
res.add(parse_video_result(result))
|
||||
|
||||
return res
|
||||
|
||||
@@ -9,12 +9,11 @@ from json import loads
|
||||
import random
|
||||
import typing as t
|
||||
from urllib.parse import urlencode
|
||||
from dateutil import parser
|
||||
|
||||
from searx.result_types import EngineResults
|
||||
from searx.engines.brave import parse_image_result, parse_news_result, parse_search_result, parse_video_result
|
||||
from searx.exceptions import SearxEngineAPIException
|
||||
from searx.network import get
|
||||
from searx.utils import html_to_text
|
||||
from searx.result_types import EngineResults
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
from searx.extended_types import SXNG_Response
|
||||
@@ -32,7 +31,8 @@ about = {
|
||||
paging = True
|
||||
|
||||
categories = ["general"]
|
||||
tusk_categ = "web"
|
||||
TuskCategType = t.Literal["web", "images", "videos", "news"]
|
||||
tusk_categ: TuskCategType = "web"
|
||||
"""Category to search in. Can be either "web", "images", "videos" or "news"."""
|
||||
|
||||
|
||||
@@ -40,7 +40,7 @@ api_url = "https://api.tusksearch.com"
|
||||
|
||||
|
||||
def setup(_: dict[str, t.Any]) -> bool | None:
|
||||
if tusk_categ not in ("web", "images", "videos", "news"):
|
||||
if tusk_categ not in t.get_args(TuskCategType):
|
||||
raise ValueError("invalid search type: %s" % tusk_categ)
|
||||
|
||||
|
||||
@@ -103,64 +103,24 @@ def request(query: str, params: "OnlineParams") -> None:
|
||||
|
||||
|
||||
def response(resp: "SXNG_Response"):
|
||||
res = EngineResults()
|
||||
|
||||
json_resp = resp.json()["results"]
|
||||
|
||||
if tusk_categ == "web":
|
||||
for result in (json_resp.get("web") or {}).get("results", []):
|
||||
res.add(
|
||||
res.types.MainResult(
|
||||
url=result["url"],
|
||||
title=html_to_text(result["title"]),
|
||||
content=html_to_text(result["description"]),
|
||||
thumbnail=(result["thumbnail"] or {}).get("src") or "",
|
||||
)
|
||||
)
|
||||
elif tusk_categ == "news":
|
||||
for result in (json_resp.get("news") or {}).get("results", []):
|
||||
publishedDate = None
|
||||
try:
|
||||
publishedDate = parser.parse(result["age"])
|
||||
except parser.ParserError:
|
||||
pass
|
||||
|
||||
res.add(
|
||||
res.types.MainResult(
|
||||
url=result["url"],
|
||||
title=html_to_text(result["title"]),
|
||||
content=html_to_text(result["description"]),
|
||||
thumbnail=result["thumbnail"]["src"],
|
||||
publishedDate=publishedDate,
|
||||
)
|
||||
)
|
||||
elif tusk_categ == "videos":
|
||||
for result in (json_resp.get("videos") or {}).get("results", []):
|
||||
publishedDate = None
|
||||
try:
|
||||
publishedDate = parser.parse(result["age"])
|
||||
except parser.ParserError:
|
||||
pass
|
||||
|
||||
res.add(
|
||||
res.types.Video(
|
||||
url=result["url"],
|
||||
title=html_to_text(result["title"]),
|
||||
content=html_to_text(result["description"]),
|
||||
thumbnail=result["thumbnail"]["src"],
|
||||
publishedDate=publishedDate,
|
||||
length=result["video"].get("duration"),
|
||||
)
|
||||
)
|
||||
elif tusk_categ == "images":
|
||||
for result in json_resp:
|
||||
res.add(
|
||||
res.types.Image(
|
||||
url=result["url"],
|
||||
title=html_to_text(result["title"]),
|
||||
img_src=result["properties"]["url"],
|
||||
thumbnail_src=result["thumbnail"]["src"],
|
||||
)
|
||||
)
|
||||
res = EngineResults()
|
||||
match tusk_categ:
|
||||
case "web":
|
||||
results = (json_resp.get("web") or {}).get("results", [])
|
||||
for result in results:
|
||||
res.add(parse_search_result(result))
|
||||
case "news":
|
||||
results = (json_resp.get("news") or {}).get("results", [])
|
||||
for result in results:
|
||||
res.add(parse_news_result(result))
|
||||
case "videos":
|
||||
results = (json_resp.get("videos") or {}).get("results", [])
|
||||
for result in results:
|
||||
res.add(parse_video_result(result))
|
||||
case "images":
|
||||
for result in json_resp:
|
||||
res.add(parse_image_result(result))
|
||||
|
||||
return res
|
||||
|
||||
Reference in New Issue
Block a user