diff --git a/.gitignore b/.gitignore index 71c6176..f1cf96c 100644 --- a/.gitignore +++ b/.gitignore @@ -36,3 +36,12 @@ dist/ .vscode/ .DS_Store Thumbs.db + +# Point alarm client artifacts +*.db +nowcast-point-alarm-filter.json +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 0233ff9..05f532a 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,108 @@ 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` 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 | +| --- | --- | +| `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. +- `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 + +```bash +python3 nowcast-point-page.py # tüm sayfalar, ekrana +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 +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 + +```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). +- `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 + +```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. 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/geo_alarm.py b/geo_alarm.py index 83b38da..c47461a 100644 --- a/geo_alarm.py +++ b/geo_alarm.py @@ -3,31 +3,30 @@ 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, + fetch_account_id, + 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 +34,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,44 +82,12 @@ 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 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 _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( @@ -227,14 +119,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 +152,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 +353,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-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-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/nowcast-point-page.py b/nowcast-point-page.py new file mode 100644 index 0000000..78b91d7 --- /dev/null +++ b/nowcast-point-page.py @@ -0,0 +1,141 @@ +#!/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; 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", + 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) + store = PointStore(args.db) + summary = fetch_pages( + service, + store, + 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(None, summary.registration_ids) + webhooks = store.webhooks_of(records) + 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, args.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-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/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/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/nowcast_common.py b/nowcast_common.py new file mode 100644 index 0000000..8999993 --- /dev/null +++ b/nowcast_common.py @@ -0,0 +1,167 @@ +"""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 ApiError, ConfigurationError, IklimClient + + +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() + + +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..23267a6 --- /dev/null +++ b/point_alarm.py @@ -0,0 +1,1017 @@ +"""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 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 common.client import RETRYABLE_STATUS_CODES +from nowcast_common import ( + as_uuid, + load_webhook_config, + read_json_object, + response_warnings, + write_csv_atomic, +) + + +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_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( + "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 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. + + Without account_id the server 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=account_id, + recipient_ids=recipient_ids, + polygon=polygon, + ) + except ApiError as exc: + 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, 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 + + +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 + 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: + 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 + + +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) + # 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 index, row in enumerate(rows, start=1): + label = f"{index}/{len(rows)}" + if str(row.get("status", "")).strip().upper() == "SUCCESS": + 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("Kayıt %s gönderilmedi: %s", label, 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": deepcopy(webhook), + "filter": deepcopy(filters), + } + self._register(label, payload, row) + self._write_csv(rows, fieldnames) + + 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(label, payload, exc) + if recovered: + self._set_result(row, "SUCCESS", f"HTTP_{exc.status_code or recovered[0]}", recovered[1], "") + LOGGER.info("Kayıt %s eşleşen kayıtla kurtarıldı: %s", label, 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("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) + 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("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("Kayıt %s başarıyla kaydedildi: %s", label, registration_id) + for warning in response_warnings(response.data): + LOGGER.warning("Kayıt %s API uyarısı: %s", label, warning) + return True + + def _try_recover( + 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("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("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 []: + 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 + + +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 index, row in enumerate(rows, start=1): + label = f"{index}/{len(rows)}" + if str(row.get("status", "")).strip().upper() == "SUCCESS": + LOGGER.info("Kayıt %s zaten SUCCESS; atlandı.", label) + continue + try: + payload = build_update_payload(row) + except ConfigurationError as 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( + "Kayıt %s: thunderstorm.speedThreshold PATCH'te backend tarafından işlenmiyor " + "(GAIA-476); değer gönderilir ancak güncellenmeyebilir.", + label, + ) + if self.dry_run: + 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())) + 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("Kayıt %s başarısız: %s", label, exc) + else: + self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", "") + LOGGER.info("Kayıt %s güncellendi: %s", label, payload["registrationId"]) + for warning in response_warnings(response.data): + 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]]: + 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 + + +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 new file mode 100644 index 0000000..938994d --- /dev/null +++ b/tests/test_point_alarm.py @@ -0,0 +1,345 @@ +from __future__ import annotations + +import csv +import io +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, + PointUnregisterProcessor, + PointUpdateProcessor, + build_update_payload, + PointStore, + fetch_pages, + flatten_registration, + load_point_alarm_filter, + lookup_rows, + read_id_list, + 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_gives_clear_error(self) -> None: + service = FakeService([registration(1)], forbid_account=True) + 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) + + 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) + + 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, {"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"]) + 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) + + +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 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(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]: + 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 + ) + 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"]["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( + 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") + + +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")) + + +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()