アマチュア無線ログブック同期API

# Upload a single QSO
curl -X POST https://api.dxqso.net/upload \
     -H "Authorization: Bearer $VENDOR_TOKEN" \
     -H "dxqso-station-token: Bearer $STATION_TOKEN" \
     -H "Content-Type: application/json" \
     -d '{"qsos":[{"call":"VK2ABC","band":"40m","mode":"FT8","qso_date":"20240802","time_on":"015115"}]}'
# → returns each record's stable qso_id:
#   {"inserted":1,"updated":0,"rejected":0,"results":[{"action":"inserted","qso_id":"898193a7-…"}]}
#   Store qso_id to reference (or later update) the record.

# Upload an ADIF file (up to 2000 QSOs)
curl -X POST https://api.dxqso.net/upload \
     -H "Authorization: Bearer $VENDOR_TOKEN" \
     -H "dxqso-station-token: Bearer $STATION_TOKEN" \
     -H "Content-Type: text/plain" \
     --data-binary @my_log.adi

# Incremental sync — fetch only QSOs changed since a date
curl "https://api.dxqso.net/logbook/download?format=json&qso_since=2025-01-01T00:00:00" \
     -H "Authorization: Bearer $VENDOR_TOKEN" \
     -H "dxqso-station-token: Bearer $STATION_TOKEN"

import requests
from typing import Optional

class DxqsoClient:
    API_BASE = "https://api.dxqso.net"

    def __init__(self, vendor_token: str, station_token: str):
        self._vendor_token  = vendor_token
        self._station_token = station_token
        self._session       = requests.Session()

    def _auth(self, next_token=None) -> dict:
        h = {"Authorization": f"Bearer {self._vendor_token}",
             "dxqso-station-token": f"Bearer {self._station_token}"}
        if next_token: h["X-Next-Token"] = next_token
        return h

    def upload_qso(self, call, band, mode, qso_date, time_on):
        qso = {"call": call, "band": band, "mode": mode,
               "qso_date": qso_date, "time_on": time_on}
        self._session.post(f"{self.API_BASE}/upload",
                           headers=self._auth(), json={"qsos": [qso]}).raise_for_status()

    def download_logbook(self, since=None):
        all_qsos, next_token = [], None
        while True:
            params = {"format": "json"}
            if since: params["qso_since"] = since
            r = self._session.get(f"{self.API_BASE}/logbook/download",
                                  headers=self._auth(next_token), params=params)
            r.raise_for_status()
            all_qsos += r.json().get("qsos", [])
            next_token = r.headers.get("X-Next-Token")
            if not next_token: break
        return all_qsos