mirror of
https://github.com/searxng/searxng.git
synced 2026-09-23 06:36:12 +00:00
[fix] yandex: image result xpath + modernization (#6728)
This commit is contained in:
@@ -1,21 +1,27 @@
|
|||||||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||||
"""Yandex (Web, images)"""
|
"""Yandex (Web, images)"""
|
||||||
|
|
||||||
from json import loads
|
import typing as t
|
||||||
|
from json import JSONDecodeError, loads
|
||||||
from urllib.parse import urlencode
|
from urllib.parse import urlencode
|
||||||
from html import unescape
|
|
||||||
from lxml import html
|
from lxml import html
|
||||||
from searx.exceptions import SearxEngineCaptchaException
|
from searx.exceptions import SearxEngineCaptchaException, SearxEngineResponseException
|
||||||
from searx.utils import humanize_bytes, eval_xpath, eval_xpath_list, extract_text, extr
|
from searx.result_types import EngineResults
|
||||||
|
from searx.utils import humanize_bytes, eval_xpath, eval_xpath_list, extract_text, html_to_text
|
||||||
|
|
||||||
|
if t.TYPE_CHECKING:
|
||||||
|
from searx import logger # logger is injected by searx.engines.set_loggers()
|
||||||
|
from searx.extended_types import SXNG_Response
|
||||||
|
from searx.search.processors import OnlineParams
|
||||||
|
|
||||||
# Engine metadata
|
# Engine metadata
|
||||||
about = {
|
about = {
|
||||||
"website": 'https://yandex.com/',
|
"website": "https://yandex.com/",
|
||||||
"wikidata_id": 'Q5281',
|
"wikidata_id": "Q5281",
|
||||||
"official_api_documentation": "?",
|
"official_api_documentation": "?",
|
||||||
"use_official_api": False,
|
"use_official_api": False,
|
||||||
"require_api_key": False,
|
"require_api_key": False,
|
||||||
"results": 'HTML',
|
"results": "HTML",
|
||||||
}
|
}
|
||||||
|
|
||||||
# Engine configuration
|
# Engine configuration
|
||||||
@@ -25,8 +31,8 @@ enable_http3 = True
|
|||||||
search_type = ""
|
search_type = ""
|
||||||
|
|
||||||
# Search URL
|
# Search URL
|
||||||
base_url_web = 'https://yandex.com/search/site/'
|
base_url_web = "https://yandex.com/search/site/"
|
||||||
base_url_images = 'https://yandex.com/images/search'
|
base_url_images = "https://yandex.com/images/search"
|
||||||
|
|
||||||
# Supported languages
|
# Supported languages
|
||||||
yandex_supported_langs = [
|
yandex_supported_langs = [
|
||||||
@@ -48,12 +54,12 @@ title_xpath = './/h3[@class="b-serp-item__title"]/a[@class="b-serp-item__title-l
|
|||||||
content_xpath = './/div[@class="b-serp-item__content"]//div[@class="b-serp-item__text"]'
|
content_xpath = './/div[@class="b-serp-item__content"]//div[@class="b-serp-item__text"]'
|
||||||
|
|
||||||
|
|
||||||
def catch_bad_response(resp):
|
def catch_bad_response(resp: "SXNG_Response") -> None:
|
||||||
if resp.headers.get('x-yandex-captcha') == 'captcha':
|
if resp.headers.get("x-yandex-captcha") == "captcha":
|
||||||
raise SearxEngineCaptchaException()
|
raise SearxEngineCaptchaException()
|
||||||
|
|
||||||
|
|
||||||
def request(query, params):
|
def request(query: str, params: "OnlineParams") -> None:
|
||||||
query_params_web = {
|
query_params_web = {
|
||||||
"tmpl_version": "releases",
|
"tmpl_version": "releases",
|
||||||
"text": query,
|
"text": query,
|
||||||
@@ -62,7 +68,7 @@ def request(query, params):
|
|||||||
"searchid": "3131712",
|
"searchid": "3131712",
|
||||||
}
|
}
|
||||||
|
|
||||||
lang = params["language"].split("-")[0]
|
lang = params["language"].split("-")[0] # type: ignore
|
||||||
if lang in yandex_supported_langs:
|
if lang in yandex_supported_langs:
|
||||||
query_params_web["lang"] = lang
|
query_params_web["lang"] = lang
|
||||||
|
|
||||||
@@ -71,82 +77,70 @@ def request(query, params):
|
|||||||
"uinfo": "sw-1920-sh-1080-ww-1125-wh-999",
|
"uinfo": "sw-1920-sh-1080-ww-1125-wh-999",
|
||||||
}
|
}
|
||||||
|
|
||||||
if params['pageno'] > 1:
|
if params["pageno"] > 1:
|
||||||
query_params_web.update({"p": params["pageno"] - 1})
|
query_params_web["p"] = params["pageno"] - 1 # type: ignore
|
||||||
query_params_images.update({"p": params["pageno"] - 1})
|
query_params_images["p"] = params["pageno"] - 1 # type: ignore
|
||||||
|
|
||||||
params["cookies"] = {'cookie': "yp=1716337604.sp.family%3A0#1685406411.szm.1:1920x1080:1920x999"}
|
params["cookies"] = {"cookie": "yp=1716337604.sp.family%3A0#1685406411.szm.1:1920x1080:1920x999"}
|
||||||
|
|
||||||
if search_type == 'web':
|
if search_type == "web":
|
||||||
params['url'] = f"{base_url_web}?{urlencode(query_params_web)}"
|
params["url"] = f"{base_url_web}?{urlencode(query_params_web)}"
|
||||||
elif search_type == 'images':
|
elif search_type == "images":
|
||||||
params['url'] = f"{base_url_images}?{urlencode(query_params_images)}"
|
params["url"] = f"{base_url_images}?{urlencode(query_params_images)}"
|
||||||
|
|
||||||
return params
|
|
||||||
|
|
||||||
|
|
||||||
def response(resp):
|
def _parse_json_results(dom: html.HtmlElement) -> dict[str, t.Any]:
|
||||||
if search_type == 'web':
|
# attempt to parse using xpath - finding element with "data-state" attribute
|
||||||
catch_bad_response(resp)
|
data_elements = dom.xpath("//*[@data-state]/@data-state")
|
||||||
|
for json_data in data_elements:
|
||||||
|
try:
|
||||||
|
json_resp = loads(json_data)
|
||||||
|
if json_resp.get("location") == "/images/search/":
|
||||||
|
return json_resp
|
||||||
|
except JSONDecodeError:
|
||||||
|
logger.debug("failed parsing data-state json")
|
||||||
|
|
||||||
dom = html.fromstring(resp.text)
|
raise SearxEngineResponseException("failed to parse JSON data")
|
||||||
|
|
||||||
results = []
|
|
||||||
|
|
||||||
for result in eval_xpath_list(dom, results_xpath):
|
def response(resp: "SXNG_Response") -> EngineResults:
|
||||||
results.append(
|
catch_bad_response(resp)
|
||||||
{
|
results = EngineResults()
|
||||||
'url': extract_text(eval_xpath(result, url_xpath)),
|
dom = html.fromstring(resp.text)
|
||||||
'title': extract_text(eval_xpath(result, title_xpath)),
|
|
||||||
'content': extract_text(eval_xpath(result, content_xpath)),
|
|
||||||
}
|
|
||||||
)
|
|
||||||
|
|
||||||
return results
|
match search_type:
|
||||||
|
case "web":
|
||||||
|
for result in eval_xpath_list(dom, results_xpath):
|
||||||
|
url = extract_text(eval_xpath(result, url_xpath))
|
||||||
|
title = extract_text(eval_xpath(result, title_xpath))
|
||||||
|
content = extract_text(eval_xpath(result, content_xpath))
|
||||||
|
results.add(results.types.MainResult(url=url, title=str(title), content=str(content)))
|
||||||
|
case "images":
|
||||||
|
json_resp = _parse_json_results(dom)
|
||||||
|
|
||||||
if search_type == 'images':
|
# build results from loaded json values
|
||||||
catch_bad_response(resp)
|
for item_data in json_resp["initialState"]["serpList"]["items"]["entities"].values():
|
||||||
|
viewerData: dict[str, t.Any] = item_data["viewerData"]
|
||||||
|
snippet: dict[str, str] = viewerData.get("snippet", {})
|
||||||
|
|
||||||
html_data = html.fromstring(resp.text)
|
# return the image with largest dimensions
|
||||||
html_sample = unescape(html.tostring(html_data, encoding='unicode'))
|
image_sources = viewerData.get("dups", []) + viewerData.get("preview", [])
|
||||||
|
image_source = max(image_sources, key=lambda x: x["h"] * x["w"])
|
||||||
|
|
||||||
content_between_tags = extr(
|
humanized_filesize = None
|
||||||
html_sample, '{"location":"/images/search/', 'advRsyaSearchColumn":null}}', default="fail"
|
if image_source.get("fileSizeInBytes"):
|
||||||
)
|
humanized_filesize = humanize_bytes(image_source["fileSizeInBytes"])
|
||||||
json_data = '{"location":"/images/search/' + content_between_tags + 'advRsyaSearchColumn":null}}'
|
|
||||||
|
|
||||||
if content_between_tags == "fail":
|
results.add(
|
||||||
content_between_tags = extr(html_sample, '{"location":"/images/search/', 'false}}}')
|
results.types.Image(
|
||||||
json_data = '{"location":"/images/search/' + content_between_tags + 'false}}}'
|
title=snippet.get("title"), # type: ignore
|
||||||
|
content=html_to_text(snippet.get("text")), # type: ignore
|
||||||
|
url=snippet.get("url"),
|
||||||
|
img_src=image_source["url"],
|
||||||
|
filesize=humanized_filesize, # type: ignore
|
||||||
|
thumbnail_src=item_data["image"],
|
||||||
|
resolution=f"{image_source['w']} x {image_source['h']}",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
json_resp = loads(json_data)
|
return results
|
||||||
|
|
||||||
results = []
|
|
||||||
for _, item_data in json_resp['initialState']['serpList']['items']['entities'].items():
|
|
||||||
title = item_data['snippet']['title']
|
|
||||||
source = item_data['snippet']['url']
|
|
||||||
|
|
||||||
image_source = item_data["viewerData"]["thumb"]
|
|
||||||
for i in item_data['viewerData']['dups'] + item_data['viewerData']['preview']:
|
|
||||||
if i["h"] > image_source["h"]:
|
|
||||||
image_source = i
|
|
||||||
|
|
||||||
humanized_filesize = None
|
|
||||||
if image_source.get("fileSizeInBytes"):
|
|
||||||
humanized_filesize = humanize_bytes(image_source["fileSizeInBytes"])
|
|
||||||
|
|
||||||
results.append(
|
|
||||||
{
|
|
||||||
'title': title,
|
|
||||||
'url': source,
|
|
||||||
'img_src': image_source["url"],
|
|
||||||
'filesize': humanized_filesize,
|
|
||||||
'thumbnail_src': item_data["image"],
|
|
||||||
'template': 'images.html',
|
|
||||||
'resolution': f'{image_source["w"]} x {image_source["h"]}',
|
|
||||||
}
|
|
||||||
)
|
|
||||||
|
|
||||||
return results
|
|
||||||
|
|
||||||
return []
|
|
||||||
|
|||||||
Reference in New Issue
Block a user