mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2025-10-27 16:59:25 +00:00
87 lines
3.6 KiB
Python
87 lines
3.6 KiB
Python
import re
|
|
from datetime import datetime
|
|
|
|
import pytz
|
|
from rss_parser import RSSParser
|
|
|
|
from alertproviders.http_alert_provider import HTTPAlertProvider
|
|
from data.alert import Alert
|
|
|
|
|
|
# Alert provider NG3K DXpedition list
|
|
class NG3K(HTTPAlertProvider):
|
|
POLL_INTERVAL_SEC = 3600
|
|
ALERTS_URL = "https://www.ng3k.com/adxo.xml"
|
|
AS_CALL_PATTERN = re.compile("as ([a-z0-9/]+)", re.IGNORECASE)
|
|
|
|
def __init__(self, provider_config):
|
|
super().__init__(provider_config, self.ALERTS_URL, self.POLL_INTERVAL_SEC)
|
|
|
|
def http_response_to_alerts(self, http_response):
|
|
new_alerts = []
|
|
rss = RSSParser.parse(http_response.content.decode())
|
|
# Iterate through source data
|
|
for source_alert in rss.channel.items:
|
|
# Deal with "the format"...
|
|
parts = source_alert.description.split(" --\n")
|
|
|
|
start_string = parts[0].split("-")[0]
|
|
end_string = parts[0].split("-")[1]
|
|
|
|
end_year = end_string.split(", ")[1].strip()
|
|
|
|
if ", " in start_string:
|
|
start_year = start_string.split(", ")[1].strip()
|
|
start_mon = start_string.split(", ")[0][0:3].strip()
|
|
start_day = start_string.split(", ")[0][4:].strip()
|
|
else:
|
|
start_year = end_year
|
|
start_mon = start_string[0:3].strip()
|
|
start_day = start_string[4:].strip()
|
|
|
|
if " " in end_string.split(", ")[0]:
|
|
end_mon = end_string.split(", ")[0].split(" ")[0].strip()
|
|
end_day = end_string.split(", ")[0].split(" ")[1].strip()
|
|
else:
|
|
end_day = end_string.split(", ")[0].strip()
|
|
end_mon = start_mon
|
|
|
|
start_timestamp = datetime.strptime(start_year + " " + start_mon + " " + start_day, "%Y %b %d").replace(
|
|
tzinfo=pytz.UTC).timestamp()
|
|
end_timestamp = datetime.strptime(end_year + " " + end_mon + " " + end_day + " 23:59", "%Y %b %d %H:%M").replace(
|
|
tzinfo=pytz.UTC).timestamp()
|
|
|
|
# Sometimes the DX callsign is "real", sometimes you just get a prefix with the real working callsigns being
|
|
# provided in the "by" field. e.g. call="JW", by="By LA7XK as JW7XK, LA6VM as JW6VM, LA9DL as JW9DL". So
|
|
# if there are "as" callsigns in the "by" field, we extract them and use them, otherwise we fall back to the
|
|
# "real" call field.
|
|
extra_parts = parts[5].split("; ")
|
|
by = extra_parts[0]
|
|
dx_calls = [x.group(1) for x in re.finditer(self.AS_CALL_PATTERN, by)]
|
|
if not dx_calls:
|
|
dx_calls = [parts[2].upper()]
|
|
|
|
# "Calls" of TBA, TBC or TBD are not real attempts at Turkish callsigns
|
|
dx_calls = list(filter(lambda a: a != "TBA" and a != "TBC" and a != "TBD" , dx_calls))
|
|
|
|
dx_country = parts[1]
|
|
qsl_info = parts[3]
|
|
bands = extra_parts[1]
|
|
modes = extra_parts[2] if len(extra_parts) > 3 else ""
|
|
comment = extra_parts[-1]
|
|
|
|
# Convert to our alert format
|
|
alert = Alert(source=self.name,
|
|
dx_calls=dx_calls,
|
|
dx_country=dx_country,
|
|
freqs_modes=bands + (("; " + modes) if modes != "" else ""),
|
|
comment=by + "; " + comment + "; " + qsl_info,
|
|
icon="globe-africa",
|
|
start_time=start_timestamp,
|
|
end_time=end_timestamp,
|
|
is_dxpedition=True)
|
|
|
|
# Add to our list.
|
|
new_alerts.append(alert)
|
|
return new_alerts
|