turboejer_to_geo

turboejer_to_geom.py

#!/usr/bin/env python3
"""Download Danske Tursejlere / Blå oplevelser map items as KML, GeoJSON or GPX.

Data source: the public JSON API used by the "Blå oplevelser" mobile app,
    https://tursejler.eu/api/buoys/
This is much more stable than the old WordPress admin-ajax route — no per-page
nonce to scrape, no session, and the response already carries decimal lat/lon.
A bare GET on /api/buoys/ returns *every* map item (buoys "Turbøje", harbours
"Havn", lunch jetties "Frokostbro"); we filter by category/owner locally.
(/api/buoys/?buoy=true would return only the buoys.)

Requests carry the same fingerprint the app itself uses — okhttp/4.9.2, the
HTTP client bundled in the v2.0.3 APK, issuing a bare GET with no extra headers
— so they are indistinguishable from normal app traffic.

Writes turboejer-YYYYMMDDhhmm.{kml,geojson,gpx} (download time, UTC) to the
current directory.

Change-check: before writing, each output is compared against the newest
prior file of the same kind (ignoring the generation timestamp). If the buoy
data is identical, the new file is skipped and the matching file is named
instead, so unchanged runs don't accumulate duplicate files. Pass --force to
write unconditionally.

Usage:
    python3 turboejer_to_geom.py            # one combined pair, entire set
    python3 turboejer_to_geom.py --split    # split per category, buoys by
                                            # provenience (tursejler/sejlunion)
    python3 turboejer_to_geom.py --split --zip   # also bundle into a .zip
"""

import argparse
import glob
import gzip
import json
import os
import re
import sys
import unicodedata
import urllib.request
import zipfile
from datetime import datetime, timezone
from xml.sax.saxutils import escape

API_URL = "https://tursejler.eu/api/buoys/"

# The app is an Expo/React-Native build; its networking goes through okhttp
# 4.9.2 (the version shipped in the v2.0.3 APK) and it fetches the buoy feed
# with a bare GET — no custom headers. Replicating okhttp's default request
# fingerprint makes our traffic match the app's. (okhttp adds Accept-Encoding:
# gzip itself and decompresses transparently; we do the same below.)
UA = "okhttp/4.9.2"
APP_HEADERS = {
    "User-Agent": UA,
    "Accept-Encoding": "gzip",
    "Connection": "Keep-Alive",
}


def _get(url: str, timeout: float) -> bytes:
    """GET *url* with the app's okhttp fingerprint, gunzipping if needed."""
    req = urllib.request.Request(url, headers=APP_HEADERS)
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        raw = resp.read()
        if resp.headers.get("Content-Encoding", "").lower() == "gzip":
            raw = gzip.decompress(raw)
    return raw


def fetch_items(timeout: float = 30.0) -> list:
    """Fetch the app's full map-item feed (buoys, harbours, bridges, …).

    The API returns every map item in one JSON array; we always pull the
    whole set and let the caller split it locally.
    """
    items = json.loads(_get(API_URL, timeout).decode("utf-8", "replace"))
    if not isinstance(items, list):
        raise RuntimeError("Unexpected API response: %.120r" % (items,))
    return items


def owner_slug(owner: str) -> str:
    """Turn an owner name into a filesystem-safe filename fragment.

    "DANSKE TURSEJLERE" -> "danske-tursejlere"; empty/unknown -> "unknown".
    """
    norm = unicodedata.normalize("NFKD", owner or "")
    norm = norm.encode("ascii", "ignore").decode("ascii").lower()
    slug = re.sub(r"[^a-z0-9]+", "-", norm).strip("-")
    return slug or "unknown"


def group_by(items: list, key: str) -> dict:
    """Group items into {value: [item, ...]} by item[key], first-seen order."""
    groups: dict = {}
    for it in items:
        groups.setdefault(it.get(key) or "", []).append(it)
    return groups


# The feed's category_name values mapped to stable English filename slugs and
# human labels. Unknown categories fall back to a slug of the raw name.
BUOY_CATEGORY = "Turbøje"
CATEGORY_INFO = {
    "Havn": ("harbours", "Havne"),
    "Frokostbro": ("bridges", "Frokostbroer"),
    "Turbøje": ("buoys", "Turbøjer"),
}

# Buoy owners shortened to a "provenience" slug, per the app's two providers.
# Anything else falls back to a slug of the owner name.
PROVENANCE = {
    "DANSKE TURSEJLERE": "tursejler",
    "DANSK SEJLUNION": "sejlunion",
}


def category_slug(name: str) -> str:
    """Filename slug for a category_name ('Havn' -> 'harbours')."""
    info = CATEGORY_INFO.get(name)
    return info[0] if info else owner_slug(name)


def category_label(name: str) -> str:
    """Human label for a category_name ('Havn' -> 'Havne')."""
    info = CATEGORY_INFO.get(name)
    return info[1] if info else (name or "Ukendt")


def provenance_slug(owner: str) -> str:
    """Short provenience slug for a buoy owner ('DANSKE TURSEJLERE' ->
    'tursejler'); unknown owners fall back to a full owner slug."""
    return PROVENANCE.get((owner or "").strip().upper()) or owner_slug(owner)


def build_outputs(items: list, split: bool) -> list:
    """Build the list of (filename-infix, subset, KML doc name) to write.

    Without --split: one combined file pair for the entire feed.
    With --split: one pair per category (harbours, bridges, …), and the buoy
    category further split by owner/provenience (buoys-tursejler, …).
    """
    if not split:
        return [("", items, "Blå oplevelser – alle kortpunkter")]

    outputs = []
    # dict.fromkeys preserves first-seen category order across the feed.
    for cat in dict.fromkeys(b.get("category_name") for b in items):
        subset = [b for b in items if b.get("category_name") == cat]
        cslug, clabel = category_slug(cat), category_label(cat)
        if cat == BUOY_CATEGORY:
            for owner, osub in group_by(subset, "owner").items():
                outputs.append(
                    ("-%s-%s" % (cslug, provenance_slug(owner)), osub,
                     "Blå oplevelser – %s (%s)" % (clabel, owner or "ukendt")))
        else:
            outputs.append(("-%s" % cslug, subset,
                            "Blå oplevelser – %s" % clabel))
    return outputs


# A filename stamp is exactly YYYYMMDDhhmm (12 digits). Glob it precisely so the
# combined-file pattern ("turboejer-<stamp>") never swallows a split file
# ("turboejer-harbours-<stamp>", "turboejer-buoys-tursejler-<stamp>", …).
STAMP_GLOB = "[0-9]" * 12

# The only per-run-volatile text is the generation datum, an ISO-8601 UTC
# timestamp. Buoy data carries no such field, so blanking it lets two runs
# compare equal when the underlying buoy data is unchanged (the buoy *count*,
# embedded separately in the KML description, still counts as a difference).
_TS_RE = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00")


def canonical(text: str) -> str:
    """Strip the volatile generation timestamp for change comparison."""
    return _TS_RE.sub("<generated>", text)


def latest_previous(out_dir: str, infix: str, ext: str,
                    exclude_stamp: str) -> str:
    """Most recent prior output for this owner/ext in out_dir, or "" if none.

    Stamps sort lexicographically the same as chronologically, so max() of the
    matching paths is the newest.
    """
    pattern = os.path.join(out_dir, "turboejer%s-%s.%s" % (infix, STAMP_GLOB, ext))
    current = os.path.join(out_dir, "turboejer%s-%s.%s" % (infix, exclude_stamp, ext))
    candidates = [p for p in glob.glob(pattern) if p != current]
    return max(candidates) if candidates else ""


def buoy_rows(b: dict) -> list:
    """Return the (label, value) description rows for a buoy, blanks dropped."""
    rows = [
        ("Number", b.get("number")),
        ("Owner", b.get("owner")),
        ("Region", b.get("region")),
        ("Area", b.get("area")),
        ("DMS", b.get("dms")),
        ("DDM", b.get("ddm")),
        ("Category", b.get("category_name")),
        ("Active", "yes" if b.get("active") else "no"),
    ]
    return [(k, v) for k, v in rows if v not in (None, "", [])]


def buoy_description(b: dict) -> str:
    """Build an HTML description block for a buoy's KML placemark."""
    return "<br/>".join("%s: %s" % (k, v) for k, v in buoy_rows(b))


def to_kml(buoys: list, updated: str,
           doc_name: str = "Blå oplevelser – alle kortpunkter") -> str:
    """Render a list of buoy dicts as a KML document string."""
    out = [
        '<?xml version="1.0" encoding="UTF-8"?>',
        '<kml xmlns="http://www.opengis.net/kml/2.2"'
        ' xmlns:atom="http://www.w3.org/2005/Atom">',
        "  <Document>",
        "    <name>%s</name>" % escape(doc_name),
        "    <atom:updated>%s</atom:updated>" % updated,
        "    <description><![CDATA[Generated %s — %d buoys]]></description>"
        % (updated, len(buoys)),
    ]
    for b in buoys:
        lat, lon = b.get("lat"), b.get("lon")
        if lat is None or lon is None:
            continue
        name = escape(str(b.get("name") or b.get("id") or "Buoy"))
        desc = buoy_description(b)
        out += [
            "    <Placemark>",
            "      <name>%s</name>" % name,
            "      <description><![CDATA[%s]]></description>" % desc,
            "      <Point><coordinates>%s,%s,0</coordinates></Point>"
            % (lon, lat),  # KML order is lon,lat,alt
            "    </Placemark>",
        ]
    out += ["  </Document>", "</kml>", ""]
    return "\n".join(out)


def to_geojson(buoys: list, updated: str) -> str:
    """Render a list of buoy dicts as an RFC-7946 GeoJSON FeatureCollection."""
    features = []
    for b in buoys:
        lat, lon = b.get("lat"), b.get("lon")
        if lat is None or lon is None:
            continue
        # Every non-geometry field becomes a GeoJSON property.
        props = {k: v for k, v in b.items() if k not in ("lat", "lon")}
        features.append({
            "type": "Feature",
            "geometry": {
                "type": "Point",
                "coordinates": [lon, lat],  # GeoJSON order is lon,lat
            },
            "properties": props,
        })
    collection = {
        "type": "FeatureCollection",
        "generated": updated,  # update datum
        "features": features,
    }
    return json.dumps(collection, ensure_ascii=False, indent=2)


def to_gpx(buoys: list, updated: str,
           doc_name: str = "Blå oplevelser – alle kortpunkter") -> str:
    """Render a list of buoy dicts as a GPX 1.1 document of waypoints."""
    out = [
        '<?xml version="1.0" encoding="UTF-8"?>',
        '<gpx version="1.1" creator="turboejer_to_geom.py"'
        ' xmlns="http://www.topografix.com/GPX/1/1"'
        ' xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"'
        ' xsi:schemaLocation="http://www.topografix.com/GPX/1/1'
        ' http://www.topografix.com/GPX/1/1/gpx.xsd">',
        "  <metadata>",
        "    <name>%s</name>" % escape(doc_name),
        "    <desc>Generated %s — %d buoys</desc>" % (updated, len(buoys)),
        "    <time>%s</time>" % updated,
        "  </metadata>",
    ]
    for b in buoys:
        lat, lon = b.get("lat"), b.get("lon")
        if lat is None or lon is None:
            continue
        name = escape(str(b.get("name") or b.get("id") or "Buoy"))
        # GPX <desc> is plain text; join the rows with newlines.
        desc = escape("\n".join("%s: %s" % (k, v) for k, v in buoy_rows(b)))
        out += [
            '    <wpt lat="%s" lon="%s">' % (lat, lon),  # GPX attr order is lat,lon
            "      <name>%s</name>" % name,
            "      <desc>%s</desc>" % desc,
            "    </wpt>",
        ]
    out += ["</gpx>", ""]
    return "\n".join(out)


def main() -> int:
    p = argparse.ArgumentParser(description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--split", action="store_true",
                   help="Write one file pair per category "
                        "(turboejer-harbours-STAMP, turboejer-bridges-STAMP, …) "
                        "and split buoys by provenience "
                        "(turboejer-buoys-tursejler-STAMP, …). "
                        "Default: one combined file pair with the entire set.")
    p.add_argument("--zip", action="store_true",
                   help="Bundle this run's resulting files (written or kept) "
                        "into turboejer-STAMP.zip in the output directory")
    p.add_argument("--force", action="store_true",
                   help="Always write, even if unchanged since the last run")
    p.add_argument("--out-dir", default=".",
                   help="Directory to read prior files from and write to "
                        "(default: current directory; created if missing)")
    args = p.parse_args()

    now = datetime.now(timezone.utc)
    stamp = now.strftime("%Y%m%d%H%M")              # filename datum: YYYYMMDDhhmm
    updated = now.replace(microsecond=0).isoformat()  # in-document datum

    try:
        items = fetch_items()
        print("Fetched %d items" % len(items), file=sys.stderr)
    except Exception as e:
        print("Error: %s" % e, file=sys.stderr)
        return 1

    # Each output is (filename infix, item subset, KML document name).
    outputs = build_outputs(items, args.split)
    if args.split:
        print("Splitting into %d file pair(s)" % len(outputs), file=sys.stderr)

    os.makedirs(args.out_dir, exist_ok=True)

    produced = []  # resulting file of this run (freshly written or kept)
    for infix, subset, doc_name in outputs:
        for ext, render in (("kml", to_kml), ("geojson", to_geojson),
                            ("gpx", to_gpx)):
            if ext == "geojson":
                content = render(subset, updated)
            else:
                content = render(subset, updated, doc_name=doc_name)

            # Change-check: if the newest prior file holds the same data
            # (ignoring the generation timestamp), keep it and skip writing.
            prev = "" if args.force else latest_previous(args.out_dir, infix,
                                                          ext, stamp)
            if prev:
                with open(prev, encoding="utf-8") as f:
                    prev_content = f.read()
                if canonical(content) == canonical(prev_content):
                    print("Unchanged (%d items) — keeping %s"
                          % (len(subset), prev), file=sys.stderr)
                    produced.append(prev)
                    continue

            path = os.path.join(args.out_dir,
                                "turboejer%s-%s.%s" % (infix, stamp, ext))
            with open(path, "w", encoding="utf-8") as f:
                f.write(content)
            print("Wrote %s (%d items)" % (path, len(subset)), file=sys.stderr)
            produced.append(path)

    if args.zip and produced:
        zip_path = os.path.join(args.out_dir, "turboejer-%s.zip" % stamp)
        with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
            for p in produced:
                zf.write(p, arcname=os.path.basename(p))
        print("Zipped %d file(s) into %s" % (len(produced), zip_path),
              file=sys.stderr)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())