weather cog, including forecasts and location id

src/extensions/weather.py

@@ -0,0 +1,136 @@
+"""
+Weather Cog.
+WIP.
+"""
+
+import logging
+
+import aiohttp
+import discord
+from discord import app_commands, Interaction
+from discord.ext import commands, tasks
+from urllib.parse import quote
+from bs4 import BeautifulSoup as bs4
+
+from bbc_feeds import weather as weather_api
+
+log = logging.getLogger(__name__)
+
+
+class WeatherCog(commands.Cog):
+    """
+    Weather cog is used to deliver localised weather information to dedicated discord channels.
+    """
+
+    def __init__(self, bot):
+        self.bot = bot
+        self.weather_task.start()
+
+    @tasks.loop(minutes=15)
+    async def weather_task(self):
+        """
+        Send weather information to the assigned weather channels.
+        """
+
+        log.info("doing weather task")
+
+    async def get_weather_embeds(self, location_id: int, limit: int) -> list[discord.Embed]:
+        """
+        Returns a list of discord embeds containing weather data for the given location.
+        """
+
+        log.info(f"getting weather for location {location_id}")
+
+        forecasts = weather_api().forecast(location_id, limit=limit)
+        if not forecasts:
+            return [
+                discord.Embed(
+                    title="Weather Forecast Not Found",
+                    description=f"I could not find the weather forecast for `{location_id}` :(",
+                    colour=discord.Colour.red()
+                )
+            ]
+
+        return [
+            discord.Embed(
+                title = forecast.title.split(", Minimum Temp")[0],
+                description=forecast.summary.replace(", ", "\n"),
+                url=forecast.link,
+                colour=discord.Colour.from_str("#FFFFFF")
+            )
+            for forecast in forecasts
+        ]
+
+    weather_group = app_commands.Group(name="weather", description="Weather related commands")
+    channels_group = app_commands.Group(parent=weather_group, name="channels", description="Weather channel commands")
+
+    @weather_group.command(name="now")
+    async def get_weather_now(self, inter: Interaction, location_id: int):
+        """
+        Returns the current weather in your local area.
+        """
+
+        embeds = await self.get_weather_embeds(location_id, limit=1)
+        await inter.response.send_message(embeds=embeds)
+
+    @weather_group.command(name="tomorrow")
+    async def get_weather_tomorrow(self, inter: Interaction, location_id: int):
+        """
+        Returns tomorrows weather in your local area.
+        """
+
+        embed = (await self.get_weather_embeds(location_id, limit=2))[1]
+        await inter.response.send_message(embed=embed)
+
+    @weather_group.command(name="3-days")
+    async def get_weather_week(self, inter: Interaction, location_id: int):
+        """
+        Returns the next 3 days worth of weather in your local area.
+        """
+
+        embeds = await self.get_weather_embeds(location_id, limit=3)
+        await inter.response.send_message(embeds=embeds)
+
+    @weather_group.command(name="location-id")
+    async def get_weather_location_id(self, inter: Interaction, location_name: str):
+        """
+        Returns a location ID used to identify weather in your local area.
+        """
+
+        await inter.response.defer()
+
+        result_string = ""
+
+        safe_location = quote(location_name)
+        get_url = f"https://www.bbc.co.uk/weather/search?s={safe_location}"
+
+        async with aiohttp.ClientSession() as session:
+            async with session.get(get_url) as response:
+                html = await response.text()
+
+                soup = bs4(html, "html.parser")
+                anchors = soup.select(".location-search-results__result__link")
+
+                for anchor in anchors:
+                    location_id = anchor.get("href")
+                    result_string += f"\n`{location_id}` {anchor.text}"
+
+        if result_string:
+            output_string = f"Here is what I found from that search:\n{result_string}"
+        else:
+            output_string = "I couldn't find any location ID's from that search"
+
+        await inter.followup.send(output_string)
+
+
+
+                
+
+
+async def setup(bot):
+    """
+    Setup the weather cog with the bot.
+    """
+
+    cog = WeatherCog(bot)
+    await bot.add_cog(cog)