137 lines
4.5 KiB
Python
Executable File
137 lines
4.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Poll Eesti.ee Sitrep API and write events to an RSS feed file."""
|
|
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from xml.sax.saxutils import escape
|
|
|
|
import requests
|
|
|
|
API_URL = "https://api.app.eesti.ee/api/sitrep/v1/full-events"
|
|
OUTPUT_FILE = "sitrep_feed.xml"
|
|
POLL_INTERVAL = 300 # seconds
|
|
FEED_TITLE = "Eesti.ee Sitrep Events"
|
|
FEED_DESC = "Estonian situational awareness (Sitrep) emergency events"
|
|
FEED_LINK = "https://api.app.eesti.ee"
|
|
|
|
|
|
def format_rfc822(dt_str: str) -> str:
|
|
dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
|
|
return dt.strftime("%a, %d %b %Y %H:%M:%S %z")
|
|
|
|
|
|
def text_for_lang(content_list: list, lang: str) -> str:
|
|
for item in content_list:
|
|
if item.get("languageCode") == lang:
|
|
return item.get("text", "")
|
|
for item in content_list:
|
|
return item.get("text", "")
|
|
return ""
|
|
|
|
|
|
def title_for_lang(content_list: list, lang: str) -> str:
|
|
for item in content_list:
|
|
if item.get("languageCode") == lang:
|
|
return item.get("title", "") or ""
|
|
for item in content_list:
|
|
return item.get("title", "") or ""
|
|
return ""
|
|
|
|
|
|
def build_feed_xml(events: list) -> str:
|
|
now_rfc822 = format_rfc822(datetime.now(timezone.utc).isoformat())
|
|
|
|
lines = [
|
|
'<?xml version="1.0" encoding="utf-8"?>',
|
|
'<rss version="2.0">',
|
|
' <channel>',
|
|
f" <title>{escape(FEED_TITLE)}</title>",
|
|
f" <link>{escape(FEED_LINK)}</link>",
|
|
f" <description>{escape(FEED_DESC)}</description>",
|
|
f" <lastBuildDate>{now_rfc822}</lastBuildDate>",
|
|
]
|
|
|
|
seen = set()
|
|
|
|
for event in events:
|
|
data = event.get("data", {})
|
|
ev = data.get("event", {})
|
|
alerts = data.get("alerts", [])
|
|
|
|
for alert in alerts:
|
|
if alert.get("type") != "MASS_SMS":
|
|
continue
|
|
content = alert.get("content", [])
|
|
if not any(c.get("text") for c in content):
|
|
continue
|
|
alert_text = text_for_lang(content, "et") or text_for_lang(content, "en")
|
|
key = (ev.get("id"), alert_text)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
|
|
alert_title = title_for_lang(content, "et") or title_for_lang(content, "en")
|
|
locs = [l.get("settlementUnit", "") for l in alert.get("ehakLocations", [])]
|
|
state = alert.get("state", "")
|
|
atype = alert.get("type", "")
|
|
|
|
item_title = ev.get("title", "Untitled Event")
|
|
if alert_title:
|
|
item_title = f"{item_title} — {alert_title}"
|
|
elif atype:
|
|
item_title = f"{item_title} — {atype}"
|
|
|
|
lines.append(" <item>")
|
|
lines.append(f" <title>{escape(item_title)}</title>")
|
|
lines.append(f" <link>{escape(FEED_LINK)}</link>")
|
|
lines.append(f' <guid isPermaLink="false">{alert.get("id", "")}</guid>')
|
|
|
|
start = alert.get("startDate", "") or ev.get("startDate", "")
|
|
if start:
|
|
lines.append(f" <pubDate>{format_rfc822(start)}</pubDate>")
|
|
|
|
desc_parts = [alert_text]
|
|
if locs:
|
|
desc_parts.append(f"Affected areas: {', '.join(locs)}")
|
|
|
|
lines.append(f" <description>{escape('<br>'.join(desc_parts))}</description>")
|
|
lines.append(f" <category>{escape(f'{atype}, {state}')}</category>")
|
|
lines.append(" </item>")
|
|
|
|
lines.append(" </channel>")
|
|
lines.append("</rss>")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def main():
|
|
last_etag = None
|
|
|
|
while True:
|
|
try:
|
|
headers = {"Accept-Language": "et"}
|
|
if last_etag:
|
|
headers["If-None-Match"] = last_etag
|
|
|
|
resp = requests.get(API_URL, headers=headers, timeout=30)
|
|
if resp.status_code == 304:
|
|
print(f"[{datetime.now():%H:%M:%S}] No changes (304)")
|
|
elif resp.status_code == 200:
|
|
events = resp.json()
|
|
xml = build_feed_xml(events)
|
|
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
|
|
f.write(xml)
|
|
print(f"[{datetime.now():%H:%M:%S}] Wrote {len(events)} events to {OUTPUT_FILE}")
|
|
else:
|
|
print(f"[{datetime.now():%H:%M:%S}] HTTP {resp.status_code}")
|
|
|
|
last_etag = resp.headers.get("ETag")
|
|
except Exception as e:
|
|
print(f"[{datetime.now():%H:%M:%S}] Error: {e}", file=sys.stderr)
|
|
|
|
time.sleep(POLL_INTERVAL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|