WIP: #9

classes/pycordguild_colors.py

@@ -0,0 +1,9 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class PycordGuildColors:
+    default: str
+    success: str
+    warning: str
+    error: str

modules/utils.py

@@ -0,0 +1,6 @@
+def hex_to_int(hex_color: str) -> int:
+    return int(hex_color.lstrip("#"), 16)
+
+
+def int_to_hex(integer_color: int) -> str:
+    return "#" + format(integer_color, "06x")

modules/weather/parser.py

@@ -0,0 +1,9 @@
+from typing import Any, Dict
+
+
+def parse_weather(api_response: Dict[str, Any]) -> str:
+    return ""
+
+
+def parse_weather_current(api_response: Dict[str, Any]) -> str:
+    return ""

modules/weather/reporter.py

@@ -0,0 +1,106 @@
+from datetime import datetime
+import logging
+from typing import Any, Dict, List
+
+from discord import Embed
+from pymongo import DESCENDING
+from pytz import timezone
+from classes.enums import MessageEvents
+from classes.pycordbot import PycordBot
+from ujson import loads
+from modules.utils import hex_to_int
+from database import col_messages
+
+from modules.weather.parser import parse_weather
+
+# Example guild key
+# "forecast": {
+#     "channel": 0,
+#     "time": "10:00:00",
+#     "delete_previous": true
+#     "locations": [{"name": "Sample Location", "location": [10.000, 20.000]}],
+# }
+
+logger = logging.getLogger(__name__)
+
+
+async def report_weather(
+    bot: PycordBot,
+    guild: PycordGuild,  # TODO
+    channel_id: int,
+    delete_previous: bool,
+    locations: List[Dict[str, Any]],
+) -> None:  # sourcery skip: aware-datetime-for-utc
+    channel = bot.get_channel(channel_id)
+
+    if channel is None:
+        logger.error(
+            "Cannot generate weather report for %s's channel %s because channel was not found.",
+            guild.id,
+            channel_id,
+        )
+        return
+
+    # Find and delete previous forecast, if needed
+    if delete_previous:
+        async for event in col_messages.find(
+            {
+                "event": MessageEvents.WEATHER_FORECAST,
+                "guild": guild.id,
+                "channel": channel_id,
+            },
+            limit=1,
+        ).sort("date", direction=DESCENDING):
+            try:
+                old_message = bot.get_message(event["message"])
+                if old_message is not None:
+                    await old_message.delete(
+                        reason="Cleanup of the old weather report (look in guild config for details)"
+                    )
+            except Exception as exc:
+                logger.warning(
+                    "Could not delete the previous weather report in %s' channel %s due to %s",
+                    guild.id,
+                    channel_id,
+                    exc,
+                )
+
+    embeds: List[Embed] = []
+
+    # Iterate through the locations and request their forecasts.
+    # Results must be parsed and added as embeds to the embeds lits.
+    for location in locations:
+        location_timezone_offset = ":".join(
+            str(
+                timezone(bot.config["bot"]["timezone"]).utcoffset(datetime.utcnow())
+            ).split(":")[:2]
+        )
+
+        api_response = await (
+            await bot.client_session.get(
+                f"https://api.openweathermap.org/data/2.5/onecall?lat={location['location'][0]}&lon={location['location'][1]}&exclude=minutely&units=metric&lang=uk&appid={bot.config['modules']['weather']['forecasts']['api_key']}&tz={location_timezone_offset}"
+            )
+        ).json(loads=loads)
+
+        parsed_weather = parse_weather(api_response)
+
+        embeds.append(
+            Embed(
+                title=location["name"],
+                description=parsed_weather,
+                color=hex_to_int(guild.colors.default),
+            )
+        )
+
+    # Add a trailing embed with OWM information
+    embeds.append(
+        Embed(
+            title=bot._("weather_report_title", "embeds"),
+            description=bot._("weather_report_description", "embeds"),
+            color=hex_to_int(guild.colors.default),
+        )
+    )
+
+    await channel.send(  # type: ignore
+        content=bot._("weather_report_content", "messages"), embeds=embeds
+    )