107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
|
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
|
||
|
|
)
|