fix: RID collector keeps late readings instead of losing every evening

RID's hourly report often lags 1-2 h in the evening. The cycle discarded the
whole report unless it already held the current hour, the web-API loop then
slept a full hour, and after 01:00 only today's report was fetched. Result:
the 19:00-24:00 readings were dropped most nights (P.1 lost 00:00 on 23 of
the last 60 days, 13 evening hours this week), the feed went stale from
~21:00 until morning and ntfy ping-status sent a stale/recovered pair daily.

- run_scraping_cycle() upserts every reading the report has; "current hour
  present" now only decides the retry cadence (returned as before).
- Yesterday's report is re-fetched while the store has holes in it (hours
  22-24 are only reachable through that day's report).
- Leader loop polls at :05 (RID_POLL_OFFSET_MINUTES) and retries every
  RID_RETRY_MINUTES (10) while RID is behind; HII/dams/rain once per clock
  hour; forecast precompute + ntfy when newer readings arrive or hourly.
- Startup re-fetches holes from the last GAP_FILL_DAYS_ON_START (14) days.
- /scraping/status: the leader shares its stats via a JSON file, so any
  worker reports them (was always 0 runs); adds feed_current,
  latest_reading, readings_saved.

.env.example

@@ -53,6 +53,11 @@ THAIWATER_API_KEY=
 
 # Data Collection Settings
 SCRAPING_INTERVAL_HOURS=1
+# RID poll minute past the hour; retry interval while RID is behind
+RID_POLL_OFFSET_MINUTES=5
+RID_RETRY_MINUTES=10
+# Days of RID holes the leader re-fetches at startup (0 = off)
+GAP_FILL_DAYS_ON_START=14
 REQUEST_TIMEOUT=30
 MAX_RETRIES=3
 RETRY_DELAY_SECONDS=60

README.md

@@ -69,6 +69,13 @@ make test                              # pytest, synthetic data, no network
 make format                            # black + isort (the CI contract)
 ```
 
+RID collection in `--web-api` mode: polled at :05 past each hour
+(`RID_POLL_OFFSET_MINUTES`); while RID has not published the current hour yet,
+retried every `RID_RETRY_MINUTES` (10). Every poll stores all readings the report
+has, and yesterday's report is re-fetched while it still has holes (RID's evening
+hours often arrive 1-2 h late). On startup the leader re-fetches holes from the
+last `GAP_FILL_DAYS_ON_START` (14) days; for older history run `--fill-gaps all`.
+
 ## API
 
 Read-only, no key, JSON. Base URL `https://water.buildfor.life`; timestamps are

src/config.py

@@ -134,6 +134,16 @@ class Config:
 
     # Scheduler settings
     SCRAPING_INTERVAL_HOURS = int(os.getenv("SCRAPING_INTERVAL_HOURS", "1"))
+    # RID is polled this many minutes past each hour; while the current hour is
+    # not published yet (the evening report often lags 1-2 h) it is retried
+    # every RID_RETRY_MINUTES. Late readings are stored either way.
+    RID_POLL_OFFSET_MINUTES = int(os.getenv("RID_POLL_OFFSET_MINUTES", "5"))
+    RID_RETRY_MINUTES = int(os.getenv("RID_RETRY_MINUTES", "10"))
+    # On startup the collection leader re-fetches RID reports with holes from
+    # the last N days (0 = off). Older history: `run.py --fill-gaps all`.
+    GAP_FILL_DAYS_ON_START = int(os.getenv("GAP_FILL_DAYS_ON_START", "14"))
+    # Leader -> other workers hand-off for /scraping/status (default: temp dir)
+    SCRAPING_STATUS_FILE = os.getenv("SCRAPING_STATUS_FILE", "")
 
     # Logging settings
     LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
@@ -209,6 +219,12 @@ class Config:
         if cls.SCRAPING_INTERVAL_HOURS <= 0:
             errors.append("SCRAPING_INTERVAL_HOURS must be positive")
 
+        if not 0 <= cls.RID_POLL_OFFSET_MINUTES < 60:
+            errors.append("RID_POLL_OFFSET_MINUTES must be between 0 and 59")
+
+        if cls.RID_RETRY_MINUTES <= 0:
+            errors.append("RID_RETRY_MINUTES must be positive")
+
         if cls.DATA_RETENTION_DAYS <= 0:
             errors.append("DATA_RETENTION_DAYS must be positive")
 
@@ -274,6 +290,9 @@ class Config:
             "API_URL": cls.API_URL,
             "REQUEST_TIMEOUT": cls.REQUEST_TIMEOUT,
             "SCRAPING_INTERVAL_HOURS": cls.SCRAPING_INTERVAL_HOURS,
+            "RID_POLL_OFFSET_MINUTES": cls.RID_POLL_OFFSET_MINUTES,
+            "RID_RETRY_MINUTES": cls.RID_RETRY_MINUTES,
+            "GAP_FILL_DAYS_ON_START": cls.GAP_FILL_DAYS_ON_START,
             "LOG_LEVEL": cls.LOG_LEVEL,
             "LOG_FILE": cls.LOG_FILE,
             "DATA_RETENTION_DAYS": cls.DATA_RETENTION_DAYS,

src/schemas.py

@@ -73,5 +73,8 @@ class ScrapingStatusResponse(BaseModel):
     last_run: Optional[datetime] = None
     next_run: Optional[datetime] = None
     total_runs: int = 0
-    successful_runs: int = 0
+    successful_runs: int = 0  # cycles that found the current hour published
-    failed_runs: int = 0
+    failed_runs: int = 0  # cycles behind (RID late) or failed
+    feed_current: Optional[bool] = None  # last cycle had the current hour
+    latest_reading: Optional[datetime] = None  # newest RID reading stored
+    readings_saved: int = 0  # readings upserted by the last cycle

src/water_scraper_v3.py

@@ -87,6 +87,7 @@ class EnhancedWaterMonitorScraper:
         self.last_successful_update = None
         self.retry_mode = False
         self.next_hourly_check = None
+        self.last_cycle: Optional[Dict] = None  # see run_scraping_cycle()
 
         # Rate limiting and request tracking
         self.rate_limiter = RateLimiter(max_requests=10, time_window_seconds=60)
@@ -495,86 +496,99 @@ class EnhancedWaterMonitorScraper:
             logger.error(f"Error getting latest data: {e}")
             return []
 
-    def _check_data_freshness(self, water_data: List[Dict]) -> bool:
+    def _cycle_source_dates(self, now: datetime.datetime) -> List[datetime.date]:
-        """Check if the fetched data contains new data for the current hour"""
+        """Report dates (the API's ``DW[TimeCurrent]``) to fetch in one cycle.
-        if not water_data:
-            return False
 
-        current_time = datetime.datetime.now()
+        Always the report holding the current hour: today's, or before 01:00
-        current_hour = current_time.hour
+        yesterday's (its "hour 24" is today's 00:00). Plus yesterday's report
+        while the store still has holes in it. RID often publishes the evening
+        hours an hour or two late, and hours 22-24 of a day are only reachable
+        through that day's report, so they must be re-fetched after midnight.
+        """
+        today = now.date()
+        yesterday = today - datetime.timedelta(days=1)
+        dates = {today if now.hour >= 1 else yesterday}
+        try:
+            dates.update(self._find_gap_fetch_dates(yesterday, today, now))
+        except Exception as e:
+            logger.warning(f"Gap detection failed, fetching the current report: {e}")
+        return sorted(dates)
 
-        # Find the most recent timestamp in the data
+    def run_scraping_cycle(self, now: Optional[datetime.datetime] = None) -> bool:
-        latest_timestamp = None
+        """Fetch the RID hourly report(s) and store every reading in them.
-        for data_point in water_data:
-            timestamp = data_point.get("timestamp")
-            if timestamp and (latest_timestamp is None or timestamp > latest_timestamp):
-                latest_timestamp = timestamp
 
-        if latest_timestamp is None:
+        Returns True when the current hour's reading is in (the feed is
-            logger.warning("No valid timestamps found in data")
+        current) and False when the caller should retry soon. Readings are
-            return False
+        saved either way: RID's report regularly lags an hour or two in the
+        evening, and an all-or-nothing freshness check used to discard every
+        reading of a late report, losing the evening hours and leaving the
+        feed stale until the next morning.
 
-        latest_hour = latest_timestamp.hour
+        Details of the last cycle are kept in ``self.last_cycle``.
-        time_diff = current_time - latest_timestamp
+        """
-        minutes_old = time_diff.total_seconds() / 60
+        now = now or datetime.datetime.now()
-
+        current_hour = now.replace(minute=0, second=0, microsecond=0)
-        logger.info(
+        status = {
-            f"Current time: {current_time.strftime('%H:%M')}, Latest data: {latest_timestamp.strftime('%H:%M')}"
+            "time": now,
-        )
+            "dates": [],
-        logger.info(
+            "fetched": 0,
-            f"Current hour: {current_hour}, Latest data hour: {latest_hour}, Age: {minutes_old:.1f} minutes"
+            "saved": 0,
-        )
+            "latest_reading": None,
-
+            "current": False,
-        # Strict check: we need data from the current hour
+            "error": None,
-        # If it's 20:xx and we only have data up to 19:xx, that's stale - go to retry mode
+        }
-        has_current_hour_data = latest_hour >= current_hour
+        self.last_cycle = status
-
-        if not has_current_hour_data:
-            logger.warning(
-                f"No new data available - expected hour {current_hour}, got {latest_hour}"
-            )
-            logger.warning("Switching to retry mode until new data becomes available")
-            return False
-        else:
-            logger.info(f"Fresh data available for current hour {current_hour}")
-            return True
-
-    def run_scraping_cycle(self) -> bool:
-        """Run a complete scraping cycle with freshness check"""
         logger.info("Starting scraping cycle...")
 
         try:
-            # Fetch current data
+            dates = self._cycle_source_dates(now)
-            water_data = self.fetch_water_data()
+            status["dates"] = [d.isoformat() for d in dates]
-            if water_data:
-                # Check if data is fresh/recent
-                is_fresh = self._check_data_freshness(water_data)
 
-                if is_fresh:
+            water_data: List[Dict] = []
-                    success = self.save_to_database(water_data)
+            for source_date in dates:
-                    if success:
+                rows = self.fetch_water_data_for_date(
-                        logger.info(
+                    datetime.datetime.combine(source_date, datetime.time.min)
-                            "Scraping cycle completed successfully with fresh data"
+                )
-                        )
+                if rows:
-                        increment_counter("scraping_cycles_successful")
+                    water_data.extend(rows)
-                        return True
+            status["fetched"] = len(water_data)
-                    else:
+
-                        logger.error("Failed to save data")
+            if not water_data:
-                        increment_counter("scraping_cycles_failed")
+                status["error"] = "no data fetched"
-                        return False
-                else:
-                    # Data exists but is stale
-                    logger.warning(
-                        "Data fetched but is stale - treating as no fresh data available"
-                    )
-                    increment_counter("scraping_cycles_failed")
-                    return False
-            else:
                 logger.warning("No data fetched")
                 increment_counter("scraping_cycles_failed")
                 return False
 
+            if not self.save_to_database(water_data):
+                status["error"] = "save failed"
+                logger.error("Failed to save data")
+                increment_counter("scraping_cycles_failed")
+                return False
+            status["saved"] = len(water_data)
+
+            latest = max(m["timestamp"] for m in water_data)
+            status["latest_reading"] = latest
+            status["current"] = latest >= current_hour
+
+            if status["current"]:
+                logger.info(
+                    f"Scraping cycle complete: {len(water_data)} readings "
+                    f"saved, latest {latest:%Y-%m-%d %H:%M}"
+                )
+                increment_counter("scraping_cycles_successful")
+            else:
+                behind_h = (current_hour - latest).total_seconds() / 3600
+                logger.warning(
+                    f"Saved {len(water_data)} readings up to "
+                    f"{latest:%Y-%m-%d %H:%M}; RID has not published "
+                    f"{current_hour:%H:%M} yet ({behind_h:.0f} h behind), "
+                    "will retry"
+                )
+                increment_counter("scraping_cycles_behind")
+            return status["current"]
+
         except Exception as e:
+            status["error"] = str(e)
             logger.error(f"Scraping cycle failed: {e}")
             increment_counter("scraping_cycles_failed")
             return False

src/web_api.py

@@ -4,8 +4,10 @@ FastAPI web interface for water monitoring system
 """
 
 import asyncio
+import json
 import os
 import secrets
+import tempfile
 import time
 from contextlib import asynccontextmanager
 from datetime import date, datetime, timedelta
@@ -110,6 +112,9 @@ app_state = {
         "failed_runs": 0,
         "last_run": None,
         "next_run": None,
+        "feed_current": None,
+        "latest_reading": None,
+        "readings_saved": 0,
     },
 }
 
@@ -391,100 +396,188 @@ async def _precompute_forecasts():
             logger.warning(f"Forecast history save failed: {e}")
 
 
+def _next_poll_time(now: datetime, feed_current: bool) -> datetime:
+    """When the collection leader should poll RID next.
+
+    On schedule: a few minutes past the next hour, once RID has usually
+    published that hour. While the current hour is still missing (RID's
+    evening report often lags 1-2 h), retry every RID_RETRY_MINUTES instead of
+    waiting a whole interval.
+    """
+    on_schedule = now.replace(minute=0, second=0, microsecond=0) + timedelta(
+        hours=Config.SCRAPING_INTERVAL_HOURS,
+        minutes=Config.RID_POLL_OFFSET_MINUTES,
+    )
+    if feed_current:
+        return on_schedule
+    return min(now + timedelta(minutes=Config.RID_RETRY_MINUTES), on_schedule)
+
+
+def _scraping_status_path() -> str:
+    return Config.SCRAPING_STATUS_FILE or os.path.join(
+        tempfile.gettempdir(),
+        f"ping-monitor-scraping-{Config.COLLECTION_LEADER_PORT}.json",
+    )
+
+
+def _record_scrape_run(start_time: datetime, feed_current: bool):
+    """Update the scrape counters from the scraper's last cycle."""
+    stats = app_state["scraping_stats"]
+    stats["total_runs"] += 1
+    stats["last_run"] = start_time
+    stats["successful_runs" if feed_current else "failed_runs"] += 1
+    scraper = app_state.get("scraper")
+    cycle = getattr(scraper, "last_cycle", None) or {}
+    stats["feed_current"] = bool(feed_current)
+    stats["readings_saved"] = cycle.get("saved", 0)
+    if cycle.get("latest_reading"):
+        stats["latest_reading"] = cycle["latest_reading"]
+
+
+def _publish_scraping_status():
+    """Share the leader's scrape stats with the other uvicorn workers.
+
+    Only the leader collects, but /scraping/status can be answered by any
+    worker, so the leader writes its stats to a small JSON file (atomically).
+    """
+    if not app_state.get("leader_lock"):
+        return
+    payload = {"is_running": app_state["is_scraping"]}
+    for key, value in app_state["scraping_stats"].items():
+        payload[key] = value.isoformat() if isinstance(value, datetime) else value
+    path = _scraping_status_path()
+    try:
+        tmp = f"{path}.tmp"
+        with open(tmp, "w", encoding="utf-8") as f:
+            json.dump(payload, f)
+        os.replace(tmp, path)
+    except OSError as e:
+        logger.warning(f"Could not write scraping status to {path}: {e}")
+
+
+def _read_scraping_status() -> Optional[Dict[str, Any]]:
+    try:
+        with open(_scraping_status_path(), encoding="utf-8") as f:
+            return json.load(f)
+    except (OSError, ValueError):
+        return None
+
+
+async def _collect_hourly_sources():
+    """HII rainfall/water level, RID dams and Open-Meteo rain (leader, hourly)."""
+    loop = asyncio.get_running_loop()
+
+    hii_collector = app_state.get("hii_collector")
+    if hii_collector:
+        try:
+            hii_counts = await loop.run_in_executor(None, hii_collector.run_cycle)
+            set_gauge("hii_rainfall_rows_saved", hii_counts["rainfall"])
+            set_gauge("hii_waterlevel_rows_saved", hii_counts["waterlevel"])
+        except Exception as e:
+            logger.error(f"HII collection failed: {e}")
+
+    reservoir_collector = app_state.get("reservoir_collector")
+    if reservoir_collector:
+        try:
+            rsv_saved = await loop.run_in_executor(None, reservoir_collector.run_cycle)
+            set_gauge("reservoir_rows_saved", rsv_saved)
+        except Exception as e:
+            logger.error(f"Reservoir collection failed: {e}")
+
+    # Persist the Open-Meteo catchment rain (observed tail + 48h forecast) so
+    # the DB carries the weather context too
+    await _persist_rain()
+
+
+async def _startup_gap_fill(scraper):
+    """Re-fetch recent RID reports that have holes (e.g. lost evening hours)."""
+    days = Config.GAP_FILL_DAYS_ON_START
+    if days <= 0:
+        return
+    try:
+        filled = await asyncio.get_running_loop().run_in_executor(
+            None, scraper.fill_data_gaps, days
+        )
+        logger.info(f"Startup gap fill ({days} days): {filled} readings re-fetched")
+    except Exception as e:
+        logger.error(f"Startup gap fill failed: {e}")
+
+
 async def background_scraping_task():
-    """Background task for periodic data scraping"""
+    """Collection loop, run by the leader worker only.
+
+    RID: polled a few minutes past each hour and, while the current hour is
+    not published yet, every RID_RETRY_MINUTES. Every poll stores all readings
+    the report has, so a late report costs latency, not data.
+    HII, dams and Open-Meteo rain: once per clock hour.
+    Forecast precompute + ntfy: whenever newer RID readings arrive, and once
+    per clock hour regardless (so a feed that stops is noticed and reported).
+    """
+    loop = asyncio.get_running_loop()
+    last_hourly_slot = None  # clock hour of the last hourly collection
+    newest_reading = None  # newest RID reading stored so far
+
+    scraper = app_state["scraper"]
+    if scraper:
+        app_state["is_scraping"] = True
+        try:
+            await _startup_gap_fill(scraper)
+        finally:
+            app_state["is_scraping"] = False
+
     while True:
         try:
-            if not app_state["is_scraping"]:
+            feed_current = False
-                app_state["is_scraping"] = True
+            scraper = app_state["scraper"]
 
-                # Run scraping cycle
+            if scraper and not app_state["is_scraping"]:
-                scraper = app_state["scraper"]
+                app_state["is_scraping"] = True
-                if scraper:
+                try:
                     logger.info("Starting background scraping cycle")
                     start_time = datetime.now()
-
                     try:
-                        # run_scraping_cycle() does blocking network/DB I/O and time.sleep
+                        # Blocking network/DB I/O: keep it off the event loop.
-                        # retries; run it in a thread so it doesn't freeze the event loop.
+                        feed_current = bool(
-                        result = await asyncio.get_event_loop().run_in_executor(
+                            await loop.run_in_executor(None, scraper.run_scraping_cycle)
-                            None, scraper.run_scraping_cycle
                         )
-
-                        # Update stats
-                        app_state["scraping_stats"]["total_runs"] += 1
-                        app_state["scraping_stats"]["last_run"] = start_time
-
-                        if result:
-                            app_state["scraping_stats"]["successful_runs"] += 1
-                            increment_counter("scraping_cycles_successful")
-                            logger.info(
-                                "Background scraping cycle completed successfully"
-                            )
-                        else:
-                            app_state["scraping_stats"]["failed_runs"] += 1
-                            increment_counter("scraping_cycles_failed")
-                            logger.warning(
-                                "Background scraping cycle completed with no new data"
-                            )
-
-                        # Update metrics
-                        set_gauge("last_scraping_timestamp", start_time.timestamp())
-
                     except Exception as e:
-                        app_state["scraping_stats"]["failed_runs"] += 1
-                        increment_counter("scraping_cycles_failed")
                         logger.error(f"Background scraping cycle failed: {e}")
+                    _record_scrape_run(start_time, feed_current)
+                    set_gauge("last_scraping_timestamp", start_time.timestamp())
 
-                    # HII rainfall/water-level snapshot (independent of RID cycle)
+                    latest = (getattr(scraper, "last_cycle", None) or {}).get(
-                    hii_collector = app_state.get("hii_collector")
+                        "latest_reading"
-                    if hii_collector:
+                    )
-                        try:
+                    new_readings = latest is not None and (
-                            hii_counts = await asyncio.get_event_loop().run_in_executor(
+                        newest_reading is None or latest > newest_reading
-                                None, hii_collector.run_cycle
+                    )
-                            )
+                    if new_readings:
-                            set_gauge("hii_rainfall_rows_saved", hii_counts["rainfall"])
+                        newest_reading = latest
-                            set_gauge(
-                                "hii_waterlevel_rows_saved", hii_counts["waterlevel"]
-                            )
-                        except Exception as e:
-                            logger.error(f"HII collection failed: {e}")
 
-                    # RID large-dam daily status (Mae Ngat storage/inflow/outflow)
+                    slot = datetime.now().replace(minute=0, second=0, microsecond=0)
-                    reservoir_collector = app_state.get("reservoir_collector")
+                    hourly_due = slot != last_hourly_slot
-                    if reservoir_collector:
+                    if hourly_due:
-                        try:
+                        last_hourly_slot = slot
-                            rsv_saved = await asyncio.get_event_loop().run_in_executor(
+                        await _collect_hourly_sources()
-                                None, reservoir_collector.run_cycle
-                            )
-                            set_gauge("reservoir_rows_saved", rsv_saved)
-                        except Exception as e:
-                            logger.error(f"Reservoir collection failed: {e}")
 
-                    # Persist the Open-Meteo catchment rain (observed tail +
+                    if new_readings or hourly_due:
-                    # 48h forecast) so the DB carries the weather context too
+                        # Primes the forecast cache (user requests never pay
-                    await _persist_rain()
+                        # for inference), records the issued forecast for
+                        # predicted-vs-actual evaluation, then pushes any
+                        # threshold / feed transitions to ntfy.
+                        await _precompute_forecasts()
+                        await _notify_transitions()
+                finally:
+                    app_state["is_scraping"] = False
 
-                    # Precompute forecasts on fresh data: primes the response
+            next_run = _next_poll_time(datetime.now(), feed_current)
-                    # cache (user requests never pay for inference) and records
+            app_state["scraping_stats"]["next_run"] = next_run
-                    # what the model predicted for later predicted-vs-actual
+            _publish_scraping_status()
-                    # evaluation.
+            logger.info(
-                    await _precompute_forecasts()
+                f"Next RID poll at {next_run:%H:%M}"
-
+                + ("" if feed_current else " (current hour not published yet)")
-                    # Push notifications for threshold crossings (uses the
-                    # forecasts just computed; no-op unless NTFY_SERVER set).
-                    await _notify_transitions()
-
-                app_state["is_scraping"] = False
-
-            # Calculate next run time
-            interval_seconds = Config.SCRAPING_INTERVAL_HOURS * 3600
-            app_state["scraping_stats"]["next_run"] = datetime.now() + timedelta(
-                seconds=interval_seconds
             )
-
+            await asyncio.sleep(max(30.0, (next_run - datetime.now()).total_seconds()))
-            # Wait for next cycle
-            await asyncio.sleep(interval_seconds)
 
         except asyncio.CancelledError:
             logger.info("Background scraping task cancelled")
@@ -1578,18 +1671,12 @@ async def trigger_scraping(background_tasks: BackgroundTasks):
             app_state["is_scraping"] = True
             logger.info("Manual scraping triggered via API")
 
+            start_time = datetime.now()
             result = scraper.run_scraping_cycle()
-
+            _record_scrape_run(start_time, result)
-            # Update stats
+            increment_counter(
-            app_state["scraping_stats"]["total_runs"] += 1
+                "manual_scraping_successful" if result else "manual_scraping_failed"
-            app_state["scraping_stats"]["last_run"] = datetime.now()
+            )
-
-            if result:
-                app_state["scraping_stats"]["successful_runs"] += 1
-                increment_counter("manual_scraping_successful")
-            else:
-                app_state["scraping_stats"]["failed_runs"] += 1
-                increment_counter("manual_scraping_failed")
 
         except Exception as e:
             app_state["scraping_stats"]["failed_runs"] += 1
@@ -1608,15 +1695,14 @@ async def get_scraping_status():
     """Get current scraping status"""
     increment_counter("api_requests", labels={"endpoint": "scraping_status"})
 
-    stats = app_state["scraping_stats"]
+    # Only the leader worker collects; the others report what it published.
+    if app_state.get("leader_lock"):
+        stats = dict(app_state["scraping_stats"], is_running=app_state["is_scraping"])
+    else:
+        stats = _read_scraping_status() or {"is_running": False}
 
     return ScrapingStatusResponse(
-        is_running=app_state["is_scraping"],
+        **{k: v for k, v in stats.items() if k in ScrapingStatusResponse.model_fields}
-        last_run=stats["last_run"],
-        next_run=stats["next_run"],
-        total_runs=stats["total_runs"],
-        successful_runs=stats["successful_runs"],
-        failed_runs=stats["failed_runs"],
     )
 
 

tests/test_scraping_cycle.py

@@ -0,0 +1,307 @@
+"""The RID scraping cycle must keep every reading of a late report.
+
+RID's hourly report often lags 1-2 h in the evening. The cycle used to discard
+a whole report unless it already held the current hour, and after 01:00 it
+only fetched today's report, so the evening hours (and the "hour 24" =
+00:00 reading) were lost for good. These tests pin the fixed behaviour with a
+mocked HTTP session and DB adapter: no network, no database.
+"""
+
+import datetime
+from unittest.mock import MagicMock
+
+import pytest
+
+import src.water_scraper_v3 as scraper_mod
+from src.water_scraper_v3 import EnhancedWaterMonitorScraper as Scraper
+
+TODAY = datetime.date(2026, 10, 17)
+YESTERDAY = TODAY - datetime.timedelta(days=1)
+
+
+def _thai(d: datetime.date) -> str:
+    return f"{d.day:02d}/{d.month:02d}/{d.year + 543}"
+
+
+def _report(last_hour: int):
+    """RID rows 1.00..24.00 with readings up to ``last_hour`` (rest null)."""
+    return [
+        {
+            "hourlytime": f"{h}.00",
+            "wlvalues8": 1.0 + h / 100 if h <= last_hour else None,
+        }
+        for h in range(1, 25)
+    ]
+
+
+@pytest.fixture
+def make_scraper(monkeypatch):
+    monkeypatch.setattr(
+        scraper_mod.DataValidator,
+        "validate_measurements",
+        staticmethod(lambda m: m),
+    )
+
+    def _factory(reports, hours_by_day):
+        """reports: {date: last published hour}; hours_by_day: DB contents."""
+        scraper = Scraper.__new__(Scraper)  # bypass __init__ (no DB/network)
+        scraper.api_url = "https://example.invalid/api"
+        scraper.rate_limiter = MagicMock()
+        scraper.request_tracker = MagicMock()
+        scraper.station_config_path = "/nonexistent/stations.json"
+        scraper.station_mapping = scraper._load_station_mapping()
+        scraper.last_cycle = None
+
+        by_thai = {_thai(d): last for d, last in reports.items()}
+
+        def post(url, data, timeout):
+            response = MagicMock()
+            response.raise_for_status.return_value = None
+            last = by_thai.get(data["DW[TimeCurrent]"])
+            response.json.return_value = {
+                "rows": _report(last) if last is not None else []
+            }
+            return response
+
+        scraper.session = MagicMock()
+        scraper.session.post.side_effect = post
+
+        scraper.db_adapter = MagicMock()
+        scraper.db_adapter.get_recorded_hours_by_day.return_value = hours_by_day
+        scraper.db_adapter.save_measurements.return_value = True
+        return scraper
+
+    return _factory
+
+
+def _fetched_dates(scraper):
+    return [
+        c.kwargs["data"]["DW[TimeCurrent]"] for c in scraper.session.post.call_args_list
+    ]
+
+
+def _saved_times(scraper):
+    saved = []
+    for c in scraper.db_adapter.save_measurements.call_args_list:
+        saved.extend(m["timestamp"] for m in c.args[0])
+    return sorted(set(saved))
+
+
+def test_late_report_is_saved_but_not_current(make_scraper):
+    # 21:30, RID has only published up to 20:00 (the case that lost data)
+    scraper = make_scraper(
+        {TODAY: 20},
+        {YESTERDAY: set(range(24)), TODAY: set(range(0, 19))},
+    )
+    now = datetime.datetime(2026, 10, 17, 21, 30)
+
+    assert scraper.run_scraping_cycle(now=now) is False
+
+    saved = _saved_times(scraper)
+    assert datetime.datetime(2026, 10, 17, 19) in saved
+    assert datetime.datetime(2026, 10, 17, 20) in saved
+    assert scraper.last_cycle["latest_reading"] == datetime.datetime(2026, 10, 17, 20)
+    assert scraper.last_cycle["current"] is False
+
+
+def test_current_hour_published_is_current(make_scraper):
+    scraper = make_scraper({TODAY: 21}, {YESTERDAY: set(range(24))})
+    now = datetime.datetime(2026, 10, 17, 21, 5)
+
+    assert scraper.run_scraping_cycle(now=now) is True
+    assert scraper.last_cycle["latest_reading"] == datetime.datetime(2026, 10, 17, 21)
+
+
+def test_holes_in_yesterday_refetch_yesterdays_report(make_scraper):
+    # 02:10: yesterday's 22:00-23:00 and today's 00:00 (yesterday's hour 24)
+    # were published late; only yesterday's report still holds them
+    scraper = make_scraper(
+        {YESTERDAY: 24, TODAY: 2},
+        {YESTERDAY: set(range(22)), TODAY: {1}},
+    )
+    now = datetime.datetime(2026, 10, 17, 2, 10)
+
+    assert scraper.run_scraping_cycle(now=now) is True
+
+    assert _fetched_dates(scraper) == [_thai(YESTERDAY), _thai(TODAY)]
+    saved = _saved_times(scraper)
+    for hour in (
+        datetime.datetime(2026, 10, 16, 22),
+        datetime.datetime(2026, 10, 16, 23),
+        datetime.datetime(2026, 10, 17, 0),
+    ):
+        assert hour in saved
+
+
+def test_complete_yesterday_is_not_refetched(make_scraper):
+    scraper = make_scraper(
+        {TODAY: 15},
+        {YESTERDAY: set(range(24)), TODAY: set(range(15))},
+    )
+    scraper.run_scraping_cycle(now=datetime.datetime(2026, 10, 17, 15, 5))
+
+    assert _fetched_dates(scraper) == [_thai(TODAY)]
+
+
+def test_before_one_am_uses_yesterdays_report(make_scraper):
+    # 00:20: today's report is empty; midnight is yesterday's "hour 24"
+    scraper = make_scraper({YESTERDAY: 24}, {YESTERDAY: set(range(24))})
+
+    assert scraper.run_scraping_cycle(now=datetime.datetime(2026, 10, 17, 0, 20))
+    assert _fetched_dates(scraper) == [_thai(YESTERDAY)]
+    assert scraper.last_cycle["latest_reading"] == datetime.datetime(2026, 10, 17, 0)
+
+
+def test_nothing_published_returns_false_without_saving(make_scraper):
+    scraper = make_scraper({}, {YESTERDAY: set(range(24))})
+
+    assert (
+        scraper.run_scraping_cycle(now=datetime.datetime(2026, 10, 17, 9, 5)) is False
+    )
+    scraper.db_adapter.save_measurements.assert_not_called()
+    assert scraper.last_cycle["error"] == "no data fetched"
+
+
+def test_gap_detection_failure_still_fetches_current_report(make_scraper):
+    scraper = make_scraper({TODAY: 9}, {})
+    scraper.db_adapter.get_recorded_hours_by_day.side_effect = RuntimeError("db down")
+
+    assert scraper.run_scraping_cycle(now=datetime.datetime(2026, 10, 17, 9, 5))
+    assert _fetched_dates(scraper) == [_thai(TODAY)]
+
+
+class TestNextPollTime:
+    @pytest.fixture(autouse=True)
+    def _config(self, monkeypatch):
+        from src.config import Config
+
+        monkeypatch.setattr(Config, "SCRAPING_INTERVAL_HOURS", 1)
+        monkeypatch.setattr(Config, "RID_POLL_OFFSET_MINUTES", 5)
+        monkeypatch.setattr(Config, "RID_RETRY_MINUTES", 10)
+
+    def test_current_feed_waits_for_next_hour_plus_offset(self):
+        from src.web_api import _next_poll_time
+
+        now = datetime.datetime(2026, 10, 17, 21, 6)
+        assert _next_poll_time(now, True) == datetime.datetime(2026, 10, 17, 22, 5)
+
+    def test_behind_feed_retries_soon(self):
+        from src.web_api import _next_poll_time
+
+        now = datetime.datetime(2026, 10, 17, 21, 6)
+        assert _next_poll_time(now, False) == datetime.datetime(2026, 10, 17, 21, 16)
+
+    def test_retry_never_skips_the_hourly_poll(self):
+        from src.web_api import _next_poll_time
+
+        now = datetime.datetime(2026, 10, 17, 21, 58)
+        assert _next_poll_time(now, False) == datetime.datetime(2026, 10, 17, 22, 5)
+
+
+def test_scraping_status_shared_through_file(tmp_path, monkeypatch):
+    from src import web_api
+    from src.config import Config
+
+    monkeypatch.setattr(Config, "SCRAPING_STATUS_FILE", str(tmp_path / "s.json"))
+    monkeypatch.setitem(web_api.app_state, "leader_lock", object())
+    monkeypatch.setitem(web_api.app_state, "is_scraping", False)
+    stats = dict(web_api.app_state["scraping_stats"])
+    stats.update(
+        total_runs=3,
+        successful_runs=2,
+        failed_runs=1,
+        feed_current=True,
+        latest_reading=datetime.datetime(2026, 10, 17, 21),
+    )
+    monkeypatch.setitem(web_api.app_state, "scraping_stats", stats)
+
+    web_api._publish_scraping_status()
+    shared = web_api._read_scraping_status()
+
+    assert shared["total_runs"] == 3
+    assert shared["latest_reading"] == "2026-10-17T21:00:00"
+    resp = web_api.ScrapingStatusResponse(
+        **{
+            k: v
+            for k, v in shared.items()
+            if k in web_api.ScrapingStatusResponse.model_fields
+        }
+    )
+    assert resp.latest_reading == datetime.datetime(2026, 10, 17, 21)
+    assert resp.feed_current is True
+
+
+def test_loop_retries_while_behind_and_runs_hourly_work_once(monkeypatch):
+    """Leader loop: 10-min retries while RID is behind; HII/dams/rain once per
+    clock hour; forecasts + ntfy only when newer readings arrive (or hourly)."""
+    import asyncio
+
+    from src import web_api
+    from src.config import Config
+
+    clock = datetime.datetime(2026, 10, 17, 21, 30)
+
+    class FixedClock(datetime.datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return clock
+
+    class FakeScraper:
+        last_cycle = None
+        results = [
+            (False, datetime.datetime(2026, 10, 17, 20)),
+            (False, datetime.datetime(2026, 10, 17, 20)),
+            (True, datetime.datetime(2026, 10, 17, 21)),
+        ]
+
+        def fill_data_gaps(self, days):
+            calls["gap_fill"] = days
+            return 0
+
+        def run_scraping_cycle(self):
+            current, latest = self.results.pop(0)
+            self.last_cycle = {"latest_reading": latest, "saved": 16}
+            return current
+
+    calls = {"hourly": 0, "precompute": 0, "notify": 0, "gap_fill": None}
+    sleeps = []
+
+    def counter(name):
+        async def _fn():
+            calls[name] += 1
+
+        return _fn
+
+    async def fake_sleep(seconds):
+        sleeps.append(seconds)
+        if len(sleeps) == 3:
+            raise asyncio.CancelledError
+
+    monkeypatch.setattr(web_api, "datetime", FixedClock)
+    monkeypatch.setattr(web_api.asyncio, "sleep", fake_sleep)
+    monkeypatch.setattr(web_api, "_collect_hourly_sources", counter("hourly"))
+    monkeypatch.setattr(web_api, "_precompute_forecasts", counter("precompute"))
+    monkeypatch.setattr(web_api, "_notify_transitions", counter("notify"))
+    monkeypatch.setattr(Config, "SCRAPING_INTERVAL_HOURS", 1)
+    monkeypatch.setattr(Config, "RID_POLL_OFFSET_MINUTES", 5)
+    monkeypatch.setattr(Config, "RID_RETRY_MINUTES", 10)
+    monkeypatch.setattr(Config, "GAP_FILL_DAYS_ON_START", 14)
+    monkeypatch.setitem(web_api.app_state, "scraper", FakeScraper())
+    monkeypatch.setitem(web_api.app_state, "leader_lock", None)
+    monkeypatch.setitem(web_api.app_state, "is_scraping", False)
+    monkeypatch.setitem(
+        web_api.app_state, "scraping_stats", dict(web_api.app_state["scraping_stats"])
+    )
+
+    asyncio.run(web_api.background_scraping_task())
+
+    assert calls == {"hourly": 1, "precompute": 2, "notify": 2, "gap_fill": 14}
+    assert sleeps == [600.0, 600.0, 2100.0]  # retry, retry, then 22:05
+    stats = web_api.app_state["scraping_stats"]
+    assert (stats["total_runs"], stats["successful_runs"], stats["failed_runs"]) == (
+        3,
+        1,
+        2,
+    )
+    assert stats["latest_reading"] == datetime.datetime(2026, 10, 17, 21)
+    assert web_api.app_state["is_scraping"] is False