mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-09-20 14:27:42 +00:00
98 lines
3.8 KiB
Python
98 lines
3.8 KiB
Python
import re
|
|
from datetime import datetime
|
|
from typing import cast
|
|
|
|
import pytz
|
|
from rss_parser import Parser
|
|
from rss_parser.models.rss import RSS
|
|
|
|
from core.constants import CALL_REGEX
|
|
from data.alert import Alert
|
|
from providers.alert.http_alert_provider import HTTPAlertProvider
|
|
|
|
|
|
class NG3K(HTTPAlertProvider):
|
|
"""Alert provider NG3K DXpedition list"""
|
|
|
|
POLL_INTERVAL_SEC = 1800
|
|
ALERTS_URL = "https://www.ng3k.com/adxo.xml"
|
|
AS_CALL_PATTERN = re.compile(rf"as ({CALL_REGEX})", re.IGNORECASE)
|
|
|
|
def __init__(self, provider_config):
|
|
super().__init__("NG3K", provider_config, self.ALERTS_URL, self.POLL_INTERVAL_SEC)
|
|
|
|
def _http_response_to_alerts(self, http_response):
|
|
new_alerts = []
|
|
rss = cast(RSS, Parser.parse(http_response.content.decode("utf-8-sig")))
|
|
# 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(f"{start_year} {start_mon} {start_day}", "%Y %b %d")
|
|
.replace(tzinfo=pytz.UTC)
|
|
.timestamp()
|
|
)
|
|
end_timestamp = (
|
|
datetime.strptime(f"{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] if len(extra_parts) > 1 else ""
|
|
modes = extra_parts[2] if len(extra_parts) > 2 else ""
|
|
comment = extra_parts[3] if len(extra_parts) > 3 else ""
|
|
|
|
# Convert to our alert format
|
|
alert = Alert(
|
|
source=self.name,
|
|
dx_calls=dx_calls,
|
|
dx_country=dx_country,
|
|
freqs_modes=bands + (f"; {modes}" if modes != "" else ""),
|
|
comment=f"{by}; {comment}; {qsl_info}",
|
|
start_time=start_timestamp,
|
|
end_time=end_timestamp,
|
|
is_dxpedition=True,
|
|
)
|
|
|
|
# Add to our list.
|
|
new_alerts.append(alert)
|
|
return new_alerts
|