|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +import asyncio |
| 4 | +from dataclasses import dataclass |
| 5 | +import urllib.parse |
| 6 | +import urllib.request |
| 7 | +import xml.etree.ElementTree as ET |
| 8 | + |
| 9 | +from .config import QRZConfig |
| 10 | +from .__init__ import __version__ |
| 11 | + |
| 12 | + |
| 13 | +class QRZLookupError(RuntimeError): |
| 14 | + pass |
| 15 | + |
| 16 | + |
| 17 | +@dataclass(slots=True) |
| 18 | +class QRZLookupResult: |
| 19 | + callsign: str |
| 20 | + fname: str = "" |
| 21 | + name: str = "" |
| 22 | + addr1: str = "" |
| 23 | + addr2: str = "" |
| 24 | + state: str = "" |
| 25 | + country: str = "" |
| 26 | + grid: str = "" |
| 27 | + county: str = "" |
| 28 | + lat: str = "" |
| 29 | + lon: str = "" |
| 30 | + dxcc: str = "" |
| 31 | + cqzone: str = "" |
| 32 | + ituzone: str = "" |
| 33 | + aliases: str = "" |
| 34 | + |
| 35 | + |
| 36 | +class QRZClient: |
| 37 | + def __init__(self, config: QRZConfig) -> None: |
| 38 | + self._config = config |
| 39 | + self._session_key = "" |
| 40 | + self._lock = asyncio.Lock() |
| 41 | + |
| 42 | + def configured(self) -> bool: |
| 43 | + return bool(self._config.username.strip() and self._config.password.strip()) |
| 44 | + |
| 45 | + async def lookup(self, callsign: str) -> QRZLookupResult | None: |
| 46 | + if not self.configured(): |
| 47 | + raise QRZLookupError("QRZ lookup is not configured on this node.") |
| 48 | + async with self._lock: |
| 49 | + return await self._lookup_locked(callsign.upper()) |
| 50 | + |
| 51 | + async def _lookup_locked(self, callsign: str) -> QRZLookupResult | None: |
| 52 | + if not self._session_key: |
| 53 | + self._session_key = await self._login_locked() |
| 54 | + payload = self._fetch_xml({"s": self._session_key, "callsign": callsign}) |
| 55 | + session = payload.get("session", {}) |
| 56 | + error = session.get("error", "") |
| 57 | + if error and any(token in error.lower() for token in ("session", "invalid", "timeout")): |
| 58 | + self._session_key = await self._login_locked() |
| 59 | + payload = self._fetch_xml({"s": self._session_key, "callsign": callsign}) |
| 60 | + session = payload.get("session", {}) |
| 61 | + error = session.get("error", "") |
| 62 | + if error: |
| 63 | + if "not found" in error.lower(): |
| 64 | + return None |
| 65 | + raise QRZLookupError(error) |
| 66 | + data = payload.get("callsign", {}) |
| 67 | + if not data: |
| 68 | + return None |
| 69 | + return QRZLookupResult( |
| 70 | + callsign=data.get("call", callsign), |
| 71 | + fname=data.get("fname", ""), |
| 72 | + name=data.get("name", ""), |
| 73 | + addr1=data.get("addr1", ""), |
| 74 | + addr2=data.get("addr2", ""), |
| 75 | + state=data.get("state", ""), |
| 76 | + country=data.get("country", ""), |
| 77 | + grid=data.get("grid", ""), |
| 78 | + county=data.get("county", ""), |
| 79 | + lat=data.get("lat", ""), |
| 80 | + lon=data.get("lon", ""), |
| 81 | + dxcc=data.get("dxcc", ""), |
| 82 | + cqzone=data.get("cqzone", ""), |
| 83 | + ituzone=data.get("ituzone", ""), |
| 84 | + aliases=data.get("aliases", ""), |
| 85 | + ) |
| 86 | + |
| 87 | + async def _login_locked(self) -> str: |
| 88 | + payload = self._fetch_xml( |
| 89 | + { |
| 90 | + "username": self._config.username.strip(), |
| 91 | + "password": self._config.password, |
| 92 | + "agent": self._config.agent.strip() or f"pyCluster/{__version__}", |
| 93 | + } |
| 94 | + ) |
| 95 | + session = payload.get("session", {}) |
| 96 | + key = session.get("key", "") |
| 97 | + error = session.get("error", "") |
| 98 | + if error: |
| 99 | + raise QRZLookupError(error) |
| 100 | + if not key: |
| 101 | + raise QRZLookupError("QRZ login did not return a session key.") |
| 102 | + return key |
| 103 | + |
| 104 | + def _fetch_xml(self, params: dict[str, str]) -> dict[str, dict[str, str]]: |
| 105 | + query = urllib.parse.urlencode(params) |
| 106 | + req = urllib.request.Request( |
| 107 | + f"{self._config.api_url}?{query}", |
| 108 | + headers={"User-Agent": self._config.agent.strip() or f"pyCluster/{__version__}"}, |
| 109 | + ) |
| 110 | + with urllib.request.urlopen(req, timeout=10) as resp: |
| 111 | + xml_bytes = resp.read() |
| 112 | + root = ET.fromstring(xml_bytes) |
| 113 | + return _xml_payload(root) |
| 114 | + |
| 115 | + |
| 116 | +def _xml_payload(root: ET.Element) -> dict[str, dict[str, str]]: |
| 117 | + out: dict[str, dict[str, str]] = {} |
| 118 | + for child in root: |
| 119 | + section = _local_tag(child.tag) |
| 120 | + values: dict[str, str] = {} |
| 121 | + for node in child: |
| 122 | + values[_local_tag(node.tag)] = (node.text or "").strip() |
| 123 | + out[section] = values |
| 124 | + return out |
| 125 | + |
| 126 | + |
| 127 | +def _local_tag(tag: str) -> str: |
| 128 | + if "}" in tag: |
| 129 | + tag = tag.rsplit("}", 1)[1] |
| 130 | + return tag.lower() |
0 commit comments