diff --git a/nowcast-point-get.py b/nowcast-point-get.py new file mode 100644 index 0000000..8f9ceb3 --- /dev/null +++ b/nowcast-point-get.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python3 +"""Look up point alarm registrations by registrationId or recipientId from a CSV/TXT list.""" + +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path + +from common import ConfigurationError, IklimClient, configure_logging +from common.errors import ApiError +from point_alarm import ( + RESULT_COLUMNS, + PointAlarmService, + PointStore, + lookup_rows, + read_id_list, + timestamped_path, + write_records, +) + + +BASE_DIR = Path(__file__).resolve().parent + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Nokta alarm kayıtlarını get-by-id veya get-by-recipient-id ile sorgular. Girdi CSV " + "(başlıkta registrationId/recipientId sütunu, ';' ayraçlı) veya TXT (satır başına bir ID; " + "boş ve '#' ile başlayan satırlar atlanır) olabilir. Sonuçlar SQLite'a da yazılır." + ) + ) + parser.add_argument("input_file", type=Path, help="ID listesi (.csv veya .txt)") + parser.add_argument( + "--by", + choices=["id", "recipient"], + default="id", + help="Sorgu türü: id=registrationId, recipient=recipientId (varsayılan: id)", + ) + parser.add_argument( + "--output", + type=Path, + required=True, + help=( + "Çıktı dosyası: .csv (noktalı virgüllü) veya .txt (JSON Lines); dosya adına zaman soneki " + "eklenir. Satır durumları: SUCCESS, NOT_FOUND (HTTP 404), FAILED (httpStatus+errorBody). " + "Bir recipientId birden çok kayıt dönerse her kayıt ayrı satırdır" + ), + ) + parser.add_argument( + "--db", type=Path, default=BASE_DIR / "iklim-sandbox.db", help="SQLite dosyası" + ) + 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) + store = None + try: + column = "registrationId" if args.by == "id" else "recipientId" + queries = read_id_list(args.input_file, column) + client = IklimClient.from_env(BASE_DIR / ".env") + service = PointAlarmService(client) + store = PointStore(args.db) + rows, webhooks = lookup_rows(service, store, queries, args.by, None) + path = timestamped_path(args.output) + write_records(path, rows, webhooks, columns=RESULT_COLUMNS) + failed = sum(1 for row in rows if row["status"] == "FAILED") + print(f"{len(queries)} sorgu, {len(rows)} satır yazıldı: {path.resolve()}") + return 1 if failed else 0 + except (ConfigurationError, ApiError, OSError) as exc: + logging.error("İşlem durduruldu: %s", exc) + return 1 + finally: + if store: + store.close() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/point_alarm.py b/point_alarm.py index 770f0cc..52b8f4c 100644 --- a/point_alarm.py +++ b/point_alarm.py @@ -510,3 +510,77 @@ def prompt_export( write_records(path, records, store.webhooks(account_id)) print(f"{len(records)} kayıt yazıldı: {path.resolve()}") return path + + +def read_id_list(path: Path, column: str) -> list[str]: + """Read IDs from a CSV (header column) or TXT (one per line; blank and '#' lines skipped).""" + suffix = path.suffix.lower() + if suffix == ".csv": + with path.open(encoding="utf-8-sig", newline="") as handle: + rows = [row for row in csv.reader(handle, delimiter=";") if row and not row[0].startswith("#")] + if not rows or column not in [name.strip() for name in rows[0]]: + raise ConfigurationError(f"CSV başlığında '{column}' sütunu bulunamadı: {path}") + index = [name.strip() for name in rows[0]].index(column) + values = [row[index].strip() for row in rows[1:] if len(row) > index] + elif suffix == ".txt": + values = [line.strip() for line in path.read_text(encoding="utf-8-sig").splitlines()] + values = [line for line in values if line and not line.startswith("#")] + else: + raise ConfigurationError(f"Girdi dosyası .csv veya .txt uzantılı olmalıdır: {path}") + ids = list(dict.fromkeys(value for value in values if value)) + if not ids: + raise ConfigurationError(f"Girdi dosyasında ID bulunamadı: {path}") + return ids + + +RESULT_COLUMNS = ["query", "status", "httpStatus"] + RECORD_COLUMNS + ["errorBody"] + + +def lookup_rows( + service: PointAlarmService, + store: PointStore, + queries: list[str], + by: str, + account_id: str | None, +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """Run get-by-id / get-by-recipient-id per query; returns (result rows, webhooks).""" + rows: list[dict[str, Any]] = [] + for query in queries: + try: + response = ( + service.get_by_id(query) if by == "id" else service.get_by_recipient_id(query) + ) + except ApiError as exc: + status = "NOT_FOUND" if exc.status_code == 404 else "FAILED" + rows.append( + { + "query": query, + "status": status, + "httpStatus": exc.status_code, + "errorBody": ( + json.dumps(exc.response_data, ensure_ascii=False) + if exc.response_data is not None + else str(exc) + ), + } + ) + LOGGER.warning("%s için sonuç %s (HTTP %s).", query, status, exc.status_code) + continue + data = response.data + items = data if isinstance(data, list) else [data] + items = [item for item in items if isinstance(item, dict)] + if not items: + rows.append({"query": query, "status": "NOT_FOUND", "httpStatus": response.status_code}) + for item in items: + store.upsert_registration(item, account_id) + rows.append( + dict( + flatten_registration(item), + query=query, + status="SUCCESS", + httpStatus=response.status_code, + ) + ) + store.commit() + webhooks = store.webhooks(account_id) if account_id else store.webhooks() + return rows, webhooks diff --git a/tests/test_point_alarm.py b/tests/test_point_alarm.py index e28e588..c952508 100644 --- a/tests/test_point_alarm.py +++ b/tests/test_point_alarm.py @@ -7,11 +7,14 @@ from datetime import datetime from pathlib import Path from common import ApiError, ConfigurationError +from common.client import ApiResponse from point_alarm import ( PointStore, fetch_pages, flatten_registration, load_point_alarm_filter, + lookup_rows, + read_id_list, mask_secrets, parse_polygon, timestamped_path, @@ -122,6 +125,35 @@ class PointAlarmTest(unittest.TestCase): with self.assertRaises(ConfigurationError): load_point_alarm_filter(bad) + def test_read_id_list(self) -> None: + txt = self.dir / "ids.txt" + txt.write_text("# c\n\na\nb\na\n", encoding="utf-8") + self.assertEqual(read_id_list(txt, "registrationId"), ["a", "b"]) + csv_file = self.dir / "ids.csv" + csv_file.write_text("x;recipientId\n1;r1\n2;r2\n", encoding="utf-8") + self.assertEqual(read_id_list(csv_file, "recipientId"), ["r1", "r2"]) + with self.assertRaises(ConfigurationError): + read_id_list(csv_file, "registrationId") + + def test_lookup_rows(self) -> None: + class Svc: + def get_by_id(self, value): + if value == "missing": + raise ApiError("nf", status_code=404, response_data={"m": 1}) + if value == "boom": + raise ApiError("e", status_code=500, response_data="bad") + return ApiResponse(200, registration(1)) + + def get_by_recipient_id(self, value): + return ApiResponse(200, [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"]) + self.assertEqual(rows[2]["errorBody"], '"bad"') + rows, _ = lookup_rows(Svc(), self.store, ["r1"], "recipient", None) + self.assertEqual(len(rows), 2) + self.assertEqual(self.store.count(), 2) + if __name__ == "__main__": unittest.main()