mozdemir
0958d26e9f
Adds point_alarm.py (flattening, secret masking, SQLite upsert store, paged fetch with 403 fallback, CSV/TXT export, shared filter loader), nowcast-point-page.py CLI and unit tests. Register will read the shared filter from nowcast-point-alarm-filter.json instead of repeating it in the CSV.
370 lines
14 KiB
Python
370 lines
14 KiB
Python
"""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,
|
||
fetch_account_id,
|
||
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:
|
||
return fetch_account_id(self.client)
|
||
|
||
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)
|