diff --git a/nowcast-point-update.example.csv b/nowcast-point-update.example.csv new file mode 100644 index 0000000..2b56269 --- /dev/null +++ b/nowcast-point-update.example.csv @@ -0,0 +1 @@ +registrationId;recipientId;lat;lng;radius;lightningType;peakCurrent;inCloudHeight;intersectsAffectedPolygon;intersectsCellPolygon;severityThreshold;speedThreshold;precipitationIntensities diff --git a/nowcast-point-update.py b/nowcast-point-update.py new file mode 100644 index 0000000..a752501 --- /dev/null +++ b/nowcast-point-update.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +"""Patch point alarm registrations from a CSV file (blank cells are not sent).""" + +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, PointUpdateProcessor + + +BASE_DIR = Path(__file__).resolve().parent + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "CSV satırlarından nokta alarm kayıtlarını kısmen günceller (PATCH). Sütunlar: " + "registrationId (zorunlu) ve isteğe bağlı recipientId;lat;lng;radius;lightningType;" + "peakCurrent;inCloudHeight;intersectsAffectedPolygon;intersectsCellPolygon;" + "severityThreshold;speedThreshold;precipitationIntensities ('|' ile ayrılır). Boş hücre " + "gönderilmez; lat ve lng birlikte verilmelidir. Bir alan boşaltılamaz (temizleme " + "desteklenmez). thunderstorm.speedThreshold backend PATCH'inde işlenmiyor (GAIA-476). " + "Sonuç sütunları status;httpStatus;errorBody aynı dosyaya yazılır." + ) + ) + parser.add_argument("csv_file", type=Path, help="İşlenecek noktalı virgüllü CSV") + parser.add_argument( + "--dry-run", + action="store_true", + help="İstek göndermeden ve CSV'yi değiştirmeden PATCH gövdelerini ekrana yazar", + ) + 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") + PointUpdateProcessor( + PointAlarmService(client), csv_path=args.csv_file, dry_run=args.dry_run + ).run() + except (ConfigurationError, ApiError, OSError) as exc: + logging.error("İşlem durduruldu: %s", exc) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/point_alarm.py b/point_alarm.py index af934b2..261d685 100644 --- a/point_alarm.py +++ b/point_alarm.py @@ -765,3 +765,181 @@ class PointRegistrationProcessor: row["httpStatus"] = http_status row["registrationId"] = registration_id row["errorBody"] = error_body + + +def _blank(value: Any) -> bool: + return value is None or str(value).strip() == "" + + +def _number(row: dict[str, Any], name: str) -> float | None: + if _blank(row.get(name)): + return None + try: + return float(str(row[name]).strip().replace(",", ".")) + except ValueError as exc: + raise ConfigurationError(f"{name} sayı olmalıdır.") from exc + + +def _choice(row: dict[str, Any], name: str, allowed: set[str]) -> str | None: + if _blank(row.get(name)): + return None + value = str(row[name]).strip().upper() + if value not in allowed: + raise ConfigurationError(f"{name} şunlardan biri olmalıdır: {', '.join(sorted(allowed))}.") + return value + + +def _flag(row: dict[str, Any], name: str) -> bool | None: + if _blank(row.get(name)): + return None + value = str(row[name]).strip().lower() + if value in {"true", "1", "evet"}: + return True + if value in {"false", "0", "hayır", "hayir"}: + return False + raise ConfigurationError(f"{name} true veya false olmalıdır.") + + +def build_update_payload(row: dict[str, Any]) -> dict[str, Any]: + """Build a PATCH body from a CSV row; blank cells are left out (null = untouched).""" + registration_id = as_uuid(str(row.get("registrationId", "")).strip(), version=4) + if not registration_id: + raise ConfigurationError("registrationId geçerli bir UUID olmalıdır.") + payload: dict[str, Any] = {"registrationId": registration_id} + recipient_id = str(row.get("recipientId") or "").strip() + if recipient_id: + if not recipient_id.isascii(): + raise ConfigurationError("recipientId US-ASCII karakterlerinden oluşmalıdır.") + payload["recipientId"] = recipient_id + + lat, lng, radius = _number(row, "lat"), _number(row, "lng"), _number(row, "radius") + if (lat is None) != (lng is None): + raise ConfigurationError("lat ve lng birlikte verilmelidir.") + boundary: dict[str, Any] = {} + if lat is not None: + if not -90 <= lat <= 90 or not -180 <= lng <= 180: + raise ConfigurationError("lat -90..90, lng -180..180 aralığında olmalıdır.") + boundary["point"] = {"lat": lat, "lng": lng} + if radius is not None: + if not 0 <= radius <= MAX_RADIUS_METERS: + raise ConfigurationError(f"radius 0..{int(MAX_RADIUS_METERS)} metre aralığında olmalıdır.") + boundary["radius"] = radius + if boundary: + payload["boundary"] = boundary + + lightning = { + key: value + for key, value in { + "type": _choice(row, "lightningType", LIGHTNING_TYPES), + "peakCurrent": _number(row, "peakCurrent"), + "inCloudHeight": _number(row, "inCloudHeight"), + }.items() + if value is not None + } + thunderstorm = { + key: value + for key, value in { + "intersectsAffectedPolygon": _flag(row, "intersectsAffectedPolygon"), + "intersectsCellPolygon": _flag(row, "intersectsCellPolygon"), + "severityThreshold": _choice(row, "severityThreshold", SEVERITIES), + "speedThreshold": _number(row, "speedThreshold"), + }.items() + if value is not None + } + precipitation: dict[str, Any] = {} + if not _blank(row.get("precipitationIntensities")): + items = [part.strip().upper() for part in str(row["precipitationIntensities"]).split("|") if part.strip()] + invalid = [item for item in items if item not in INTENSITIES] + if invalid or not items: + raise ConfigurationError( + "precipitationIntensities '|' ile ayrılmış geçerli değerler içermelidir: " + + ", ".join(sorted(INTENSITIES)) + ) + precipitation["intensities"] = items + filters = { + name: section + for name, section in ( + ("lightning", lightning), + ("thunderstorm", thunderstorm), + ("precipitation", precipitation), + ) + if section + } + if filters: + payload["filter"] = filters + if len(payload) == 1: + raise ConfigurationError("Güncellenecek alan yok (registrationId dışında tüm hücreler boş).") + return payload + + +UPDATE_REQUIRED_COLUMNS = ["registrationId"] +UPDATE_COLUMNS = ["registrationId", "recipientId", "lat", "lng", "radius"] + FILTER_COLUMNS + + +class PointUpdateProcessor: + """PATCH one registration per CSV row; blank cells are not sent.""" + + def __init__(self, service: PointAlarmService, *, csv_path: Path, dry_run: bool = False) -> None: + self.service = service + self.csv_path = csv_path.resolve() + self.dry_run = dry_run + + def run(self) -> None: + rows, fieldnames = self._read_csv() + 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: + payload = build_update_payload(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(rows, fieldnames) + continue + if payload.get("filter", {}).get("thunderstorm", {}).get("speedThreshold") is not None: + LOGGER.warning( + "Satır %s: thunderstorm.speedThreshold PATCH'te backend tarafından işlenmiyor " + "(GAIA-476); değer gönderilir ancak güncellenmeyebilir.", + row_number, + ) + if self.dry_run: + print(f"[dry-run] satır {row_number}: PATCH {ALARMS_PATH}/update {json.dumps(payload, ensure_ascii=False)}") + continue + try: + response = self.service.update(payload, str(uuid4())) + except ApiError as exc: + 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) + else: + self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", "") + LOGGER.info("Satır %s güncellendi: %s", row_number, payload["registrationId"]) + for warning in response_warnings(response.data): + LOGGER.warning("Satır %s API uyarısı: %s", row_number, warning) + self._write(rows, fieldnames) + + 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 or "registrationId" not in reader.fieldnames: + raise ConfigurationError("CSV başlığında 'registrationId' sütunu bulunamadı.") + fieldnames = [n for n in reader.fieldnames if n not in ("status", "httpStatus", "errorBody")] + [ + "status", "httpStatus", "errorBody", + ] + return list(reader), fieldnames + + def _write(self, rows: list[dict[str, Any]], fieldnames: list[str]) -> None: + if not self.dry_run: + write_csv_atomic(self.csv_path, rows, fieldnames) + + @staticmethod + def _set_result(row: dict[str, Any], status: str, http_status: str, error_body: str) -> None: + row["status"] = status + row["httpStatus"] = http_status + row["errorBody"] = error_body diff --git a/tests/test_point_alarm.py b/tests/test_point_alarm.py index eee8631..a5ba9e4 100644 --- a/tests/test_point_alarm.py +++ b/tests/test_point_alarm.py @@ -15,6 +15,8 @@ from common import ApiError, ConfigurationError from common.client import ApiResponse from point_alarm import ( PointRegistrationProcessor, + PointUpdateProcessor, + build_update_payload, PointStore, fetch_pages, flatten_registration, @@ -243,5 +245,54 @@ class RegisterTest(unittest.TestCase): self.assertEqual(rows[0]["status"], "FAILED") +class UpdateTest(unittest.TestCase): + def test_payload_skips_blank_cells(self) -> None: + payload = build_update_payload( + {"registrationId": REG_ID, "lat": "39.9", "lng": "32.8", "radius": "", + "severityThreshold": "high", "precipitationIntensities": "light|heavy"} + ) + self.assertEqual(payload["boundary"], {"point": {"lat": 39.9, "lng": 32.8}}) + self.assertEqual(payload["filter"]["thunderstorm"], {"severityThreshold": "HIGH"}) + self.assertEqual(payload["filter"]["precipitation"]["intensities"], ["LIGHT", "HEAVY"]) + + def test_payload_errors(self) -> None: + for row in ( + {"registrationId": REG_ID}, + {"registrationId": REG_ID, "lat": "39"}, + {"registrationId": "x", "radius": "5"}, + {"registrationId": REG_ID, "radius": "60000"}, + {"registrationId": REG_ID, "lightningType": "BAD"}, + ): + with self.assertRaises(ConfigurationError, msg=row): + build_update_payload(row) + + def test_dry_run_does_not_call_or_write(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "u.csv" + text = f"registrationId;radius\n{REG_ID};500\n" + path.write_text(text, encoding="utf-8") + + class Boom: + def update(self, *_): + raise AssertionError("must not be called") + + PointUpdateProcessor(Boom(), csv_path=path, dry_run=True).run() + self.assertEqual(path.read_text(encoding="utf-8"), text) + + def test_update_records_result(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "u.csv" + path.write_text(f"registrationId;radius\n{REG_ID};500\n", encoding="utf-8") + + class Svc: + def update(self, payload, key): + return ApiResponse(200, {}) + + PointUpdateProcessor(Svc(), csv_path=path).run() + with path.open(encoding="utf-8-sig", newline="") as handle: + row = next(csv.DictReader(handle, delimiter=";")) + self.assertEqual((row["status"], row["httpStatus"]), ("SUCCESS", "HTTP_200")) + + if __name__ == "__main__": unittest.main()