summaryrefslogtreecommitdiff
path: root/searx/engines/boardreader.py
blob: b5baa5c3fe858079d7fe13301ec20a358f08dada (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Boardreader (forum search)"""

import re

from datetime import datetime
from urllib.parse import urlencode
import typing as t
import gettext
import babel

from searx.locales import language_tag
from searx.enginelib import EngineCache
from searx.enginelib.traits import EngineTraits
from searx.engines.json_engine import safe_search_map
from searx.exceptions import SearxEngineAPIException
from searx.network import get, raise_for_httperror
from searx.result_types import EngineResults
from searx.utils import extr, js_obj_str_to_python, html_to_text

if t.TYPE_CHECKING:
    from searx.extended_types import SXNG_Response
    from searx.search.processors import OnlineParams


about = {
    "website": "https://boardreader.com",
    "official_api_documentation": None,
    "use_official_api": False,
    "require_api_key": False,
    "results": "JSON",
}

categories = ["general", "social media"]
paging = True
time_range_support = True
language_support = True

base_url = "https://boardreader.com"
time_range_map = {"day": "1", "week": "7", "month": "30", "year": "365"}

CACHE: EngineCache
CACHE_SESSION_ID_KEY = "session_id_key"

KEYWORD_RE = re.compile(r"\[\/?Keyword\]")


def init(engine_settings: dict[str, t.Any]) -> bool:
    global CACHE  # pylint: disable=global-statement
    CACHE = EngineCache(engine_name=engine_settings["name"])
    return True


def _get_session_id() -> str:
    cached: str | None = CACHE.get(CACHE_SESSION_ID_KEY)
    if cached:
        return cached

    resp = get(base_url)
    if resp.status_code != 200:
        raise_for_httperror(resp)

    session_id = extr(resp.text, "'currentSessionId', '", "'")
    if not session_id:
        raise SearxEngineAPIException("failed to obtain session id")

    CACHE.set(CACHE_SESSION_ID_KEY, session_id)
    return session_id


def request(query: str, params: "OnlineParams"):
    session_id = _get_session_id()

    language: str = traits.get_language(
        params["searxng_locale"], default="All"
    )  # pyright: ignore[reportAssignmentType]
    args = {
        "query": query,
        "page": params["pageno"],
        "language": language,
        "session_id": session_id,
    }
    if params["time_range"]:
        args["period"] = safe_search_map[params["time_range"]]  # pyright: ignore[reportArgumentType]

    params["url"] = f"{base_url}/return.php?{urlencode(args)}"
    return params


def _remove_keyword_marker(text: str) -> str:
    """
    Convert text like "[Keyword]ABCDE[/Keyword]" to "ABCDE".
    """
    return html_to_text(KEYWORD_RE.sub("", text))


def response(resp: "SXNG_Response") -> EngineResults:
    res = EngineResults()

    result: dict[str, str]
    for result in resp.json()["SearchResults"]:
        res.add(
            res.types.MainResult(
                title=_remove_keyword_marker(result["Subject"]),
                content=_remove_keyword_marker(result["Text"]),
                url=result["Url"],
                publishedDate=datetime.fromisoformat(result["Published"]),
                metadata=gettext.gettext("Posted by {author}").format(author=result["Author"]),
            )
        )

    return res


def fetch_traits(engine_traits: EngineTraits):
    # load main page to be able to find location of JavaScript source code
    resp = get(base_url)
    if resp.status_code != 200:
        raise_for_httperror(resp)

    # load actual JavaScript code
    script_name = "main." + extr(resp.text, "main.", ".js") + ".js"
    script_resp = get(f"{base_url}/{script_name}")
    if script_resp.status_code != 200:
        raise_for_httperror(resp)

    # find list of languages (JavaScript object)
    js_object_string = extr(script_resp.text, "languageValues=", "}],") + "}]"
    languages: list[dict[str, str]] = js_obj_str_to_python(js_object_string)

    # finally, add all parsed languages to the engine traits
    language: dict[str, str]
    for language in languages:
        search_value = language["value"]
        for code in language["codes"]:
            try:
                locale = babel.Locale.parse(code)
            except babel.UnknownLocaleError:
                continue

            sxng_lang = language_tag(locale)
            if sxng_lang not in engine_traits.languages:
                engine_traits.languages[sxng_lang] = search_value

    # "All" is the search value to unset the search language
    engine_traits.all_locale = "All"