mirror of
https://github.com/searxng/searxng.git
synced 2026-07-17 21:41:24 +00:00
9e25585aec
Refactor engines that parse ISO 8601 dates with strptime to use
fromisoformat instead. In most cases this is a direct replacement of
strptime(text, "format") with fromisoformat(text).
For engines where the source has a trailing "Z" that strptime consumed
as a literal (e.g. "%Y-%m-%dT%H:%M:%S.%fZ" in huggingface.py), add
rstrip("Z") to keep the output naive and preserve the existing behavior.
In sogou.py the date is extracted with a regular expression, which can
yield strings like "2026-7-11". strptime accepts this via its format
string, but fromisoformat does not. To preserve the existing behavior
and satisfy the format fromisoformat expects, add zero-padding for the
month and day.
Closes: #6098
---------
Signed-off-by: OneVth <onebrotravel@gmail.com>
221 lines
6.8 KiB
Python
221 lines
6.8 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
"""Baidu_
|
|
|
|
.. _Baidu: https://www.baidu.com
|
|
"""
|
|
|
|
# There exits a https://github.com/ohblue/baidu-serp-api/
|
|
# but we don't use it here (may we can learn from).
|
|
|
|
import typing as t
|
|
|
|
from urllib.parse import urlencode
|
|
from datetime import datetime
|
|
from html import unescape
|
|
import time
|
|
import json
|
|
|
|
from searx.exceptions import SearxEngineAPIException, SearxEngineCaptchaException, SearxEngineAccessDeniedException
|
|
from searx.enginelib import EngineCache
|
|
from searx.network import get as http_get
|
|
from searx.utils import html_to_text
|
|
|
|
about = {
|
|
"website": "https://www.baidu.com",
|
|
"wikidata_id": "Q14772",
|
|
"official_api_documentation": None,
|
|
"use_official_api": False,
|
|
"require_api_key": False,
|
|
"results": "JSON",
|
|
}
|
|
language = "zh"
|
|
|
|
paging = True
|
|
categories = []
|
|
results_per_page = 10
|
|
|
|
baidu_category = 'general'
|
|
|
|
time_range_support = True
|
|
time_range_dict = {"day": 86400, "week": 604800, "month": 2592000, "year": 31536000}
|
|
|
|
image_base_url = "https://image.baidu.com/"
|
|
|
|
COOKIE_CACHE_KEY = "cookie"
|
|
COOKIE_CACHE_EXPIRATION_SECONDS = 3600
|
|
|
|
CACHE: EngineCache
|
|
"""Stores cookies from Baidu image search warmup."""
|
|
|
|
|
|
def setup(engine_settings: dict[str, t.Any]) -> bool:
|
|
global CACHE # pylint: disable=global-statement
|
|
CACHE = EngineCache(engine_settings["name"])
|
|
return True
|
|
|
|
|
|
def get_image_cookies(headers: dict[str, str]) -> dict[str, str]:
|
|
cookies: dict[str, str] | None = CACHE.get(COOKIE_CACHE_KEY)
|
|
if cookies:
|
|
return cookies
|
|
|
|
warmup = http_get(image_base_url, headers=headers, timeout=10)
|
|
cookies = dict(warmup.cookies.items())
|
|
CACHE.set(key=COOKIE_CACHE_KEY, value=cookies, expire=COOKIE_CACHE_EXPIRATION_SECONDS)
|
|
return cookies
|
|
|
|
|
|
def init(_):
|
|
if baidu_category not in ('general', 'images', 'it'):
|
|
raise SearxEngineAPIException(f"Unsupported category: {baidu_category}")
|
|
|
|
|
|
def request(query, params):
|
|
page_num = params["pageno"]
|
|
|
|
category_config = {
|
|
'general': {
|
|
'endpoint': 'https://www.baidu.com/s',
|
|
'params': {
|
|
"wd": query,
|
|
"rn": results_per_page,
|
|
"pn": (page_num - 1) * results_per_page,
|
|
"tn": "json",
|
|
},
|
|
},
|
|
'images': {
|
|
'endpoint': 'https://image.baidu.com/search/acjson',
|
|
'params': {
|
|
"word": query,
|
|
"rn": results_per_page,
|
|
"pn": (page_num - 1) * results_per_page,
|
|
"tn": "resultjson_com",
|
|
},
|
|
},
|
|
'it': {
|
|
'endpoint': 'https://kaifa.baidu.com/rest/v1/search',
|
|
'params': {
|
|
"wd": query,
|
|
"pageSize": results_per_page,
|
|
"pageNum": page_num,
|
|
"paramList": f"page_num={page_num},page_size={results_per_page}",
|
|
"position": 0,
|
|
},
|
|
},
|
|
}
|
|
|
|
query_params = category_config[baidu_category]['params']
|
|
query_url = category_config[baidu_category]['endpoint']
|
|
|
|
if params.get("time_range") in time_range_dict:
|
|
now = int(time.time())
|
|
past = now - time_range_dict[params["time_range"]]
|
|
|
|
if baidu_category == 'general':
|
|
query_params["gpc"] = f"stf={past},{now}|stftype=1"
|
|
|
|
if baidu_category == 'it':
|
|
query_params["paramList"] += f",timestamp_range={past}-{now}"
|
|
|
|
if baidu_category == 'images':
|
|
params["cookies"] = get_image_cookies(params["headers"])
|
|
|
|
params["url"] = f"{query_url}?{urlencode(query_params)}"
|
|
params["allow_redirects"] = False
|
|
return params
|
|
|
|
|
|
def response(resp):
|
|
# Detect Baidu Captcha, it will redirect to wappass.baidu.com
|
|
if 'wappass.baidu.com/static/captcha' in resp.headers.get('Location', ''):
|
|
raise SearxEngineCaptchaException()
|
|
|
|
text = resp.text
|
|
if baidu_category == 'images':
|
|
# baidu's JSON encoder wrongly quotes / and ' characters by \\ and \'
|
|
text = text.replace(r"\/", "/").replace(r"\'", "'")
|
|
data = json.loads(text, strict=False)
|
|
if data.get("antiFlag") == 1:
|
|
raise SearxEngineAccessDeniedException(data.get("message", "Forbid spider access"))
|
|
parsers = {'general': parse_general, 'images': parse_images, 'it': parse_it}
|
|
|
|
return parsers[baidu_category](data)
|
|
|
|
|
|
def parse_general(data):
|
|
results = []
|
|
if not data.get("feed", {}).get("entry"):
|
|
raise SearxEngineAPIException("Invalid response")
|
|
|
|
for entry in data["feed"]["entry"]:
|
|
if not entry.get("title") or not entry.get("url"):
|
|
continue
|
|
|
|
published_date = None
|
|
if entry.get("time"):
|
|
try:
|
|
published_date = datetime.fromtimestamp(entry["time"])
|
|
except (ValueError, TypeError):
|
|
published_date = None
|
|
|
|
# title and content sometimes containing characters such as & ' " etc...
|
|
title = unescape(entry["title"])
|
|
content = unescape(entry.get("abs", ""))
|
|
|
|
results.append(
|
|
{
|
|
"title": title,
|
|
"url": entry["url"],
|
|
"content": content,
|
|
"publishedDate": published_date,
|
|
}
|
|
)
|
|
return results
|
|
|
|
|
|
def parse_images(data):
|
|
results = []
|
|
if "data" in data:
|
|
for item in data["data"]:
|
|
if not item:
|
|
# the last item in the JSON list is empty, the JSON string ends with "}, {}]"
|
|
continue
|
|
replace_url = item.get("replaceUrl", [{}])[0]
|
|
width = item.get("width")
|
|
height = item.get("height")
|
|
img_date = item.get("bdImgnewsDate")
|
|
publishedDate = None
|
|
if img_date:
|
|
publishedDate = datetime.fromisoformat(img_date)
|
|
results.append(
|
|
{
|
|
"template": "images.html",
|
|
"url": replace_url.get("FromURL"),
|
|
"thumbnail_src": item.get("thumbURL"),
|
|
"img_src": replace_url.get("ObjURL"),
|
|
"title": html_to_text(item.get("fromPageTitle")),
|
|
"source": item.get("fromURLHost"),
|
|
"resolution": f"{width} x {height}",
|
|
"img_format": item.get("type"),
|
|
"filesize": item.get("filesize"),
|
|
"publishedDate": publishedDate,
|
|
}
|
|
)
|
|
return results
|
|
|
|
|
|
def parse_it(data):
|
|
results = []
|
|
if not data.get("data", {}).get("documents", {}).get("data"):
|
|
raise SearxEngineAPIException("Invalid response")
|
|
|
|
for entry in data["data"]["documents"]["data"]:
|
|
results.append(
|
|
{
|
|
'title': entry["techDocDigest"]["title"],
|
|
'url': entry["techDocDigest"]["url"],
|
|
'content': entry["techDocDigest"]["summary"],
|
|
}
|
|
)
|
|
return results
|