diff --git a/nowcast-point-register.py b/nowcast-point-register.py new file mode 100644 index 0000000..e834c53 --- /dev/null +++ b/nowcast-point-register.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 +"""Create point 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 point_alarm import PointAlarmService, PointRegistrationProcessor + + +BASE_DIR = Path(__file__).resolve().parent + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "CSV satırlarından iklim.co nokta alarm kayıtları oluşturur. CSV sütunları: " + "recipientId;lat;lng;radius (radius metre, 0-50000). Filtre ortak JSON dosyasından " + "okunur. Sonuç sütunları (status;httpStatus;registrationId;errorBody) aynı dosyaya " + "yazılır; SUCCESS satırlar yeniden çalıştırmada atlanır." + ) + ) + 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( + "--filter-config", + type=Path, + default=BASE_DIR / "nowcast-point-alarm-filter.json", + help="Her kayda uygulanan filtre JSON dosyası (varsayılan: nowcast-point-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") + PointRegistrationProcessor( + PointAlarmService(client), + csv_path=args.csv_file, + webhook_config_path=args.webhook_config, + filter_config_path=args.filter_config, + ).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/nowcast-point-registrations.example.csv b/nowcast-point-registrations.example.csv new file mode 100644 index 0000000..55e5efa --- /dev/null +++ b/nowcast-point-registrations.example.csv @@ -0,0 +1,2 @@ +recipientId;lat;lng;radius +recipient-001;39.9208;32.8541;1000 diff --git a/point_alarm.py b/point_alarm.py index 52b8f4c..af934b2 100644 --- a/point_alarm.py +++ b/point_alarm.py @@ -7,15 +7,26 @@ import json import logging import sqlite3 import sys +from copy import deepcopy from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Iterable, TextIO from urllib.parse import quote +from uuid import uuid4 from common import ApiError, ConfigurationError, IklimClient from common.client import ApiResponse -from nowcast_common import fetch_account_id, read_json_object +from common.client import RETRYABLE_STATUS_CODES +from common.config import env_bool +from nowcast_common import ( + as_uuid, + fetch_account_id, + load_webhook_config, + read_json_object, + response_warnings, + write_csv_atomic, +) LOGGER = logging.getLogger(__name__) @@ -567,6 +578,8 @@ def lookup_rows( LOGGER.warning("%s için sonuç %s (HTTP %s).", query, status, exc.status_code) continue data = response.data + if isinstance(data, dict) and "registrations" in data: + data = data["registrations"] items = data if isinstance(data, list) else [data] items = [item for item in items if isinstance(item, dict)] if not items: @@ -584,3 +597,171 @@ def lookup_rows( store.commit() webhooks = store.webhooks(account_id) if account_id else store.webhooks() return rows, webhooks + + +REGISTER_REQUIRED_COLUMNS = ["recipientId", "lat", "lng", "radius"] +REGISTER_RESULT_COLUMNS = ["status", "httpStatus", "registrationId", "errorBody"] + + +def error_body_text(exc: ApiError) -> str: + if exc.response_data is None: + return str(exc) + if isinstance(exc.response_data, str): + return exc.response_data + return json.dumps(exc.response_data, ensure_ascii=False) + + +def parse_point_row(row: dict[str, Any]) -> tuple[str, float, float, float]: + """Validate one register CSV row and return (recipientId, lat, lng, radius).""" + 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.") + try: + lat = float(str(row.get("lat", "")).strip().replace(",", ".")) + lng = float(str(row.get("lng", "")).strip().replace(",", ".")) + radius = float(str(row.get("radius", "")).strip().replace(",", ".")) + except ValueError as exc: + raise ConfigurationError("lat, lng ve radius sayı olmalıdır.") from exc + if not -90 <= lat <= 90 or not -180 <= lng <= 180: + raise ConfigurationError("lat -90..90, lng -180..180 aralığında olmalıdır.") + if not 0 <= radius <= MAX_RADIUS_METERS: + raise ConfigurationError(f"radius 0..{int(MAX_RADIUS_METERS)} metre aralığında olmalıdır.") + return recipient_id, lat, lng, radius + + +class PointRegistrationProcessor: + """Register one point alarm per CSV row, recording the outcome back into the CSV.""" + + def __init__( + self, + service: PointAlarmService, + *, + csv_path: Path, + webhook_config_path: Path, + filter_config_path: Path, + ) -> None: + self.service = service + self.csv_path = csv_path.resolve() + self.webhook_config_path = webhook_config_path.resolve() + self.filter_config_path = filter_config_path.resolve() + + def run(self) -> None: + rows, fieldnames = self._read_csv() + filters = load_point_alarm_filter(self.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ı.") + 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, lat, lng, radius = parse_point_row(row) + except ConfigurationError as exc: + LOGGER.error("Satır %s gönderilmedi: %s", row_number, exc) + self._set_result(row, "FAILED", "NOT_SENT", "", str(exc)) + self._write_csv(rows, fieldnames) + continue + payload = { + "recipientId": recipient_id, + "boundary": {"point": {"lat": lat, "lng": lng}, "radius": radius}, + "webhook": {"accountId": account_id} if webhook_initialized else deepcopy(full_webhook), + "filter": deepcopy(filters), + } + registered = self._register(row_number, payload, row) + if registered and not webhook_initialized: + webhook_initialized = True + self.service.client.token_store.set_webhook_initialized(True) + self._write_csv(rows, fieldnames) + + def _register(self, row_number: int, payload: dict[str, Any], row: dict[str, Any]) -> bool: + try: + response = self.service.register(payload, str(uuid4())) + except ApiError as exc: + recovered = self._try_recover(row_number, payload, exc) + if recovered: + self._set_result(row, "SUCCESS", f"HTTP_{exc.status_code or recovered[0]}", recovered[1], "") + LOGGER.info("Satır %s eşleşen kayıtla kurtarıldı: %s", row_number, recovered[1]) + return True + self._set_result( + row, + "FAILED", + f"HTTP_{exc.status_code}" if exc.status_code else "NETWORK_ERROR", + "", + error_body_text(exc), + ) + LOGGER.error("Satır %s başarısız: %s", row_number, exc) + return False + data = response.data if isinstance(response.data, dict) else {} + registration_id = as_uuid(data.get("registrationId"), version=4) + if not registration_id: + self._set_result( + row, "FAILED", f"HTTP_{response.status_code}", "", + "Yanıtta UUID v4 registrationId bulunamadı: " + json.dumps(response.data, ensure_ascii=False), + ) + LOGGER.error("Satır %s yanıtında registrationId yok.", row_number) + return False + self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", registration_id, "") + LOGGER.info("Satır %s başarıyla kaydedildi: %s", row_number, registration_id) + for warning in response_warnings(response.data): + LOGGER.warning("Satır %s API uyarısı: %s", row_number, warning) + return True + + def _try_recover( + self, row_number: int, payload: dict[str, Any], error: ApiError + ) -> tuple[int, str] | None: + """After an ambiguous failure, look for an identical registration (recipient + point + radius).""" + status = error.status_code + if status is not None and status != 409 and status not in RETRYABLE_STATUS_CODES: + return None + LOGGER.warning("Satır %s sonucu belirsiz; recipientId ile kontrol ediliyor.", row_number) + boundary = payload["boundary"] + try: + response = self.service.get_by_recipient_id(payload["recipientId"]) + except ApiError as exc: + LOGGER.error("Satır %s için recipient sorgusu başarısız: %s", row_number, exc) + return None + data = response.data if isinstance(response.data, dict) else {} + for registration in data.get("registrations") or []: + flat = flatten_registration(registration) if isinstance(registration, dict) else {} + registration_id = as_uuid(flat.get("registrationId"), version=4) + if ( + registration_id + and flat.get("recipientId") == payload["recipientId"] + and flat.get("lat") == boundary["point"]["lat"] + and flat.get("lng") == boundary["point"]["lng"] + and flat.get("radius") == boundary["radius"] + ): + return response.status_code, registration_id + return None + + 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 handle: + reader = csv.DictReader(handle, delimiter=";") + if not reader.fieldnames: + raise ConfigurationError("CSV başlık satırı bulunamadı.") + missing = [name for name in REGISTER_REQUIRED_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 REGISTER_RESULT_COLUMNS + ] + REGISTER_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, error_body: str + ) -> None: + row["status"] = status + row["httpStatus"] = http_status + row["registrationId"] = registration_id + row["errorBody"] = error_body diff --git a/tests/test_point_alarm.py b/tests/test_point_alarm.py index c952508..eee8631 100644 --- a/tests/test_point_alarm.py +++ b/tests/test_point_alarm.py @@ -1,14 +1,20 @@ from __future__ import annotations import csv +import json +import os import tempfile import unittest from datetime import datetime from pathlib import Path +from unittest.mock import patch + +from common.security import TokenStore from common import ApiError, ConfigurationError from common.client import ApiResponse from point_alarm import ( + PointRegistrationProcessor, PointStore, fetch_pages, flatten_registration, @@ -145,7 +151,7 @@ class PointAlarmTest(unittest.TestCase): return ApiResponse(200, registration(1)) def get_by_recipient_id(self, value): - return ApiResponse(200, [registration(1), registration(2)]) + return ApiResponse(200, {"registrations": [registration(1), registration(2)]}) rows, _ = lookup_rows(Svc(), self.store, ["ok", "missing", "boom"], "id", None) self.assertEqual([r["status"] for r in rows], ["SUCCESS", "NOT_FOUND", "FAILED"]) @@ -155,5 +161,87 @@ class PointAlarmTest(unittest.TestCase): self.assertEqual(self.store.count(), 2) +REG_ID = "f7587d9e-2481-4b4c-818d-c8d1946851b7" + + +class FakeRegisterService: + def __init__(self, env_file: Path, results: list) -> None: + self.client = type("C", (), {"token_store": TokenStore(env_file)})() + self.results = list(results) + self.payloads: list[dict] = [] + self.recipient_hits: list = [] + + def get_account_id(self): + return "acc" + + def register(self, payload, key): + self.payloads.append(payload) + result = self.results.pop(0) + if isinstance(result, Exception): + raise result + return result + + def get_by_recipient_id(self, recipient_id): + return ApiResponse(200, {"registrations": self.recipient_hits}) + + +class RegisterTest(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.dir = Path(self.tmp.name) + (self.dir / ".env").write_text("", encoding="utf-8") + self.filter = self.dir / "filter.json" + self.filter.write_text(json.dumps({"lightning": {"type": "PULSE_IN_CLOUD"}}), encoding="utf-8") + self.webhook = self.dir / "webhook.json" + self.webhook.write_text("{}", encoding="utf-8") + self.csv = self.dir / "r.csv" + + def run_processor(self, rows: str | None, service) -> list[dict]: + if rows is not None: + self.csv.write_text("recipientId;lat;lng;radius\n" + rows, encoding="utf-8") + processor = PointRegistrationProcessor( + service, csv_path=self.csv, webhook_config_path=self.webhook, filter_config_path=self.filter + ) + with patch.dict(os.environ, {"IKLIM_WEBHOOK_INITIALIZED": "true"}): + processor.run() + with self.csv.open(encoding="utf-8-sig", newline="") as handle: + return list(csv.DictReader(handle, delimiter=";")) + + def test_success_failure_and_skip(self) -> None: + service = FakeRegisterService( + self.dir / ".env", + [ + ApiResponse(201, {"registrationId": REG_ID}), + ApiError("bad", status_code=400, response_data={"message": "duplicate"}), + ], + ) + rows = self.run_processor("a;39.9;32.8;1000\nb;39.9;32.8;1000\nc;99;32.8;1000\n", service) + self.assertEqual([r["status"] for r in rows], ["SUCCESS", "FAILED", "FAILED"]) + self.assertEqual(rows[0]["registrationId"], REG_ID) + self.assertIn("duplicate", rows[1]["errorBody"]) + self.assertEqual(rows[2]["httpStatus"], "NOT_SENT") + self.assertEqual(service.payloads[0]["webhook"], {"accountId": "acc"}) + self.assertEqual(service.payloads[0]["filter"]["lightning"]["type"], "PULSE_IN_CLOUD") + # Re-run: SUCCESS row is skipped, the failed one is retried. + service2 = FakeRegisterService( + self.dir / ".env", [ApiResponse(201, {"registrationId": REG_ID})] * 2 + ) + rows = self.run_processor(None, service2) + self.assertEqual(len(service2.payloads), 1) + self.assertEqual(rows[0]["registrationId"], REG_ID) + + def test_recovery_requires_matching_point(self) -> None: + service = FakeRegisterService(self.dir / ".env", [ApiError("t", status_code=503)]) + service.recipient_hits = [dict(registration(1, "a"), registrationId=REG_ID)] + rows = self.run_processor("a;39.9;32.8;1000\n", service) + self.assertEqual(rows[0]["status"], "SUCCESS") + self.assertEqual(rows[0]["registrationId"], REG_ID) + other = FakeRegisterService(self.dir / ".env", [ApiError("t", status_code=503)]) + other.recipient_hits = [dict(registration(1, "a"), registrationId=REG_ID)] + rows = self.run_processor("a;40.0;32.8;1000\n", other) + self.assertEqual(rows[0]["status"], "FAILED") + + if __name__ == "__main__": unittest.main()