summaryrefslogtreecommitdiff
path: root/searx/engines/duckduckgo_weather.py
blob: 4d52effcd80b36efc3ff97e44e8a9651dd194613 (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
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
DuckDuckGo Weather
~~~~~~~~~~~~~~~~~~
"""

import typing as t
from json import loads
from urllib.parse import quote

from dateutil import parser as date_parser

from searx.engines.duckduckgo import fetch_traits  # pylint: disable=unused-import
from searx.engines.duckduckgo import get_ddg_lang

from searx.result_types import EngineResults
from searx.extended_types import SXNG_Response
from searx import weather


about = {
    "website": 'https://duckduckgo.com/',
    "wikidata_id": 'Q12805',
    "official_api_documentation": None,
    "use_official_api": True,
    "require_api_key": False,
    "results": "JSON",
}

send_accept_language_header = True

# engine dependent config
categories = ["weather"]
base_url = "https://duckduckgo.com/js/spice/forecast/{query}/{lang}"

# adapted from https://gist.github.com/mikesprague/048a93b832e2862050356ca233ef4dc1
WEATHERKIT_TO_CONDITION: dict[str, weather.WeatherConditionType] = {
    "BlowingDust": "fog",
    "Clear": "clear sky",
    "Cloudy": "cloudy",
    "Foggy": "fog",
    "Haze": "fog",
    "MostlyClear": "clear sky",
    "MostlyCloudy": "partly cloudy",
    "PartlyCloudy": "partly cloudy",
    "Smoky": "fog",
    "Breezy": "partly cloudy",
    "Windy": "partly cloudy",
    "Drizzle": "light rain",
    "HeavyRain": "heavy rain",
    "IsolatedThunderstorms": "rain and thunder",
    "Rain": "rain",
    "SunShowers": "rain",
    "ScatteredThunderstorms": "heavy rain and thunder",
    "StrongStorms": "heavy rain and thunder",
    "Thunderstorms": "rain and thunder",
    "Frigid": "clear sky",
    "Hail": "heavy rain",
    "Hot": "clear sky",
    "Flurries": "light snow",
    "Sleet": "sleet",
    "Snow": "light snow",
    "SunFlurries": "light snow",
    "WintryMix": "sleet",
    "Blizzard": "heavy snow",
    "BlowingSnow": "heavy snow",
    "FreezingDrizzle": "light sleet",
    "FreezingRain": "sleet",
    "HeavySnow": "heavy snow",
    "Hurricane": "rain and thunder",
    "TropicalStorm": "rain and thunder",
}


def _weather_data(location: weather.GeoLocation, data: dict[str, t.Any]):

    return EngineResults.types.WeatherAnswer.Item(
        location=location,
        temperature=weather.Temperature(val=data['temperature'], unit="°C"),
        condition=WEATHERKIT_TO_CONDITION[data["conditionCode"]],
        feels_like=weather.Temperature(val=data['temperatureApparent'], unit="°C"),
        wind_from=weather.Compass(data["windDirection"]),
        wind_speed=weather.WindSpeed(val=data["windSpeed"], unit="mi/h"),
        pressure=weather.Pressure(val=data["pressure"], unit="hPa"),
        humidity=weather.RelativeHumidity(data["humidity"] * 100),
        cloud_cover=data["cloudCover"] * 100,
    )


def request(query: str, params: dict[str, t.Any]):

    eng_region = traits.get_region(params['searxng_locale'], traits.all_locale)
    eng_lang = get_ddg_lang(traits, params['searxng_locale'])

    # !ddw paris :es-AR --> {'ad': 'es_AR', 'ah': 'ar-es', 'l': 'ar-es'}
    params['cookies']['ad'] = eng_lang
    params['cookies']['ah'] = eng_region
    params['cookies']['l'] = eng_region
    logger.debug("cookies: %s", params['cookies'])

    params["url"] = base_url.format(query=quote(query), lang=eng_lang.split('_')[0])
    return params


def response(resp: SXNG_Response):
    res = EngineResults()

    if resp.text.strip() == "ddg_spice_forecast();":
        return res

    json_data = loads(resp.text[resp.text.find('\n') + 1 : resp.text.rfind('\n') - 2])

    geoloc = weather.GeoLocation.by_query(resp.search_params["query"])

    weather_answer = EngineResults.types.WeatherAnswer(
        current=_weather_data(geoloc, json_data["currentWeather"]),
        service="duckduckgo weather",
    )

    for forecast in json_data['forecastHourly']['hours']:
        forecast_time = date_parser.parse(forecast['forecastStart'])
        forecast_data = _weather_data(geoloc, forecast)
        forecast_data.datetime = weather.DateTime(forecast_time)
        weather_answer.forecasts.append(forecast_data)

    res.add(weather_answer)
    return res