"""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