"""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 row_number, row in enumerate(rows, start=2): if str(row.get("status", "")).strip().upper() == "SUCCESS": LOGGER.info("Satır %s zaten SUCCESS; atlandı.", row_number) continue try: recipient_id, lat, lng, radius = parse_point_row(row) except ConfigurationError as exc: LOGGER.error("Satır %s gönderilmedi: %s", row_number, exc) self._set_result(row, "FAILED", "NOT_SENT", "", str(exc)) self._write_csv(rows, fieldnames) continue payload = { "recipientId": recipient_id, "boundary": {"point": {"lat": lat, "lng": lng}, "radius": radius}, "webhook": deepcopy(webhook), "filter": deepcopy(filters), } self._register(row_number, payload, row) self._write_csv(rows, fieldnames) def _register(self, row_number: int, payload: dict[str, Any], row: dict[str, Any]) -> bool: try: response = self.service.register(payload, str(uuid4())) except ApiError as exc: recovered = self._try_recover(row_number, payload, exc) if recovered: self._set_result(row, "SUCCESS", f"HTTP_{exc.status_code or recovered[0]}", recovered[1], "") LOGGER.info("Satır %s eşleşen kayıtla kurtarıldı: %s", row_number, recovered[1]) return True self._set_result( row, "FAILED", f"HTTP_{exc.status_code}" if exc.status_code else "NETWORK_ERROR", "", error_body_text(exc), ) LOGGER.error("Satır %s başarısız: %s", row_number, exc) return False data = response.data if isinstance(response.data, dict) else {} registration_id = as_uuid(data.get("registrationId"), version=4) if not registration_id: self._set_result( row, "FAILED", f"HTTP_{response.status_code}", "", "Yanıtta UUID v4 registrationId bulunamadı: " + json.dumps(response.data, ensure_ascii=False), ) LOGGER.error("Satır %s yanıtında registrationId yok.", row_number) return False self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", registration_id, "") LOGGER.info("Satır %s başarıyla kaydedildi: %s", row_number, registration_id) for warning in response_warnings(response.data): LOGGER.warning("Satır %s API uyarısı: %s", row_number, warning) return True def _try_recover( self, row_number: int, payload: dict[str, Any], error: ApiError ) -> tuple[int, str] | None: """After an ambiguous failure, look for an identical registration (recipient + point + radius).""" status = error.status_code if status is not None and status != 409 and status not in RETRYABLE_STATUS_CODES: return None LOGGER.warning("Satır %s sonucu belirsiz; recipientId ile kontrol ediliyor.", row_number) boundary = payload["boundary"] try: response = self.service.get_by_recipient_id(payload["recipientId"]) except ApiError as exc: LOGGER.error("Satır %s için recipient sorgusu başarısız: %s", row_number, exc) return None data = response.data if isinstance(response.data, dict) else {} for registration in data.get("registrations") or []: flat = flatten_registration(registration) if isinstance(registration, dict) else {} registration_id = as_uuid(flat.get("registrationId"), version=4) if ( registration_id and flat.get("recipientId") == payload["recipientId"] and flat.get("lat") == boundary["point"]["lat"] and flat.get("lng") == boundary["point"]["lng"] and flat.get("radius") == boundary["radius"] ): return response.status_code, registration_id return None def _read_csv(self) -> tuple[list[dict[str, Any]], list[str]]: if not self.csv_path.is_file(): raise ConfigurationError(f"CSV dosyası bulunamadı: {self.csv_path}") with self.csv_path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle, delimiter=";") if not reader.fieldnames: raise ConfigurationError("CSV başlık satırı bulunamadı.") missing = [name for name in REGISTER_REQUIRED_COLUMNS if name not in reader.fieldnames] if missing: raise ConfigurationError("CSV içinde eksik zorunlu sütunlar: " + ", ".join(missing)) fieldnames = [ name for name in reader.fieldnames if name not in REGISTER_RESULT_COLUMNS ] + REGISTER_RESULT_COLUMNS rows = list(reader) return rows, fieldnames def _write_csv(self, rows: list[dict[str, Any]], fieldnames: list[str]) -> None: write_csv_atomic(self.csv_path, rows, fieldnames) @staticmethod def _set_result( row: dict[str, Any], status: str, http_status: str, registration_id: str, error_body: str ) -> None: row["status"] = status row["httpStatus"] = http_status row["registrationId"] = registration_id row["errorBody"] = error_body def _blank(value: Any) -> bool: return value is None or str(value).strip() == "" def _number(row: dict[str, Any], name: str) -> float | None: if _blank(row.get(name)): return None try: return float(str(row[name]).strip().replace(",", ".")) except ValueError as exc: raise ConfigurationError(f"{name} sayı olmalıdır.") from exc def _choice(row: dict[str, Any], name: str, allowed: set[str]) -> str | None: if _blank(row.get(name)): return None value = str(row[name]).strip().upper() if value not in allowed: raise ConfigurationError(f"{name} şunlardan biri olmalıdır: {', '.join(sorted(allowed))}.") return value def _flag(row: dict[str, Any], name: str) -> bool | None: if _blank(row.get(name)): return None value = str(row[name]).strip().lower() if value in {"true", "1", "evet"}: return True if value in {"false", "0", "hayır", "hayir"}: return False raise ConfigurationError(f"{name} true veya false olmalıdır.") def build_update_payload(row: dict[str, Any]) -> dict[str, Any]: """Build a PATCH body from a CSV row; blank cells are left out (null = untouched).""" registration_id = as_uuid(str(row.get("registrationId", "")).strip(), version=4) if not registration_id: raise ConfigurationError("registrationId geçerli bir UUID olmalıdır.") payload: dict[str, Any] = {"registrationId": registration_id} recipient_id = str(row.get("recipientId") or "").strip() if recipient_id: if not recipient_id.isascii(): raise ConfigurationError("recipientId US-ASCII karakterlerinden oluşmalıdır.") payload["recipientId"] = recipient_id lat, lng, radius = _number(row, "lat"), _number(row, "lng"), _number(row, "radius") if (lat is None) != (lng is None): raise ConfigurationError("lat ve lng birlikte verilmelidir.") boundary: dict[str, Any] = {} if lat is not None: if not -90 <= lat <= 90 or not -180 <= lng <= 180: raise ConfigurationError("lat -90..90, lng -180..180 aralığında olmalıdır.") boundary["point"] = {"lat": lat, "lng": lng} if radius is not None: if not 0 <= radius <= MAX_RADIUS_METERS: raise ConfigurationError(f"radius 0..{int(MAX_RADIUS_METERS)} metre aralığında olmalıdır.") boundary["radius"] = radius if boundary: payload["boundary"] = boundary lightning = { key: value for key, value in { "type": _choice(row, "lightningType", LIGHTNING_TYPES), "peakCurrent": _number(row, "peakCurrent"), "inCloudHeight": _number(row, "inCloudHeight"), }.items() if value is not None } thunderstorm = { key: value for key, value in { "intersectsAffectedPolygon": _flag(row, "intersectsAffectedPolygon"), "intersectsCellPolygon": _flag(row, "intersectsCellPolygon"), "severityThreshold": _choice(row, "severityThreshold", SEVERITIES), "speedThreshold": _number(row, "speedThreshold"), }.items() if value is not None } precipitation: dict[str, Any] = {} if not _blank(row.get("precipitationIntensities")): items = [part.strip().upper() for part in str(row["precipitationIntensities"]).split("|") if part.strip()] invalid = [item for item in items if item not in INTENSITIES] if invalid or not items: raise ConfigurationError( "precipitationIntensities '|' ile ayrılmış geçerli değerler içermelidir: " + ", ".join(sorted(INTENSITIES)) ) precipitation["intensities"] = items filters = { name: section for name, section in ( ("lightning", lightning), ("thunderstorm", thunderstorm), ("precipitation", precipitation), ) if section } if filters: payload["filter"] = filters if len(payload) == 1: raise ConfigurationError("Güncellenecek alan yok (registrationId dışında tüm hücreler boş).") return payload UPDATE_REQUIRED_COLUMNS = ["registrationId"] UPDATE_COLUMNS = ["registrationId", "recipientId", "lat", "lng", "radius"] + FILTER_COLUMNS class PointUpdateProcessor: """PATCH one registration per CSV row; blank cells are not sent.""" def __init__(self, service: PointAlarmService, *, csv_path: Path, dry_run: bool = False) -> None: self.service = service self.csv_path = csv_path.resolve() self.dry_run = dry_run def run(self) -> None: rows, fieldnames = self._read_csv() for row_number, row in enumerate(rows, start=2): if str(row.get("status", "")).strip().upper() == "SUCCESS": LOGGER.info("Satır %s zaten SUCCESS; atlandı.", row_number) continue try: payload = build_update_payload(row) except ConfigurationError as exc: LOGGER.error("Satır %s gönderilmedi: %s", row_number, exc) self._set_result(row, "FAILED", "NOT_SENT", str(exc)) self._write(rows, fieldnames) continue if payload.get("filter", {}).get("thunderstorm", {}).get("speedThreshold") is not None: LOGGER.warning( "Satır %s: thunderstorm.speedThreshold PATCH'te backend tarafından işlenmiyor " "(GAIA-476); değer gönderilir ancak güncellenmeyebilir.", row_number, ) if self.dry_run: print(f"[dry-run] satır {row_number}: PATCH {ALARMS_PATH}/update {json.dumps(payload, ensure_ascii=False)}") continue try: response = self.service.update(payload, str(uuid4())) except ApiError as exc: self._set_result( row, "FAILED", f"HTTP_{exc.status_code}" if exc.status_code else "NETWORK_ERROR", error_body_text(exc), ) LOGGER.error("Satır %s başarısız: %s", row_number, exc) else: self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", "") LOGGER.info("Satır %s güncellendi: %s", row_number, payload["registrationId"]) for warning in response_warnings(response.data): LOGGER.warning("Satır %s API uyarısı: %s", row_number, warning) self._write(rows, fieldnames) def _read_csv(self) -> tuple[list[dict[str, Any]], list[str]]: if not self.csv_path.is_file(): raise ConfigurationError(f"CSV dosyası bulunamadı: {self.csv_path}") with self.csv_path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle, delimiter=";") if not reader.fieldnames or "registrationId" not in reader.fieldnames: raise ConfigurationError("CSV başlığında 'registrationId' sütunu bulunamadı.") fieldnames = [n for n in reader.fieldnames if n not in ("status", "httpStatus", "errorBody")] + [ "status", "httpStatus", "errorBody", ] return list(reader), fieldnames def _write(self, rows: list[dict[str, Any]], fieldnames: list[str]) -> None: if not self.dry_run: write_csv_atomic(self.csv_path, rows, fieldnames) @staticmethod def _set_result(row: dict[str, Any], status: str, http_status: str, error_body: str) -> None: row["status"] = status row["httpStatus"] = http_status row["errorBody"] = error_body 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