[feat] engines: add www.acfun.cn

Co-authored-by: Bnyro <bnyro@tutanota.com>

searx/engines/acfun.py

@@ -0,0 +1,108 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""Acfun search engine for searxng"""
+
+from urllib.parse import urlencode
+import re
+import json
+from datetime import datetime, timedelta
+from lxml import html
+
+from searx.utils import extract_text
+
+# Metadata
+about = {
+    "website": "https://www.acfun.cn/",
+    "wikidata_id": "Q3077675",
+    "use_official_api": False,
+    "require_api_key": False,
+    "results": "HTML",
+}
+
+# Engine Configuration
+categories = ["videos"]
+paging = True
+
+# Base URL
+base_url = "https://www.acfun.cn"
+
+
+def request(query, params):
+    query_params = {"keyword": query, "pCursor": params["pageno"]}
+    params["url"] = f"{base_url}/search?{urlencode(query_params)}"
+    return params
+
+
+def response(resp):
+    results = []
+
+    matches = re.findall(r'bigPipe\.onPageletArrive\((\{.*?\})\);', resp.text, re.DOTALL)
+    if not matches:
+        return results
+
+    for match in matches:
+        try:
+            json_data = json.loads(match)
+            raw_html = json_data.get("html", "")
+            if not raw_html:
+                continue
+
+            tree = html.fromstring(raw_html)
+
+            video_blocks = tree.xpath('//div[contains(@class, "search-video")]')
+            if not video_blocks:
+                continue
+
+            for video_block in video_blocks:
+                video_info = extract_video_data(video_block)
+                if video_info and video_info["title"] and video_info["url"]:
+                    results.append(video_info)
+
+        except json.JSONDecodeError:
+            continue
+
+    return results
+
+
+def extract_video_data(video_block):
+    try:
+        data_exposure_log = video_block.get('data-exposure-log')
+        video_data = json.loads(data_exposure_log)
+
+        content_id = video_data.get("content_id", "")
+        title = video_data.get("title", "")
+
+        url = f"{base_url}/v/ac{content_id}"
+        iframe_src = f"{base_url}/player/ac{content_id}"
+
+        create_time = extract_text(video_block.xpath('.//span[contains(@class, "info__create-time")]'))
+        video_cover = extract_text(video_block.xpath('.//div[contains(@class, "video__cover")]/a/img/@src')[0])
+        video_duration = extract_text(video_block.xpath('.//span[contains(@class, "video__duration")]'))
+        video_intro = extract_text(video_block.xpath('.//div[contains(@class, "video__main__intro")]'))
+
+        published_date = None
+        if create_time:
+            try:
+                published_date = datetime.strptime(create_time.strip(), "%Y-%m-%d")
+            except (ValueError, TypeError):
+                pass
+
+        length = None
+        if video_duration:
+            try:
+                timediff = datetime.strptime(video_duration.strip(), "%M:%S")
+                length = timedelta(minutes=timediff.minute, seconds=timediff.second)
+            except (ValueError, TypeError):
+                pass
+
+        return {
+            "title": title,
+            "url": url,
+            "content": video_intro,
+            "thumbnail": video_cover,
+            "length": length,
+            "publishedDate": published_date,
+            "iframe_src": iframe_src,
+        }
+
+    except (json.JSONDecodeError, AttributeError, TypeError, ValueError):
+        return None

searx/engines/chinaso.py

@@ -0,0 +1,142 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""ChinaSo: A search engine from ChinaSo."""
+
+from urllib.parse import urlencode
+from datetime import datetime
+
+from searx.exceptions import SearxEngineAPIException
+from searx.utils import html_to_text
+
+about = {
+    "website": "https://www.chinaso.com/",
+    "wikidata_id": "Q10846064",
+    "use_official_api": False,
+    "require_api_key": False,
+    "results": "JSON",
+}
+
+paging = True
+time_range_support = True
+results_per_page = 10
+categories = []
+chinaso_category = 'news'
+"""ChinaSo supports news, videos, images search.
+
+- ``news``: search for news
+- ``videos``: search for videos
+- ``images``: search for images
+"""
+
+time_range_dict = {'day': '24h', 'week': '1w', 'month': '1m', 'year': '1y'}
+
+base_url = "https://www.chinaso.com"
+
+
+def init(_):
+    if chinaso_category not in ('news', 'videos', 'images'):
+        raise SearxEngineAPIException(f"Unsupported category: {chinaso_category}")
+
+
+def request(query, params):
+    query_params = {"q": query}
+
+    if time_range_dict.get(params['time_range']):
+        query_params["stime"] = time_range_dict[params['time_range']]
+        query_params["etime"] = 'now'
+
+    category_config = {
+        'news': {'endpoint': '/v5/general/v1/web/search', 'params': {'pn': params["pageno"], 'ps': results_per_page}},
+        'images': {
+            'endpoint': '/v5/general/v1/search/image',
+            'params': {'start_index': (params["pageno"] - 1) * results_per_page, 'rn': results_per_page},
+        },
+        'videos': {
+            'endpoint': '/v5/general/v1/search/video',
+            'params': {'start_index': (params["pageno"] - 1) * results_per_page, 'rn': results_per_page},
+        },
+    }
+
+    query_params.update(category_config[chinaso_category]['params'])
+
+    params["url"] = f"{base_url}{category_config[chinaso_category]['endpoint']}?{urlencode(query_params)}"
+
+    return params
+
+
+def response(resp):
+    try:
+        data = resp.json()
+    except Exception as e:
+        raise SearxEngineAPIException(f"Invalid response: {e}") from e
+
+    parsers = {'news': parse_news, 'images': parse_images, 'videos': parse_videos}
+
+    return parsers[chinaso_category](data)
+
+
+def parse_news(data):
+    results = []
+    if not data.get("data", {}).get("data"):
+        raise SearxEngineAPIException("Invalid response")
+
+    for entry in data["data"]["data"]:
+        published_date = None
+        if entry.get("timestamp"):
+            try:
+                published_date = datetime.fromtimestamp(int(entry["timestamp"]))
+            except (ValueError, TypeError):
+                pass
+
+        results.append(
+            {
+                'title': html_to_text(entry["title"]),
+                'url': entry["url"],
+                'content': html_to_text(entry["snippet"]),
+                'publishedDate': published_date,
+            }
+        )
+    return results
+
+
+def parse_images(data):
+    results = []
+    if not data.get("data", {}).get("arrRes"):
+        raise SearxEngineAPIException("Invalid response")
+
+    for entry in data["data"]["arrRes"]:
+        results.append(
+            {
+                'url': entry["web_url"],
+                'title': html_to_text(entry["title"]),
+                'content': html_to_text(entry["ImageInfo"]),
+                'template': 'images.html',
+                'img_src': entry["url"].replace("http://", "https://"),
+                'thumbnail_src': entry["largeimage"].replace("http://", "https://"),
+            }
+        )
+    return results
+
+
+def parse_videos(data):
+    results = []
+    if not data.get("data", {}).get("arrRes"):
+        raise SearxEngineAPIException("Invalid response")
+
+    for entry in data["data"]["arrRes"]:
+        published_date = None
+        if entry.get("VideoPubDate"):
+            try:
+                published_date = datetime.fromtimestamp(int(entry["VideoPubDate"]))
+            except (ValueError, TypeError):
+                pass
+
+        results.append(
+            {
+                'url': entry["url"],
+                'title': html_to_text(entry["raw_title"]),
+                'template': 'videos.html',
+                'publishedDate': published_date,
+                'thumbnail': entry["image_src"].replace("http://", "https://"),
+            }
+        )
+    return results

searx/engines/google.py

@@ -10,10 +10,14 @@ engines:
 - :ref:`google autocomplete`
 
 """
+from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
 import re
+import random
+import string
+import time
 from urllib.parse import urlencode
 from lxml import html
 import babel
@@ -64,11 +68,31 @@ filter_mapping = {0: 'off', 1: 'medium', 2: 'high'}
 # from the links not the links itself.
 suggestion_xpath = '//div[contains(@class, "EIaa9b")]//a'
 
-# UI_ASYNC = 'use_ac:true,_fmt:html' # returns a HTTP 500 when user search for
-#                                    # celebrities like '!google natasha allegri'
-#                                    # or '!google chris evans'
-UI_ASYNC = 'use_ac:true,_fmt:prog'
-"""Format of the response from UI's async request."""
+
+_arcid_range = string.ascii_letters + string.digits + "_-"
+_arcid_random: tuple[str, int] | None = None
+
+
+def ui_async(start: int) -> str:
+    """Format of the response from UI's async request.
+
+    - ``arc_id:<...>,use_ac:true,_fmt:prog``
+
+    The arc_id is random generated every hour.
+    """
+    global _arcid_random  # pylint: disable=global-statement
+
+    use_ac = "use_ac:true"
+    # _fmt:html returns a HTTP 500 when user search for celebrities like
+    # '!google natasha allegri' or '!google chris evans'
+    _fmt = "_fmt:prog"
+
+    # create a new random arc_id every hour
+    if not _arcid_random or (int(time.time()) - _arcid_random[1]) > 3600:
+        _arcid_random = (''.join(random.choices(_arcid_range, k=23)), int(time.time()))
+    arc_id = f"arc_id:srp_{_arcid_random[0]}_1{start:02}"
+
+    return ",".join([arc_id, use_ac, _fmt])
 
 
 def get_google_info(params, eng_traits):
@@ -258,8 +282,10 @@ def detect_google_sorry(resp):
 def request(query, params):
     """Google search request"""
     # pylint: disable=line-too-long
-    offset = (params['pageno'] - 1) * 10
+    start = (params['pageno'] - 1) * 10
+    str_async = ui_async(start)
     google_info = get_google_info(params, traits)
+    logger.debug("ARC_ID: %s", str_async)
 
     # https://www.google.de/search?q=corona&hl=de&lr=lang_de&start=0&tbs=qdr%3Ad&safe=medium
     query_url = (
@@ -272,7 +298,7 @@ def request(query, params):
                 'q': query,
                 **google_info['params'],
                 'filter': '0',
-                'start': offset,
+                'start': start,
                 # 'vet': '12ahUKEwik3ZbIzfn7AhXMX_EDHbUDBh0QxK8CegQIARAC..i',
                 # 'ved': '2ahUKEwik3ZbIzfn7AhXMX_EDHbUDBh0Q_skCegQIARAG',
                 # 'cs' : 1,
@@ -284,7 +310,7 @@ def request(query, params):
                 # 'sstk': 'AcOHfVkD7sWCSAheZi-0tx_09XDO55gTWY0JNq3_V26cNN-c8lfD45aZYPI8s_Bqp8s57AHz5pxchDtAGCA_cikAWSjy9kw3kgg'
                 # formally known as use_mobile_ui
                 'asearch': 'arc',
-                'async': UI_ASYNC,
+                'async': str_async,
             }
         )
     )
@@ -303,15 +329,20 @@ def request(query, params):
 # =26;[3,"dimg_ZNMiZPCqE4apxc8P3a2tuAQ_137"]a87;data:image/jpeg;base64,/9j/4AAQSkZJRgABA
 # ...6T+9Nl4cnD+gr9OK8I56/tX3l86nWYw//2Q==26;
 RE_DATA_IMAGE = re.compile(r'"(dimg_[^"]*)"[^;]*;(data:image[^;]*;[^;]*);')
+RE_DATA_IMAGE_end = re.compile(r'"(dimg_[^"]*)"[^;]*;(data:image[^;]*;[^;]*)$')
 
 
-def _parse_data_images(dom):
+def parse_data_images(text: str):
     data_image_map = {}
-    for img_id, data_image in RE_DATA_IMAGE.findall(dom.text_content()):
+
+    for img_id, data_image in RE_DATA_IMAGE.findall(text):
         end_pos = data_image.rfind('=')
         if end_pos > 0:
             data_image = data_image[: end_pos + 1]
         data_image_map[img_id] = data_image
+    last = RE_DATA_IMAGE_end.search(text)
+    if last:
+        data_image_map[last.group(1)] = last.group(2)
     logger.debug('data:image objects --> %s', list(data_image_map.keys()))
     return data_image_map
 
@@ -320,12 +351,12 @@ def response(resp) -> EngineResults:
     """Get response from google's search request"""
     # pylint: disable=too-many-branches, too-many-statements
     detect_google_sorry(resp)
+    data_image_map = parse_data_images(resp.text)
 
     results = EngineResults()
 
     # convert the text to dom
     dom = html.fromstring(resp.text)
-    data_image_map = _parse_data_images(dom)
 
     # results --> answer
     answer_list = eval_xpath(dom, '//div[contains(@class, "LGOjhe")]')

searx/engines/google_videos.py

@@ -12,6 +12,7 @@
    https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/Data_URIs
 
 """
+from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
@@ -32,6 +33,8 @@ from searx.engines.google import (
     filter_mapping,
     suggestion_xpath,
     detect_google_sorry,
+    ui_async,
+    parse_data_images,
 )
 from searx.enginelib.traits import EngineTraits
 from searx.utils import get_embeded_stream_url
@@ -67,6 +70,7 @@ def request(query, params):
     """Google-Video search request"""
 
     google_info = get_google_info(params, traits)
+    start = (params['pageno'] - 1) * 10
 
     query_url = (
         'https://'
@@ -80,7 +84,7 @@ def request(query, params):
                 'start': 10 * params['pageno'],
                 **google_info['params'],
                 'asearch': 'arc',
-                'async': 'use_ac:true,_fmt:html',
+                'async': ui_async(start),
             }
         )
     )
@@ -101,6 +105,7 @@ def response(resp):
     results = []
 
     detect_google_sorry(resp)
+    data_image_map = parse_data_images(resp.text)
 
     # convert the text to dom
     dom = html.fromstring(resp.text)
@@ -109,8 +114,13 @@ def response(resp):
     for result in eval_xpath_list(dom, '//div[contains(@class, "g ")]'):
 
         thumbnail = eval_xpath_getindex(result, './/img/@src', 0, None)
-        if thumbnail is None:
-            continue
+        if thumbnail:
+            if thumbnail.startswith('data:image'):
+                img_id = eval_xpath_getindex(result, './/img/@id', 0, None)
+                if img_id:
+                    thumbnail = data_image_map.get(img_id)
+        else:
+            thumbnail = None
 
         title = extract_text(eval_xpath_getindex(result, './/a/h3[1]', 0))
         url = eval_xpath_getindex(result, './/a/h3[1]/../@href', 0)

searx/engines/iqiyi.py

@@ -0,0 +1,83 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""iQiyi: A search engine for retrieving videos from iQiyi."""
+
+from urllib.parse import urlencode
+from datetime import datetime, timedelta
+
+from searx.exceptions import SearxEngineAPIException
+
+about = {
+    "website": "https://www.iqiyi.com/",
+    "wikidata_id": "Q15913890",
+    "use_official_api": False,
+    "require_api_key": False,
+    "results": "JSON",
+}
+
+paging = True
+time_range_support = True
+categories = ["videos"]
+
+time_range_dict = {'day': '1', 'week': '2', 'month': '3'}
+
+base_url = "https://mesh.if.iqiyi.com"
+
+
+def request(query, params):
+    query_params = {"key": query, "pageNum": params["pageno"], "pageSize": 25}
+
+    if time_range_dict.get(params['time_range']):
+        query_params["sitePublishDate"] = time_range_dict[params['time_range']]
+
+    params["url"] = f"{base_url}/portal/lw/search/homePageV3?{urlencode(query_params)}"
+    return params
+
+
+def response(resp):
+    try:
+        data = resp.json()
+    except Exception as e:
+        raise SearxEngineAPIException(f"Invalid response: {e}") from e
+    results = []
+
+    if "data" not in data or "templates" not in data["data"]:
+        raise SearxEngineAPIException("Invalid response")
+
+    for entry in data["data"]["templates"]:
+        album_info = entry.get("albumInfo", {})
+
+        published_date = None
+        release_time = album_info.get("releaseTime", {}).get("value")
+        if release_time:
+            try:
+                published_date = datetime.strptime(release_time, "%Y-%m-%d")
+            except (ValueError, TypeError):
+                pass
+
+        length = None
+        subscript_content = album_info.get("subscriptContent")
+        if subscript_content:
+            try:
+                time_parts = subscript_content.split(":")
+                if len(time_parts) == 2:
+                    minutes, seconds = map(int, time_parts)
+                    length = timedelta(minutes=minutes, seconds=seconds)
+                elif len(time_parts) == 3:
+                    hours, minutes, seconds = map(int, time_parts)
+                    length = timedelta(hours=hours, minutes=minutes, seconds=seconds)
+            except (ValueError, TypeError):
+                pass
+
+        results.append(
+            {
+                'url': album_info.get("pageUrl", "").replace("http://", "https://"),
+                'title': album_info.get("title", ""),
+                'content': album_info.get("brief", {}).get("value", ""),
+                'template': 'videos.html',
+                'length': length,
+                'publishedDate': published_date,
+                'thumbnail': album_info.get("img", ""),
+            }
+        )
+
+    return results

searx/settings.yml

@@ -352,6 +352,11 @@ engines:
     shortcut: 9g
     disabled: true
 
+  - name: acfun
+    engine: acfun
+    shortcut: acf
+    disabled: true
+
   - name: adobe stock
     engine: adobe_stock
     shortcut: asi
@@ -573,6 +578,24 @@ engines:
     # to show premium or plus results too:
     # skip_premium: false
 
+  - name: chinaso news
+    chinaso_category: news
+    engine: chinaso
+    shortcut: chinaso
+    disabled: true
+
+  - name: chinaso images
+    chinaso_category: images
+    engine: chinaso
+    shortcut: chinasoi
+    disabled: true
+
+  - name: chinaso videos
+    chinaso_category: videos
+    engine: chinaso
+    shortcut: chinasov
+    disabled: true
+
   - name: cloudflareai
     engine: cloudflareai
     shortcut: cfai
@@ -1100,6 +1123,11 @@ engines:
     shortcut: ip
     disabled: true
 
+  - name: iqiyi
+    engine: iqiyi
+    shortcut: iq
+    disabled: true
+
   - name: jisho
     engine: jisho
     shortcut: js