"""Geo alarm API operations and CSV batch registration workflow.""" from __future__ import annotations import csv import logging import unicodedata from copy import deepcopy from pathlib import Path from typing import Any from urllib.parse import quote from uuid import uuid4 from common import ApiError, ConfigurationError, IklimClient from common.config import env_bool from common.client import RETRYABLE_STATUS_CODES from nowcast_common import ( read_json_object, response_warnings, as_uuid, load_webhook_config, write_csv_atomic, ) LOGGER = logging.getLogger(__name__) RESULT_COLUMNS = ["status", "httpStatus", "registrationId"] REQUIRED_CSV_COLUMNS = ["city", "district", "recipientId"] def normalize_turkish_name(value: str) -> str: trimmed = unicodedata.normalize("NFKC", value.strip()) return trimmed.translate(str.maketrans({"I": "ı", "İ": "i"})).casefold() def load_nowcast_geo_alarm_filter(path: Path) -> dict[str, Any]: config = read_json_object(path, "Nowcast geo alarm filter config") if not config: raise ConfigurationError("Nowcast geo alarm filter config boş olamaz.") lightning = config.get("lightning") if lightning is not None: if not isinstance(lightning, dict): raise ConfigurationError("filter.lightning JSON nesnesi olmalıdır.") event_type = lightning.get("type") if event_type is not None and event_type not in { "FLASH_CLOUD_TO_GROUND", "PULSE_IN_CLOUD", }: raise ConfigurationError( "filter.lightning.type belirtilirse FLASH_CLOUD_TO_GROUND veya " "PULSE_IN_CLOUD olmalıdır." ) thunderstorm = config.get("thunderstorm") if thunderstorm is not None: if not isinstance(thunderstorm, dict): raise ConfigurationError("filter.thunderstorm JSON nesnesi olmalıdır.") severity = thunderstorm.get("severityThreshold") if severity not in {"LOW", "MEDIUM", "HIGH"}: raise ConfigurationError( "filter.thunderstorm.severityThreshold LOW, MEDIUM veya HIGH olmalıdır." ) precipitation = config.get("precipitation") if precipitation is not None: if not isinstance(precipitation, dict): raise ConfigurationError("filter.precipitation JSON nesnesi olmalıdır.") intensities = precipitation.get("intensities") allowed = {"DRIZZLE", "LIGHT", "MODERATE", "HEAVY", "VERY_HEAVY", "EXTREME"} 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 allowed] if invalid: raise ConfigurationError( "Geçersiz precipitation intensity: " + ", ".join(map(str, invalid)) ) return config 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 as_uuid(account_id): raise ApiError("/v1/users/me yanıtında geçerli account.id bulunamadı.") return account_id def list_cities(self) -> list[dict[str, Any]]: data = self.client.request_json( "GET", "/v1/alarms/geometries/cities" ).data if not isinstance(data, list): raise ApiError("Şehir listesi beklenen JSON dizisini döndürmedi.") return data def list_districts(self, city_id: int) -> list[dict[str, Any]]: data = self.client.request_json( "GET", f"/v1/alarms/geometries/districts/{city_id}" ).data if not isinstance(data, list): raise ApiError("İlçe listesi beklenen JSON dizisini döndürmedi.") return data def create_registration( self, payload: dict[str, Any], idempotency_key: str ) -> tuple[int, str, list[str]]: response = self.client.request_json( "POST", "/v1/alarms/geometries/register", json_body=payload, idempotency_key=idempotency_key, ) registration_id = ( response.data.get("registrationId") if isinstance(response.data, dict) else None ) 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) def find_by_recipient_id( self, recipient_id: str ) -> tuple[int, str | None, list[str]]: encoded_recipient = quote(recipient_id, safe="") response = self.client.request_json( "GET", f"/v1/alarms/geometries/get-by-recipient-id/{encoded_recipient}", ) registrations = ( response.data.get("registrations") if isinstance(response.data, dict) else None ) if not isinstance(registrations, list): raise ApiError( "Recipient sorgusu beklenen registrations listesini döndürmedi.", status_code=response.status_code, response_data=response.data, ) for registration in registrations: if not isinstance(registration, dict): continue if registration.get("recipientId") != recipient_id: continue registration_id = as_uuid(registration.get("registrationId"), version=4) if registration_id: return ( response.status_code, registration_id, response_warnings(response.data, registration), ) return response.status_code, None, response_warnings(response.data) def _name_index(items: list[dict[str, Any]], item_type: str) -> dict[str, int]: result: dict[str, int] = {} for item in items: name = item.get("name") if isinstance(item, dict) else None item_id = item.get("id") if isinstance(item, dict) else None if not isinstance(name, str) or not isinstance(item_id, int): raise ApiError(f"{item_type} listesinde geçersiz name/id alanı bulundu.") normalized = normalize_turkish_name(name) if normalized in result and result[normalized] != item_id: raise ApiError(f"{item_type} listesinde birden fazla '{name}' eşleşmesi var.") result[normalized] = item_id return result class CsvRegistrationProcessor: def __init__( self, service: GeoAlarmService, *, csv_path: Path, webhook_config_path: Path, nowcast_filter_config_path: Path, ) -> None: self.service = service self.csv_path = csv_path.resolve() self.webhook_config_path = webhook_config_path.resolve() self.nowcast_filter_config_path = nowcast_filter_config_path.resolve() self.district_indexes: dict[int, dict[str, int]] = {} def run(self) -> None: rows, fieldnames = self._read_csv() filters = load_nowcast_geo_alarm_filter(self.nowcast_filter_config_path) webhook_initialized = env_bool("IKLIM_WEBHOOK_INITIALIZED", False) full_webhook = ( None if webhook_initialized else load_webhook_config(self.webhook_config_path) ) account_id = self.service.get_account_id() LOGGER.info("Kullanıcı hesabı doğrulandı; accountId API'den alındı.") city_index = _name_index(self.service.list_cities(), "Şehir") 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, city_id, district_id = self._resolve_row( row, row_number, city_index ) except (ConfigurationError, ApiError) as exc: LOGGER.error("Satır %s gönderilmedi: %s", row_number, exc) self._set_result(row, "FAILED", "NOT_SENT", "") self._write_csv(rows, fieldnames) continue webhook_payload = ( {"accountId": account_id} if webhook_initialized else deepcopy(full_webhook) ) payload = { "recipientId": recipient_id, "boundary": { "type": "ADMINISTRATIVE", "cityId": city_id, "districtId": district_id, }, "webhook": webhook_payload, "filter": deepcopy(filters), } idempotency_key = str(uuid4()) try: http_status, registration_id, warnings = ( self.service.create_registration(payload, idempotency_key) ) except ApiError as exc: recovered = self._try_recover(recipient_id, row_number, exc) if recovered: recovery_http_status, registration_id, warnings = recovered recorded_status = exc.status_code or recovery_http_status self._set_result( row, "SUCCESS", f"HTTP_{recorded_status}", registration_id, ) LOGGER.info( "Satır %s recipient sorgusuyla kurtarıldı: %s", row_number, registration_id, ) self._log_response_warnings(row_number, warnings) if not webhook_initialized: webhook_initialized = True self.service.client.token_store.set_webhook_initialized(True) else: http_result = ( f"HTTP_{exc.status_code}" if exc.status_code else "NETWORK_ERROR" ) self._set_result(row, "FAILED", http_result, "") LOGGER.error("Satır %s başarısız: %s", row_number, exc) else: self._set_result( row, "SUCCESS", f"HTTP_{http_status}", registration_id ) LOGGER.info("Satır %s başarıyla kaydedildi: %s", row_number, registration_id) self._log_response_warnings(row_number, warnings) if not webhook_initialized: webhook_initialized = True self.service.client.token_store.set_webhook_initialized(True) self._write_csv(rows, fieldnames) def _resolve_row( self, row: dict[str, Any], row_number: int, city_index: dict[str, int], ) -> tuple[str, int, int]: 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.") city_name = str(row.get("city", "")) city_id = city_index.get(normalize_turkish_name(city_name)) if city_id is None: raise ConfigurationError(f"İl bulunamadı: '{city_name.strip()}'.") if city_id not in self.district_indexes: districts = self.service.list_districts(city_id) self.district_indexes[city_id] = _name_index(districts, "İlçe") district_name = str(row.get("district", "")) district_id = self.district_indexes[city_id].get( normalize_turkish_name(district_name) ) if district_id is None: raise ConfigurationError( f"'{city_name.strip()}' içinde ilçe bulunamadı: " f"'{district_name.strip()}'." ) return recipient_id, city_id, district_id def _try_recover( self, recipient_id: str, row_number: int, original_error: ApiError ) -> tuple[int, str, list[str]] | None: status = original_error.status_code if status != 409 and status not in RETRYABLE_STATUS_CODES and status is not None: return None LOGGER.warning( "Satır %s sonucu belirsiz/çakışmalı; recipientId ile kontrol ediliyor.", row_number, ) try: recovery_status, registration_id, warnings = ( self.service.find_by_recipient_id(recipient_id) ) except ApiError as recovery_error: LOGGER.error( "Satır %s için recipient sorgusu başarısız: %s", row_number, recovery_error, ) return None if not registration_id: return None return recovery_status, registration_id, warnings 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 csv_file: reader = csv.DictReader(csv_file, delimiter=";") if not reader.fieldnames: raise ConfigurationError("CSV başlık satırı bulunamadı.") missing = [name for name in REQUIRED_CSV_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 RESULT_COLUMNS ] + 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 ) -> None: row["status"] = status row["httpStatus"] = http_status row["registrationId"] = registration_id @staticmethod def _log_response_warnings(row_number: int, warnings: list[str]) -> None: for warning in warnings: LOGGER.warning("Satır %s API uyarısı: %s", row_number, warning)