commit 2067d8c2982beffd058e6ce1ccb895ac682dfbbc Author: Murat ÖZDEMİR Date: Wed Sep 9 16:21:13 2026 +0300 Initial commit diff --git a/README.md b/README.md new file mode 100644 index 0000000..0233ff9 --- /dev/null +++ b/README.md @@ -0,0 +1,153 @@ +# iklim.co Python sandbox istemcileri + +Bu klasör HMAC imzalı istek, otomatik login/refresh ve `.env` token saklama +özelliklerini ortak bir istemci katmanında toplar. Geo alarm registration istemcisi +bu ortak yapıyı kullanarak noktalı virgüllü CSV satırlarını sırayla işler. + +## Dosyalar + +- `common/`: tekrar kullanılabilir API, HMAC, retry ve authentication katmanı +- `login.py`: manuel login komutu +- `nowcast-geo-register.py`: CSV geo alarm registration istemcisi +- `.env.example`: güvenli ortam değişkeni şablonu +- `webhook.example.json`: dört webhook authentication seçeneğini içeren şablon +- `nowcast-geo-alarm-filter.example.json`: ortak nowcast geo alarm filtreleri şablonu +- `nowcast-geo-alarm-registrations.example.csv`: CSV şablonu + +Gerçek değerler için `.env`, `webhook.json` ve +`nowcast-geo-alarm-filter.json` kullanılır. `.env` ile `webhook.json` kaynak kod +deposuna eklenmemelidir. + +## Kurulum + +```bash +cd iklim-sandbox +python3 -m venv .venv +source .venv/bin/activate +python -m pip install -r requirements.txt +``` + +Yeni kurulumda şablonları kopyalayın ve gerçek değerlerle düzenleyin: + +```bash +cp .env.example .env +cp webhook.example.json webhook.json +cp nowcast-geo-alarm-filter.example.json nowcast-geo-alarm-filter.json +cp nowcast-geo-alarm-registrations.example.csv nowcast-geo-alarm-registrations.csv +``` + +## CSV biçimi + +Girdi UTF-8 ve noktalı virgül ayracına sahip olmalıdır: + +```csv +city;district;recipientId +Ankara;Çankaya;recipient-001 +İstanbul;Kadıköy;recipient-002 +``` + +Script aynı dosyaya şu üç sonuç sütununu ekler veya mevcut değerlerini günceller: + +```csv +city;district;recipientId;status;httpStatus;registrationId +Ankara;Çankaya;recipient-001;SUCCESS;HTTP_200;f7587d9e-2481-4b4c-818d-c8d1946851b7 +``` + +- İl ve ilçe adları trim edilip Türkçe büyük/küçük harf kurallarıyla kesin eşleştirilir. +- Bulunamayan konumlar `FAILED;NOT_SENT;` olarak kaydedilir. +- Daha önce `SUCCESS` olmuş satırlar konsola bilgi yazılarak atlanır. +- Her satırdan sonra CSV aynı dizinde geçici dosyayla atomik olarak güncellenir. +- `recipientId` boş olamaz ve US-ASCII karakterlerinden oluşmalıdır. + +## İl ve ilçe sorgu cache'i + +Şehir listesi her script çalıştırmasında yalnızca bir kez API'den alınır ve isim-ID +eşleştirmesi bellekte tutulur. İlçe listesi de her benzersiz şehir için yalnızca ilk +karşılaşmada alınır; aynı şehirdeki sonraki CSV satırlarında bellekteki sonuç +kullanılır. Cache diske yazılmaz ve script kapandığında temizlenir. + +## Webhook yapılandırması + +`webhook.json` içindeki `authentication.selected` şu değerlerden biri olmalıdır: + +- `BASIC` +- `JWT_TOKEN` +- `API_KEY` +- `HMAC_SIGNATURE` + +Yalnızca seçilen seçeneğin `options` altındaki alanları API'ye gönderilir. İlk +initialization isteğinde tam webhook config gönderilir ve `accountId` kesinlikle +gönderilmez. İlk başarılı kayıttan sonra `.env` içindeki +`IKLIM_WEBHOOK_INITIALIZED=true` yapılır. Sonraki kayıtlar webhook altında yalnızca +API'nin `/v1/users/me` yanıtındaki `account.id` değerini gönderir. Webhook'u yeniden +tanımlamak için bu değeri elle `false` yapın. + +Test/prod ortamını veya kullanıcı hesabını değiştirirken de yeni ortamın webhook'u +ilk kayıtla tanımlayabilmesi için `IKLIM_WEBHOOK_INITIALIZED=false` yapın. + +## Nowcast geo alarm filter yapılandırması + +Tüm CSV satırlarına uygulanacak ortak nowcast filtreleri +`nowcast-geo-alarm-filter.json` içinde tutulur. Yıldırım, thunderstorm ve yağış +filtrelerinden ihtiyaç duyulmayan nesneler dosyadan çıkarılabilir. API isteğindeki +alan adı doküman gereği yine `filter` olarak gönderilir. + +## Otomatik authentication + +Ortak istemci her yetkili istekten önce access token'ı kontrol eder: + +1. Access token yoksa `.env` kullanıcı bilgileriyle login olur. +2. Access token'ın bitmesine 60 saniyeden az kaldıysa geçerli refresh token ile yeniler. +3. Refresh token yoksa/geçersizse yeniden login olur. +4. Sunucu `401` döndürürse refresh veya login yapıp isteği bir kez daha gönderir. +5. Yeni tokenları ve JWT `exp` değerlerini `.env` dosyasına kaydeder. + +Yetkili isteklerde `Authorization: Bearer ` ile HMAC başlıkları birlikte +gönderilir. Tokenlar ve parolalar konsola yazılmaz. + +## Retry davranışı + +Ağ hataları ile `429`, `500`, `502`, `503` ve `504` yanıtları ilk istekten sonra en +fazla üç kez, varsayılan olarak 2, 4 ve 8 saniye beklenerek tekrar edilir. `429` +yanıtındaki `Retry-After` varsa ona uyulur. Bir operasyonun retry isteklerinde aynı +idempotency key; yeni timestamp, nonce ve HMAC imzası kullanılır. + +Belirsiz sonuç veya `409` sonrasında kayıt `recipientId` ile sorgulanır. Kayıt +bulunursa `registrationId` kurtarılır ve satır `SUCCESS` yapılır. + +## Debug HTTP logları + +Her çalıştırmada maskelenmiş HTTP istek ve yanıtları aşağıdaki dosyaya yazılır: + +```text +logs/iklim-api-debug.log +``` + +Log; istek metodu, URL, deneme numarası, başlıklar, istek gövdesi, HTTP durum kodu, +yanıt başlıkları ve yanıt gövdesini içerir. Parola, kullanıcı adı, JWT, refresh +token, API key, webhook secret, `Authorization`, cookie ve `X-Signature` değerleri +maskelenir. Dosya 5 MB'a ulaştığında döndürülür ve en fazla üç eski dosya tutulur. +Başarılı API yanıtındaki `warning`/`warnings` alanları ayrıca konsola yazılır. + +## Çalıştırma + +Manuel login: + +```bash +python login.py +``` + +CSV registration: + +```bash +python nowcast-geo-register.py nowcast-geo-alarm-registrations.csv +``` + +Farklı config dosyaları vermek için: + +```bash +python nowcast-geo-register.py nowcast-geo-alarm-registrations.csv \ + --webhook-config another-webhook.json \ + --nowcast-filter-config another-nowcast-geo-alarm-filter.json \ + --debug-log logs/custom-debug.log +``` diff --git a/common/__init__.py b/common/__init__.py new file mode 100644 index 0000000..e170edb --- /dev/null +++ b/common/__init__.py @@ -0,0 +1,15 @@ +"""Reusable iklim.co API client components.""" + +from .client import ApiResponse, IklimClient +from .config import Settings +from .errors import ApiError, ConfigurationError +from .logging_config import configure_logging + +__all__ = [ + "ApiError", + "ApiResponse", + "ConfigurationError", + "IklimClient", + "Settings", + "configure_logging", +] diff --git a/common/client.py b/common/client.py new file mode 100644 index 0000000..bc5292e --- /dev/null +++ b/common/client.py @@ -0,0 +1,384 @@ +"""Reusable authenticated HTTP client for iklim.co APIs.""" + +from __future__ import annotations + +import json +import logging +import re +import time +import uuid +from dataclasses import dataclass +from email.utils import parsedate_to_datetime +from pathlib import Path +from typing import Any, Callable + +import requests + +from .config import Settings +from .errors import ApiError +from .security import TokenStore, create_signed_headers, token_is_valid + + +LOGGER = logging.getLogger(__name__) +RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504} +SENSITIVE_KEY_PARTS = { + "apikey", + "authorization", + "cookie", + "password", + "secret", + "signature", + "token", + "username", +} +REDACTED = "" + + +@dataclass(frozen=True) +class ApiResponse: + status_code: int + data: Any + + +def _error_message(response: requests.Response, data: Any) -> str: + if isinstance(data, dict): + message = data.get("message") or data.get("messages") or data.get("error") + if isinstance(message, list): + return "; ".join(str(item) for item in message) + if message: + return str(message) + return response.reason or "Bilinmeyen API hatası" + + +def _normalized_key(value: object) -> str: + return "".join(character for character in str(value).lower() if character.isalnum()) + + +def _redact(value: Any, parent_key: str = "") -> Any: + if isinstance(value, dict): + redact_all_values = _normalized_key(parent_key) == "requestheaders" + result = {} + for key, item in value.items(): + normalized = _normalized_key(key) + sensitive = redact_all_values or any( + part in normalized for part in SENSITIVE_KEY_PARTS + ) + result[key] = REDACTED if sensitive else _redact(item, str(key)) + return result + if isinstance(value, list): + return [_redact(item, parent_key) for item in value] + return value + + +def _redact_text(value: str) -> str: + value = re.sub(r"(?i)Bearer\s+\S+", f"Bearer {REDACTED}", value) + return re.sub( + r"\b[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b", + REDACTED, + value, + ) + + +def _request_body_for_log(body: str) -> Any: + if not body: + return None + try: + return _redact(json.loads(body)) + except json.JSONDecodeError: + return _redact_text(body[:20_000]) + + +def _response_body_for_log(response: requests.Response) -> Any: + if not response.content: + return None + try: + return _redact(response.json()) + except requests.exceptions.JSONDecodeError: + return _redact_text(response.text[:20_000]) + + +class AuthenticationManager: + LOGIN_PATH = "/v1/auth/login" + REFRESH_PATH = "/v1/auth/refresh" + + def __init__(self, client: "IklimClient", token_store: TokenStore) -> None: + self.client = client + self.token_store = token_store + + def ensure_access_token(self) -> str: + access_token = self.token_store.access_token + if not access_token: + LOGGER.info("Access token bulunamadı; otomatik login yapılıyor.") + self.login() + return self._required_saved_access_token() + + if token_is_valid(access_token, self.client.settings.token_expiry_skew_seconds): + return access_token + + LOGGER.info("Access token süresi dolmuş veya dolmak üzere.") + refresh_token = self.token_store.refresh_token + if token_is_valid( + refresh_token, self.client.settings.token_expiry_skew_seconds + ): + self.refresh() + else: + LOGGER.info("Refresh token geçerli değil; otomatik login yapılıyor.") + self.login() + return self._required_saved_access_token() + + def recover_after_unauthorized(self) -> None: + refresh_token = self.token_store.refresh_token + if token_is_valid( + refresh_token, self.client.settings.token_expiry_skew_seconds + ): + LOGGER.info("HTTP 401 alındı; access token yenileniyor.") + self.refresh() + else: + LOGGER.info("HTTP 401 alındı ve refresh token geçersiz; yeniden login yapılıyor.") + self.login() + + def login(self) -> dict[str, Any]: + username, password = self.client.settings.credentials() + LOGGER.info("iklim.co API login işlemi başlatıldı.") + response = self.client.request_json( + "POST", + self.LOGIN_PATH, + json_body={"username": username, "password": password}, + authenticated=False, + ) + self._save_auth_response(response.data) + LOGGER.info("Login başarılı; tokenlar .env dosyasına kaydedildi.") + return response.data + + def refresh(self) -> dict[str, Any]: + refresh_token = self.token_store.refresh_token + if not refresh_token: + LOGGER.info("Refresh token bulunamadı; otomatik login yapılıyor.") + return self.login() + + LOGGER.info("Access token refresh işlemi başlatıldı.") + try: + response = self.client.request_json( + "POST", + self.REFRESH_PATH, + json_body={"refreshToken": refresh_token}, + authenticated=False, + ) + except ApiError as exc: + if exc.status_code in {400, 401, 403}: + LOGGER.info("Refresh token API tarafından reddedildi; yeniden login yapılıyor.") + return self.login() + raise + + self._save_auth_response(response.data) + LOGGER.info("Token yenileme başarılı; yeni tokenlar .env dosyasına kaydedildi.") + return response.data + + def _save_auth_response(self, data: Any) -> None: + if not isinstance(data, dict): + raise ApiError("Kimlik doğrulama endpoint'i beklenen JSON nesnesini döndürmedi.") + self.token_store.save(data) + + def _required_saved_access_token(self) -> str: + access_token = self.token_store.access_token + if not access_token: + raise ApiError("Login/refresh sonrasında access token kaydedilemedi.") + return access_token + + +class IklimClient: + def __init__( + self, + settings: Settings, + *, + session: requests.Session | None = None, + sleep: Callable[[float], None] = time.sleep, + ) -> None: + self.settings = settings + self.session = session or requests.Session() + self.sleep = sleep + self.token_store = TokenStore(settings.env_file) + self.auth = AuthenticationManager(self, self.token_store) + + @classmethod + def from_env(cls, env_file: Path | str | None = None) -> "IklimClient": + settings = Settings.from_env() if env_file is None else Settings.from_env(env_file) + return cls(settings) + + def request_json( + self, + method: str, + path_with_query: str, + *, + json_body: Any = None, + authenticated: bool = True, + idempotency_key: str | None = None, + ) -> ApiResponse: + method = method.upper() + if not path_with_query.startswith("/"): + raise ValueError("API yolu '/' ile başlamalıdır.") + + body = ( + "" + if json_body is None + else json.dumps(json_body, ensure_ascii=False, separators=(",", ":")) + ) + operation_idempotency_key = idempotency_key or str(uuid.uuid4()) + + response: requests.Response | None = None + for auth_round in range(2): + access_token = self.auth.ensure_access_token() if authenticated else None + response = self._send_with_retries( + method=method, + path_with_query=path_with_query, + body=body, + access_token=access_token, + idempotency_key=operation_idempotency_key, + ) + if response.status_code != 401 or not authenticated: + break + if auth_round == 0: + self.auth.recover_after_unauthorized() + + if response is None: + raise ApiError("API isteği için yanıt alınamadı.") + + data = self._decode_response(response) + if not response.ok: + raise ApiError( + f"HTTP {response.status_code}: {_error_message(response, data)}", + status_code=response.status_code, + response_data=data, + ) + return ApiResponse(response.status_code, data) + + def _send_with_retries( + self, + *, + method: str, + path_with_query: str, + body: str, + access_token: str | None, + idempotency_key: str, + ) -> requests.Response: + for attempt in range(self.settings.max_retries + 1): + headers = create_signed_headers( + method=method, + path_with_query=path_with_query, + body=body, + secret=self.settings.hmac_secret, + accept_language=self.settings.accept_language, + idempotency_key=idempotency_key, + ) + if access_token: + headers["Authorization"] = f"Bearer {access_token}" + + url = f"{self.settings.base_url}{path_with_query}" + LOGGER.debug( + "HTTP request | %s", + json.dumps( + { + "attempt": attempt + 1, + "method": method, + "url": url, + "headers": _redact(headers), + "body": _request_body_for_log(body), + }, + ensure_ascii=False, + default=str, + ), + ) + + try: + response = self.session.request( + method, + url, + data=None if not body else body.encode("utf-8"), + headers=headers, + timeout=self.settings.timeout_seconds, + ) + except requests.RequestException as exc: + LOGGER.debug( + "HTTP network error | %s", + json.dumps( + { + "attempt": attempt + 1, + "method": method, + "url": url, + "error": _redact_text(str(exc)), + }, + ensure_ascii=False, + ), + ) + if attempt >= self.settings.max_retries: + raise ApiError( + f"Ağ hatası; maksimum deneme sayısına ulaşıldı: {exc}" + ) from exc + self._wait_before_retry(attempt, None) + continue + + LOGGER.debug( + "HTTP response | %s", + json.dumps( + { + "attempt": attempt + 1, + "method": method, + "url": url, + "status": response.status_code, + "headers": _redact(dict(response.headers)), + "body": _response_body_for_log(response), + }, + ensure_ascii=False, + default=str, + ), + ) + + if ( + response.status_code in RETRYABLE_STATUS_CODES + and attempt < self.settings.max_retries + ): + self._wait_before_retry(attempt, response) + continue + return response + + raise ApiError("API isteği beklenmeyen biçimde tamamlanamadı.") + + def _wait_before_retry( + self, attempt: int, response: requests.Response | None + ) -> None: + delay = self.settings.retry_base_seconds * (2**attempt) + if response is not None and response.status_code == 429: + delay = self._retry_after_seconds(response, delay) + LOGGER.warning( + "İstek yeniden denenecek (%s/%s); %.1f saniye bekleniyor.", + attempt + 1, + self.settings.max_retries, + delay, + ) + self.sleep(delay) + + @staticmethod + def _retry_after_seconds(response: requests.Response, fallback: float) -> float: + raw_value = response.headers.get("Retry-After", "").strip() + if not raw_value: + return fallback + try: + return max(0.0, float(raw_value)) + except ValueError: + try: + retry_at = parsedate_to_datetime(raw_value) + return max(0.0, retry_at.timestamp() - time.time()) + except (TypeError, ValueError, OverflowError): + return fallback + + @staticmethod + def _decode_response(response: requests.Response) -> Any: + if not response.content: + return None + try: + return response.json() + except requests.exceptions.JSONDecodeError as exc: + raise ApiError( + f"API JSON olmayan bir yanıt döndürdü (HTTP {response.status_code}).", + status_code=response.status_code, + ) from exc diff --git a/common/config.py b/common/config.py new file mode 100644 index 0000000..5fcd3e7 --- /dev/null +++ b/common/config.py @@ -0,0 +1,110 @@ +"""Environment-backed configuration shared by every service client.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path +from urllib.parse import urlsplit + +from dotenv import load_dotenv + +from .errors import ConfigurationError + + +DEFAULT_ENV_FILE = Path(__file__).resolve().parent.parent / ".env" + + +def _required(name: str) -> str: + value = os.getenv(name, "").strip() + if not value: + raise ConfigurationError(f".env içinde {name} eksik veya boş.") + if value.lower() == "change-me": + raise ConfigurationError(f".env içindeki {name} placeholder değeri değiştirilmemiş.") + return value + + +def _positive_float(name: str, default: str) -> float: + raw_value = os.getenv(name, default).strip() + try: + value = float(raw_value) + except ValueError as exc: + raise ConfigurationError(f"{name} sayısal bir değer olmalıdır.") from exc + if value <= 0: + raise ConfigurationError(f"{name} sıfırdan büyük olmalıdır.") + return value + + +def _non_negative_int(name: str, default: str) -> int: + raw_value = os.getenv(name, default).strip() + try: + value = int(raw_value) + except ValueError as exc: + raise ConfigurationError(f"{name} tam sayı olmalıdır.") from exc + if value < 0: + raise ConfigurationError(f"{name} negatif olamaz.") + return value + + +def env_bool(name: str, default: bool = False) -> bool: + raw_value = os.getenv(name) + if raw_value is None or not raw_value.strip(): + return default + normalized = raw_value.strip().lower() + if normalized in {"1", "true", "yes", "on"}: + return True + if normalized in {"0", "false", "no", "off"}: + return False + raise ConfigurationError(f"{name} true veya false olmalıdır.") + + +@dataclass(frozen=True) +class Settings: + env_file: Path + base_url: str + hmac_secret: str + accept_language: str + timeout_seconds: float + token_expiry_skew_seconds: int + max_retries: int + retry_base_seconds: float + + @classmethod + def from_env(cls, env_file: Path = DEFAULT_ENV_FILE) -> "Settings": + env_file = Path(env_file).resolve() + if not env_file.is_file(): + raise ConfigurationError(f".env dosyası bulunamadı: {env_file}") + + load_dotenv(env_file, override=True) + environment = _required("IKLIM_API_ENV").lower() + url_variables = { + "test": "IKLIM_API_TEST_URL", + "prod": "IKLIM_API_PROD_URL", + } + if environment not in url_variables: + raise ConfigurationError("IKLIM_API_ENV yalnızca test veya prod olabilir.") + + base_url = _required(url_variables[environment]).rstrip("/") + parsed_url = urlsplit(base_url) + if parsed_url.scheme != "https" or not parsed_url.netloc: + raise ConfigurationError("Seçilen API adresi geçerli bir HTTPS adresi olmalıdır.") + + language = os.getenv("IKLIM_ACCEPT_LANGUAGE", "en").strip().lower() or "en" + if language not in {"tr", "en"}: + raise ConfigurationError("IKLIM_ACCEPT_LANGUAGE yalnızca tr veya en olabilir.") + + return cls( + env_file=env_file, + base_url=base_url, + hmac_secret=_required("IKLIM_HMAC_SECRET"), + accept_language=language, + timeout_seconds=_positive_float("IKLIM_TIMEOUT_SECONDS", "30"), + token_expiry_skew_seconds=_non_negative_int( + "IKLIM_TOKEN_EXPIRY_SKEW_SECONDS", "60" + ), + max_retries=_non_negative_int("IKLIM_MAX_RETRIES", "3"), + retry_base_seconds=_positive_float("IKLIM_RETRY_BASE_SECONDS", "2"), + ) + + def credentials(self) -> tuple[str, str]: + return _required("IKLIM_USERNAME"), _required("IKLIM_PASSWORD") diff --git a/common/errors.py b/common/errors.py new file mode 100644 index 0000000..7886618 --- /dev/null +++ b/common/errors.py @@ -0,0 +1,24 @@ +"""Shared exceptions for iklim.co clients.""" + +from __future__ import annotations + +from typing import Any + + +class ConfigurationError(ValueError): + """Raised when a required local setting is missing or invalid.""" + + +class ApiError(RuntimeError): + """Raised after an API request fails and eligible retries are exhausted.""" + + def __init__( + self, + message: str, + *, + status_code: int | None = None, + response_data: Any = None, + ) -> None: + super().__init__(message) + self.status_code = status_code + self.response_data = response_data diff --git a/common/logging_config.py b/common/logging_config.py new file mode 100644 index 0000000..f808278 --- /dev/null +++ b/common/logging_config.py @@ -0,0 +1,37 @@ +"""Console and rotating debug-file logging configuration.""" + +from __future__ import annotations + +import logging +from logging.handlers import RotatingFileHandler +from pathlib import Path + + +def configure_logging(debug_log_file: Path) -> Path: + debug_log_file = debug_log_file.resolve() + debug_log_file.parent.mkdir(parents=True, exist_ok=True) + + root_logger = logging.getLogger() + root_logger.setLevel(logging.DEBUG) + root_logger.handlers.clear() + + formatter = logging.Formatter( + "%(asctime)s | %(levelname)s | %(name)s | %(message)s" + ) + console_handler = logging.StreamHandler() + console_handler.setLevel(logging.INFO) + console_handler.setFormatter(formatter) + + file_handler = RotatingFileHandler( + debug_log_file, + maxBytes=5 * 1024 * 1024, + backupCount=3, + encoding="utf-8", + ) + file_handler.setLevel(logging.DEBUG) + file_handler.setFormatter(formatter) + + root_logger.addHandler(console_handler) + root_logger.addHandler(file_handler) + logging.getLogger(__name__).info("Debug log dosyası: %s", debug_log_file) + return debug_log_file diff --git a/common/security.py b/common/security.py new file mode 100644 index 0000000..d6144a8 --- /dev/null +++ b/common/security.py @@ -0,0 +1,111 @@ +"""HMAC request signing and local JWT token persistence.""" + +from __future__ import annotations + +import base64 +import binascii +import hashlib +import hmac +import json +import os +import time +import uuid +from pathlib import Path +from typing import Any + +from dotenv import set_key + +from .errors import ApiError + + +def jwt_expiration(token: str) -> int: + """Read a JWT exp claim; this does not replace server-side verification.""" + try: + encoded_payload = token.split(".")[1] + padding = "=" * (-len(encoded_payload) % 4) + payload = json.loads( + base64.urlsafe_b64decode(encoded_payload + padding).decode("utf-8") + ) + return int(payload["exp"]) + except ( + IndexError, + KeyError, + TypeError, + ValueError, + UnicodeDecodeError, + binascii.Error, + ) as exc: + raise ApiError("API geçerli bir JWT token döndürmedi.") from exc + + +def token_is_valid(token: str | None, skew_seconds: int) -> bool: + if not token: + return False + try: + return jwt_expiration(token) > int(time.time()) + skew_seconds + except ApiError: + return False + + +def create_signed_headers( + *, + method: str, + path_with_query: str, + body: str, + secret: str, + accept_language: str, + idempotency_key: str, +) -> dict[str, str]: + timestamp = str(int(time.time() * 1000)) + data_to_sign = f"{method.upper()}|{path_with_query}|{timestamp}|{body}" + signature = hmac.new( + secret.encode("utf-8"), + data_to_sign.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return { + "Accept": "application/json", + "Accept-Language": accept_language, + "Content-Type": "application/json", + "X-Signature": signature, + "X-Timestamp": timestamp, + "X-Nonce": str(uuid.uuid4()), + "X-Idempotency-Key": idempotency_key, + } + + +class TokenStore: + def __init__(self, env_file: Path) -> None: + self.env_file = env_file + + @property + def access_token(self) -> str | None: + return os.getenv("IKLIM_ACCESS_TOKEN", "").strip() or None + + @property + def refresh_token(self) -> str | None: + return os.getenv("IKLIM_REFRESH_TOKEN", "").strip() or None + + def save(self, response_data: dict[str, Any]) -> None: + access_token = response_data.get("accessToken") + refresh_token = response_data.get("refreshToken") + if not isinstance(access_token, str) or not access_token: + raise ApiError("Kimlik doğrulama yanıtında accessToken bulunamadı.") + if not isinstance(refresh_token, str) or not refresh_token: + raise ApiError("Kimlik doğrulama yanıtında refreshToken bulunamadı.") + + values = { + "IKLIM_ACCESS_TOKEN": access_token, + "IKLIM_REFRESH_TOKEN": refresh_token, + "IKLIM_ACCESS_TOKEN_EXPIRES_AT": str(jwt_expiration(access_token)), + "IKLIM_REFRESH_TOKEN_EXPIRES_AT": str(jwt_expiration(refresh_token)), + "IKLIM_TOKENS_UPDATED_AT": str(int(time.time())), + } + for key, value in values.items(): + set_key(self.env_file, key, value, quote_mode="never") + os.environ[key] = value + + def set_webhook_initialized(self, initialized: bool) -> None: + value = "true" if initialized else "false" + set_key(self.env_file, "IKLIM_WEBHOOK_INITIALIZED", value, quote_mode="never") + os.environ["IKLIM_WEBHOOK_INITIALIZED"] = value diff --git a/geo_alarm.py b/geo_alarm.py new file mode 100644 index 0000000..138248b --- /dev/null +++ b/geo_alarm.py @@ -0,0 +1,498 @@ +"""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 not in {"FLASH_CLOUD_TO_GROUND", "PULSE_IN_CLOUD"}: + 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: + 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) diff --git a/login.py b/login.py new file mode 100644 index 0000000..20ddd2f --- /dev/null +++ b/login.py @@ -0,0 +1,35 @@ +#!/usr/bin/env python3 +"""Manually trigger login using the reusable iklim.co client.""" + +from __future__ import annotations + +import logging +import sys +from pathlib import Path +from typing import Any + +import requests + +from common import ApiError, ConfigurationError, IklimClient, configure_logging + + +BASE_DIR = Path(__file__).resolve().parent + + +def login() -> dict[str, Any]: + client = IklimClient.from_env(BASE_DIR / ".env") + return client.auth.login() + + +def main() -> int: + configure_logging(BASE_DIR / "logs" / "iklim-api-debug.log") + try: + login() + except (ConfigurationError, ApiError, requests.RequestException) as exc: + logging.error("Login başarısız: %s", exc) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/nowcast-geo-alarm-filter.example.json b/nowcast-geo-alarm-filter.example.json new file mode 100644 index 0000000..4fdcb7a --- /dev/null +++ b/nowcast-geo-alarm-filter.example.json @@ -0,0 +1,16 @@ +{ + "lightning": { + "type": "FLASH_CLOUD_TO_GROUND", + "peakCurrent": -12.5, + "inCloudHeight": 4500 + }, + "thunderstorm": { + "intersectsAffectedPolygon": true, + "intersectsCellPolygon": false, + "severityThreshold": "MEDIUM", + "speedThreshold": 20 + }, + "precipitation": { + "intensities": ["LIGHT", "MODERATE"] + } +} diff --git a/nowcast-geo-alarm-registrations.example.csv b/nowcast-geo-alarm-registrations.example.csv new file mode 100644 index 0000000..78d617b --- /dev/null +++ b/nowcast-geo-alarm-registrations.example.csv @@ -0,0 +1,3 @@ +city;district;recipientId +Ankara;Çankaya;recipient-001 +İstanbul;Kadıköy;recipient-002 diff --git a/nowcast-geo-register.py b/nowcast-geo-register.py new file mode 100644 index 0000000..11a9806 --- /dev/null +++ b/nowcast-geo-register.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 +"""Create geo alarm registrations for every eligible row in a CSV file.""" + +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path + +from common import ApiError, ConfigurationError, IklimClient, configure_logging +from geo_alarm import CsvRegistrationProcessor, GeoAlarmService + + +BASE_DIR = Path(__file__).resolve().parent + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="CSV satırlarından iklim.co coğrafi alarm kayıtları oluşturur." + ) + parser.add_argument("csv_file", type=Path, help="İşlenecek noktalı virgüllü CSV") + parser.add_argument( + "--webhook-config", + type=Path, + default=BASE_DIR / "webhook.json", + help="Webhook JSON dosyası (varsayılan: webhook.json)", + ) + parser.add_argument( + "--nowcast-filter-config", + type=Path, + default=BASE_DIR / "nowcast-geo-alarm-filter.json", + help=( + "Nowcast geo alarm filter JSON dosyası " + "(varsayılan: nowcast-geo-alarm-filter.json)" + ), + ) + parser.add_argument( + "--debug-log", + type=Path, + default=BASE_DIR / "logs" / "iklim-api-debug.log", + help="Maskelenmiş HTTP debug log dosyası", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + configure_logging(args.debug_log) + try: + client = IklimClient.from_env(BASE_DIR / ".env") + processor = CsvRegistrationProcessor( + GeoAlarmService(client), + csv_path=args.csv_file, + webhook_config_path=args.webhook_config, + nowcast_filter_config_path=args.nowcast_filter_config, + ) + processor.run() + except (ConfigurationError, ApiError, OSError) as exc: + logging.error("İşlem durduruldu: %s", exc) + return 1 + + logging.info("CSV işlemi tamamlandı: %s", args.csv_file.resolve()) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..319ff09 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +python-dotenv>=1.0,<2.0 +requests>=2.31,<3.0 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..37285a8 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import os +import tempfile +import time +import unittest +from pathlib import Path +from unittest.mock import patch + +import requests + +from common.client import IklimClient +from common.config import Settings +from common.security import create_signed_headers + + +def jwt(expiration: int) -> str: + def encode(value: dict[str, object]) -> str: + return base64.urlsafe_b64encode( + json.dumps(value, separators=(",", ":")).encode() + ).decode().rstrip("=") + + return f"{encode({'alg': 'none'})}.{encode({'exp': expiration})}.signature" + + +def response(status: int, body: object, headers: dict[str, str] | None = None): + result = requests.Response() + result.status_code = status + result._content = json.dumps(body).encode() + result.headers.update(headers or {}) + result.reason = "test response" + return result + + +class FakeSession: + def __init__(self, responses: list[requests.Response]) -> None: + self.responses = iter(responses) + self.calls: list[dict[str, object]] = [] + + def request(self, method, url, **kwargs): + self.calls.append({"method": method, "url": url, **kwargs}) + return next(self.responses) + + +class ClientTest(unittest.TestCase): + @staticmethod + def settings(env_file: Path, *, max_retries: int = 0) -> Settings: + return Settings( + env_file=env_file, + base_url="https://api-test.iklim.co", + hmac_secret="test-secret", + accept_language="tr", + timeout_seconds=30, + token_expiry_skew_seconds=60, + max_retries=max_retries, + retry_base_seconds=2, + ) + + def test_retry_reuses_idempotency_and_regenerates_nonce(self) -> None: + with tempfile.TemporaryDirectory() as directory: + env_file = Path(directory) / ".env" + env_file.write_text("", encoding="utf-8") + settings = self.settings(env_file, max_retries=3) + session = FakeSession( + [response(500, {"message": "temporary"}), response(200, {"ok": True})] + ) + waits: list[float] = [] + with patch.dict( + os.environ, + {"IKLIM_ACCESS_TOKEN": jwt(int(time.time()) + 3600)}, + clear=True, + ): + client = IklimClient(settings, session=session, sleep=waits.append) + result = client.request_json("POST", "/v1/test", json_body={"x": 1}) + + self.assertEqual(result.data, {"ok": True}) + self.assertEqual(waits, [2]) + first_headers = session.calls[0]["headers"] + second_headers = session.calls[1]["headers"] + self.assertEqual( + first_headers["X-Idempotency-Key"], + second_headers["X-Idempotency-Key"], + ) + self.assertNotEqual(first_headers["X-Nonce"], second_headers["X-Nonce"]) + self.assertTrue(first_headers["Authorization"].startswith("Bearer ")) + + def test_hmac_matches_postman_formula(self) -> None: + body = '{"username":"name@domain.com","password":"password"}' + timestamp = "1752751106704" + with patch("common.security.time.time", return_value=1752751106.704), patch( + "common.security.uuid.uuid4", + return_value="684a0dca-bd6a-4056-a449-2567f9847f9c", + ): + headers = create_signed_headers( + method="POST", + path_with_query="/v1/auth/login", + body=body, + secret="test-secret", + accept_language="tr", + idempotency_key="777edc03-ad49-4c17-be6b-9baf05a1b9e0", + ) + expected = hmac.new( + b"test-secret", + f"POST|/v1/auth/login|{timestamp}|{body}".encode(), + hashlib.sha256, + ).hexdigest() + self.assertEqual(headers["X-Signature"], expected) + self.assertEqual(headers["X-Timestamp"], timestamp) + + def test_unauthorized_refreshes_token_and_retries_request(self) -> None: + with tempfile.TemporaryDirectory() as directory: + env_file = Path(directory) / ".env" + env_file.write_text("", encoding="utf-8") + settings = self.settings(env_file) + now = int(time.time()) + old_access = jwt(now + 3600) + old_refresh = jwt(now + 7200) + new_access = jwt(now + 10800) + new_refresh = jwt(now + 14400) + session = FakeSession( + [ + response(401, {"message": "expired on server"}), + response( + 200, + {"accessToken": new_access, "refreshToken": new_refresh}, + ), + response(200, {"ok": True}), + ] + ) + + with patch.dict( + os.environ, + { + "IKLIM_ACCESS_TOKEN": old_access, + "IKLIM_REFRESH_TOKEN": old_refresh, + }, + clear=True, + ): + client = IklimClient(settings, session=session, sleep=lambda _: None) + result = client.request_json("GET", "/v1/protected") + self.assertEqual(os.environ["IKLIM_ACCESS_TOKEN"], new_access) + + self.assertEqual(result.data, {"ok": True}) + self.assertTrue(session.calls[0]["headers"]["Authorization"].endswith(old_access)) + self.assertNotIn("Authorization", session.calls[1]["headers"]) + self.assertTrue(session.calls[2]["headers"]["Authorization"].endswith(new_access)) + self.assertTrue(session.calls[1]["url"].endswith("/v1/auth/refresh")) + + def test_expired_access_token_is_refreshed_before_request(self) -> None: + with tempfile.TemporaryDirectory() as directory: + env_file = Path(directory) / ".env" + env_file.write_text("", encoding="utf-8") + now = int(time.time()) + new_access = jwt(now + 3600) + new_refresh = jwt(now + 7200) + session = FakeSession( + [ + response( + 200, + {"accessToken": new_access, "refreshToken": new_refresh}, + ), + response(200, {"ok": True}), + ] + ) + with patch.dict( + os.environ, + { + "IKLIM_ACCESS_TOKEN": jwt(now - 1), + "IKLIM_REFRESH_TOKEN": jwt(now + 3600), + }, + clear=True, + ): + client = IklimClient( + self.settings(env_file), session=session, sleep=lambda _: None + ) + result = client.request_json("GET", "/v1/protected") + + self.assertEqual(result.data, {"ok": True}) + self.assertTrue(session.calls[0]["url"].endswith("/v1/auth/refresh")) + self.assertNotIn("Authorization", session.calls[0]["headers"]) + self.assertTrue(session.calls[1]["headers"]["Authorization"].endswith(new_access)) + + def test_missing_tokens_trigger_automatic_login(self) -> None: + with tempfile.TemporaryDirectory() as directory: + env_file = Path(directory) / ".env" + env_file.write_text("", encoding="utf-8") + now = int(time.time()) + access = jwt(now + 3600) + refresh = jwt(now + 7200) + session = FakeSession( + [ + response(200, {"accessToken": access, "refreshToken": refresh}), + response(200, {"ok": True}), + ] + ) + with patch.dict( + os.environ, + {"IKLIM_USERNAME": "user@example.com", "IKLIM_PASSWORD": "secret"}, + clear=True, + ): + client = IklimClient( + self.settings(env_file), session=session, sleep=lambda _: None + ) + result = client.request_json("GET", "/v1/protected") + + self.assertEqual(result.data, {"ok": True}) + self.assertTrue(session.calls[0]["url"].endswith("/v1/auth/login")) + self.assertNotIn("Authorization", session.calls[0]["headers"]) + self.assertTrue(session.calls[1]["headers"]["Authorization"].endswith(access)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_geo_alarm.py b/tests/test_geo_alarm.py new file mode 100644 index 0000000..9824e46 --- /dev/null +++ b/tests/test_geo_alarm.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +import csv +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from common.security import TokenStore +from common import ApiError +from geo_alarm import CsvRegistrationProcessor, normalize_turkish_name + + +REGISTRATION_ONE = "f7587d9e-2481-4b4c-818d-c8d1946851b7" +REGISTRATION_TWO = "57d1e2ba-bf2d-4ef3-b078-523060b2000c" +ACCOUNT_ID = "2f04f1b5-1c0a-4c4e-b0a7-0ba6e9b7f2e1" + + +class FakeClient: + def __init__(self, env_file: Path) -> None: + self.token_store = TokenStore(env_file) + + +class FakeService: + def __init__(self, env_file: Path) -> None: + self.client = FakeClient(env_file) + self.payloads = [] + + def get_account_id(self): + return ACCOUNT_ID + + def list_cities(self): + return [{"id": 6, "name": "Ankara"}, {"id": 34, "name": "İstanbul"}] + + def list_districts(self, city_id): + return { + 6: [{"id": 557, "name": "Çankaya"}], + 34: [{"id": 347, "name": "Kadıköy"}], + }[city_id] + + def create_registration(self, payload, idempotency_key): + self.payloads.append(payload) + registration_id = REGISTRATION_ONE if len(self.payloads) == 1 else REGISTRATION_TWO + warnings = ["test warning"] if len(self.payloads) == 1 else [] + return 200, registration_id, warnings + + +class RecoveringService(FakeService): + def create_registration(self, payload, idempotency_key): + self.payloads.append(payload) + raise ApiError("duplicate", status_code=409) + + def find_by_recipient_id(self, recipient_id): + return 200, REGISTRATION_ONE, [] + + +class GeoAlarmTest(unittest.TestCase): + def test_turkish_exact_normalization(self) -> None: + self.assertEqual(normalize_turkish_name(" İSTANBUL "), "istanbul") + self.assertEqual(normalize_turkish_name("IĞDIR"), "ığdır") + self.assertEqual(normalize_turkish_name("ÇANKAYA"), "çankaya") + self.assertNotEqual(normalize_turkish_name("I"), normalize_turkish_name("İ")) + + def test_csv_processing_and_webhook_transition(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + env_file = root / ".env" + env_file.write_text("IKLIM_WEBHOOK_INITIALIZED=false\n", encoding="utf-8") + csv_file = root / "nowcast-geo-alarm-registrations.csv" + csv_file.write_text( + "city;district;recipientId\n" + " anKARA ; ÇANKAYA ;recipient-001\n" + "Ankara;Olmayan;recipient-invalid\n" + "İSTANBUL;KADIKÖY;recipient-002\n", + encoding="utf-8", + ) + webhook_file = root / "webhook.json" + webhook_file.write_text( + json.dumps( + { + "url": "https://customer.example/webhook", + "httpMethod": "POST", + "contentType": "application/json", + "authentication": { + "selected": "BASIC", + "options": { + "BASIC": {"username": "user", "password": "pass"} + }, + }, + } + ), + encoding="utf-8", + ) + nowcast_filter_file = root / "nowcast-geo-alarm-filter.json" + nowcast_filter_file.write_text( + json.dumps({"precipitation": {"intensities": ["LIGHT"]}}), + encoding="utf-8", + ) + service = FakeService(env_file) + + with patch.dict( + os.environ, {"IKLIM_WEBHOOK_INITIALIZED": "false"}, clear=True + ): + CsvRegistrationProcessor( + service, + csv_path=csv_file, + webhook_config_path=webhook_file, + nowcast_filter_config_path=nowcast_filter_file, + ).run() + + with csv_file.open("r", encoding="utf-8-sig", newline="") as handle: + rows = list(csv.DictReader(handle, delimiter=";")) + + self.assertEqual( + list(rows[0]), + ["city", "district", "recipientId", "status", "httpStatus", "registrationId"], + ) + self.assertEqual(rows[0]["status"], "SUCCESS") + self.assertEqual(rows[0]["registrationId"], REGISTRATION_ONE) + self.assertEqual(rows[1]["status"], "FAILED") + self.assertEqual(rows[1]["httpStatus"], "NOT_SENT") + self.assertEqual(rows[2]["status"], "SUCCESS") + self.assertEqual(service.payloads[0]["webhook"]["url"], "https://customer.example/webhook") + self.assertNotIn("accountId", service.payloads[0]["webhook"]) + self.assertEqual(service.payloads[1]["webhook"], {"accountId": ACCOUNT_ID}) + self.assertIn("IKLIM_WEBHOOK_INITIALIZED=true", env_file.read_text()) + + def test_conflict_is_recovered_by_recipient_id(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + env_file = root / ".env" + env_file.write_text("IKLIM_WEBHOOK_INITIALIZED=false\n", encoding="utf-8") + csv_file = root / "nowcast-geo-alarm-registrations.csv" + csv_file.write_text( + "city;district;recipientId\nAnkara;Çankaya;recipient-001\n", + encoding="utf-8", + ) + webhook_file = root / "webhook.json" + webhook_file.write_text( + json.dumps( + { + "url": "https://customer.example/webhook", + "httpMethod": "POST", + "contentType": "application/json", + "authentication": { + "selected": "BASIC", + "options": { + "BASIC": {"username": "user", "password": "pass"} + }, + }, + } + ), + encoding="utf-8", + ) + nowcast_filter_file = root / "nowcast-geo-alarm-filter.json" + nowcast_filter_file.write_text( + json.dumps({"precipitation": {"intensities": ["LIGHT"]}}), + encoding="utf-8", + ) + + with patch.dict( + os.environ, {"IKLIM_WEBHOOK_INITIALIZED": "false"}, clear=True + ): + CsvRegistrationProcessor( + RecoveringService(env_file), + csv_path=csv_file, + webhook_config_path=webhook_file, + nowcast_filter_config_path=nowcast_filter_file, + ).run() + + with csv_file.open("r", encoding="utf-8-sig", newline="") as handle: + row = next(csv.DictReader(handle, delimiter=";")) + self.assertEqual(row["status"], "SUCCESS") + self.assertEqual(row["httpStatus"], "HTTP_409") + self.assertEqual(row["registrationId"], REGISTRATION_ONE) + + +if __name__ == "__main__": + unittest.main() diff --git a/webhook.example.json b/webhook.example.json new file mode 100644 index 0000000..51412c4 --- /dev/null +++ b/webhook.example.json @@ -0,0 +1,33 @@ +{ + "url": "https://change-me.example/webhook", + "httpMethod": "POST", + "contentType": "application/json", + "requestHeaders": {}, + "deliveryPolicy": { + "timeout": 60, + "maxRetries": 3, + "retryDelay": 10 + }, + "authentication": { + "selected": "BASIC", + "options": { + "BASIC": { + "username": "change-me", + "password": "change-me" + }, + "JWT_TOKEN": { + "jwtToken": "change-me" + }, + "API_KEY": { + "apiKey": "change-me", + "asQueryParameter": false, + "asHeader": true + }, + "HMAC_SIGNATURE": { + "clientId": "change-me", + "secret": "change-me", + "algorithm": "HmacSHA256" + } + } + } +}