import logging from datetime import datetime from typing import cast from xml.parsers.expat import ExpatError import pytz from rss_parser import Parser as RSSParser from rss_parser.models.rss import RSS from core.enums import ActivityName from data.activity_ref import ActivityRef from data.alert import Alert from providers.alert.http_alert_provider import HTTPAlertProvider logger = logging.getLogger(__name__) 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 = [] try: 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 dx_call = None ref = None ref_name = None try: title_split = source_alert.title.split(" on ") dx_call = title_split[0] 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]) except Exception: logger.warning(f"Could not parse WOTA alert title: {source_alert.description}", exc_info=True) # Pick apart the description comment = None freqs_modes = None try: desc_split = source_alert.description.split(". ") freqs_modes = desc_split[0].replace("Frequencies/modes:", "").strip() if len(desc_split) > 1: comment = desc_split[1].strip() except Exception: logger.warning(f"Could not parse WOTA alert description: {source_alert.description}", exc_info=True) 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=[ActivityRef(id=ref, sig=ActivityName.WOTA, name=ref_name)] if ref else [], start_time=time.timestamp(), ) # Add to our list. new_alerts.append(alert) except ExpatError: logger.warning("WOTA alert RSS feed was fetched but was invalid") return new_alerts