From 4d9c982cdadedf9b582e32f9d30044e49b6aa703 Mon Sep 17 00:00:00 2001 From: Bnyro Date: Thu, 10 Sep 2026 12:32:22 +0200 Subject: [PATCH] [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. --- searx/engines/brave.py | 18 ++++---- searx/engines/braveapi.py | 41 +++--------------- searx/engines/tusksearch.py | 84 ++++++++++--------------------------- 3 files changed, 36 insertions(+), 107 deletions(-) diff --git a/searx/engines/brave.py b/searx/engines/brave.py index 98d34ce0b..a8f2b3087 100644 --- a/searx/engines/brave.py +++ b/searx/engines/brave.py @@ -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", {}) diff --git a/searx/engines/braveapi.py b/searx/engines/braveapi.py index 83ea1800c..2053a6c64 100644 --- a/searx/engines/braveapi.py +++ b/searx/engines/braveapi.py @@ -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 diff --git a/searx/engines/tusksearch.py b/searx/engines/tusksearch.py index dca81849e..962e3cb5c 100644 --- a/searx/engines/tusksearch.py +++ b/searx/engines/tusksearch.py @@ -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