iklim.co/iklim-sandbox
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: number point alarm register and update log lines by record instead of file line

Browse Source
This commit is contained in:
mozdemir
2026-10-23 19:47:55 +03:00
parent e17c06da4a
commit d2b430a524
1 changed files with 25 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
point_alarm.py
+25 -23
View File
@@ -652,14 +652,15 @@ class PointRegistrationProcessor:
webhook = load_webhook_config(self.webhook_config_path) webhook = load_webhook_config(self.webhook_config_path)
if not isinstance(webhook.get("deliveryPolicy"), dict): if not isinstance(webhook.get("deliveryPolicy"), dict):
raise ConfigurationError("Nokta alarm kaydı için webhook.deliveryPolicy zorunludur.") raise ConfigurationError("Nokta alarm kaydı için webhook.deliveryPolicy zorunludur.")
for row_number, row in enumerate(rows, start=2): for index, row in enumerate(rows, start=1):
label = f"{index}/{len(rows)}"
if str(row.get("status", "")).strip().upper() == "SUCCESS": if str(row.get("status", "")).strip().upper() == "SUCCESS":
LOGGER.info("Satır %s zaten SUCCESS; atlandı.", row_number) LOGGER.info("Kayıt %s zaten SUCCESS; atlandı.", label)
continue continue
try: try:
recipient_id, lat, lng, radius = parse_point_row(row) recipient_id, lat, lng, radius = parse_point_row(row)
except ConfigurationError as exc: except ConfigurationError as exc:
LOGGER.error("Satır %s gönderilmedi: %s", row_number, exc) LOGGER.error("Kayıt %s gönderilmedi: %s", label, exc)
self._set_result(row, "FAILED", "NOT_SENT", "", str(exc)) self._set_result(row, "FAILED", "NOT_SENT", "", str(exc))
self._write_csv(rows, fieldnames) self._write_csv(rows, fieldnames)
continue continue
@@ -669,17 +670,17 @@ class PointRegistrationProcessor:
"webhook": deepcopy(webhook), "webhook": deepcopy(webhook),
"filter": deepcopy(filters), "filter": deepcopy(filters),
} }
self._register(row_number, payload, row) self._register(label, payload, row)
self._write_csv(rows, fieldnames) self._write_csv(rows, fieldnames)
def _register(self, row_number: int, payload: dict[str, Any], row: dict[str, Any]) -> bool: def _register(self, label: str, payload: dict[str, Any], row: dict[str, Any]) -> bool:
try: try:
response = self.service.register(payload, str(uuid4())) response = self.service.register(payload, str(uuid4()))
except ApiError as exc: except ApiError as exc:
recovered = self._try_recover(row_number, payload, exc) recovered = self._try_recover(label, payload, exc)
if recovered: if recovered:
self._set_result(row, "SUCCESS", f"HTTP_{exc.status_code or recovered[0]}", recovered[1], "") 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]) LOGGER.info("Kayıt %s eşleşen kayıtla kurtarıldı: %s", label, recovered[1])
return True return True
self._set_result( self._set_result(
row, row,
@@ -688,7 +689,7 @@ class PointRegistrationProcessor:
"", "",
error_body_text(exc), error_body_text(exc),
) )
LOGGER.error("Satır %s başarısız: %s", row_number, exc) LOGGER.error("Kayıt %s başarısız: %s", label, exc)
return False return False
data = response.data if isinstance(response.data, dict) else {} data = response.data if isinstance(response.data, dict) else {}
registration_id = as_uuid(data.get("registrationId"), version=4) registration_id = as_uuid(data.get("registrationId"), version=4)
@@ -697,27 +698,27 @@ class PointRegistrationProcessor:
row, "FAILED", f"HTTP_{response.status_code}", "", row, "FAILED", f"HTTP_{response.status_code}", "",
"Yanıtta UUID v4 registrationId bulunamadı: " + json.dumps(response.data, ensure_ascii=False), "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) LOGGER.error("Kayıt %s yanıtında registrationId yok.", label)
return False return False
self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", registration_id, "") 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) LOGGER.info("Kayıt %s başarıyla kaydedildi: %s", label, registration_id)
for warning in response_warnings(response.data): for warning in response_warnings(response.data):
LOGGER.warning("Satır %s API uyarısı: %s", row_number, warning) LOGGER.warning("Kayıt %s API uyarısı: %s", label, warning)
return True return True
def _try_recover( def _try_recover(
self, row_number: int, payload: dict[str, Any], error: ApiError self, label: str, payload: dict[str, Any], error: ApiError
) -> tuple[int, str] | None: ) -> tuple[int, str] | None:
"""After an ambiguous failure, look for an identical registration (recipient + point + radius).""" """After an ambiguous failure, look for an identical registration (recipient + point + radius)."""
status = error.status_code status = error.status_code
if status is not None and status != 409 and status not in RETRYABLE_STATUS_CODES: if status is not None and status != 409 and status not in RETRYABLE_STATUS_CODES:
return None return None
LOGGER.warning("Satır %s sonucu belirsiz; recipientId ile kontrol ediliyor.", row_number) LOGGER.warning("Kayıt %s sonucu belirsiz; recipientId ile kontrol ediliyor.", label)
boundary = payload["boundary"] boundary = payload["boundary"]
try: try:
response = self.service.get_by_recipient_id(payload["recipientId"]) response = self.service.get_by_recipient_id(payload["recipientId"])
except ApiError as exc: except ApiError as exc:
LOGGER.error("Satır %s için recipient sorgusu başarısız: %s", row_number, exc) LOGGER.error("Kayıt %s için recipient sorgusu başarısız: %s", label, exc)
return None return None
data = response.data if isinstance(response.data, dict) else {} data = response.data if isinstance(response.data, dict) else {}
for registration in data.get("registrations") or []: for registration in data.get("registrations") or []:
@@ -881,25 +882,26 @@ class PointUpdateProcessor:
def run(self) -> None: def run(self) -> None:
rows, fieldnames = self._read_csv() rows, fieldnames = self._read_csv()
for row_number, row in enumerate(rows, start=2): for index, row in enumerate(rows, start=1):
label = f"{index}/{len(rows)}"
if str(row.get("status", "")).strip().upper() == "SUCCESS": if str(row.get("status", "")).strip().upper() == "SUCCESS":
LOGGER.info("Satır %s zaten SUCCESS; atlandı.", row_number) LOGGER.info("Kayıt %s zaten SUCCESS; atlandı.", label)
continue continue
try: try:
payload = build_update_payload(row) payload = build_update_payload(row)
except ConfigurationError as exc: except ConfigurationError as exc:
LOGGER.error("Satır %s gönderilmedi: %s", row_number, exc) LOGGER.error("Kayıt %s gönderilmedi: %s", label, exc)
self._set_result(row, "FAILED", "NOT_SENT", str(exc)) self._set_result(row, "FAILED", "NOT_SENT", str(exc))
self._write(rows, fieldnames) self._write(rows, fieldnames)
continue continue
if payload.get("filter", {}).get("thunderstorm", {}).get("speedThreshold") is not None: if payload.get("filter", {}).get("thunderstorm", {}).get("speedThreshold") is not None:
LOGGER.warning( LOGGER.warning(
"Satır %s: thunderstorm.speedThreshold PATCH'te backend tarafından işlenmiyor " "Kayıt %s: thunderstorm.speedThreshold PATCH'te backend tarafından işlenmiyor "
"(GAIA-476); değer gönderilir ancak güncellenmeyebilir.", "(GAIA-476); değer gönderilir ancak güncellenmeyebilir.",
row_number, label,
) )
if self.dry_run: if self.dry_run:
print(f"[dry-run] satır {row_number}: PATCH {ALARMS_PATH}/update {json.dumps(payload, ensure_ascii=False)}") print(f"[dry-run] kayıt {label}: PATCH {ALARMS_PATH}/update {json.dumps(payload, ensure_ascii=False)}")
continue continue
try: try:
response = self.service.update(payload, str(uuid4())) response = self.service.update(payload, str(uuid4()))
@@ -909,12 +911,12 @@ class PointUpdateProcessor:
f"HTTP_{exc.status_code}" if exc.status_code else "NETWORK_ERROR", f"HTTP_{exc.status_code}" if exc.status_code else "NETWORK_ERROR",
error_body_text(exc), error_body_text(exc),
) )
LOGGER.error("Satır %s başarısız: %s", row_number, exc) LOGGER.error("Kayıt %s başarısız: %s", label, exc)
else: else:
self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", "") self._set_result(row, "SUCCESS", f"HTTP_{response.status_code}", "")
LOGGER.info("Satır %s güncellendi: %s", row_number, payload["registrationId"]) LOGGER.info("Kayıt %s güncellendi: %s", label, payload["registrationId"])
for warning in response_warnings(response.data): for warning in response_warnings(response.data):
LOGGER.warning("Satır %s API uyarısı: %s", row_number, warning) LOGGER.warning("Kayıt %s API uyarısı: %s", label, warning)
self._write(rows, fieldnames) self._write(rows, fieldnames)
def _read_csv(self) -> tuple[list[dict[str, Any]], list[str]]: def _read_csv(self) -> tuple[list[dict[str, Any]], list[str]]: