Refactor of caching & data storage part 9 #118

providers/alert/wota.py

@@ -0,0 +1,64 @@
+from datetime import datetime
+from typing import cast
+
+import pytz
+from rss_parser import Parser as RSSParser
+from rss_parser.models.rss import RSS
+
+from providers.alert.http_alert_provider import HTTPAlertProvider
+from data.alert import Alert
+from data.sig_ref import SIGRef
+
+
+class WOTA(HTTPAlertProvider):
+    """Alert provider for Wainwrights on the Air"""
+
+    POLL_INTERVAL_SEC = 1800
+    ALERTS_URL = "https://www.wota.org.uk/alerts_rss.php"
+    RSS_DATE_TIME_FORMAT = "%a, %d %b %Y %H:%M:%S %z"
+
+    def __init__(self, provider_config):
+        super().__init__("WOTA", provider_config, self.ALERTS_URL, self.POLL_INTERVAL_SEC)
+
+    def _http_response_to_alerts(self, http_response):
+        new_alerts = []
+        rss = cast(RSS, RSSParser.parse(http_response.content.decode("utf-8-sig")))
+        # Iterate through source data
+        for source_alert in rss.channel.items:
+
+            # Reject GUID missing or zero
+            if not source_alert.guid or not source_alert.guid.content or source_alert.guid.content == "http://www.wota.org.uk/alerts/0":
+                continue
+
+            # Pick apart the title
+            title_split = source_alert.title.split(" on ")
+            dx_call = title_split[0]
+            ref = None
+            ref_name = None
+            if len(title_split) > 1:
+                ref_split = title_split[1].split(" - ")
+                ref = str(ref_split[0])
+                if len(ref_split) > 1:
+                    ref_name = str(ref_split[1])
+
+            # Pick apart the description
+            desc_split = source_alert.description.split(". ")
+            freqs_modes = desc_split[0].replace("Frequencies/modes:", "").strip()
+            comment = None
+            if len(desc_split) > 1:
+                comment = desc_split[1].strip()
+
+            time = datetime.strptime(source_alert.pub_date.content, self.RSS_DATE_TIME_FORMAT).astimezone(pytz.UTC)
+
+            # Convert to our alert format
+            alert = Alert(source=self.name,
+                          source_id=source_alert.guid.content,
+                          dx_calls=[dx_call],
+                          freqs_modes=freqs_modes,
+                          comment=comment,
+                          sig_refs=[SIGRef(id=ref, sig="WOTA", name=ref_name)] if ref else [],
+                          start_time=time.timestamp())
+
+            # Add to our list.
+            new_alerts.append(alert)
+        return new_alerts