diff --git a/geo_alarm.py b/geo_alarm.py index 83b38da..88e23d4 100644 --- a/geo_alarm.py +++ b/geo_alarm.py @@ -3,31 +3,29 @@ 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, + 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 +33,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,32 +81,6 @@ 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 @@ -193,7 +90,7 @@ class GeoAlarmService: 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): + if not as_uuid(account_id): raise ApiError("/v1/users/me yanıtında geçerli account.id bulunamadı.") return account_id @@ -227,14 +124,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 +157,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 +358,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_common.py b/nowcast_common.py new file mode 100644 index 0000000..f66a031 --- /dev/null +++ b/nowcast_common.py @@ -0,0 +1,156 @@ +"""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 ConfigurationError + + +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()