From 0958d26e9f149bea9d27b2985958cefae0c931d2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:11:34 +0300 Subject: [PATCH] feat: add nowcast point alarm page client with SQLite store Adds point_alarm.py (flattening, secret masking, SQLite upsert store, paged fetch with 403 fallback, CSV/TXT export, shared filter loader), nowcast-point-page.py CLI and unit tests. Register will read the shared filter from nowcast-point-alarm-filter.json instead of repeating it in the CSV. --- .gitignore | 5 + geo_alarm.py | 9 +- nowcast-point-alarm-filter.example.json | 16 + nowcast-point-page.py | 139 +++++++ nowcast_common.py | 13 +- point_alarm.py | 512 ++++++++++++++++++++++++ tests/test_point_alarm.py | 127 ++++++ 7 files changed, 813 insertions(+), 8 deletions(-) create mode 100644 nowcast-point-alarm-filter.example.json create mode 100644 nowcast-point-page.py create mode 100644 point_alarm.py create mode 100644 tests/test_point_alarm.py diff --git a/.gitignore b/.gitignore index 71c6176..895bca1 100644 --- a/.gitignore +++ b/.gitignore @@ -36,3 +36,8 @@ dist/ .vscode/ .DS_Store Thumbs.db + +# Point alarm client artifacts +*.db +nowcast-point-alarm-filter.json +nowcast-point-*-backup-* diff --git a/geo_alarm.py b/geo_alarm.py index 88e23d4..c47461a 100644 --- a/geo_alarm.py +++ b/geo_alarm.py @@ -18,6 +18,7 @@ from nowcast_common import ( read_json_object, response_warnings, as_uuid, + fetch_account_id, load_webhook_config, write_csv_atomic, ) @@ -86,13 +87,7 @@ class GeoAlarmService: self.client = client def get_account_id(self) -> str: - response = self.client.request_json("GET", "/v1/users/me") - data = response.data - account = data.get("account") if isinstance(data, dict) else None - account_id = account.get("id") if isinstance(account, dict) else None - if not as_uuid(account_id): - raise ApiError("/v1/users/me yanıtında geçerli account.id bulunamadı.") - return account_id + return fetch_account_id(self.client) def list_cities(self) -> list[dict[str, Any]]: data = self.client.request_json( diff --git a/nowcast-point-alarm-filter.example.json b/nowcast-point-alarm-filter.example.json new file mode 100644 index 0000000..4fdcb7a --- /dev/null +++ b/nowcast-point-alarm-filter.example.json @@ -0,0 +1,16 @@ +{ + "lightning": { + "type": "FLASH_CLOUD_TO_GROUND", + "peakCurrent": -12.5, + "inCloudHeight": 4500 + }, + "thunderstorm": { + "intersectsAffectedPolygon": true, + "intersectsCellPolygon": false, + "severityThreshold": "MEDIUM", + "speedThreshold": 20 + }, + "precipitation": { + "intensities": ["LIGHT", "MODERATE"] + } +} diff --git a/nowcast-point-page.py b/nowcast-point-page.py new file mode 100644 index 0000000..fd477c1 --- /dev/null +++ b/nowcast-point-page.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +"""List point alarm registrations page by page into SQLite, screen or CSV/TXT.""" + +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path + +from common import ApiError, ConfigurationError, IklimClient, configure_logging +from point_alarm import ( + MAX_PAGE_SIZE, + PointAlarmService, + PointStore, + fetch_pages, + parse_polygon, + print_records, + prompt_export, + timestamped_path, + write_records, +) + + +BASE_DIR = Path(__file__).resolve().parent + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "iklim.co nokta alarm kayıtlarını /v1/alarms/points/page ile listeler. " + "Sonuçlar SQLite'a (registrationId anahtarlı) yazılır; varsayılan olarak ekrana " + "basılır, --output ile CSV/TXT'ye yazılır." + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + "--account-id", + help="Sorgulanacak hesap ID'si (varsayılan: /v1/users/me içindeki account.id)", + ) + parser.add_argument( + "--recipient-id", + action="append", + default=[], + help="Alıcı ID'sine göre süz; birden fazla kez verilebilir", + ) + parser.add_argument( + "--polygon", + help="Poligon kesişim filtresi: 'lng,lat;lng,lat;...' (en az 4 nokta, kapalı halka)", + ) + parser.add_argument( + "--page-number", + type=int, + help="Yalnızca bu sayfayı getir (0'dan başlar); verilmezse tüm sayfalar gezilir", + ) + parser.add_argument( + "--page-size", + type=int, + default=MAX_PAGE_SIZE, + help=f"Sayfa boyutu, 1-{MAX_PAGE_SIZE} (varsayılan: {MAX_PAGE_SIZE})", + ) + parser.add_argument( + "--output", + type=Path, + help=( + "Çıktı dosyası (.csv noktalı virgüllü, .txt JSON Lines); dosya adına " + "zaman soneki eklenir (dosya-YYYYmmdd-HHMMSS.csv). Verilmezse ekrana yazılır" + ), + ) + parser.add_argument( + "--db", + type=Path, + default=BASE_DIR / "iklim-sandbox.db", + help="SQLite dosyası (varsayılan: iklim-sandbox.db)", + ) + parser.add_argument( + "--debug-log", + type=Path, + default=BASE_DIR / "logs" / "iklim-api-debug.log", + help="Maskelenmiş HTTP debug log dosyası", + ) + args = parser.parse_args() + if not 1 <= args.page_size <= MAX_PAGE_SIZE: + parser.error(f"--page-size 1-{MAX_PAGE_SIZE} aralığında olmalıdır") + if args.page_number is not None and args.page_number < 0: + parser.error("--page-number 0 veya daha büyük olmalıdır") + return args + + +def main() -> int: + args = parse_args() + configure_logging(args.debug_log) + store = None + try: + polygon = parse_polygon(args.polygon) if args.polygon else None + client = IklimClient.from_env(BASE_DIR / ".env") + service = PointAlarmService(client) + account_id = args.account_id or service.get_account_id() + store = PointStore(args.db) + summary = fetch_pages( + service, + store, + account_id=account_id, + recipient_ids=args.recipient_id or None, + polygon=polygon, + page_size=args.page_size, + page_number=args.page_number, + ) + records = store.records(summary.account_id, summary.registration_ids) + webhooks = store.webhooks(summary.account_id) + if args.output: + path = timestamped_path(args.output) + write_records(path, records, webhooks) + print(f"{len(records)} kayıt yazıldı: {path.resolve()}") + else: + print_records(records, webhooks) + if args.page_number is None and summary.missing: + logging.warning( + "Sunucu %s kayıt bildirdi, %s benzersiz kayıt alındı (%s eksik). Sayfalama sırasında " + "kayıt eklenmiş/silinmiş olabilir; komutu yeniden çalıştırın.", + summary.total_records, + summary.distinct, + summary.missing, + ) + if summary.duplicates: + logging.warning("Sayfalar arasında %s tekrar eden kayıt görüldü.", summary.duplicates) + if not args.output: + prompt_export(store, summary.account_id) + except (ConfigurationError, ApiError, OSError) as exc: + logging.error("İşlem durduruldu: %s", exc) + return 1 + finally: + if store: + store.close() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/nowcast_common.py b/nowcast_common.py index f66a031..8999993 100644 --- a/nowcast_common.py +++ b/nowcast_common.py @@ -12,7 +12,7 @@ from typing import Any from urllib.parse import urlsplit from uuid import UUID -from common import ConfigurationError +from common import ApiError, ConfigurationError, IklimClient AUTHENTICATION_FIELDS = { @@ -154,3 +154,14 @@ def write_csv_atomic( finally: if temp_path and temp_path.exists(): temp_path.unlink() + + +def fetch_account_id(client: IklimClient) -> str: + """Return the caller's account id from /v1/users/me.""" + response = client.request_json("GET", "/v1/users/me") + data = response.data + account = data.get("account") if isinstance(data, dict) else None + account_id = account.get("id") if isinstance(account, dict) else None + if not as_uuid(account_id): + raise ApiError("/v1/users/me yanıtında geçerli account.id bulunamadı.") + return account_id diff --git a/point_alarm.py b/point_alarm.py new file mode 100644 index 0000000..770f0cc --- /dev/null +++ b/point_alarm.py @@ -0,0 +1,512 @@ +"""Point alarm API operations, record flattening and the local SQLite store.""" + +from __future__ import annotations + +import csv +import json +import logging +import sqlite3 +import sys +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Iterable, TextIO +from urllib.parse import quote + +from common import ApiError, ConfigurationError, IklimClient +from common.client import ApiResponse +from nowcast_common import fetch_account_id, read_json_object + + +LOGGER = logging.getLogger(__name__) + +ALARMS_PATH = "/v1/alarms/points" +MAX_RADIUS_METERS = 50000.0 +MAX_PAGE_SIZE = 100 +SECRET_KEYS = {"password", "jwttoken", "apikey", "secret"} +MASK = "***" + +LIGHTNING_TYPES = {"FLASH_CLOUD_TO_GROUND", "PULSE_IN_CLOUD"} +SEVERITIES = {"LOW", "MEDIUM", "HIGH"} +INTENSITIES = {"DRIZZLE", "LIGHT", "MODERATE", "HEAVY", "VERY_HEAVY", "EXTREME"} + +FILTER_COLUMNS = [ + "lightningType", + "peakCurrent", + "inCloudHeight", + "intersectsAffectedPolygon", + "intersectsCellPolygon", + "severityThreshold", + "speedThreshold", + "precipitationIntensities", +] +RECORD_COLUMNS = ["registrationId", "recipientId", "lat", "lng", "radius"] + FILTER_COLUMNS +STORE_COLUMN_TYPES = { + "registrationId": "TEXT PRIMARY KEY", + "accountId": "TEXT", + "recipientId": "TEXT", + "lat": "REAL", + "lng": "REAL", + "radius": "REAL", + "lightningType": "TEXT", + "peakCurrent": "REAL", + "inCloudHeight": "REAL", + "intersectsAffectedPolygon": "TEXT", + "intersectsCellPolygon": "TEXT", + "severityThreshold": "TEXT", + "speedThreshold": "REAL", + "precipitationIntensities": "TEXT", + "fetchedAt": "TEXT", +} + + +def load_point_alarm_filter(path: Path) -> dict[str, Any]: + """Load and validate the shared filter applied to every registered point.""" + config = read_json_object(path, "Nowcast point alarm filter config") + if not config: + raise ConfigurationError("Nowcast point alarm filter config boş olamaz.") + for section in ("lightning", "thunderstorm", "precipitation"): + if config.get(section) is not None and not isinstance(config[section], dict): + raise ConfigurationError(f"filter.{section} JSON nesnesi olmalıdır.") + lightning = config.get("lightning") or {} + if lightning.get("type") is not None and lightning["type"] not in LIGHTNING_TYPES: + raise ConfigurationError( + "filter.lightning.type FLASH_CLOUD_TO_GROUND veya PULSE_IN_CLOUD olmalıdır." + ) + thunderstorm = config.get("thunderstorm") + if thunderstorm is not None and thunderstorm.get("severityThreshold") not in SEVERITIES: + raise ConfigurationError( + "filter.thunderstorm.severityThreshold LOW, MEDIUM veya HIGH olmalıdır." + ) + precipitation = config.get("precipitation") + if precipitation is not None: + intensities = precipitation.get("intensities") + if not isinstance(intensities, list) or not intensities: + raise ConfigurationError( + "filter.precipitation.intensities boş olmayan bir liste olmalıdır." + ) + invalid = [item for item in intensities if item not in INTENSITIES] + if invalid: + raise ConfigurationError( + "Geçersiz precipitation intensity: " + ", ".join(map(str, invalid)) + ) + return config + + +def now_iso() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def mask_secrets(value: Any) -> Any: + """Return a deep copy of value with webhook credentials replaced by a mask.""" + if isinstance(value, dict): + return { + key: MASK if str(key).lower() in SECRET_KEYS else mask_secrets(item) + for key, item in value.items() + } + if isinstance(value, list): + return [mask_secrets(item) for item in value] + return value + + +def _bool_text(value: Any) -> str | None: + if value is None: + return None + return "true" if value else "false" + + +def flatten_registration(registration: dict[str, Any]) -> dict[str, Any]: + """Flatten an API point registration into one value per RECORD_COLUMNS entry.""" + boundary = registration.get("boundary") or {} + point = boundary.get("point") or {} + filters = registration.get("filter") or {} + lightning = filters.get("lightning") or {} + thunderstorm = filters.get("thunderstorm") or {} + precipitation = filters.get("precipitation") or {} + intensities = precipitation.get("intensities") + return { + "registrationId": registration.get("registrationId"), + "recipientId": registration.get("recipientId"), + "lat": point.get("lat"), + "lng": point.get("lng"), + "radius": boundary.get("radius"), + "lightningType": lightning.get("type"), + "peakCurrent": lightning.get("peakCurrent"), + "inCloudHeight": lightning.get("inCloudHeight"), + "intersectsAffectedPolygon": _bool_text(thunderstorm.get("intersectsAffectedPolygon")), + "intersectsCellPolygon": _bool_text(thunderstorm.get("intersectsCellPolygon")), + "severityThreshold": thunderstorm.get("severityThreshold"), + "speedThreshold": thunderstorm.get("speedThreshold"), + "precipitationIntensities": ( + "|".join(str(item) for item in intensities) if intensities else None + ), + } + + +def registration_webhook(registration: dict[str, Any]) -> dict[str, Any] | None: + webhook = registration.get("webhook") + return mask_secrets(webhook) if isinstance(webhook, dict) else None + + +def timestamped_path(path: Path, now: datetime | None = None) -> Path: + stamp = (now or datetime.now()).strftime("%Y%m%d-%H%M%S") + return path.with_name(f"{path.stem}-{stamp}{path.suffix}") + + +def output_format(path: Path) -> str: + suffix = path.suffix.lower() + if suffix == ".csv": + return "csv" + if suffix == ".txt": + return "txt" + raise ConfigurationError(f"Çıktı dosyası .csv veya .txt uzantılı olmalıdır: {path}") + + +def _cell(value: Any) -> Any: + return "" if value is None else value + + +def write_records( + path: Path, + records: list[dict[str, Any]], + webhooks: list[dict[str, Any]], + *, + columns: list[str] | None = None, +) -> None: + """Write records as CSV (semicolon, webhook preamble) or TXT (JSON Lines).""" + columns = columns or RECORD_COLUMNS + kind = output_format(path) + with path.open("w", encoding="utf-8-sig" if kind == "csv" else "utf-8", newline="") as handle: + if kind == "csv": + writer = csv.writer(handle, delimiter=";") + for webhook in webhooks: + writer.writerow( + ["#webhook", _cell(webhook.get("accountId")), json.dumps(webhook, ensure_ascii=False)] + ) + writer.writerow(columns) + for record in records: + writer.writerow([_cell(record.get(column)) for column in columns]) + else: + for webhook in webhooks: + handle.write(json.dumps({"webhook": webhook}, ensure_ascii=False) + "\n") + for record in records: + handle.write( + json.dumps({column: record.get(column) for column in columns}, ensure_ascii=False) + + "\n" + ) + + +def filter_summary(record: dict[str, Any]) -> str: + parts: list[str] = [] + lightning = [ + str(record[name]) + for name in ("lightningType", "peakCurrent", "inCloudHeight") + if record.get(name) is not None + ] + if lightning: + parts.append("L[" + ",".join(lightning) + "]") + thunderstorm = [ + f"{name}={record[name]}" + for name in ( + "intersectsAffectedPolygon", + "intersectsCellPolygon", + "severityThreshold", + "speedThreshold", + ) + if record.get(name) is not None + ] + if thunderstorm: + parts.append("T[" + ",".join(thunderstorm) + "]") + if record.get("precipitationIntensities"): + parts.append("P[" + str(record["precipitationIntensities"]) + "]") + return " ".join(parts) or "-" + + +def print_records( + records: list[dict[str, Any]], + webhooks: list[dict[str, Any]], + out: TextIO | None = None, +) -> None: + out = out or sys.stdout + for webhook in webhooks: + out.write(f"Webhook (hesap {webhook.get('accountId', '-')}): {json.dumps(webhook, ensure_ascii=False)}\n") + headers = ["registrationId", "recipientId", "lat", "lng", "radius", "filters"] + rows = [ + [ + str(_cell(record.get("registrationId"))), + str(_cell(record.get("recipientId"))), + str(_cell(record.get("lat"))), + str(_cell(record.get("lng"))), + str(_cell(record.get("radius"))), + filter_summary(record), + ] + for record in records + ] + widths = [max(len(headers[i]), *(len(row[i]) for row in rows)) if rows else len(headers[i]) for i in range(len(headers))] + out.write(" ".join(headers[i].ljust(widths[i]) for i in range(len(headers))) + "\n") + for row in rows: + out.write(" ".join(row[i].ljust(widths[i]) for i in range(len(headers))) + "\n") + out.write(f"Toplam: {len(records)} kayıt\n") + + +class PointStore: + """SQLite store keyed by registrationId so repeated fetches never duplicate rows.""" + + def __init__(self, path: Path | str) -> None: + self.path = Path(path) + self.connection = sqlite3.connect(self.path) + self.connection.row_factory = sqlite3.Row + columns = ", ".join(f"{name} {kind}" for name, kind in STORE_COLUMN_TYPES.items()) + self.connection.execute(f"CREATE TABLE IF NOT EXISTS registrations ({columns})") + self.connection.execute( + "CREATE TABLE IF NOT EXISTS webhooks " + "(accountId TEXT PRIMARY KEY, webhook TEXT NOT NULL, fetchedAt TEXT)" + ) + self.connection.commit() + + def close(self) -> None: + self.connection.close() + + def upsert_registration(self, registration: dict[str, Any], account_id: str | None) -> str | None: + """Store one API registration; return its registrationId (None if it has none).""" + record = flatten_registration(registration) + registration_id = record["registrationId"] + if not registration_id: + return None + webhook = registration_webhook(registration) + owner = (webhook or {}).get("accountId") or account_id + values = dict(record, accountId=owner, fetchedAt=now_iso()) + names = list(STORE_COLUMN_TYPES) + updates = ", ".join(f"{name}=excluded.{name}" for name in names if name != "registrationId") + self.connection.execute( + f"INSERT INTO registrations ({', '.join(names)}) " + f"VALUES ({', '.join(':' + name for name in names)}) " + f"ON CONFLICT(registrationId) DO UPDATE SET {updates}", + values, + ) + if webhook and owner: + self.connection.execute( + "INSERT INTO webhooks (accountId, webhook, fetchedAt) VALUES (?, ?, ?) " + "ON CONFLICT(accountId) DO UPDATE SET webhook=excluded.webhook, " + "fetchedAt=excluded.fetchedAt", + (owner, json.dumps(webhook, ensure_ascii=False), now_iso()), + ) + return registration_id + + def commit(self) -> None: + self.connection.commit() + + def count(self, account_id: str | None = None) -> int: + if account_id: + row = self.connection.execute( + "SELECT COUNT(*) FROM registrations WHERE accountId = ?", (account_id,) + ).fetchone() + else: + row = self.connection.execute("SELECT COUNT(*) FROM registrations").fetchone() + return int(row[0]) + + def records( + self, account_id: str | None = None, registration_ids: Iterable[str] | None = None + ) -> list[dict[str, Any]]: + query = "SELECT * FROM registrations" + clauses: list[str] = [] + params: list[Any] = [] + if account_id: + clauses.append("accountId = ?") + params.append(account_id) + ids = list(registration_ids) if registration_ids is not None else None + if ids is not None: + if not ids: + return [] + clauses.append(f"registrationId IN ({', '.join('?' * len(ids))})") + params.extend(ids) + if clauses: + query += " WHERE " + " AND ".join(clauses) + query += " ORDER BY recipientId, registrationId" + return [dict(row) for row in self.connection.execute(query, params)] + + def webhooks(self, account_id: str | None = None) -> list[dict[str, Any]]: + if account_id: + rows = self.connection.execute( + "SELECT webhook FROM webhooks WHERE accountId = ?", (account_id,) + ) + else: + rows = self.connection.execute("SELECT webhook FROM webhooks ORDER BY accountId") + return [json.loads(row["webhook"]) for row in rows] + + +class PointAlarmService: + def __init__(self, client: IklimClient) -> None: + self.client = client + + def get_account_id(self) -> str: + return fetch_account_id(self.client) + + def page( + self, + *, + page_number: int, + page_size: int, + account_id: str | None = None, + recipient_ids: list[str] | None = None, + polygon: list[list[float]] | None = None, + ) -> dict[str, Any]: + body: dict[str, Any] = {"pageNumber": page_number, "pageSize": page_size} + if account_id: + body["accountId"] = account_id + if recipient_ids: + body["filterByRecipientIds"] = recipient_ids + if polygon: + body["filterByPolygonIntersection"] = polygon + data = self.client.request_json("POST", f"{ALARMS_PATH}/page", json_body=body).data + if not isinstance(data, dict): + raise ApiError("Sayfa yanıtı beklenen JSON nesnesini döndürmedi.") + return data + + def get_by_id(self, registration_id: str) -> ApiResponse: + return self.client.request_json( + "GET", f"{ALARMS_PATH}/get-by-id/{quote(registration_id, safe='')}" + ) + + def get_by_recipient_id(self, recipient_id: str) -> ApiResponse: + return self.client.request_json( + "GET", f"{ALARMS_PATH}/get-by-recipient-id/{quote(recipient_id, safe='')}" + ) + + def register(self, payload: dict[str, Any], idempotency_key: str) -> ApiResponse: + return self.client.request_json( + "POST", f"{ALARMS_PATH}/register", json_body=payload, idempotency_key=idempotency_key + ) + + def update(self, payload: dict[str, Any], idempotency_key: str) -> ApiResponse: + return self.client.request_json( + "PATCH", f"{ALARMS_PATH}/update", json_body=payload, idempotency_key=idempotency_key + ) + + def unregister(self, registration_id: str) -> ApiResponse: + return self.client.request_json( + "DELETE", f"{ALARMS_PATH}/unregister/{quote(registration_id, safe='')}" + ) + + +@dataclass +class FetchSummary: + total_records: int = 0 + received: int = 0 + registration_ids: list[str] = field(default_factory=list) + account_id: str | None = None + + @property + def distinct(self) -> int: + return len(self.registration_ids) + + @property + def duplicates(self) -> int: + return self.received - self.distinct + + @property + def missing(self) -> int: + return max(self.total_records - self.distinct, 0) + + +def fetch_pages( + service: PointAlarmService, + store: PointStore, + *, + account_id: str | None, + recipient_ids: list[str] | None = None, + polygon: list[list[float]] | None = None, + page_size: int = MAX_PAGE_SIZE, + page_number: int | None = None, +) -> FetchSummary: + """Fetch one page (page_number given) or every page into the store. + + A 403 for an explicit accountId (caller below POWER_USER) is retried once without + it; the server then scopes the listing to the caller's own account. + """ + summary = FetchSummary(account_id=account_id) + seen: set[str] = set() + current = page_number if page_number is not None else 0 + while True: + try: + data = service.page( + page_number=current, + page_size=page_size, + account_id=summary.account_id, + recipient_ids=recipient_ids, + polygon=polygon, + ) + except ApiError as exc: + if exc.status_code == 403 and summary.account_id: + LOGGER.warning( + "accountId ile sorgu 403 verdi; accountId olmadan, sunucunun kendi hesabınıza " + "daraltmasıyla yeniden deneniyor." + ) + summary.account_id = None + continue + raise + registrations = [item for item in (data.get("registrations") or []) if isinstance(item, dict)] + summary.total_records = int(data.get("totalRecords") or 0) + summary.received += len(registrations) + for registration in registrations: + registration_id = store.upsert_registration(registration, summary.account_id) + if registration_id and registration_id not in seen: + seen.add(registration_id) + summary.registration_ids.append(registration_id) + store.commit() + LOGGER.info( + "Sayfa %s alındı: %s kayıt (toplam %s).", current, len(registrations), summary.total_records + ) + if page_number is not None or not registrations: + break + if (current + 1) * page_size >= summary.total_records: + break + current += 1 + return summary + + +def parse_polygon(text: str) -> list[list[float]]: + """Parse 'lng,lat;lng,lat;...' into a closed ring of at least four points.""" + ring: list[list[float]] = [] + for pair in text.split(";"): + parts = pair.strip().split(",") + if len(parts) != 2: + raise ConfigurationError(f"Poligon noktası 'lng,lat' biçiminde olmalıdır: '{pair.strip()}'.") + try: + lng, lat = float(parts[0]), float(parts[1]) + except ValueError as exc: + raise ConfigurationError(f"Poligon koordinatı sayı olmalıdır: '{pair.strip()}'.") from exc + if not (-180 <= lng <= 180 and -90 <= lat <= 90): + raise ConfigurationError(f"Poligon koordinatı aralık dışında: '{pair.strip()}'.") + ring.append([lng, lat]) + if len(ring) < 4 or ring[0] != ring[-1]: + raise ConfigurationError( + "Poligon en az 4 nokta içermeli ve kapalı olmalıdır (ilk ve son nokta aynı)." + ) + return ring + + +def prompt_export( + store: PointStore, + account_id: str | None, + *, + input_fn: Callable[[str], str] = input, + interactive: bool | None = None, +) -> Path | None: + """Offer to dump the SQLite content to a CSV/TXT file; returns the written path.""" + if interactive is None: + interactive = sys.stdin.isatty() + if not interactive: + return None + answer = input_fn("SQLite verisi CSV/TXT dosyasına aktarılsın mı? [e/H]: ").strip().lower() + if answer not in {"e", "evet", "y", "yes"}: + return None + raw = input_fn("Dosya yolu (.csv veya .txt): ").strip() + if not raw: + return None + path = timestamped_path(Path(raw)) + output_format(path) + records = store.records(account_id) + write_records(path, records, store.webhooks(account_id)) + print(f"{len(records)} kayıt yazıldı: {path.resolve()}") + return path diff --git a/tests/test_point_alarm.py b/tests/test_point_alarm.py new file mode 100644 index 0000000..e28e588 --- /dev/null +++ b/tests/test_point_alarm.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import csv +import tempfile +import unittest +from datetime import datetime +from pathlib import Path + +from common import ApiError, ConfigurationError +from point_alarm import ( + PointStore, + fetch_pages, + flatten_registration, + load_point_alarm_filter, + mask_secrets, + parse_polygon, + timestamped_path, + write_records, +) + + +def registration(index: int, recipient: str = "r1") -> dict: + return { + "registrationId": f"id-{index}", + "recipientId": recipient, + "boundary": {"point": {"lat": 39.9, "lng": 32.8}, "radius": 1000}, + "filter": { + "lightning": {"type": "PULSE_IN_CLOUD", "peakCurrent": 5}, + "thunderstorm": {"intersectsCellPolygon": True, "severityThreshold": "LOW"}, + "precipitation": {"intensities": ["LIGHT", "HEAVY"]}, + }, + "webhook": {"accountId": "acc", "url": "https://x", "authentication": {"password": "p"}}, + } + + +class FakeService: + def __init__(self, registrations: list[dict], forbid_account: bool = False) -> None: + self.registrations = registrations + self.forbid_account = forbid_account + self.calls: list[dict] = [] + + def page(self, *, page_number, page_size, account_id=None, **_): + self.calls.append({"page": page_number, "account": account_id}) + if self.forbid_account and account_id: + raise ApiError("forbidden", status_code=403) + start = page_number * page_size + return { + "registrations": self.registrations[start : start + page_size], + "totalRecords": len(self.registrations), + } + + +class PointAlarmTest(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.dir = Path(self.tmp.name) + self.store = PointStore(self.dir / "t.db") + self.addCleanup(self.store.close) + + def test_flatten(self) -> None: + flat = flatten_registration(registration(1)) + self.assertEqual(flat["precipitationIntensities"], "LIGHT|HEAVY") + self.assertEqual(flat["intersectsCellPolygon"], "true") + self.assertIsNone(flat["speedThreshold"]) + + def test_mask(self) -> None: + masked = mask_secrets(registration(1)["webhook"]) + self.assertEqual(masked["authentication"]["password"], "***") + self.assertEqual(masked["url"], "https://x") + + def test_fetch_all_pages_upserts_without_duplicates(self) -> None: + service = FakeService([registration(i) for i in range(5)]) + summary = fetch_pages(service, self.store, account_id="acc", page_size=2) + self.assertEqual(summary.distinct, 5) + self.assertEqual(summary.missing, 0) + fetch_pages(service, self.store, account_id="acc", page_size=2) + self.assertEqual(self.store.count(), 5) + self.assertEqual(self.store.webhooks("acc")[0]["authentication"]["password"], "***") + + def test_single_page(self) -> None: + service = FakeService([registration(i) for i in range(5)]) + summary = fetch_pages(service, self.store, account_id="acc", page_size=2, page_number=1) + self.assertEqual(summary.received, 2) + self.assertEqual(len(service.calls), 1) + + def test_forbidden_account_retries_without_it(self) -> None: + service = FakeService([registration(1)], forbid_account=True) + summary = fetch_pages(service, self.store, account_id="acc") + self.assertEqual(summary.distinct, 1) + self.assertIsNone(service.calls[-1]["account"]) + + def test_csv_and_txt_output(self) -> None: + fetch_pages(FakeService([registration(1)]), self.store, account_id="acc") + records = self.store.records("acc") + write_records(self.dir / "o.csv", records, self.store.webhooks("acc")) + with (self.dir / "o.csv").open(encoding="utf-8-sig", newline="") as handle: + rows = list(csv.reader(handle, delimiter=";")) + self.assertEqual(rows[0][0], "#webhook") + self.assertEqual(rows[1][0], "registrationId") + self.assertEqual(rows[2][0], "id-1") + write_records(self.dir / "o.txt", records, []) + self.assertIn('"registrationId": "id-1"', (self.dir / "o.txt").read_text(encoding="utf-8")) + with self.assertRaises(ConfigurationError): + write_records(self.dir / "o.json", records, []) + + def test_timestamped_path(self) -> None: + path = timestamped_path(Path("a/out.csv"), datetime(2026, 10, 23, 9, 5, 7)) + self.assertEqual(path, Path("a/out-20261023-090507.csv")) + + def test_polygon(self) -> None: + ring = parse_polygon("0,0;1,0;1,1;0,0") + self.assertEqual(ring[0], [0.0, 0.0]) + with self.assertRaises(ConfigurationError): + parse_polygon("0,0;1,0;1,1") + + def test_filter_validation(self) -> None: + good = load_point_alarm_filter(Path("nowcast-point-alarm-filter.example.json")) + self.assertIn("lightning", good) + bad = self.dir / "bad.json" + bad.write_text('{"thunderstorm": {"severityThreshold": "X"}}', encoding="utf-8") + with self.assertRaises(ConfigurationError): + load_point_alarm_filter(bad) + + +if __name__ == "__main__": + unittest.main()