Announcements work + task

.gitignore

@@ -8,6 +8,7 @@ files/stats.json
 files/lost.json
 files/locked.json
 files/database.json
+files/ufora_notifications.json
 .idea/
 __pycache__
 .env

cogs/tasks.py

@@ -6,6 +6,7 @@ from functions import timeFormatters
 from functions.config import config
 from functions.database import currency, poke, prison, birthdays, stats
 from functions.scraping import getMatchweek
+from functions import ufora_notifications
 import json
 import random
 import requests
@@ -24,6 +25,7 @@ class Tasks(commands.Cog):
         self.updateMessageCounts.start()
         self.sendReminders.start()
         self.updateMatchweek.start()
+        self.uforaAnnouncements.start()
 
     @tasks.loop(hours=1.0)
     async def bankInterest(self):
@@ -238,6 +240,25 @@ class Tasks(commands.Cog):
     async def beforeUpdateMatchweek(self):
         await self.client.wait_until_ready()
 
+    @tasks.loop(minutes=2)
+    async def uforaAnnouncements(self):
+        """
+        Task that checks for new Ufora announcements every few minutes
+        """
+        # Get new notifications
+        announcements = ufora_notifications.run()
+
+        if announcements:
+            # TODO change to COC Bot testing if it works
+            announcements_channel = self.client.get_channel(int(constants.ZandbakSpeeltuin))
+
+            for an in announcements:
+                await announcements_channel.send(embed=an.to_embed())
+
+    @uforaAnnouncements.before_loop
+    async def beforeUforaAnnouncements(self):
+        await self.client.wait_until_ready()
+
     def getCurrentHour(self):
         return timeFormatters.dateTimeNow().hour
 

data/constants.py

@@ -20,6 +20,7 @@ CoCGeneral = "626699611813314561"
 DeZandbak = "728361030404538488"
 ErrorLogs = "762668505455132722"
 FeatureRequests = "762668473313787964"
+ZandbakSpeeltuin = "769248992957038612"
 
 mods = {
     626699611192688641: [384457911377854467, 171671190631481345],

data/embeds.py

@@ -1,5 +0,0 @@
-from discord import embeds
-
-
-class UforaNotification:
-    pass

data/embeds/__init__.py

@@ -0,0 +1 @@
+from .ufora import UforaNotification

data/embeds/ufora.py

@@ -0,0 +1,81 @@
+from datetime import datetime, timedelta
+from discord import Embed, Colour
+from functions.timeFormatters import intToWeekday
+import pytz
+import re
+
+
+class UforaNotification:
+    def __init__(self, content: dict, course):
+        self._content: dict = content
+        self._course = course
+        self._notif_id, self._course_id = self._find_ids(self._content["id"])
+        self._view_url = self._create_url()
+        self._title = self._clean_content(self._content["title"])
+        self._description = self._get_description()
+        self._published = self._get_published()
+
+    def to_embed(self):
+        embed = Embed(colour=Colour.from_rgb(30, 100, 200))
+
+        embed.set_author(name=self._course)
+        embed.title = self._title
+        embed.url = self._view_url
+        embed.description = self._description
+        embed.set_footer(text=self._published)
+
+        return embed
+
+    def _create_url(self):
+        if self._notif_id is None or self._course_id is None:
+            return self._content["link"]
+
+        return "https://ufora.ugent.be/d2l/le/news/{0}/{1}/view?ou={0}".format(self._notif_id, self._course_id)
+
+    def _find_ids(self, url: str):
+        match = re.search(r"[0-9]+-[0-9]+$", url)
+
+        if not match:
+            return None, None
+
+        spl = match[0].split("-")
+
+        return spl[0], spl[1]
+
+    def _get_description(self):
+        desc = self._clean_content(self._content["summary"])
+
+        if len(desc) > 500:
+            return desc[:500]
+
+        return desc
+
+    def _clean_content(self, text: str):
+        html_table = {
+            "&": '&',
+            """: '"',
+            "apos;": "'",
+            ">": ">",
+            "&lt;": "<"
+        }
+
+        # Unescape HTML
+        for key, value in html_table.items():
+            text = text.replace(key, value)
+
+        # Remove HTML tags
+        return re.sub(r"<[^>]*>", "", text)
+
+    def _get_published(self):
+        time_string = "%a, %d %b %Y %H:%M:%S %Z"
+        dt = datetime.strptime(self._content["published"], time_string)\
+            .astimezone(pytz.timezone("Europe/Brussels"))
+
+        # Apply timezone offset
+        dt = dt + timedelta(hours=dt.utcoffset().seconds//3600)
+
+        return "{} {}/{}/{} om {}:{}:{}".format(
+            intToWeekday(dt.weekday()),
+            dt.day, dt.month, dt.year,
+            dt.hour, dt.minute, dt.second
+        )

functions/timeFormatters.py

@@ -1,8 +1,7 @@
 import datetime
-import time
-
 import dateutil.relativedelta
 import pytz
+import time
 
 
 def epochToDate(epochTimeStamp, strFormat="%d/%m/%Y om %H:%M:%S"):

functions/ufora_notifications.py

@@ -1,9 +1,6 @@
 import feedparser
-from data.constants import BotTesting
+from data.embeds import UforaNotification
-
+import json
-
-# TODO when it works, move to announcements channel
-notifications_channel = BotTesting
 
 
 course_urls = {
@@ -17,4 +14,40 @@ course_urls = {
     "Systeemprogrammeren": "https://ufora.ugent.be/d2l/le/news/rss/222035/course?token=aehhv6utkf46t8cc102e0&ou=222035",
     "Webdevelopment": "https://ufora.ugent.be/d2l/le/news/rss/223449/course?token=aehhv6utkf46t8cc102e0&ou=223449",
     "Wetenschappelijk Rekenen": "https://ufora.ugent.be/d2l/le/news/rss/236404/course?token=aehhv6utkf46t8cc102e0&ou=236404"
-}
+}
+
+
+def run():
+    """
+    Check for new notifications
+    """
+    # List of new notifications
+    new_notifications = []
+
+    # List of old notifications
+    with open("files/ufora_notifications.json", "r") as fp:
+        notifications = json.load(fp)
+
+    for course, url in course_urls.items():
+        # Automatically append new/missing courses
+        if course not in notifications:
+            notifications[course] = []
+
+        # Get the updated feed
+        feed = feedparser.parse(url)
+
+        # Filter out old notifications
+        feed = list(filter(lambda f: f["id"] not in notifications[course], feed.entries))
+
+        if feed:
+            for item in feed:
+                notifications[course].append(item["id"])
+
+                new_notifications.append(UforaNotification(item, course))
+
+    # Update list of notifications
+    if new_notifications:
+        with open("files/ufora_notifications.json", "w") as fp:
+            json.dump(notifications, fp)
+
+    return new_notifications