add water history data

src/ems_water_grp/backfill_water_pg_1day.py

@@ -163,6 +163,26 @@ def normalize_items(items: List[dict]) -> List[Tuple[pendulum.DateTime, Decimal]
     return points
 
 
+def group_items_by_day(data: List[dict]) -> Dict[str, List[dict]]:
+    day_node_map = defaultdict(lambda: defaultdict(list))
+    for node in data:
+        node_id = node.get("id")
+        for point_time, value in normalize_items(node.get("items") or []):
+            day_key = point_time.to_date_string()
+            day_node_map[day_key][node_id].append({
+                "time": fmt_dt(point_time),
+                "value": str(value),
+            })
+
+    return {
+        day_key: [
+            {"id": node_id, "items": items}
+            for node_id, items in node_map.items()
+        ]
+        for day_key, node_map in day_node_map.items()
+    }
+
+
 class DeviceResolver:
     """
     Resolve Influx node id -> monitor id -> water/pg master id.
@@ -284,9 +304,15 @@ class WaterPgDailyBackfill:
             water_master_table: str,
             pg_master_table: str,
             dry_run: bool,
+            request_interval: float,
+            max_retries: int,
+            fetch_mode: str,
     ):
         self.db = db
         self.dry_run = dry_run
+        self.request_interval = request_interval
+        self.max_retries = max_retries
+        self.fetch_mode = fetch_mode
         self.http = HttpxClient()
         self.resolver = DeviceResolver(
             db=db,
@@ -305,20 +331,41 @@ class WaterPgDailyBackfill:
             day_end: pendulum.DateTime,
     ) -> List[dict]:
         payload = build_period_payload(day_start, day_end)
+        last_error = None
+        for attempt in range(1, self.max_retries + 2):
             resp_str, status_code = await self.http.post(
                 INFLUX_PERIOD_URL,
                 headers={"Content-Type": "application/json"},
                 json=payload,
             )
-        if status_code != 200:
-            raise RuntimeError(f"Influx fetch failed: HTTP {status_code}, {resp_str}")
 
+            try:
                 resp = json.loads(resp_str)
-        if resp.get("code") != 0:
-            raise RuntimeError(f"Influx fetch failed: {resp}")
+            except json.JSONDecodeError:
+                resp = {}
 
+            limited = (
+                status_code == 429
+                or status_code >= 500
+                or "请求频繁" in resp_str
+            )
+            if status_code == 200 and resp.get("code") == 0:
                 return resp.get("data") or []
 
+            last_error = f"HTTP {status_code}, {resp_str}"
+            if limited and attempt <= self.max_retries:
+                sleep_seconds = self.request_interval * attempt
+                logger.warning(
+                    f"Influx request limited, retry={attempt}/"
+                    f"{self.max_retries}, sleep={sleep_seconds}s, {last_error}"
+                )
+                await asyncio.sleep(sleep_seconds)
+                continue
+
+            break
+
+        raise RuntimeError(f"Influx fetch failed: {last_error}")
+
     def calc_day_rows(
             self,
             create_time: pendulum.DateTime,
@@ -417,6 +464,45 @@ class WaterPgDailyBackfill:
         if not devices:
             raise RuntimeError("No water/pg devices resolved from MySQL")
 
+        if self.fetch_mode == "range":
+            await self.run_range_mode(start, end, devices)
+            return
+
+        await self.run_day_mode(start, end, devices)
+
+    async def run_range_mode(
+            self,
+            start: pendulum.DateTime,
+            end: pendulum.DateTime,
+            devices: Dict[str, dict],
+    ):
+        total_water = 0
+        total_pg = 0
+        logger.info(f"Fetch whole period {fmt_dt(start)} ~ {fmt_dt(end)}")
+        data = await self.fetch_period(start, end)
+        day_data_map = group_items_by_day(data)
+
+        for create_time, day_start, day_end in iter_days(start, end):
+            day_key = create_time.to_date_string()
+            day_data = day_data_map.get(day_key, [])
+            water_rows, pg_rows = self.calc_day_rows(create_time, day_data, devices)
+            await self.save_rows(water_rows, pg_rows)
+
+            total_water += len(water_rows)
+            total_pg += len(pg_rows)
+            logger.info(
+                f"Saved day={create_time.to_date_string()}, "
+                f"water={len(water_rows)}, pg={len(pg_rows)}"
+            )
+
+        logger.info(f"Backfill finished. total_water={total_water}, total_pg={total_pg}")
+
+    async def run_day_mode(
+            self,
+            start: pendulum.DateTime,
+            end: pendulum.DateTime,
+            devices: Dict[str, dict],
+    ):
         total_water = 0
         total_pg = 0
         for create_time, day_start, day_end in iter_days(start, end):
@@ -431,6 +517,8 @@ class WaterPgDailyBackfill:
                 f"Saved day={create_time.to_date_string()}, "
                 f"water={len(water_rows)}, pg={len(pg_rows)}"
             )
+            if self.request_interval > 0:
+                await asyncio.sleep(self.request_interval)
 
         logger.info(f"Backfill finished. total_water={total_water}, total_pg={total_pg}")
 
@@ -446,6 +534,24 @@ def build_arg_parser() -> argparse.ArgumentParser:
     parser.add_argument("--water-master-table", default="water")
     parser.add_argument("--pg-master-table", default="water_pg")
     parser.add_argument("--dry-run", action="store_true", help="Fetch and calculate only")
+    parser.add_argument(
+        "--request-interval",
+        type=float,
+        default=3.0,
+        help="Seconds to sleep between daily Influx requests",
+    )
+    parser.add_argument(
+        "--max-retries",
+        type=int,
+        default=5,
+        help="Max retries for limited/5xx Influx requests",
+    )
+    parser.add_argument(
+        "--fetch-mode",
+        choices=("range", "day"),
+        default="range",
+        help="range: fetch whole period once; day: fetch one day per request",
+    )
     return parser
 
 
@@ -462,6 +568,9 @@ async def async_main():
         water_master_table=args.water_master_table,
         pg_master_table=args.pg_master_table,
         dry_run=args.dry_run,
+        request_interval=args.request_interval,
+        max_retries=args.max_retries,
+        fetch_mode=args.fetch_mode,
     )
     try:
         await service.run(parse_dt(args.start), parse_dt(args.end))