From 7a875caa73ea9a1553166e837ed5b19704538113 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:08:15 +0300 Subject: [PATCH 01/14] refactor: extract shared nowcast helpers into nowcast_common Move webhook config loading, UUID and warning helpers and the atomic semicolon CSV writer out of geo_alarm so the upcoming point alarm clients can reuse them. Behavior is unchanged and the existing geo tests still pass. --- geo_alarm.py | 162 +++++----------------------------------------- nowcast_common.py | 156 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 173 insertions(+), 145 deletions(-) create mode 100644 nowcast_common.py diff --git a/geo_alarm.py b/geo_alarm.py index 83b38da..88e23d4 100644 --- a/geo_alarm.py +++ b/geo_alarm.py @@ -3,31 +3,29 @@ from __future__ import annotations import csv -import json import logging -import os -import tempfile import unicodedata from copy import deepcopy from pathlib import Path from typing import Any -from urllib.parse import quote, urlsplit -from uuid import UUID, uuid4 +from urllib.parse import quote +from uuid import uuid4 from common import ApiError, ConfigurationError, IklimClient from common.config import env_bool from common.client import RETRYABLE_STATUS_CODES +from nowcast_common import ( + read_json_object, + response_warnings, + as_uuid, + load_webhook_config, + write_csv_atomic, +) LOGGER = logging.getLogger(__name__) RESULT_COLUMNS = ["status", "httpStatus", "registrationId"] REQUIRED_CSV_COLUMNS = ["city", "district", "recipientId"] -AUTHENTICATION_FIELDS = { - "BASIC": ("username", "password"), - "JWT_TOKEN": ("jwtToken",), - "API_KEY": ("apiKey", "asQueryParameter", "asHeader"), - "HMAC_SIGNATURE": ("clientId", "secret", "algorithm"), -} def normalize_turkish_name(value: str) -> str: @@ -35,83 +33,8 @@ def normalize_turkish_name(value: str) -> str: return trimmed.translate(str.maketrans({"I": "ı", "İ": "i"})).casefold() -def _read_json_object(path: Path, description: str) -> dict[str, Any]: - try: - data = json.loads(path.read_text(encoding="utf-8")) - except FileNotFoundError as exc: - raise ConfigurationError(f"{description} dosyası bulunamadı: {path}") from exc - except json.JSONDecodeError as exc: - raise ConfigurationError( - f"{description} geçerli JSON değil ({path}:{exc.lineno})." - ) from exc - if not isinstance(data, dict): - raise ConfigurationError(f"{description} kök değeri JSON nesnesi olmalıdır.") - return data - - -def _required_config_value(data: dict[str, Any], key: str, location: str) -> Any: - value = data.get(key) - if value is None or (isinstance(value, str) and not value.strip()): - raise ConfigurationError(f"{location}.{key} eksik veya boş.") - if isinstance(value, str) and "change-me" in value.strip().lower(): - raise ConfigurationError(f"{location}.{key} placeholder değeri değiştirilmemiş.") - return value - - -def load_webhook_config(path: Path) -> dict[str, Any]: - config = _read_json_object(path, "Webhook config") - url = str(_required_config_value(config, "url", "webhook")) - parsed_url = urlsplit(url) - if parsed_url.scheme != "https" or not parsed_url.netloc: - raise ConfigurationError("webhook.url geçerli bir HTTPS adresi olmalıdır.") - - http_method = str(_required_config_value(config, "httpMethod", "webhook")).upper() - if http_method not in {"POST", "PUT", "GET"}: - raise ConfigurationError("webhook.httpMethod POST, PUT veya GET olmalıdır.") - - authentication = config.get("authentication") - if not isinstance(authentication, dict): - raise ConfigurationError("webhook.authentication JSON nesnesi olmalıdır.") - selected = str( - _required_config_value(authentication, "selected", "webhook.authentication") - ).upper() - if selected not in AUTHENTICATION_FIELDS: - allowed = ", ".join(AUTHENTICATION_FIELDS) - raise ConfigurationError( - f"webhook.authentication.selected şu değerlerden biri olmalıdır: {allowed}." - ) - options = authentication.get("options") - if not isinstance(options, dict) or not isinstance(options.get(selected), dict): - raise ConfigurationError( - f"webhook.authentication.options.{selected} JSON nesnesi olmalıdır." - ) - selected_options = options[selected] - auth_payload: dict[str, Any] = {"type": selected} - for field in AUTHENTICATION_FIELDS[selected]: - auth_payload[field] = _required_config_value( - selected_options, field, f"webhook.authentication.options.{selected}" - ) - - payload: dict[str, Any] = { - "url": url, - "httpMethod": http_method, - "contentType": str( - _required_config_value(config, "contentType", "webhook") - ), - "authentication": auth_payload, - } - for optional_field in ("requestHeaders", "deliveryPolicy"): - if optional_field in config: - if not isinstance(config[optional_field], dict): - raise ConfigurationError( - f"webhook.{optional_field} JSON nesnesi olmalıdır." - ) - payload[optional_field] = deepcopy(config[optional_field]) - return payload - - def load_nowcast_geo_alarm_filter(path: Path) -> dict[str, Any]: - config = _read_json_object(path, "Nowcast geo alarm filter config") + config = read_json_object(path, "Nowcast geo alarm filter config") if not config: raise ConfigurationError("Nowcast geo alarm filter config boş olamaz.") @@ -158,32 +81,6 @@ def load_nowcast_geo_alarm_filter(path: Path) -> dict[str, Any]: return config -def _uuid(value: Any, *, version: int | None = None) -> str | None: - if not isinstance(value, str): - return None - try: - parsed = UUID(value) - except ValueError: - return None - if version is not None and parsed.version != version: - return None - return value - - -def _response_warnings(*objects: Any) -> list[str]: - warnings: list[str] = [] - for data in objects: - if not isinstance(data, dict): - continue - for key in ("warning", "warnings"): - value = data.get(key) - if isinstance(value, str) and value.strip(): - warnings.append(value.strip()) - elif isinstance(value, list): - warnings.extend(str(item) for item in value if str(item).strip()) - return warnings - - class GeoAlarmService: def __init__(self, client: IklimClient) -> None: self.client = client @@ -193,7 +90,7 @@ class GeoAlarmService: 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 _uuid(account_id): + if not as_uuid(account_id): raise ApiError("/v1/users/me yanıtında geçerli account.id bulunamadı.") return account_id @@ -227,14 +124,14 @@ class GeoAlarmService: if isinstance(response.data, dict) else None ) - registration_id = _uuid(registration_id, version=4) + registration_id = as_uuid(registration_id, version=4) if not registration_id: raise ApiError( "Başarılı registration yanıtında UUID v4 registrationId bulunamadı.", status_code=response.status_code, response_data=response.data, ) - return response.status_code, registration_id, _response_warnings(response.data) + return response.status_code, registration_id, response_warnings(response.data) def find_by_recipient_id( self, recipient_id: str @@ -260,14 +157,14 @@ class GeoAlarmService: continue if registration.get("recipientId") != recipient_id: continue - registration_id = _uuid(registration.get("registrationId"), version=4) + registration_id = as_uuid(registration.get("registrationId"), version=4) if registration_id: return ( response.status_code, registration_id, - _response_warnings(response.data, registration), + response_warnings(response.data, registration), ) - return response.status_code, None, _response_warnings(response.data) + return response.status_code, None, response_warnings(response.data) def _name_index(items: list[dict[str, Any]], item_type: str) -> dict[str, int]: @@ -461,32 +358,7 @@ class CsvRegistrationProcessor: return rows, fieldnames def _write_csv(self, rows: list[dict[str, Any]], fieldnames: list[str]) -> None: - temp_path: Path | None = None - try: - with tempfile.NamedTemporaryFile( - "w", - encoding="utf-8-sig", - newline="", - dir=self.csv_path.parent, - prefix=f".{self.csv_path.name}.", - suffix=".tmp", - delete=False, - ) as temp_file: - temp_path = Path(temp_file.name) - writer = csv.DictWriter( - temp_file, - fieldnames=fieldnames, - delimiter=";", - extrasaction="ignore", - ) - writer.writeheader() - writer.writerows(rows) - temp_file.flush() - os.fsync(temp_file.fileno()) - os.replace(temp_path, self.csv_path) - finally: - if temp_path and temp_path.exists(): - temp_path.unlink() + write_csv_atomic(self.csv_path, rows, fieldnames) @staticmethod def _set_result( diff --git a/nowcast_common.py b/nowcast_common.py new file mode 100644 index 0000000..f66a031 --- /dev/null +++ b/nowcast_common.py @@ -0,0 +1,156 @@ +"""Helpers shared by the nowcast alarm clients (geo and point).""" + +from __future__ import annotations + +import csv +import json +import os +import tempfile +from copy import deepcopy +from pathlib import Path +from typing import Any +from urllib.parse import urlsplit +from uuid import UUID + +from common import ConfigurationError + + +AUTHENTICATION_FIELDS = { + "BASIC": ("username", "password"), + "JWT_TOKEN": ("jwtToken",), + "API_KEY": ("apiKey", "asQueryParameter", "asHeader"), + "HMAC_SIGNATURE": ("clientId", "secret", "algorithm"), +} + + +def read_json_object(path: Path, description: str) -> dict[str, Any]: + try: + data = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise ConfigurationError(f"{description} dosyası bulunamadı: {path}") from exc + except json.JSONDecodeError as exc: + raise ConfigurationError( + f"{description} geçerli JSON değil ({path}:{exc.lineno})." + ) from exc + if not isinstance(data, dict): + raise ConfigurationError(f"{description} kök değeri JSON nesnesi olmalıdır.") + return data + + +def required_config_value(data: dict[str, Any], key: str, location: str) -> Any: + value = data.get(key) + if value is None or (isinstance(value, str) and not value.strip()): + raise ConfigurationError(f"{location}.{key} eksik veya boş.") + if isinstance(value, str) and "change-me" in value.strip().lower(): + raise ConfigurationError(f"{location}.{key} placeholder değeri değiştirilmemiş.") + return value + + +def load_webhook_config(path: Path) -> dict[str, Any]: + config = read_json_object(path, "Webhook config") + url = str(required_config_value(config, "url", "webhook")) + parsed_url = urlsplit(url) + if parsed_url.scheme != "https" or not parsed_url.netloc: + raise ConfigurationError("webhook.url geçerli bir HTTPS adresi olmalıdır.") + + http_method = str(required_config_value(config, "httpMethod", "webhook")).upper() + if http_method not in {"POST", "PUT", "GET"}: + raise ConfigurationError("webhook.httpMethod POST, PUT veya GET olmalıdır.") + + authentication = config.get("authentication") + if not isinstance(authentication, dict): + raise ConfigurationError("webhook.authentication JSON nesnesi olmalıdır.") + selected = str( + required_config_value(authentication, "selected", "webhook.authentication") + ).upper() + if selected not in AUTHENTICATION_FIELDS: + allowed = ", ".join(AUTHENTICATION_FIELDS) + raise ConfigurationError( + f"webhook.authentication.selected şu değerlerden biri olmalıdır: {allowed}." + ) + options = authentication.get("options") + if not isinstance(options, dict) or not isinstance(options.get(selected), dict): + raise ConfigurationError( + f"webhook.authentication.options.{selected} JSON nesnesi olmalıdır." + ) + selected_options = options[selected] + auth_payload: dict[str, Any] = {"type": selected} + for field in AUTHENTICATION_FIELDS[selected]: + auth_payload[field] = required_config_value( + selected_options, field, f"webhook.authentication.options.{selected}" + ) + + payload: dict[str, Any] = { + "url": url, + "httpMethod": http_method, + "contentType": str( + required_config_value(config, "contentType", "webhook") + ), + "authentication": auth_payload, + } + for optional_field in ("requestHeaders", "deliveryPolicy"): + if optional_field in config: + if not isinstance(config[optional_field], dict): + raise ConfigurationError( + f"webhook.{optional_field} JSON nesnesi olmalıdır." + ) + payload[optional_field] = deepcopy(config[optional_field]) + return payload + + +def as_uuid(value: Any, *, version: int | None = None) -> str | None: + if not isinstance(value, str): + return None + try: + parsed = UUID(value) + except ValueError: + return None + if version is not None and parsed.version != version: + return None + return value + + +def response_warnings(*objects: Any) -> list[str]: + warnings: list[str] = [] + for data in objects: + if not isinstance(data, dict): + continue + for key in ("warning", "warnings"): + value = data.get(key) + if isinstance(value, str) and value.strip(): + warnings.append(value.strip()) + elif isinstance(value, list): + warnings.extend(str(item) for item in value if str(item).strip()) + return warnings + + +def write_csv_atomic( + path: Path, rows: list[dict[str, Any]], fieldnames: list[str] +) -> None: + """Rewrite a semicolon separated CSV through a temp file in the same directory.""" + temp_path: Path | None = None + try: + with tempfile.NamedTemporaryFile( + "w", + encoding="utf-8-sig", + newline="", + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + delete=False, + ) as temp_file: + temp_path = Path(temp_file.name) + writer = csv.DictWriter( + temp_file, + fieldnames=fieldnames, + delimiter=";", + extrasaction="ignore", + ) + writer.writeheader() + writer.writerows(rows) + temp_file.flush() + os.fsync(temp_file.fileno()) + os.replace(temp_path, path) + finally: + if temp_path and temp_path.exists(): + temp_path.unlink() 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 02/14] 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() From 32a587a1c61eb6046b0ac52bcd8c210bee16eaf3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:12:03 +0300 Subject: [PATCH 03/14] feat: add nowcast point alarm get client for id and recipient lookups Reads IDs from CSV or TXT, writes per-query SUCCESS, NOT_FOUND or FAILED rows to CSV or JSON Lines and mirrors results into SQLite. --- nowcast-point-get.py | 89 +++++++++++++++++++++++++++++++++++++++ point_alarm.py | 74 ++++++++++++++++++++++++++++++++ tests/test_point_alarm.py | 32 ++++++++++++++ 3 files changed, 195 insertions(+) create mode 100644 nowcast-point-get.py diff --git a/nowcast-point-get.py b/nowcast-point-get.py new file mode 100644 index 0000000..8f9ceb3 --- /dev/null +++ b/nowcast-point-get.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python3 +"""Look up point alarm registrations by registrationId or recipientId from a CSV/TXT list.""" + +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path + +from common import ConfigurationError, IklimClient, configure_logging +from common.errors import ApiError +from point_alarm import ( + RESULT_COLUMNS, + PointAlarmService, + PointStore, + lookup_rows, + read_id_list, + timestamped_path, + write_records, +) + + +BASE_DIR = Path(__file__).resolve().parent + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Nokta alarm kayıtlarını get-by-id veya get-by-recipient-id ile sorgular. Girdi CSV " + "(başlıkta registrationId/recipientId sütunu, ';' ayraçlı) veya TXT (satır başına bir ID; " + "boş ve '#' ile başlayan satırlar atlanır) olabilir. Sonuçlar SQLite'a da yazılır." + ) + ) + parser.add_argument("input_file", type=Path, help="ID listesi (.csv veya .txt)") + parser.add_argument( + "--by", + choices=["id", "recipient"], + default="id", + help="Sorgu türü: id=registrationId, recipient=recipientId (varsayılan: id)", + ) + parser.add_argument( + "--output", + type=Path, + required=True, + help=( + "Çıktı dosyası: .csv (noktalı virgüllü) veya .txt (JSON Lines); dosya adına zaman soneki " + "eklenir. Satır durumları: SUCCESS, NOT_FOUND (HTTP 404), FAILED (httpStatus+errorBody). " + "Bir recipientId birden çok kayıt dönerse her kayıt ayrı satırdır" + ), + ) + parser.add_argument( + "--db", type=Path, default=BASE_DIR / "iklim-sandbox.db", help="SQLite dosyası" + ) + parser.add_argument( + "--debug-log", + type=Path, + default=BASE_DIR / "logs" / "iklim-api-debug.log", + help="Maskelenmiş HTTP debug log dosyası", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + configure_logging(args.debug_log) + store = None + try: + column = "registrationId" if args.by == "id" else "recipientId" + queries = read_id_list(args.input_file, column) + client = IklimClient.from_env(BASE_DIR / ".env") + service = PointAlarmService(client) + store = PointStore(args.db) + rows, webhooks = lookup_rows(service, store, queries, args.by, None) + path = timestamped_path(args.output) + write_records(path, rows, webhooks, columns=RESULT_COLUMNS) + failed = sum(1 for row in rows if row["status"] == "FAILED") + print(f"{len(queries)} sorgu, {len(rows)} satır yazıldı: {path.resolve()}") + return 1 if failed else 0 + except (ConfigurationError, ApiError, OSError) as exc: + logging.error("İşlem durduruldu: %s", exc) + return 1 + finally: + if store: + store.close() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/point_alarm.py b/point_alarm.py index 770f0cc..52b8f4c 100644 --- a/point_alarm.py +++ b/point_alarm.py @@ -510,3 +510,77 @@ def prompt_export( write_records(path, records, store.webhooks(account_id)) print(f"{len(records)} kayıt yazıldı: {path.resolve()}") return path + + +def read_id_list(path: Path, column: str) -> list[str]: + """Read IDs from a CSV (header column) or TXT (one per line; blank and '#' lines skipped).""" + suffix = path.suffix.lower() + if suffix == ".csv": + with path.open(encoding="utf-8-sig", newline="") as handle: + rows = [row for row in csv.reader(handle, delimiter=";") if row and not row[0].startswith("#")] + if not rows or column not in [name.strip() for name in rows[0]]: + raise ConfigurationError(f"CSV başlığında '{column}' sütunu bulunamadı: {path}") + index = [name.strip() for name in rows[0]].index(column) + values = [row[index].strip() for row in rows[1:] if len(row) > index] + elif suffix == ".txt": + values = [line.strip() for line in path.read_text(encoding="utf-8-sig").splitlines()] + values = [line for line in values if line and not line.startswith("#")] + else: + raise ConfigurationError(f"Girdi dosyası .csv veya .txt uzantılı olmalıdır: {path}") + ids = list(dict.fromkeys(value for value in values if value)) + if not ids: + raise ConfigurationError(f"Girdi dosyasında ID bulunamadı: {path}") + return ids + + +RESULT_COLUMNS = ["query", "status", "httpStatus"] + RECORD_COLUMNS + ["errorBody"] + + +def lookup_rows( + service: PointAlarmService, + store: PointStore, + queries: list[str], + by: str, + account_id: str | None, +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """Run get-by-id / get-by-recipient-id per query; returns (result rows, webhooks).""" + rows: list[dict[str, Any]] = [] + for query in queries: + try: + response = ( + service.get_by_id(query) if by == "id" else service.get_by_recipient_id(query) + ) + except ApiError as exc: + status = "NOT_FOUND" if exc.status_code == 404 else "FAILED" + rows.append( + { + "query": query, + "status": status, + "httpStatus": exc.status_code, + "errorBody": ( + json.dumps(exc.response_data, ensure_ascii=False) + if exc.response_data is not None + else str(exc) + ), + } + ) + LOGGER.warning("%s için sonuç %s (HTTP %s).", query, status, exc.status_code) + continue + data = response.data + items = data if isinstance(data, list) else [data] + items = [item for item in items if isinstance(item, dict)] + if not items: + rows.append({"query": query, "status": "NOT_FOUND", "httpStatus": response.status_code}) + for item in items: + store.upsert_registration(item, account_id) + rows.append( + dict( + flatten_registration(item), + query=query, + status="SUCCESS", + httpStatus=response.status_code, + ) + ) + store.commit() + webhooks = store.webhooks(account_id) if account_id else store.webhooks() + return rows, webhooks diff --git a/tests/test_point_alarm.py b/tests/test_point_alarm.py index e28e588..c952508 100644 --- a/tests/test_point_alarm.py +++ b/tests/test_point_alarm.py @@ -7,11 +7,14 @@ from datetime import datetime from pathlib import Path from common import ApiError, ConfigurationError +from common.client import ApiResponse from point_alarm import ( PointStore, fetch_pages, flatten_registration, load_point_alarm_filter, + lookup_rows, + read_id_list, mask_secrets, parse_polygon, timestamped_path, @@ -122,6 +125,35 @@ class PointAlarmTest(unittest.TestCase): with self.assertRaises(ConfigurationError): load_point_alarm_filter(bad) + def test_read_id_list(self) -> None: + txt = self.dir / "ids.txt" + txt.write_text("# c\n\na\nb\na\n", encoding="utf-8") + self.assertEqual(read_id_list(txt, "registrationId"), ["a", "b"]) + csv_file = self.dir / "ids.csv" + csv_file.write_text("x;recipientId\n1;r1\n2;r2\n", encoding="utf-8") + self.assertEqual(read_id_list(csv_file, "recipientId"), ["r1", "r2"]) + with self.assertRaises(ConfigurationError): + read_id_list(csv_file, "registrationId") + + def test_lookup_rows(self) -> None: + class Svc: + def get_by_id(self, value): + if value == "missing": + raise ApiError("nf", status_code=404, response_data={"m": 1}) + if value == "boom": + raise ApiError("e", status_code=500, response_data="bad") + return ApiResponse(200, registration(1)) + + def get_by_recipient_id(self, value): + return ApiResponse(200, [registration(1), registration(2)]) + + rows, _ = lookup_rows(Svc(), self.store, ["ok", "missing", "boom"], "id", None) + self.assertEqual([r["status"] for r in rows], ["SUCCESS", "NOT_FOUND", "FAILED"]) + self.assertEqual(rows[2]["errorBody"], '"bad"') + rows, _ = lookup_rows(Svc(), self.store, ["r1"], "recipient", None) + self.assertEqual(len(rows), 2) + self.assertEqual(self.store.count(), 2) + if __name__ == "__main__": unittest.main() From ba249128a11d1aedcc0f8abbb077853aa79c2af5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:13:01 +0300 Subject: [PATCH 04/14] feat: add nowcast point alarm register client Registers one point alarm per CSV row (recipientId, lat, lng, radius) with the shared filter JSON and the existing webhook bootstrap, records status, httpStatus, registrationId and errorBody back into the CSV and recovers ambiguous failures only on an identical recipient, point and radius match. --- nowcast-point-register.py | 68 +++++++++ nowcast-point-registrations.example.csv | 2 + point_alarm.py | 183 +++++++++++++++++++++++- tests/test_point_alarm.py | 90 +++++++++++- 4 files changed, 341 insertions(+), 2 deletions(-) create mode 100644 nowcast-point-register.py create mode 100644 nowcast-point-registrations.example.csv diff --git a/nowcast-point-register.py b/nowcast-point-register.py new file mode 100644 index 0000000..e834c53 --- /dev/null +++ b/nowcast-point-register.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 +"""Create point alarm registrations for every eligible row in a CSV file.""" + +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 PointAlarmService, PointRegistrationProcessor + + +BASE_DIR = Path(__file__).resolve().parent + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "CSV satırlarından iklim.co nokta alarm kayıtları oluşturur. CSV sütunları: " + "recipientId;lat;lng;radius (radius metre, 0-50000). Filtre ortak JSON dosyasından " + "okunur. Sonuç sütunları (status;httpStatus;registrationId;errorBody) aynı dosyaya " + "yazılır; SUCCESS satırlar yeniden çalıştırmada atlanır." + ) + ) + parser.add_argument("csv_file", type=Path, help="İşlenecek noktalı virgüllü CSV") + parser.add_argument( + "--webhook-config", + type=Path, + default=BASE_DIR / "webhook.json", + help="Webhook JSON dosyası (varsayılan: webhook.json)", + ) + parser.add_argument( + "--filter-config", + type=Path, + default=BASE_DIR / "nowcast-point-alarm-filter.json", + help="Her kayda uygulanan filtre JSON dosyası (varsayılan: nowcast-point-alarm-filter.json)", + ) + parser.add_argument( + "--debug-log", + type=Path, + default=BASE_DIR / "logs" / "iklim-api-debug.log", + help="Maskelenmiş HTTP debug log dosyası", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + configure_logging(args.debug_log) + try: + client = IklimClient.from_env(BASE_DIR / ".env") + PointRegistrationProcessor( + PointAlarmService(client), + csv_path=args.csv_file, + webhook_config_path=args.webhook_config, + filter_config_path=args.filter_config, + ).run() + except (ConfigurationError, ApiError, OSError) as exc: + logging.error("İşlem durduruldu: %s", exc) + return 1 + logging.info("CSV işlemi tamamlandı: %s", args.csv_file.resolve()) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/nowcast-point-registrations.example.csv b/nowcast-point-registrations.example.csv new file mode 100644 index 0000000..55e5efa --- /dev/null +++ b/nowcast-point-registrations.example.csv @@ -0,0 +1,2 @@ +recipientId;lat;lng;radius +recipient-001;39.9208;32.8541;1000 diff --git a/point_alarm.py b/point_alarm.py index 52b8f4c..af934b2 100644 --- a/point_alarm.py +++ b/point_alarm.py @@ -7,15 +7,26 @@ import json import logging import sqlite3 import sys +from copy import deepcopy 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 uuid import uuid4 from common import ApiError, ConfigurationError, IklimClient from common.client import ApiResponse -from nowcast_common import fetch_account_id, read_json_object +from common.client import RETRYABLE_STATUS_CODES +from common.config import env_bool +from nowcast_common import ( + as_uuid, + fetch_account_id, + load_webhook_config, + read_json_object, + response_warnings, + write_csv_atomic, +) LOGGER = logging.getLogger(__name__) @@ -567,6 +578,8 @@ def lookup_rows( LOGGER.warning("%s için sonuç %s (HTTP %s).", query, status, exc.status_code) continue data = response.data + if isinstance(data, dict) and "registrations" in data: + data = data["registrations"] items = data if isinstance(data, list) else [data] items = [item for item in items if isinstance(item, dict)] if not items: @@ -584,3 +597,171 @@ def lookup_rows( store.commit() webhooks = store.webhooks(account_id) if account_id else store.webhooks() return rows, webhooks + + +REGISTER_REQUIRED_COLUMNS = ["recipientId", "lat", "lng", "radius"] +REGISTER_RESULT_COLUMNS = ["status", "httpStatus", "registrationId", "errorBody"] + + +def error_body_text(exc: ApiError) -> str: + if exc.response_data is None: + return str(exc) + if isinstance(exc.response_data, str): + return exc.response_data + return json.dumps(exc.response_data, ensure_ascii=False) + + +def parse_point_row(row: dict[str, Any]) -> tuple[str, float, float, float]: + """Validate one register CSV row and return (recipientId, lat, lng, radius).""" + recipient_id = str(row.get("recipientId", "")).strip() + if not recipient_id: + raise ConfigurationError("recipientId boş.") + if not recipient_id.isascii(): + raise ConfigurationError("recipientId US-ASCII karakterlerinden oluşmalıdır.") + try: + lat = float(str(row.get("lat", "")).strip().replace(",", ".")) + lng = float(str(row.get("lng", "")).strip().replace(",", ".")) + radius = float(str(row.get("radius", "")).strip().replace(",", ".")) + except ValueError as exc: + raise ConfigurationError("lat, lng ve radius sayı olmalıdır.") from exc + if not -90 <= lat <= 90 or not -180 <= lng <= 180: + raise ConfigurationError("lat -90..90, lng -180..180 aralığında olmalıdır.") + if not 0 <= radius <= MAX_RADIUS_METERS: + raise ConfigurationError(f"radius 0..{int(MAX_RADIUS_METERS)} metre aralığında olmalıdır.") + return recipient_id, lat, lng, radius + + +class PointRegistrationProcessor: + """Register one point alarm per CSV row, recording the outcome back into the CSV.""" + + def __init__( + self, + service: PointAlarmService, + *, + csv_path: Path, + webhook_config_path: Path, + filter_config_path: Path, + ) -> None: + self.service = service + self.csv_path = csv_path.resolve() + self.webhook_config_path = webhook_config_path.resolve() + self.filter_config_path = filter_config_path.resolve() + + def run(self) -> None: + rows, fieldnames = self._read_csv() + filters = load_point_alarm_filter(self.filter_config_path) + webhook_initialized = env_bool("IKLIM_WEBHOOK_INITIALIZED", False) + full_webhook = None if webhook_initialized else load_webhook_config(self.webhook_config_path) + account_id = self.service.get_account_id() + LOGGER.info("Kullanıcı hesabı doğrulandı; accountId API'den alındı.") + for row_number, row in enumerate(rows, start=2): + if str(row.get("status", "")).strip().upper() == "SUCCESS": + LOGGER.info("Satır %s zaten SUCCESS; atlandı.", row_number) + continue + try: + recipient_id, lat, lng, radius = parse_point_row(row) + except ConfigurationError as exc: + LOGGER.error("Satır %s gönderilmedi: %s", row_number, exc) + self._set_result(row, "FAILED", "NOT_SENT", "", str(exc)) + self._write_csv(rows, fieldnames) + continue + payload = { + "recipientId": recipient_id, + "boundary": {"point": {"lat": lat, "lng": lng}, "radius": radius}, + "webhook": {"accountId": account_id} if webhook_initialized else deepcopy(full_webhook), + "filter": deepcopy(filters), + } + registered = self._register(row_number, payload, row) + if registered and not webhook_initialized: + webhook_initialized = True + self.service.client.token_store.set_webhook_initialized(True) + self._write_csv(rows, fieldnames) + + def _register(self, row_number: int, payload: dict[str, Any], row: dict[str, Any]) -> bool: + try: + response = self.service.register(payload, str(uuid4())) + except ApiError as exc: + recovered = self._try_recover(row_number, payload, exc) + if recovered: + self._set_result(row, "SUCCESS", f"HTTP_{exc.status_code or recovered[0]}", recovered[1], "") + LOGGER.info("Satır %s eşleşen kayıtla kurtarıldı: %s", row_number, recovered[1]) + return True + self._set_result( + row, + "FAILED", + f"HTTP_{exc.status_code}" if exc.status_code else "NETWORK_ERROR", + "", + error_body_text(exc), + ) + LOGGER.error("Satır %s başarısız: %s", row_number, exc) + return False + data = response.data if isinstance(response.data, dict) else {} + registration_id = as_uuid(data.get("registrationId"), version=4) + if not registration_id: + self._set_result( + row, "FAILED", f"HTTP_{response.status_code}", "", + "Yanıtta UUID v4 registrationId bulunamadı: " + json.dumps(response.data, ensure_ascii=False), + ) + LOGGER.error("Satır %s yanıtında registrationId yok.", row_number) + return False + self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", registration_id, "") + LOGGER.info("Satır %s başarıyla kaydedildi: %s", row_number, registration_id) + for warning in response_warnings(response.data): + LOGGER.warning("Satır %s API uyarısı: %s", row_number, warning) + return True + + def _try_recover( + self, row_number: int, payload: dict[str, Any], error: ApiError + ) -> tuple[int, str] | None: + """After an ambiguous failure, look for an identical registration (recipient + point + radius).""" + status = error.status_code + if status is not None and status != 409 and status not in RETRYABLE_STATUS_CODES: + return None + LOGGER.warning("Satır %s sonucu belirsiz; recipientId ile kontrol ediliyor.", row_number) + boundary = payload["boundary"] + try: + response = self.service.get_by_recipient_id(payload["recipientId"]) + except ApiError as exc: + LOGGER.error("Satır %s için recipient sorgusu başarısız: %s", row_number, exc) + return None + data = response.data if isinstance(response.data, dict) else {} + for registration in data.get("registrations") or []: + flat = flatten_registration(registration) if isinstance(registration, dict) else {} + registration_id = as_uuid(flat.get("registrationId"), version=4) + if ( + registration_id + and flat.get("recipientId") == payload["recipientId"] + and flat.get("lat") == boundary["point"]["lat"] + and flat.get("lng") == boundary["point"]["lng"] + and flat.get("radius") == boundary["radius"] + ): + return response.status_code, registration_id + return None + + def _read_csv(self) -> tuple[list[dict[str, Any]], list[str]]: + if not self.csv_path.is_file(): + raise ConfigurationError(f"CSV dosyası bulunamadı: {self.csv_path}") + with self.csv_path.open("r", encoding="utf-8-sig", newline="") as handle: + reader = csv.DictReader(handle, delimiter=";") + if not reader.fieldnames: + raise ConfigurationError("CSV başlık satırı bulunamadı.") + missing = [name for name in REGISTER_REQUIRED_COLUMNS if name not in reader.fieldnames] + if missing: + raise ConfigurationError("CSV içinde eksik zorunlu sütunlar: " + ", ".join(missing)) + fieldnames = [ + name for name in reader.fieldnames if name not in REGISTER_RESULT_COLUMNS + ] + REGISTER_RESULT_COLUMNS + rows = list(reader) + return rows, fieldnames + + def _write_csv(self, rows: list[dict[str, Any]], fieldnames: list[str]) -> None: + write_csv_atomic(self.csv_path, rows, fieldnames) + + @staticmethod + def _set_result( + row: dict[str, Any], status: str, http_status: str, registration_id: str, error_body: str + ) -> None: + row["status"] = status + row["httpStatus"] = http_status + row["registrationId"] = registration_id + row["errorBody"] = error_body diff --git a/tests/test_point_alarm.py b/tests/test_point_alarm.py index c952508..eee8631 100644 --- a/tests/test_point_alarm.py +++ b/tests/test_point_alarm.py @@ -1,14 +1,20 @@ from __future__ import annotations import csv +import json +import os import tempfile import unittest from datetime import datetime from pathlib import Path +from unittest.mock import patch + +from common.security import TokenStore from common import ApiError, ConfigurationError from common.client import ApiResponse from point_alarm import ( + PointRegistrationProcessor, PointStore, fetch_pages, flatten_registration, @@ -145,7 +151,7 @@ class PointAlarmTest(unittest.TestCase): return ApiResponse(200, registration(1)) def get_by_recipient_id(self, value): - return ApiResponse(200, [registration(1), registration(2)]) + return ApiResponse(200, {"registrations": [registration(1), registration(2)]}) rows, _ = lookup_rows(Svc(), self.store, ["ok", "missing", "boom"], "id", None) self.assertEqual([r["status"] for r in rows], ["SUCCESS", "NOT_FOUND", "FAILED"]) @@ -155,5 +161,87 @@ class PointAlarmTest(unittest.TestCase): self.assertEqual(self.store.count(), 2) +REG_ID = "f7587d9e-2481-4b4c-818d-c8d1946851b7" + + +class FakeRegisterService: + def __init__(self, env_file: Path, results: list) -> None: + self.client = type("C", (), {"token_store": TokenStore(env_file)})() + self.results = list(results) + self.payloads: list[dict] = [] + self.recipient_hits: list = [] + + def get_account_id(self): + return "acc" + + def register(self, payload, key): + self.payloads.append(payload) + result = self.results.pop(0) + if isinstance(result, Exception): + raise result + return result + + def get_by_recipient_id(self, recipient_id): + return ApiResponse(200, {"registrations": self.recipient_hits}) + + +class RegisterTest(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.dir = Path(self.tmp.name) + (self.dir / ".env").write_text("", encoding="utf-8") + self.filter = self.dir / "filter.json" + self.filter.write_text(json.dumps({"lightning": {"type": "PULSE_IN_CLOUD"}}), encoding="utf-8") + self.webhook = self.dir / "webhook.json" + self.webhook.write_text("{}", encoding="utf-8") + self.csv = self.dir / "r.csv" + + def run_processor(self, rows: str | None, service) -> list[dict]: + if rows is not None: + self.csv.write_text("recipientId;lat;lng;radius\n" + rows, encoding="utf-8") + processor = PointRegistrationProcessor( + service, csv_path=self.csv, webhook_config_path=self.webhook, filter_config_path=self.filter + ) + with patch.dict(os.environ, {"IKLIM_WEBHOOK_INITIALIZED": "true"}): + processor.run() + with self.csv.open(encoding="utf-8-sig", newline="") as handle: + return list(csv.DictReader(handle, delimiter=";")) + + def test_success_failure_and_skip(self) -> None: + service = FakeRegisterService( + self.dir / ".env", + [ + ApiResponse(201, {"registrationId": REG_ID}), + ApiError("bad", status_code=400, response_data={"message": "duplicate"}), + ], + ) + rows = self.run_processor("a;39.9;32.8;1000\nb;39.9;32.8;1000\nc;99;32.8;1000\n", service) + self.assertEqual([r["status"] for r in rows], ["SUCCESS", "FAILED", "FAILED"]) + self.assertEqual(rows[0]["registrationId"], REG_ID) + self.assertIn("duplicate", rows[1]["errorBody"]) + self.assertEqual(rows[2]["httpStatus"], "NOT_SENT") + self.assertEqual(service.payloads[0]["webhook"], {"accountId": "acc"}) + self.assertEqual(service.payloads[0]["filter"]["lightning"]["type"], "PULSE_IN_CLOUD") + # Re-run: SUCCESS row is skipped, the failed one is retried. + service2 = FakeRegisterService( + self.dir / ".env", [ApiResponse(201, {"registrationId": REG_ID})] * 2 + ) + rows = self.run_processor(None, service2) + self.assertEqual(len(service2.payloads), 1) + self.assertEqual(rows[0]["registrationId"], REG_ID) + + def test_recovery_requires_matching_point(self) -> None: + service = FakeRegisterService(self.dir / ".env", [ApiError("t", status_code=503)]) + service.recipient_hits = [dict(registration(1, "a"), registrationId=REG_ID)] + rows = self.run_processor("a;39.9;32.8;1000\n", service) + self.assertEqual(rows[0]["status"], "SUCCESS") + self.assertEqual(rows[0]["registrationId"], REG_ID) + other = FakeRegisterService(self.dir / ".env", [ApiError("t", status_code=503)]) + other.recipient_hits = [dict(registration(1, "a"), registrationId=REG_ID)] + rows = self.run_processor("a;40.0;32.8;1000\n", other) + self.assertEqual(rows[0]["status"], "FAILED") + + if __name__ == "__main__": unittest.main() From 8587669ea5b17bd3732b1a8c08184debefedc755 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:13:35 +0300 Subject: [PATCH 05/14] feat: add nowcast point alarm update client with dry-run Patches registrations from CSV rows, sending only non-blank cells, validating lat/lng pairing and ranges locally, warning about the speedThreshold PATCH gap (GAIA-476) and recording status, httpStatus and errorBody back into the CSV. --- nowcast-point-update.example.csv | 1 + nowcast-point-update.py | 60 +++++++++++ point_alarm.py | 178 +++++++++++++++++++++++++++++++ tests/test_point_alarm.py | 51 +++++++++ 4 files changed, 290 insertions(+) create mode 100644 nowcast-point-update.example.csv create mode 100644 nowcast-point-update.py diff --git a/nowcast-point-update.example.csv b/nowcast-point-update.example.csv new file mode 100644 index 0000000..2b56269 --- /dev/null +++ b/nowcast-point-update.example.csv @@ -0,0 +1 @@ +registrationId;recipientId;lat;lng;radius;lightningType;peakCurrent;inCloudHeight;intersectsAffectedPolygon;intersectsCellPolygon;severityThreshold;speedThreshold;precipitationIntensities diff --git a/nowcast-point-update.py b/nowcast-point-update.py new file mode 100644 index 0000000..a752501 --- /dev/null +++ b/nowcast-point-update.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +"""Patch point alarm registrations from a CSV file (blank cells are not sent).""" + +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 PointAlarmService, PointUpdateProcessor + + +BASE_DIR = Path(__file__).resolve().parent + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "CSV satırlarından nokta alarm kayıtlarını kısmen günceller (PATCH). Sütunlar: " + "registrationId (zorunlu) ve isteğe bağlı recipientId;lat;lng;radius;lightningType;" + "peakCurrent;inCloudHeight;intersectsAffectedPolygon;intersectsCellPolygon;" + "severityThreshold;speedThreshold;precipitationIntensities ('|' ile ayrılır). Boş hücre " + "gönderilmez; lat ve lng birlikte verilmelidir. Bir alan boşaltılamaz (temizleme " + "desteklenmez). thunderstorm.speedThreshold backend PATCH'inde işlenmiyor (GAIA-476). " + "Sonuç sütunları status;httpStatus;errorBody aynı dosyaya yazılır." + ) + ) + parser.add_argument("csv_file", type=Path, help="İşlenecek noktalı virgüllü CSV") + parser.add_argument( + "--dry-run", + action="store_true", + help="İstek göndermeden ve CSV'yi değiştirmeden PATCH gövdelerini ekrana yazar", + ) + parser.add_argument( + "--debug-log", + type=Path, + default=BASE_DIR / "logs" / "iklim-api-debug.log", + help="Maskelenmiş HTTP debug log dosyası", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + configure_logging(args.debug_log) + try: + client = IklimClient.from_env(BASE_DIR / ".env") + PointUpdateProcessor( + PointAlarmService(client), csv_path=args.csv_file, dry_run=args.dry_run + ).run() + except (ConfigurationError, ApiError, OSError) as exc: + logging.error("İşlem durduruldu: %s", exc) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/point_alarm.py b/point_alarm.py index af934b2..261d685 100644 --- a/point_alarm.py +++ b/point_alarm.py @@ -765,3 +765,181 @@ class PointRegistrationProcessor: row["httpStatus"] = http_status row["registrationId"] = registration_id row["errorBody"] = error_body + + +def _blank(value: Any) -> bool: + return value is None or str(value).strip() == "" + + +def _number(row: dict[str, Any], name: str) -> float | None: + if _blank(row.get(name)): + return None + try: + return float(str(row[name]).strip().replace(",", ".")) + except ValueError as exc: + raise ConfigurationError(f"{name} sayı olmalıdır.") from exc + + +def _choice(row: dict[str, Any], name: str, allowed: set[str]) -> str | None: + if _blank(row.get(name)): + return None + value = str(row[name]).strip().upper() + if value not in allowed: + raise ConfigurationError(f"{name} şunlardan biri olmalıdır: {', '.join(sorted(allowed))}.") + return value + + +def _flag(row: dict[str, Any], name: str) -> bool | None: + if _blank(row.get(name)): + return None + value = str(row[name]).strip().lower() + if value in {"true", "1", "evet"}: + return True + if value in {"false", "0", "hayır", "hayir"}: + return False + raise ConfigurationError(f"{name} true veya false olmalıdır.") + + +def build_update_payload(row: dict[str, Any]) -> dict[str, Any]: + """Build a PATCH body from a CSV row; blank cells are left out (null = untouched).""" + registration_id = as_uuid(str(row.get("registrationId", "")).strip(), version=4) + if not registration_id: + raise ConfigurationError("registrationId geçerli bir UUID olmalıdır.") + payload: dict[str, Any] = {"registrationId": registration_id} + recipient_id = str(row.get("recipientId") or "").strip() + if recipient_id: + if not recipient_id.isascii(): + raise ConfigurationError("recipientId US-ASCII karakterlerinden oluşmalıdır.") + payload["recipientId"] = recipient_id + + lat, lng, radius = _number(row, "lat"), _number(row, "lng"), _number(row, "radius") + if (lat is None) != (lng is None): + raise ConfigurationError("lat ve lng birlikte verilmelidir.") + boundary: dict[str, Any] = {} + if lat is not None: + if not -90 <= lat <= 90 or not -180 <= lng <= 180: + raise ConfigurationError("lat -90..90, lng -180..180 aralığında olmalıdır.") + boundary["point"] = {"lat": lat, "lng": lng} + if radius is not None: + if not 0 <= radius <= MAX_RADIUS_METERS: + raise ConfigurationError(f"radius 0..{int(MAX_RADIUS_METERS)} metre aralığında olmalıdır.") + boundary["radius"] = radius + if boundary: + payload["boundary"] = boundary + + lightning = { + key: value + for key, value in { + "type": _choice(row, "lightningType", LIGHTNING_TYPES), + "peakCurrent": _number(row, "peakCurrent"), + "inCloudHeight": _number(row, "inCloudHeight"), + }.items() + if value is not None + } + thunderstorm = { + key: value + for key, value in { + "intersectsAffectedPolygon": _flag(row, "intersectsAffectedPolygon"), + "intersectsCellPolygon": _flag(row, "intersectsCellPolygon"), + "severityThreshold": _choice(row, "severityThreshold", SEVERITIES), + "speedThreshold": _number(row, "speedThreshold"), + }.items() + if value is not None + } + precipitation: dict[str, Any] = {} + if not _blank(row.get("precipitationIntensities")): + items = [part.strip().upper() for part in str(row["precipitationIntensities"]).split("|") if part.strip()] + invalid = [item for item in items if item not in INTENSITIES] + if invalid or not items: + raise ConfigurationError( + "precipitationIntensities '|' ile ayrılmış geçerli değerler içermelidir: " + + ", ".join(sorted(INTENSITIES)) + ) + precipitation["intensities"] = items + filters = { + name: section + for name, section in ( + ("lightning", lightning), + ("thunderstorm", thunderstorm), + ("precipitation", precipitation), + ) + if section + } + if filters: + payload["filter"] = filters + if len(payload) == 1: + raise ConfigurationError("Güncellenecek alan yok (registrationId dışında tüm hücreler boş).") + return payload + + +UPDATE_REQUIRED_COLUMNS = ["registrationId"] +UPDATE_COLUMNS = ["registrationId", "recipientId", "lat", "lng", "radius"] + FILTER_COLUMNS + + +class PointUpdateProcessor: + """PATCH one registration per CSV row; blank cells are not sent.""" + + def __init__(self, service: PointAlarmService, *, csv_path: Path, dry_run: bool = False) -> None: + self.service = service + self.csv_path = csv_path.resolve() + self.dry_run = dry_run + + def run(self) -> None: + rows, fieldnames = self._read_csv() + for row_number, row in enumerate(rows, start=2): + if str(row.get("status", "")).strip().upper() == "SUCCESS": + LOGGER.info("Satır %s zaten SUCCESS; atlandı.", row_number) + continue + try: + payload = build_update_payload(row) + except ConfigurationError as exc: + LOGGER.error("Satır %s gönderilmedi: %s", row_number, exc) + self._set_result(row, "FAILED", "NOT_SENT", str(exc)) + self._write(rows, fieldnames) + continue + if payload.get("filter", {}).get("thunderstorm", {}).get("speedThreshold") is not None: + LOGGER.warning( + "Satır %s: thunderstorm.speedThreshold PATCH'te backend tarafından işlenmiyor " + "(GAIA-476); değer gönderilir ancak güncellenmeyebilir.", + row_number, + ) + if self.dry_run: + print(f"[dry-run] satır {row_number}: PATCH {ALARMS_PATH}/update {json.dumps(payload, ensure_ascii=False)}") + continue + try: + response = self.service.update(payload, str(uuid4())) + except ApiError as exc: + self._set_result( + row, "FAILED", + f"HTTP_{exc.status_code}" if exc.status_code else "NETWORK_ERROR", + error_body_text(exc), + ) + LOGGER.error("Satır %s başarısız: %s", row_number, exc) + else: + self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", "") + LOGGER.info("Satır %s güncellendi: %s", row_number, payload["registrationId"]) + for warning in response_warnings(response.data): + LOGGER.warning("Satır %s API uyarısı: %s", row_number, warning) + self._write(rows, fieldnames) + + def _read_csv(self) -> tuple[list[dict[str, Any]], list[str]]: + if not self.csv_path.is_file(): + raise ConfigurationError(f"CSV dosyası bulunamadı: {self.csv_path}") + with self.csv_path.open("r", encoding="utf-8-sig", newline="") as handle: + reader = csv.DictReader(handle, delimiter=";") + if not reader.fieldnames or "registrationId" not in reader.fieldnames: + raise ConfigurationError("CSV başlığında 'registrationId' sütunu bulunamadı.") + fieldnames = [n for n in reader.fieldnames if n not in ("status", "httpStatus", "errorBody")] + [ + "status", "httpStatus", "errorBody", + ] + return list(reader), fieldnames + + def _write(self, rows: list[dict[str, Any]], fieldnames: list[str]) -> None: + if not self.dry_run: + write_csv_atomic(self.csv_path, rows, fieldnames) + + @staticmethod + def _set_result(row: dict[str, Any], status: str, http_status: str, error_body: str) -> None: + row["status"] = status + row["httpStatus"] = http_status + row["errorBody"] = error_body diff --git a/tests/test_point_alarm.py b/tests/test_point_alarm.py index eee8631..a5ba9e4 100644 --- a/tests/test_point_alarm.py +++ b/tests/test_point_alarm.py @@ -15,6 +15,8 @@ from common import ApiError, ConfigurationError from common.client import ApiResponse from point_alarm import ( PointRegistrationProcessor, + PointUpdateProcessor, + build_update_payload, PointStore, fetch_pages, flatten_registration, @@ -243,5 +245,54 @@ class RegisterTest(unittest.TestCase): self.assertEqual(rows[0]["status"], "FAILED") +class UpdateTest(unittest.TestCase): + def test_payload_skips_blank_cells(self) -> None: + payload = build_update_payload( + {"registrationId": REG_ID, "lat": "39.9", "lng": "32.8", "radius": "", + "severityThreshold": "high", "precipitationIntensities": "light|heavy"} + ) + self.assertEqual(payload["boundary"], {"point": {"lat": 39.9, "lng": 32.8}}) + self.assertEqual(payload["filter"]["thunderstorm"], {"severityThreshold": "HIGH"}) + self.assertEqual(payload["filter"]["precipitation"]["intensities"], ["LIGHT", "HEAVY"]) + + def test_payload_errors(self) -> None: + for row in ( + {"registrationId": REG_ID}, + {"registrationId": REG_ID, "lat": "39"}, + {"registrationId": "x", "radius": "5"}, + {"registrationId": REG_ID, "radius": "60000"}, + {"registrationId": REG_ID, "lightningType": "BAD"}, + ): + with self.assertRaises(ConfigurationError, msg=row): + build_update_payload(row) + + def test_dry_run_does_not_call_or_write(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "u.csv" + text = f"registrationId;radius\n{REG_ID};500\n" + path.write_text(text, encoding="utf-8") + + class Boom: + def update(self, *_): + raise AssertionError("must not be called") + + PointUpdateProcessor(Boom(), csv_path=path, dry_run=True).run() + self.assertEqual(path.read_text(encoding="utf-8"), text) + + def test_update_records_result(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "u.csv" + path.write_text(f"registrationId;radius\n{REG_ID};500\n", encoding="utf-8") + + class Svc: + def update(self, payload, key): + return ApiResponse(200, {}) + + PointUpdateProcessor(Svc(), csv_path=path).run() + with path.open(encoding="utf-8-sig", newline="") as handle: + row = next(csv.DictReader(handle, delimiter=";")) + self.assertEqual((row["status"], row["httpStatus"]), ("SUCCESS", "HTTP_200")) + + if __name__ == "__main__": unittest.main() From 4d270aea053432723b67a11c3196ba78c235074d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:14:08 +0300 Subject: [PATCH 06/14] feat: add nowcast point alarm unregister client with confirmation and backup Shows environment, count and an irreversibility warning, requires typing the confirmation word (plus the environment name on prod), backs every registration up through get-by-id before deleting and aborts the whole run if the backup cannot be completed. --- common/config.py | 2 + nowcast-point-unregister.py | 86 +++++++++++++++++++++++++++++++++++++ point_alarm.py | 75 ++++++++++++++++++++++++++++++++ tests/test_point_alarm.py | 45 +++++++++++++++++++ 4 files changed, 208 insertions(+) create mode 100644 nowcast-point-unregister.py diff --git a/common/config.py b/common/config.py index 5fcd3e7..575631f 100644 --- a/common/config.py +++ b/common/config.py @@ -68,6 +68,7 @@ class Settings: token_expiry_skew_seconds: int max_retries: int retry_base_seconds: float + environment: str = "" @classmethod def from_env(cls, env_file: Path = DEFAULT_ENV_FILE) -> "Settings": @@ -96,6 +97,7 @@ class Settings: return cls( env_file=env_file, base_url=base_url, + environment=environment, hmac_secret=_required("IKLIM_HMAC_SECRET"), accept_language=language, timeout_seconds=_positive_float("IKLIM_TIMEOUT_SECONDS", "30"), diff --git a/nowcast-point-unregister.py b/nowcast-point-unregister.py new file mode 100644 index 0000000..d3c3233 --- /dev/null +++ b/nowcast-point-unregister.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +"""Delete point alarm registrations listed in a CSV/TXT file, with confirmation and backup.""" + +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 ( + PointAlarmService, + PointUnregisterProcessor, + read_id_list, + timestamped_path, + write_records, +) + + +BASE_DIR = Path(__file__).resolve().parent + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Dosyadaki registrationId'leri siler (DELETE /v1/alarms/points/unregister/{id}). " + "Girdi CSV (registrationId sütunu) veya TXT (satır başına bir ID). Başlamadan önce " + "ortam, silinecek sayı ve 'geri alınamaz' uyarısı gösterilir; onay için SİL yazılmalıdır " + "(prod ortamında ek olarak 'prod' yazılır). Bir hesabın son kaydı silinirse webhook da " + "silinir. Silmeden önce her kayıt get-by-id ile yedeklenir; yedek yazılamazsa hiçbir " + "kayıt silinmez." + ) + ) + parser.add_argument("input_file", type=Path, help="ID listesi (.csv veya .txt)") + parser.add_argument( + "--backup", + type=Path, + default=BASE_DIR / "nowcast-point-unregister-backup.csv", + help="Yedek dosyası (.csv/.txt, zaman soneki eklenir; varsayılan: nowcast-point-unregister-backup.csv)", + ) + parser.add_argument("--no-backup", action="store_true", help="Silmeden önce yedek almayı kapatır") + parser.add_argument( + "--output", + type=Path, + help="Sonuç dosyası (.csv/.txt, zaman soneki eklenir); verilmezse yalnızca özet yazılır", + ) + parser.add_argument( + "--debug-log", + type=Path, + default=BASE_DIR / "logs" / "iklim-api-debug.log", + help="Maskelenmiş HTTP debug log dosyası", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + configure_logging(args.debug_log) + try: + ids = read_id_list(args.input_file, "registrationId") + client = IklimClient.from_env(BASE_DIR / ".env") + processor = PointUnregisterProcessor( + PointAlarmService(client), + ids=ids, + environment=client.settings.environment, + base_url=client.settings.base_url, + backup_path=None if args.no_backup else timestamped_path(args.backup), + ) + results = processor.run() + if not results: + return 1 + failed = sum(1 for row in results if row["status"] == "FAILED") + print(f"{len(results) - failed} silindi, {failed} başarısız.") + if args.output: + path = timestamped_path(args.output) + write_records(path, results, [], columns=["registrationId", "status", "httpStatus", "errorBody"]) + print(f"Sonuçlar yazıldı: {path.resolve()}") + return 1 if failed else 0 + except (ConfigurationError, ApiError, OSError) as exc: + logging.error("İşlem durduruldu: %s", exc) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/point_alarm.py b/point_alarm.py index 261d685..965692e 100644 --- a/point_alarm.py +++ b/point_alarm.py @@ -943,3 +943,78 @@ class PointUpdateProcessor: row["status"] = status row["httpStatus"] = http_status row["errorBody"] = error_body + + +CONFIRM_WORD = "SİL" + + +class PointUnregisterProcessor: + """Delete registrations listed in a CSV/TXT after a typed confirmation and a backup.""" + + def __init__( + self, + service: PointAlarmService, + *, + ids: list[str], + environment: str, + base_url: str, + backup_path: Path | None, + input_fn: Callable[[str], str] = input, + out: TextIO | None = None, + ) -> None: + self.service = service + self.ids = ids + self.environment = environment + self.base_url = base_url + self.backup_path = backup_path + self.input_fn = input_fn + self.out = out or sys.stdout + + def confirm(self) -> bool: + say = lambda text: self.out.write(text + "\n") + say(f"Ortam: {self.environment or '-'} ({self.base_url})") + say(f"Silinecek kayıt sayısı: {len(self.ids)}") + say("Bu işlem geri alınamaz. Bir hesabın SON kaydı silinirse hesabın webhook tanımı da silinir.") + if self.environment == "prod": + if self.input_fn("PROD ortamında işlem yapıyorsunuz. Onaylamak için 'prod' yazın: ").strip() != "prod": + say("Ortam adı eşleşmedi; işlem iptal edildi.") + return False + if self.input_fn(f"Devam etmek için '{CONFIRM_WORD}' yazın: ").strip() != CONFIRM_WORD: + say("Onay verilmedi; işlem iptal edildi.") + return False + return True + + def run(self) -> list[dict[str, Any]]: + if not self.confirm(): + return [] + backups: list[dict[str, Any]] = [] + results: list[dict[str, Any]] = [] + webhooks: list[dict[str, Any]] = [] + if self.backup_path: + for registration_id in self.ids: + try: + data = self.service.get_by_id(registration_id).data + except ApiError as exc: + if exc.status_code == 404: + LOGGER.warning("%s bulunamadı; yedeklenemedi.", registration_id) + continue + # An unreadable record cannot be backed up, so nothing is deleted. + raise + if isinstance(data, dict): + backups.append(flatten_registration(data)) + webhook = registration_webhook(data) + if webhook and webhook not in webhooks: + webhooks.append(webhook) + # A backup that cannot be written must stop the run before anything is deleted. + write_records(self.backup_path, backups, webhooks) + LOGGER.info("Yedek yazıldı: %s (%s kayıt)", self.backup_path, len(backups)) + for registration_id in self.ids: + try: + response = self.service.unregister(registration_id) + except ApiError as exc: + results.append({"registrationId": registration_id, "status": "FAILED", "httpStatus": exc.status_code, "errorBody": error_body_text(exc)}) + LOGGER.error("%s silinemedi: %s", registration_id, exc) + else: + results.append({"registrationId": registration_id, "status": "SUCCESS", "httpStatus": response.status_code, "errorBody": ""}) + LOGGER.info("%s silindi.", registration_id) + return results diff --git a/tests/test_point_alarm.py b/tests/test_point_alarm.py index a5ba9e4..a715aef 100644 --- a/tests/test_point_alarm.py +++ b/tests/test_point_alarm.py @@ -1,6 +1,7 @@ from __future__ import annotations import csv +import io import json import os import tempfile @@ -15,6 +16,7 @@ from common import ApiError, ConfigurationError from common.client import ApiResponse from point_alarm import ( PointRegistrationProcessor, + PointUnregisterProcessor, PointUpdateProcessor, build_update_payload, PointStore, @@ -294,5 +296,48 @@ class UpdateTest(unittest.TestCase): self.assertEqual((row["status"], row["httpStatus"]), ("SUCCESS", "HTTP_200")) +class UnregisterTest(unittest.TestCase): + class Svc: + def __init__(self) -> None: + self.deleted: list[str] = [] + + def get_by_id(self, value): + return ApiResponse(200, dict(registration(1), registrationId=value)) + + def unregister(self, value): + self.deleted.append(value) + return ApiResponse(204, None) + + def make(self, svc, answers, env="test", backup=None): + queue = list(answers) + return PointUnregisterProcessor( + svc, ids=[REG_ID], environment=env, base_url="https://x", + backup_path=backup, input_fn=lambda _: queue.pop(0), out=io.StringIO(), + ) + + def test_wrong_confirmation_deletes_nothing(self) -> None: + svc = self.Svc() + self.assertEqual(self.make(svc, ["sil"]).run(), []) + self.assertEqual(svc.deleted, []) + + def test_prod_needs_environment_name(self) -> None: + svc = self.Svc() + self.assertEqual(self.make(svc, ["test"], env="prod").run(), []) + self.assertEqual(svc.deleted, []) + self.assertEqual(len(self.make(svc, ["prod", "SİL"], env="prod").run()), 1) + + def test_backup_written_before_delete_and_failure_blocks(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + svc = self.Svc() + backup = Path(tmp) / "b.csv" + results = self.make(svc, ["SİL"], backup=backup).run() + self.assertEqual(results[0]["status"], "SUCCESS") + self.assertIn(REG_ID, backup.read_text(encoding="utf-8-sig")) + svc2 = self.Svc() + with self.assertRaises(OSError): + self.make(svc2, ["SİL"], backup=Path(tmp) / "no" / "b.csv").run() + self.assertEqual(svc2.deleted, []) + + if __name__ == "__main__": unittest.main() From 25569efc80a9a772bc75e8c2c0a55a62dd151d57 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:14:31 +0300 Subject: [PATCH 07/14] docs: document nowcast point alarm clients and paging caveats --- README.md | 56 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/README.md b/README.md index 0233ff9..acc4841 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,8 @@ bu ortak yapıyı kullanarak noktalı virgüllü CSV satırlarını sırayla iş - `common/`: tekrar kullanılabilir API, HMAC, retry ve authentication katmanı - `login.py`: manuel login komutu - `nowcast-geo-register.py`: CSV geo alarm registration istemcisi +- `nowcast-point-page.py`, `nowcast-point-get.py`, `nowcast-point-register.py`, `nowcast-point-update.py`, `nowcast-point-unregister.py`: nokta alarm (`/v1/alarms/points`) istemcileri +- `nowcast-point-alarm-filter.example.json`, `nowcast-point-registrations.example.csv`, `nowcast-point-update.example.csv`: nokta alarm şablonları - `.env.example`: güvenli ortam değişkeni şablonu - `webhook.example.json`: dört webhook authentication seçeneğini içeren şablon - `nowcast-geo-alarm-filter.example.json`: ortak nowcast geo alarm filtreleri şablonu @@ -151,3 +153,57 @@ python nowcast-geo-register.py nowcast-geo-alarm-registrations.csv \ --nowcast-filter-config another-nowcast-geo-alarm-filter.json \ --debug-log logs/custom-debug.log ``` + +## Nokta alarm (nowcast point) istemcileri + +Nokta alarm istemcileri `BE-NowcastPointAlarm` servisinin `/v1/alarms/points` endpoint'lerini çağırır. Tüm komutlarda `--help` seçeneklerin tam dökümünü verir. Gerçek filtre değerleri için `nowcast-point-alarm-filter.example.json` dosyasını `nowcast-point-alarm-filter.json` olarak kopyalayın (git'e eklenmez). Webhook için mevcut `webhook.json` ve `IKLIM_WEBHOOK_INITIALIZED` mekanizması aynen kullanılır. + +| Komut | Ne yapar | +| --- | --- | +| `nowcast-point-page.py` | Kayıtları sayfa sayfa listeler; SQLite'a (`iklim-sandbox.db`, `registrationId` anahtarlı) yazar, ekrana basar veya `--output` ile CSV/TXT üretir | +| `nowcast-point-get.py` | CSV/TXT içindeki `registrationId` (`--by id`) veya `recipientId` (`--by recipient`) değerlerini sorgular | +| `nowcast-point-register.py` | CSV satırlarından kayıt oluşturur | +| `nowcast-point-update.py` | CSV satırlarıyla kayıtları kısmen günceller (`--dry-run` destekler) | +| `nowcast-point-unregister.py` | Listedeki kayıtları siler (onay ve yedekle) | + +### Önemli uyarılar + +- **Page endpoint'inde sıralama (ORDER BY) yoktur.** Tüm sayfalar gezilirken kayıtların sırası garanti değildir; bu yüzden sayfa kayması yaşanabilir (bir kayıt iki kez gelebilir, bir kayıt hiç gelmeyebilir). Backend tarafında `createdAt` ardından `id` ile sıralama önerildi (yeni kayıtlar sona eklenir, `id` eşit zamanlarda sırayı sabitler). +- **Tüm kayıtlar çekilirken aynı anda register, update veya unregister yapmayın.** Veri değişirse sayfalar kayar. `nowcast-point-page.py` sonunda sunucunun bildirdiği `totalRecords` ile alınan benzersiz kayıt sayısını karşılaştırır ve fark varsa uyarı verir; bu durumda komutu yeniden çalıştırın. SQLite `registrationId` ile upsert yaptığı için tekrar çalıştırmak kayıt çoğaltmaz. +- `accountId` verilmezse ADMIN kullanıcılar tüm hesapları görür. Betik her zaman `/v1/users/me` içindeki `account.id` değerini gönderir (`--account-id` ile değiştirilebilir). Yetki yoksa (403) uyarı verip `accountId` olmadan bir kez daha dener. +- Backend her `RuntimeException`'ı HTTP 400 yapar; kayıt bulunamaması da 400 döner. Yalnızca gerçek 404 `NOT_FOUND` olarak işaretlenir. + +### page + +```bash +python3 nowcast-point-page.py # tüm sayfalar, ekrana +python3 nowcast-point-page.py --output kayitlar.csv # kayitlar-YYYYmmdd-HHMMSS.csv +python3 nowcast-point-page.py --page-number 2 --page-size 50 # yalnızca tek sayfa +python3 nowcast-point-page.py --recipient-id musteri-1 --recipient-id musteri-2 +python3 nowcast-point-page.py --polygon "32.8,39.9;32.9,39.9;32.9,40.0;32.8,39.9" +``` + +- Poligon `lng,lat` çiftlerinden oluşan **düz halkadır**, en az 4 nokta ve kapalı olmalıdır. Filtre, kayıt noktasının poligonla kesişimine bakar; yarıçap yok sayılır. +- Çıktıda webhook bilgisi dosya başında bir kez yer alır (CSV'de `#webhook` satırı); `password`, `jwtToken`, `apiKey` ve `secret` alanları `***` ile maskelenir. Her filtre ayrı bir sütundur. API yanıtında `createdAt`/`updatedAt` bulunmadığı için çıktıda da yoktur. +- Etkileşimli terminalde ve `--output` verilmediğinde sonunda "SQLite verisi CSV/TXT dosyasına aktarılsın mı?" sorulur. + +### get + +Girdi CSV ise başlıkta `registrationId` (veya `recipientId`) sütunu aranır; TXT ise satır başına bir ID yazılır, boş satırlar ve `#` ile başlayanlar atlanır. `--output` zorunludur; uzantı `.csv` ise noktalı virgüllü CSV, `.txt` ise JSON Lines yazılır. Satır durumları `SUCCESS`, `NOT_FOUND` (HTTP 404) ve `FAILED`'dır (`httpStatus` ve `errorBody` dolar). Bir `recipientId` birden çok kayıt dönerse her kayıt ayrı satırdır. + +### register + +CSV sütunları yalnızca `recipientId;lat;lng;radius` değerleridir (`radius` metre, 0-50000). Filtre CSV'de tekrarlanmaz; ortak `nowcast-point-alarm-filter.json` dosyasından okunur ve her kayda uygulanır (`--filter-config` ile değiştirilir). Sonuç sütunları aynı dosyaya yazılır: `status;httpStatus;registrationId;errorBody`. HTTP 200 dışındaki yanıtlarda cevap gövdesi `errorBody` içine yazılır. `SUCCESS` satırlar yeniden çalıştırmada atlanır. Sonucu belirsiz kalan istek yalnızca `recipientId`, nokta ve yarıçap birebir eşleşirse mevcut kayıtla kurtarılır. Aynı hesap, `recipientId`, nokta ve yarıçap ikinci kez kaydedilemez (400). + +### update + +CSV'de `registrationId` zorunludur; `recipientId;lat;lng;radius` ve filtre sütunları (`lightningType;peakCurrent;inCloudHeight;intersectsAffectedPolygon;intersectsCellPolygon;severityThreshold;speedThreshold;precipitationIntensities`, liste `|` ile ayrılır) isteğe bağlıdır. **Boş hücre gönderilmez**; `lat` ve `lng` birlikte verilmelidir. Önce `--dry-run` ile gövdeleri kontrol edin. + +- Backend PATCH'i bir alanı **temizleyemez** (null = dokunma). +- `thunderstorm.speedThreshold` backend PATCH'inde işlenmiyor, değer gönderilse de güncellenmeyebilir (GAIA-476). +- Backend sınır güncellemesinde doğrulama yapmaz; yarıçap ve koordinat aralıkları istemcide yerelde doğrulanır. +- Webhook güncellemesi bu sürümde desteklenmez. + +### unregister + +Başlamadan önce ortam adı ve URL, silinecek kayıt sayısı ve "geri alınamaz" uyarısı gösterilir; bir hesabın **son kaydı** silinirse webhook tanımı da silinir. Onay için `SİL` yazılmalıdır (`--yes` yoktur). `prod` ortamında ek olarak ortam adı (`prod`) yazdırılır. Silmeden önce her kayıt `get-by-id` ile yedeklenir (`--backup`, kapatmak için `--no-backup`); yedek yazılamazsa hiçbir kayıt silinmez. From aaa0c99eb29ad89670b21a84105f30cb74ee452d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:31:42 +0300 Subject: [PATCH 08/14] fix: stop sending accountId implicitly from the page client The page client no longer resolves accountId from /v1/users/me. accountId is only sent when passed with --account-id, and a 403 for it now raises a clear error telling the user to omit it because the server scopes the listing to the caller's own account. --- .gitignore | 1 + README.md | 2 +- nowcast-point-page.py | 14 ++++++++------ point_alarm.py | 24 +++++++++++++----------- tests/test_point_alarm.py | 7 ++++--- 5 files changed, 27 insertions(+), 21 deletions(-) diff --git a/.gitignore b/.gitignore index 895bca1..b88fa8b 100644 --- a/.gitignore +++ b/.gitignore @@ -41,3 +41,4 @@ Thumbs.db *.db nowcast-point-alarm-filter.json nowcast-point-*-backup-* +kayitlar-*.csv diff --git a/README.md b/README.md index acc4841..4e2bc89 100644 --- a/README.md +++ b/README.md @@ -170,7 +170,7 @@ Nokta alarm istemcileri `BE-NowcastPointAlarm` servisinin `/v1/alarms/points` en - **Page endpoint'inde sıralama (ORDER BY) yoktur.** Tüm sayfalar gezilirken kayıtların sırası garanti değildir; bu yüzden sayfa kayması yaşanabilir (bir kayıt iki kez gelebilir, bir kayıt hiç gelmeyebilir). Backend tarafında `createdAt` ardından `id` ile sıralama önerildi (yeni kayıtlar sona eklenir, `id` eşit zamanlarda sırayı sabitler). - **Tüm kayıtlar çekilirken aynı anda register, update veya unregister yapmayın.** Veri değişirse sayfalar kayar. `nowcast-point-page.py` sonunda sunucunun bildirdiği `totalRecords` ile alınan benzersiz kayıt sayısını karşılaştırır ve fark varsa uyarı verir; bu durumda komutu yeniden çalıştırın. SQLite `registrationId` ile upsert yaptığı için tekrar çalıştırmak kayıt çoğaltmaz. -- `accountId` verilmezse ADMIN kullanıcılar tüm hesapları görür. Betik her zaman `/v1/users/me` içindeki `account.id` değerini gönderir (`--account-id` ile değiştirilebilir). Yetki yoksa (403) uyarı verip `accountId` olmadan bir kez daha dener. +- `nowcast-point-page.py` varsayılan olarak `accountId` göndermez; sunucu listeyi çağıranın kendi hesabına daraltır ve normal kullanıcılar için doğru kullanım budur. `--account-id` yalnızca en az POWER_USER yetkisiyle verilebilir; aksi halde HTTP 403 döner ve betik `--account-id` vermemenizi söyleyen bir hata yazar. ADMIN kullanıcı `accountId` vermezse tüm hesapları görür. - Backend her `RuntimeException`'ı HTTP 400 yapar; kayıt bulunamaması da 400 döner. Yalnızca gerçek 404 `NOT_FOUND` olarak işaretlenir. ### page diff --git a/nowcast-point-page.py b/nowcast-point-page.py index fd477c1..78b91d7 100644 --- a/nowcast-point-page.py +++ b/nowcast-point-page.py @@ -36,7 +36,10 @@ def parse_args() -> argparse.Namespace: ) parser.add_argument( "--account-id", - help="Sorgulanacak hesap ID'si (varsayılan: /v1/users/me içindeki account.id)", + help=( + "Sorgulanacak hesap ID'si; yalnızca en az POWER_USER yetkisiyle verilebilir (aksi halde " + "HTTP 403). Verilmezse sunucu kendi hesabınıza daraltır; normal kullanıcılar vermemelidir" + ), ) parser.add_argument( "--recipient-id", @@ -95,19 +98,18 @@ def main() -> int: 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, + account_id=args.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) + records = store.records(None, summary.registration_ids) + webhooks = store.webhooks_of(records) if args.output: path = timestamped_path(args.output) write_records(path, records, webhooks) @@ -125,7 +127,7 @@ def main() -> int: 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) + prompt_export(store, args.account_id) except (ConfigurationError, ApiError, OSError) as exc: logging.error("İşlem durduruldu: %s", exc) return 1 diff --git a/point_alarm.py b/point_alarm.py index 965692e..0f34cdc 100644 --- a/point_alarm.py +++ b/point_alarm.py @@ -336,6 +336,11 @@ class PointStore: query += " ORDER BY recipientId, registrationId" return [dict(row) for row in self.connection.execute(query, params)] + def webhooks_of(self, records: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Webhooks of the accounts that own the given records.""" + owners = sorted({record["accountId"] for record in records if record.get("accountId")}) + return [webhook for owner in owners for webhook in self.webhooks(owner)] + def webhooks(self, account_id: str | None = None) -> list[dict[str, Any]]: if account_id: rows = self.connection.execute( @@ -432,8 +437,7 @@ def fetch_pages( ) -> 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. + Without account_id the server scopes the listing to the caller's own account. """ summary = FetchSummary(account_id=account_id) seen: set[str] = set() @@ -443,24 +447,22 @@ def fetch_pages( data = service.page( page_number=current, page_size=page_size, - account_id=summary.account_id, + account_id=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 + if exc.status_code == 403 and account_id: + raise ConfigurationError( + "accountId parametresi için en az POWER_USER yetkisi gerekir (HTTP 403). " + "--account-id vermeyin; sistem hesabınızı kendisi tespit eder." + ) from exc 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) + registration_id = store.upsert_registration(registration, account_id) if registration_id and registration_id not in seen: seen.add(registration_id) summary.registration_ids.append(registration_id) diff --git a/tests/test_point_alarm.py b/tests/test_point_alarm.py index a715aef..4d7dbe7 100644 --- a/tests/test_point_alarm.py +++ b/tests/test_point_alarm.py @@ -97,11 +97,12 @@ class PointAlarmTest(unittest.TestCase): self.assertEqual(summary.received, 2) self.assertEqual(len(service.calls), 1) - def test_forbidden_account_retries_without_it(self) -> None: + def test_forbidden_account_gives_clear_error(self) -> None: service = FakeService([registration(1)], forbid_account=True) - summary = fetch_pages(service, self.store, account_id="acc") + with self.assertRaisesRegex(ConfigurationError, "--account-id vermeyin"): + fetch_pages(service, self.store, account_id="acc") + summary = fetch_pages(service, self.store, account_id=None) 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") From 96482c0dd141ca2b07272cf78bb7c51aee4827a1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:33:04 +0300 Subject: [PATCH 09/14] docs: add accountId example to the page client usage --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index 4e2bc89..f83db9a 100644 --- a/README.md +++ b/README.md @@ -179,6 +179,7 @@ Nokta alarm istemcileri `BE-NowcastPointAlarm` servisinin `/v1/alarms/points` en python3 nowcast-point-page.py # tüm sayfalar, ekrana python3 nowcast-point-page.py --output kayitlar.csv # kayitlar-YYYYmmdd-HHMMSS.csv python3 nowcast-point-page.py --page-number 2 --page-size 50 # yalnızca tek sayfa +python3 nowcast-point-page.py --account-id 2f04f1b5-1c0a-4c4e-b0a7-0ba6e9b7f2e1 # yalnızca POWER_USER ve üstü python3 nowcast-point-page.py --recipient-id musteri-1 --recipient-id musteri-2 python3 nowcast-point-page.py --polygon "32.8,39.9;32.9,39.9;32.9,40.0;32.8,39.9" ``` From a2ccaa287397d8967282022507294119d18b96aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:34:14 +0300 Subject: [PATCH 10/14] docs: use point-alarm-registrations as the example output name --- .gitignore | 2 +- README.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.gitignore b/.gitignore index b88fa8b..2731b54 100644 --- a/.gitignore +++ b/.gitignore @@ -41,4 +41,4 @@ Thumbs.db *.db nowcast-point-alarm-filter.json nowcast-point-*-backup-* -kayitlar-*.csv +point-alarm-registrations-*.csv diff --git a/README.md b/README.md index f83db9a..7fa7356 100644 --- a/README.md +++ b/README.md @@ -177,7 +177,7 @@ Nokta alarm istemcileri `BE-NowcastPointAlarm` servisinin `/v1/alarms/points` en ```bash python3 nowcast-point-page.py # tüm sayfalar, ekrana -python3 nowcast-point-page.py --output kayitlar.csv # kayitlar-YYYYmmdd-HHMMSS.csv +python3 nowcast-point-page.py --output point-alarm-registrations.csv # point-alarm-registrations-YYYYmmdd-HHMMSS.csv python3 nowcast-point-page.py --page-number 2 --page-size 50 # yalnızca tek sayfa python3 nowcast-point-page.py --account-id 2f04f1b5-1c0a-4c4e-b0a7-0ba6e9b7f2e1 # yalnızca POWER_USER ve üstü python3 nowcast-point-page.py --recipient-id musteri-1 --recipient-id musteri-2 From 8d45cddcedd46ad2044e3cccde0ac0f3e1a05368 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:34:58 +0300 Subject: [PATCH 11/14] docs: add usage examples for every point alarm client --- README.md | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/README.md b/README.md index 7fa7356..46d9e6c 100644 --- a/README.md +++ b/README.md @@ -190,14 +190,55 @@ python3 nowcast-point-page.py --polygon "32.8,39.9;32.9,39.9;32.9,40.0;32.8,39.9 ### get +```bash +python3 nowcast-point-get.py idler.txt --output sonuc.csv # registrationId listesi (TXT), CSV çıktı +python3 nowcast-point-get.py idler.csv --output sonuc.txt # registrationId sütunlu CSV, JSON Lines çıktı +python3 nowcast-point-get.py alicilar.txt --by recipient --output sonuc.csv # recipientId ile sorgu +``` + +TXT girdi örneği (`idler.txt`): + +```text +# bu satır atlanır +f7587d9e-2481-4b4c-818d-c8d1946851b7 +57d1e2ba-bf2d-4ef3-b078-523060b2000c +``` + Girdi CSV ise başlıkta `registrationId` (veya `recipientId`) sütunu aranır; TXT ise satır başına bir ID yazılır, boş satırlar ve `#` ile başlayanlar atlanır. `--output` zorunludur; uzantı `.csv` ise noktalı virgüllü CSV, `.txt` ise JSON Lines yazılır. Satır durumları `SUCCESS`, `NOT_FOUND` (HTTP 404) ve `FAILED`'dır (`httpStatus` ve `errorBody` dolar). Bir `recipientId` birden çok kayıt dönerse her kayıt ayrı satırdır. ### register +```bash +cp nowcast-point-alarm-filter.example.json nowcast-point-alarm-filter.json # filtreleri düzenleyin +python3 nowcast-point-register.py nowcast-point-registrations.example.csv +python3 nowcast-point-register.py kayitlar.csv --filter-config baska-filtre.json --webhook-config baska-webhook.json +``` + +Girdi CSV örneği (sonuç sütunları işlem sonunda aynı dosyaya eklenir): + +```csv +recipientId;lat;lng;radius +musteri-001;39.9208;32.8541;1000 +musteri-002;41.0082;28.9784;5000 +``` + CSV sütunları yalnızca `recipientId;lat;lng;radius` değerleridir (`radius` metre, 0-50000). Filtre CSV'de tekrarlanmaz; ortak `nowcast-point-alarm-filter.json` dosyasından okunur ve her kayda uygulanır (`--filter-config` ile değiştirilir). Sonuç sütunları aynı dosyaya yazılır: `status;httpStatus;registrationId;errorBody`. HTTP 200 dışındaki yanıtlarda cevap gövdesi `errorBody` içine yazılır. `SUCCESS` satırlar yeniden çalıştırmada atlanır. Sonucu belirsiz kalan istek yalnızca `recipientId`, nokta ve yarıçap birebir eşleşirse mevcut kayıtla kurtarılır. Aynı hesap, `recipientId`, nokta ve yarıçap ikinci kez kaydedilemez (400). ### update +```bash +python3 nowcast-point-update.py guncelle.csv --dry-run # gövdeleri gösterir, istek göndermez +python3 nowcast-point-update.py guncelle.csv # gerçekten günceller +``` + +Girdi CSV örneği (boş hücre gönderilmez): + +```csv +registrationId;recipientId;lat;lng;radius;lightningType;peakCurrent;inCloudHeight;intersectsAffectedPolygon;intersectsCellPolygon;severityThreshold;speedThreshold;precipitationIntensities +f7587d9e-2481-4b4c-818d-c8d1946851b7;;;;2500;;;;;;HIGH;;LIGHT|MODERATE +57d1e2ba-bf2d-4ef3-b078-523060b2000c;;40.0;33.0;;PULSE_IN_CLOUD;;;;true;;; +``` + CSV'de `registrationId` zorunludur; `recipientId;lat;lng;radius` ve filtre sütunları (`lightningType;peakCurrent;inCloudHeight;intersectsAffectedPolygon;intersectsCellPolygon;severityThreshold;speedThreshold;precipitationIntensities`, liste `|` ile ayrılır) isteğe bağlıdır. **Boş hücre gönderilmez**; `lat` ve `lng` birlikte verilmelidir. Önce `--dry-run` ile gövdeleri kontrol edin. - Backend PATCH'i bir alanı **temizleyemez** (null = dokunma). @@ -207,4 +248,13 @@ CSV'de `registrationId` zorunludur; `recipientId;lat;lng;radius` ve filtre sütu ### unregister +```bash +python3 nowcast-point-unregister.py silinecekler.txt # yedek alır, SİL onayı ister +python3 nowcast-point-unregister.py silinecekler.csv --output sonuc.csv # sonuçları dosyaya da yazar +python3 nowcast-point-unregister.py silinecekler.txt --backup yedek/point-yedek.csv # yedek yolu +python3 nowcast-point-unregister.py silinecekler.txt --no-backup # yedeksiz (önerilmez) +``` + +Girdi `registrationId` içeren bir CSV veya satır başına bir ID içeren TXT olabilir. + Başlamadan önce ortam adı ve URL, silinecek kayıt sayısı ve "geri alınamaz" uyarısı gösterilir; bir hesabın **son kaydı** silinirse webhook tanımı da silinir. Onay için `SİL` yazılmalıdır (`--yes` yoktur). `prod` ortamında ek olarak ortam adı (`prod`) yazdırılır. Silmeden önce her kayıt `get-by-id` ile yedeklenir (`--backup`, kapatmak için `--no-backup`); yedek yazılamazsa hiçbir kayıt silinmez. From f04fd728e0e4e5637592594599b4d964b48c0986 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:37:30 +0300 Subject: [PATCH 12/14] chore: ignore working copies of the point alarm example CSVs --- .gitignore | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.gitignore b/.gitignore index 2731b54..36adfde 100644 --- a/.gitignore +++ b/.gitignore @@ -42,3 +42,5 @@ Thumbs.db nowcast-point-alarm-filter.json nowcast-point-*-backup-* point-alarm-registrations-*.csv +nowcast-point-registrations.csv +nowcast-point-update.csv From e17c06da4a4ae0f491c8d2007fa4c4f886961657 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:44:58 +0300 Subject: [PATCH 13/14] fix: send the full webhook and skip /users/me in the point alarm register client The backend requires a webhook with a delivery policy on every point alarm register call and resolves the account from the caller, so the register client no longer calls /v1/users/me (account is null for some users), no longer sends the accountId-only webhook and no longer relies on IKLIM_WEBHOOK_INITIALIZED. --- .gitignore | 1 + README.md | 2 +- point_alarm.py | 21 +++++++-------------- tests/test_point_alarm.py | 15 ++++++++------- 4 files changed, 17 insertions(+), 22 deletions(-) diff --git a/.gitignore b/.gitignore index 36adfde..f1cf96c 100644 --- a/.gitignore +++ b/.gitignore @@ -44,3 +44,4 @@ nowcast-point-*-backup-* point-alarm-registrations-*.csv nowcast-point-registrations.csv nowcast-point-update.csv +nowcast-point-alarm-records-*.csv diff --git a/README.md b/README.md index 46d9e6c..05f532a 100644 --- a/README.md +++ b/README.md @@ -156,7 +156,7 @@ python nowcast-geo-register.py nowcast-geo-alarm-registrations.csv \ ## Nokta alarm (nowcast point) istemcileri -Nokta alarm istemcileri `BE-NowcastPointAlarm` servisinin `/v1/alarms/points` endpoint'lerini çağırır. Tüm komutlarda `--help` seçeneklerin tam dökümünü verir. Gerçek filtre değerleri için `nowcast-point-alarm-filter.example.json` dosyasını `nowcast-point-alarm-filter.json` olarak kopyalayın (git'e eklenmez). Webhook için mevcut `webhook.json` ve `IKLIM_WEBHOOK_INITIALIZED` mekanizması aynen kullanılır. +Nokta alarm istemcileri `BE-NowcastPointAlarm` servisinin `/v1/alarms/points` endpoint'lerini çağırır. Tüm komutlarda `--help` seçeneklerin tam dökümünü verir. Gerçek filtre değerleri için `nowcast-point-alarm-filter.example.json` dosyasını `nowcast-point-alarm-filter.json` olarak kopyalayın (git'e eklenmez). Webhook için mevcut `webhook.json` kullanılır, ancak geo alarmdan farklı olarak her register isteğinde `webhook.json` içindeki tam webhook (`deliveryPolicy` dahil) gönderilir; `IKLIM_WEBHOOK_INITIALIZED` ve `/v1/users/me` kullanılmaz. Backend nokta alarm kaydında webhook ve `deliveryPolicy` zorunlu tutar, hesabı kullanıcıdan kendisi bulur ve hesapta kayıtlı webhook varsa onu yeniden kullanır. | Komut | Ne yapar | | --- | --- | diff --git a/point_alarm.py b/point_alarm.py index 0f34cdc..4e64d8f 100644 --- a/point_alarm.py +++ b/point_alarm.py @@ -18,10 +18,8 @@ from uuid import uuid4 from common import ApiError, ConfigurationError, IklimClient from common.client import ApiResponse from common.client import RETRYABLE_STATUS_CODES -from common.config import env_bool from nowcast_common import ( as_uuid, - fetch_account_id, load_webhook_config, read_json_object, response_warnings, @@ -355,9 +353,6 @@ 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, *, @@ -652,10 +647,11 @@ class PointRegistrationProcessor: def run(self) -> None: rows, fieldnames = self._read_csv() filters = load_point_alarm_filter(self.filter_config_path) - webhook_initialized = env_bool("IKLIM_WEBHOOK_INITIALIZED", False) - full_webhook = None if webhook_initialized else load_webhook_config(self.webhook_config_path) - account_id = self.service.get_account_id() - LOGGER.info("Kullanıcı hesabı doğrulandı; accountId API'den alındı.") + # The backend requires a webhook with a delivery policy on every register call; it reuses + # the account's stored webhook and resolves the account from the caller, so accountId is not sent. + webhook = load_webhook_config(self.webhook_config_path) + if not isinstance(webhook.get("deliveryPolicy"), dict): + raise ConfigurationError("Nokta alarm kaydı için webhook.deliveryPolicy zorunludur.") for row_number, row in enumerate(rows, start=2): if str(row.get("status", "")).strip().upper() == "SUCCESS": LOGGER.info("Satır %s zaten SUCCESS; atlandı.", row_number) @@ -670,13 +666,10 @@ class PointRegistrationProcessor: payload = { "recipientId": recipient_id, "boundary": {"point": {"lat": lat, "lng": lng}, "radius": radius}, - "webhook": {"accountId": account_id} if webhook_initialized else deepcopy(full_webhook), + "webhook": deepcopy(webhook), "filter": deepcopy(filters), } - registered = self._register(row_number, payload, row) - if registered and not webhook_initialized: - webhook_initialized = True - self.service.client.token_store.set_webhook_initialized(True) + self._register(row_number, payload, row) self._write_csv(rows, fieldnames) def _register(self, row_number: int, payload: dict[str, Any], row: dict[str, Any]) -> bool: diff --git a/tests/test_point_alarm.py b/tests/test_point_alarm.py index 4d7dbe7..938994d 100644 --- a/tests/test_point_alarm.py +++ b/tests/test_point_alarm.py @@ -176,9 +176,6 @@ class FakeRegisterService: self.payloads: list[dict] = [] self.recipient_hits: list = [] - def get_account_id(self): - return "acc" - def register(self, payload, key): self.payloads.append(payload) result = self.results.pop(0) @@ -199,7 +196,11 @@ class RegisterTest(unittest.TestCase): self.filter = self.dir / "filter.json" self.filter.write_text(json.dumps({"lightning": {"type": "PULSE_IN_CLOUD"}}), encoding="utf-8") self.webhook = self.dir / "webhook.json" - self.webhook.write_text("{}", encoding="utf-8") + self.webhook.write_text(json.dumps({ + "url": "https://example.com/hook", "httpMethod": "POST", "contentType": "application/json", + "authentication": {"selected": "JWT_TOKEN", "options": {"JWT_TOKEN": {"jwtToken": "t"}}}, + "deliveryPolicy": {"maxAttempts": 3}, + }), encoding="utf-8") self.csv = self.dir / "r.csv" def run_processor(self, rows: str | None, service) -> list[dict]: @@ -208,8 +209,7 @@ class RegisterTest(unittest.TestCase): processor = PointRegistrationProcessor( service, csv_path=self.csv, webhook_config_path=self.webhook, filter_config_path=self.filter ) - with patch.dict(os.environ, {"IKLIM_WEBHOOK_INITIALIZED": "true"}): - processor.run() + processor.run() with self.csv.open(encoding="utf-8-sig", newline="") as handle: return list(csv.DictReader(handle, delimiter=";")) @@ -226,7 +226,8 @@ class RegisterTest(unittest.TestCase): self.assertEqual(rows[0]["registrationId"], REG_ID) self.assertIn("duplicate", rows[1]["errorBody"]) self.assertEqual(rows[2]["httpStatus"], "NOT_SENT") - self.assertEqual(service.payloads[0]["webhook"], {"accountId": "acc"}) + self.assertEqual(service.payloads[0]["webhook"]["deliveryPolicy"], {"maxAttempts": 3}) + self.assertNotIn("accountId", service.payloads[0]["webhook"]) self.assertEqual(service.payloads[0]["filter"]["lightning"]["type"], "PULSE_IN_CLOUD") # Re-run: SUCCESS row is skipped, the failed one is retried. service2 = FakeRegisterService( From d2b430a524724d7bca1c207fb7a578a32de4b667 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Murat=20=C3=96ZDEM=C4=B0R?= Date: Fri, 23 Oct 2026 19:47:55 +0300 Subject: [PATCH 14/14] fix: number point alarm register and update log lines by record instead of file line --- point_alarm.py | 48 +++++++++++++++++++++++++----------------------- 1 file changed, 25 insertions(+), 23 deletions(-) diff --git a/point_alarm.py b/point_alarm.py index 4e64d8f..23267a6 100644 --- a/point_alarm.py +++ b/point_alarm.py @@ -652,14 +652,15 @@ class PointRegistrationProcessor: webhook = load_webhook_config(self.webhook_config_path) if not isinstance(webhook.get("deliveryPolicy"), dict): raise ConfigurationError("Nokta alarm kaydı için webhook.deliveryPolicy zorunludur.") - for row_number, row in enumerate(rows, start=2): + for index, row in enumerate(rows, start=1): + label = f"{index}/{len(rows)}" if str(row.get("status", "")).strip().upper() == "SUCCESS": - LOGGER.info("Satır %s zaten SUCCESS; atlandı.", row_number) + LOGGER.info("Kayıt %s zaten SUCCESS; atlandı.", label) continue try: recipient_id, lat, lng, radius = parse_point_row(row) except ConfigurationError as exc: - LOGGER.error("Satır %s gönderilmedi: %s", row_number, exc) + LOGGER.error("Kayıt %s gönderilmedi: %s", label, exc) self._set_result(row, "FAILED", "NOT_SENT", "", str(exc)) self._write_csv(rows, fieldnames) continue @@ -669,17 +670,17 @@ class PointRegistrationProcessor: "webhook": deepcopy(webhook), "filter": deepcopy(filters), } - self._register(row_number, payload, row) + self._register(label, payload, row) self._write_csv(rows, fieldnames) - def _register(self, row_number: int, payload: dict[str, Any], row: dict[str, Any]) -> bool: + def _register(self, label: str, payload: dict[str, Any], row: dict[str, Any]) -> bool: try: response = self.service.register(payload, str(uuid4())) except ApiError as exc: - recovered = self._try_recover(row_number, payload, exc) + recovered = self._try_recover(label, payload, exc) if recovered: self._set_result(row, "SUCCESS", f"HTTP_{exc.status_code or recovered[0]}", recovered[1], "") - LOGGER.info("Satır %s eşleşen kayıtla kurtarıldı: %s", row_number, recovered[1]) + LOGGER.info("Kayıt %s eşleşen kayıtla kurtarıldı: %s", label, recovered[1]) return True self._set_result( row, @@ -688,7 +689,7 @@ class PointRegistrationProcessor: "", error_body_text(exc), ) - LOGGER.error("Satır %s başarısız: %s", row_number, exc) + LOGGER.error("Kayıt %s başarısız: %s", label, exc) return False data = response.data if isinstance(response.data, dict) else {} registration_id = as_uuid(data.get("registrationId"), version=4) @@ -697,27 +698,27 @@ class PointRegistrationProcessor: row, "FAILED", f"HTTP_{response.status_code}", "", "Yanıtta UUID v4 registrationId bulunamadı: " + json.dumps(response.data, ensure_ascii=False), ) - LOGGER.error("Satır %s yanıtında registrationId yok.", row_number) + LOGGER.error("Kayıt %s yanıtında registrationId yok.", label) return False self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", registration_id, "") - LOGGER.info("Satır %s başarıyla kaydedildi: %s", row_number, registration_id) + LOGGER.info("Kayıt %s başarıyla kaydedildi: %s", label, registration_id) for warning in response_warnings(response.data): - LOGGER.warning("Satır %s API uyarısı: %s", row_number, warning) + LOGGER.warning("Kayıt %s API uyarısı: %s", label, warning) return True def _try_recover( - self, row_number: int, payload: dict[str, Any], error: ApiError + self, label: str, payload: dict[str, Any], error: ApiError ) -> tuple[int, str] | None: """After an ambiguous failure, look for an identical registration (recipient + point + radius).""" status = error.status_code if status is not None and status != 409 and status not in RETRYABLE_STATUS_CODES: return None - LOGGER.warning("Satır %s sonucu belirsiz; recipientId ile kontrol ediliyor.", row_number) + LOGGER.warning("Kayıt %s sonucu belirsiz; recipientId ile kontrol ediliyor.", label) boundary = payload["boundary"] try: response = self.service.get_by_recipient_id(payload["recipientId"]) except ApiError as exc: - LOGGER.error("Satır %s için recipient sorgusu başarısız: %s", row_number, exc) + LOGGER.error("Kayıt %s için recipient sorgusu başarısız: %s", label, exc) return None data = response.data if isinstance(response.data, dict) else {} for registration in data.get("registrations") or []: @@ -881,25 +882,26 @@ class PointUpdateProcessor: def run(self) -> None: rows, fieldnames = self._read_csv() - for row_number, row in enumerate(rows, start=2): + for index, row in enumerate(rows, start=1): + label = f"{index}/{len(rows)}" if str(row.get("status", "")).strip().upper() == "SUCCESS": - LOGGER.info("Satır %s zaten SUCCESS; atlandı.", row_number) + LOGGER.info("Kayıt %s zaten SUCCESS; atlandı.", label) continue try: payload = build_update_payload(row) except ConfigurationError as exc: - LOGGER.error("Satır %s gönderilmedi: %s", row_number, exc) + LOGGER.error("Kayıt %s gönderilmedi: %s", label, exc) self._set_result(row, "FAILED", "NOT_SENT", str(exc)) self._write(rows, fieldnames) continue if payload.get("filter", {}).get("thunderstorm", {}).get("speedThreshold") is not None: LOGGER.warning( - "Satır %s: thunderstorm.speedThreshold PATCH'te backend tarafından işlenmiyor " + "Kayıt %s: thunderstorm.speedThreshold PATCH'te backend tarafından işlenmiyor " "(GAIA-476); değer gönderilir ancak güncellenmeyebilir.", - row_number, + label, ) if self.dry_run: - print(f"[dry-run] satır {row_number}: PATCH {ALARMS_PATH}/update {json.dumps(payload, ensure_ascii=False)}") + print(f"[dry-run] kayıt {label}: PATCH {ALARMS_PATH}/update {json.dumps(payload, ensure_ascii=False)}") continue try: response = self.service.update(payload, str(uuid4())) @@ -909,12 +911,12 @@ class PointUpdateProcessor: f"HTTP_{exc.status_code}" if exc.status_code else "NETWORK_ERROR", error_body_text(exc), ) - LOGGER.error("Satır %s başarısız: %s", row_number, exc) + LOGGER.error("Kayıt %s başarısız: %s", label, exc) else: self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", "") - LOGGER.info("Satır %s güncellendi: %s", row_number, payload["registrationId"]) + LOGGER.info("Kayıt %s güncellendi: %s", label, payload["registrationId"]) for warning in response_warnings(response.data): - LOGGER.warning("Satır %s API uyarısı: %s", row_number, warning) + LOGGER.warning("Kayıt %s API uyarısı: %s", label, warning) self._write(rows, fieldnames) def _read_csv(self) -> tuple[list[dict[str, Any]], list[str]]: