[mod] engine: pexels - added type hints and migrate to Image-Results (#6739)

Code style somewhat unified, type annotations added, ``LegacyResult`` replaced
by ``Image``, but no functional modifications.

Signed-off-by: Markus Heiser <markus.heiser@darmarit.de>

searx/engines/mediathekviewweb.py

@@ -1,73 +1,80 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 """MediathekViewWeb (API)"""
 
+import typing as t
 import datetime
-from json import loads, dumps
+
+from searx.result_types import EngineResults
+
+if t.TYPE_CHECKING:
+    from searx.extended_types import SXNG_Response
+    from searx.search.processors import OnlineParams
 
 about = {
-    "website": 'https://mediathekviewweb.de/',
-    "wikidata_id": 'Q27877380',
-    "official_api_documentation": 'https://gist.github.com/bagbag/a2888478d27de0e989cf777f81fb33de',
+    "website": "https://mediathekviewweb.de/",
+    "wikidata_id": "Q27877380",
+    "official_api_documentation": "https://gist.github.com/bagbag/a2888478d27de0e989cf777f81fb33de",
     "use_official_api": True,
     "require_api_key": False,
-    "results": 'JSON',
+    "results": "JSON",
 }
 
 language = "de"
-categories = ['videos']
+categories = ["videos"]
 paging = True
 time_range_support = False
 safesearch = False
 
 
-def request(query, params):
+def request(query: str, params: "OnlineParams"):
 
-    params['url'] = 'https://mediathekviewweb.de/api/query'
-    params['method'] = 'POST'
-    params['headers']['Content-type'] = 'text/plain'
-    params['data'] = dumps(
-        {
-            'queries': [
-                {
-                    'fields': [
-                        'title',
-                        'topic',
-                    ],
-                    'query': query,
-                },
-            ],
-            'sortBy': 'timestamp',
-            'sortOrder': 'desc',
-            'future': True,
-            'offset': (params['pageno'] - 1) * 10,
-            'size': 10,
-        }
-    )
-    return params
+    params["url"] = "https://mediathekviewweb.de/api/query"
+    params["method"] = "POST"
+    params["headers"]["Content-type"] = "text/plain"
+    params["json"] = {
+        "queries": [
+            {
+                "fields": [
+                    "title",
+                    "topic",
+                    "description",
+                ],
+                "query": query,
+            },
+        ],
+        "sortBy": "timestamp",
+        "sortOrder": "desc",
+        "future": True,
+        "offset": (params["pageno"] - 1) * 10,
+        "size": 10,
+    }
 
 
-def response(resp):
+def response(resp: "SXNG_Response") -> EngineResults:
 
-    resp = loads(resp.text)
+    json_resp: dict[str, t.Any] = resp.json()
 
-    mwv_result = resp['result']
-    mwv_result_list = mwv_result['results']
+    mwv_result = json_resp["result"]
+    mwv_result_list = mwv_result["results"]
 
-    results = []
+    res = EngineResults()
 
     for item in mwv_result_list:
+        item["hms"] = str(datetime.timedelta(seconds=item["duration"]))
 
-        item['hms'] = str(datetime.timedelta(seconds=item['duration']))
+        video_url = item["url_video_hd"] or item["url_video"] or item["url_video_low"] or item["url_video"]
+        if not video_url:
+            continue
 
-        results.append(
-            {
-                'url': item['url_video_hd'].replace("http://", "https://"),
-                'title': "%(channel)s: %(title)s (%(hms)s)" % item,
-                'length': item['hms'],
-                'content': "%(description)s" % item,
-                'iframe_src': item['url_video_hd'].replace("http://", "https://"),
-                'template': 'videos.html',
-            }
+        res.add(
+            res.types.LegacyResult(
+                url=video_url,
+                title="%(channel)s: %(title)s (%(hms)s)" % item,
+                length=item["hms"],
+                content="%(description)s" % item,
+                iframe_src=video_url,
+                template="videos.html",
+            )
         )
 
-    return results
+    return res

searx/engines/pexels.py

@@ -13,18 +13,21 @@ from searx.enginelib import EngineCache
 from searx.exceptions import SearxEngineAPIException, SearxEngineAccessDeniedException
 from searx.network import get
 
-# about
+if t.TYPE_CHECKING:
+    from extended_types import SXNG_Response
+    from search.processors.online import OnlineParams
+
 about = {
-    "website": 'https://www.pexels.com',
-    "wikidata_id": 'Q101240504',
-    "official_api_documentation": 'https://www.pexels.com/api/',
+    "website": "https://www.pexels.com",
+    "wikidata_id": "Q101240504",
+    "official_api_documentation": "https://www.pexels.com/api/",
     "use_official_api": False,
     "require_api_key": False,
-    "results": 'JSON',
+    "results": "JSON",
 }
 
-base_url = 'https://www.pexels.com'
-categories = ['images']
+base_url = "https://www.pexels.com"
+categories = ["images"]
 
 api_key = "H2jk9uKnhRmL6WPwh89zBezWvr"
 """
@@ -34,7 +37,7 @@ results_per_page = 20
 
 paging = True
 time_range_support = True
-time_range_map = {'day': 'last_24_hours', 'week': 'last_week', 'month': 'last_month', 'year': 'last_year'}
+time_range_map = {"day": "last_24_hours", "week": "last_week", "month": "last_month", "year": "last_year"}
 
 SECRET_KEY_RE = re.compile('"secret-key":\b*"(.*?)"')
 SECRET_KEY_DB_KEY = "secret-key"
@@ -73,14 +76,14 @@ def _get_secret_key():
     raise SearxEngineAPIException("failed to obtain secret key")
 
 
-def request(query, params):
+def request(query: str, params: "OnlineParams"):
     args = {
-        'query': query,
-        'page': params['pageno'],
-        'per_page': results_per_page,
+        "query": query,
+        "page": params["pageno"],
+        "per_page": results_per_page,
     }
-    if params['time_range']:
-        args['date_from'] = time_range_map[params['time_range']]
+    if params["time_range"]:
+        args["date_from"] = time_range_map[params["time_range"]]
 
     params["url"] = f"{base_url}/en-us/api/v3/search/photos?{urlencode(args)}"
 
@@ -97,24 +100,21 @@ def request(query, params):
     params["headers"]["secret-key"] = secret_key
 
 
-def response(resp):
+def response(resp: "SXNG_Response") -> EngineResults:
     res = EngineResults()
     json_data = resp.json()
 
-    for result in json_data.get('data', []):
+    for result in json_data.get("data", []):
         attrs = result["attributes"]
         res.add(
-            res.types.LegacyResult(
-                {
-                    'template': 'images.html',
-                    'url': f"{base_url}/photo/{attrs['slug']}-{attrs['id']}/",
-                    'title': attrs["title"],
-                    'content': attrs["description"],
-                    'thumbnail_src': attrs["image"]["small"],
-                    'img_src': attrs["image"]["download_link"],
-                    'resolution': f"{attrs['width']}x{attrs['height']}",
-                    'author': f"{attrs['user']['username']}",
-                }
+            res.types.Image(
+                url=f"{base_url}/photo/{attrs['slug']}-{attrs['id']}/",
+                title=attrs["title"],
+                content=attrs["description"],
+                thumbnail_src=attrs["image"]["small"],
+                img_src=attrs["image"]["download_link"],
+                resolution=f"{attrs['width']}x{attrs['height']}",
+                author=f"{attrs['user']['username']}",
             )
         )
 

searx/network/__init__.py

@@ -201,6 +201,8 @@ def delete(url: str, **kwargs: t.Any) -> SXNG_Response:
 
 async def stream_chunk_to_queue(network, queue, method: str, url: str, **kwargs: t.Any):
     try:
+        # prevent curl from decompressing the response https://github.com/searxng/searxng/pull/6730
+        kwargs.setdefault('accept_encoding', None)
         async with await network.stream(method, url, **kwargs) as response:
             queue.put(response)
             async for chunk in response.aiter_content():