"""Geo alarm API operations and CSV batch registration workflow.""" 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 common import ApiError, ConfigurationError, IklimClient from common.config import env_bool from common.client import RETRYABLE_STATUS_CODES 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: trimmed = unicodedata.normalize("NFKC", value.strip()) 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") 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 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 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 = _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 = _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: 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() @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)