From 04ebdaa931dedc24bbcf8d49f1d098a4f33c06b9 Mon Sep 17 00:00:00 2001 From: grabowski Date: Sat, 17 Oct 2026 21:45:09 +0700 Subject: [PATCH] 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 | 5 + README.md | 7 + src/config.py | 19 +++ src/schemas.py | 7 +- src/water_scraper_v3.py | 148 +++++++++-------- src/web_api.py | 280 +++++++++++++++++++++----------- tests/test_scraping_cycle.py | 307 +++++++++++++++++++++++++++++++++++ 7 files changed, 607 insertions(+), 166 deletions(-) create mode 100644 tests/test_scraping_cycle.py diff --git a/.env.example b/.env.example index a5722e7..e058280 100644 --- a/.env.example +++ b/.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 diff --git a/README.md b/README.md index 36b5928..551c035 100644 --- a/README.md +++ b/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 diff --git a/src/config.py b/src/config.py index 1ea484d..b65cd85 100644 --- a/src/config.py +++ b/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, diff --git a/src/schemas.py b/src/schemas.py index 3be02b4..121d393 100644 --- a/src/schemas.py +++ b/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 - failed_runs: int = 0 + successful_runs: int = 0 # cycles that found the current hour published + 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 diff --git a/src/water_scraper_v3.py b/src/water_scraper_v3.py index 62e69eb..310584b 100644 --- a/src/water_scraper_v3.py +++ b/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: - """Check if the fetched data contains new data for the current hour""" - if not water_data: - return False + def _cycle_source_dates(self, now: datetime.datetime) -> List[datetime.date]: + """Report dates (the API's ``DW[TimeCurrent]``) to fetch in one cycle. - current_time = datetime.datetime.now() - current_hour = current_time.hour + Always the report holding the current hour: today's, or before 01:00 + 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 - latest_timestamp = None - 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 + def run_scraping_cycle(self, now: Optional[datetime.datetime] = None) -> bool: + """Fetch the RID hourly report(s) and store every reading in them. - if latest_timestamp is None: - logger.warning("No valid timestamps found in data") - return False + Returns True when the current hour's reading is in (the feed is + current) and False when the caller should retry soon. Readings are + 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 - time_diff = current_time - latest_timestamp - minutes_old = time_diff.total_seconds() / 60 - - logger.info( - f"Current time: {current_time.strftime('%H:%M')}, Latest data: {latest_timestamp.strftime('%H:%M')}" - ) - logger.info( - f"Current hour: {current_hour}, Latest data hour: {latest_hour}, Age: {minutes_old:.1f} minutes" - ) - - # Strict check: we need data from the current hour - # 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 - - 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""" + Details of the last cycle are kept in ``self.last_cycle``. + """ + now = now or datetime.datetime.now() + current_hour = now.replace(minute=0, second=0, microsecond=0) + status = { + "time": now, + "dates": [], + "fetched": 0, + "saved": 0, + "latest_reading": None, + "current": False, + "error": None, + } + self.last_cycle = status logger.info("Starting scraping cycle...") try: - # Fetch current data - water_data = self.fetch_water_data() - if water_data: - # Check if data is fresh/recent - is_fresh = self._check_data_freshness(water_data) + dates = self._cycle_source_dates(now) + status["dates"] = [d.isoformat() for d in dates] - if is_fresh: - success = self.save_to_database(water_data) - if success: - logger.info( - "Scraping cycle completed successfully with fresh data" - ) - increment_counter("scraping_cycles_successful") - return True - else: - logger.error("Failed to save data") - increment_counter("scraping_cycles_failed") - 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: + water_data: List[Dict] = [] + for source_date in dates: + rows = self.fetch_water_data_for_date( + datetime.datetime.combine(source_date, datetime.time.min) + ) + if rows: + water_data.extend(rows) + status["fetched"] = len(water_data) + + if not water_data: + status["error"] = "no data fetched" 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 diff --git a/src/web_api.py b/src/web_api.py index 997c626..fc09158 100644 --- a/src/web_api.py +++ b/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"]: - app_state["is_scraping"] = True + feed_current = False + scraper = app_state["scraper"] - # Run scraping cycle - scraper = app_state["scraper"] - if scraper: + if scraper and not app_state["is_scraping"]: + app_state["is_scraping"] = True + try: logger.info("Starting background scraping cycle") start_time = datetime.now() - try: - # run_scraping_cycle() does blocking network/DB I/O and time.sleep - # retries; run it in a thread so it doesn't freeze the event loop. - result = await asyncio.get_event_loop().run_in_executor( - None, scraper.run_scraping_cycle + # Blocking network/DB I/O: keep it off the event loop. + feed_current = bool( + await loop.run_in_executor(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) - hii_collector = app_state.get("hii_collector") - if hii_collector: - try: - hii_counts = await asyncio.get_event_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}") + latest = (getattr(scraper, "last_cycle", None) or {}).get( + "latest_reading" + ) + new_readings = latest is not None and ( + newest_reading is None or latest > newest_reading + ) + if new_readings: + newest_reading = latest - # RID large-dam daily status (Mae Ngat storage/inflow/outflow) - reservoir_collector = app_state.get("reservoir_collector") - if reservoir_collector: - try: - rsv_saved = await asyncio.get_event_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}") + slot = datetime.now().replace(minute=0, second=0, microsecond=0) + hourly_due = slot != last_hourly_slot + if hourly_due: + last_hourly_slot = slot + await _collect_hourly_sources() - # Persist the Open-Meteo catchment rain (observed tail + - # 48h forecast) so the DB carries the weather context too - await _persist_rain() + if new_readings or hourly_due: + # Primes the forecast cache (user requests never pay + # 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 - # cache (user requests never pay for inference) and records - # what the model predicted for later predicted-vs-actual - # evaluation. - await _precompute_forecasts() - - # 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 + next_run = _next_poll_time(datetime.now(), feed_current) + app_state["scraping_stats"]["next_run"] = next_run + _publish_scraping_status() + logger.info( + f"Next RID poll at {next_run:%H:%M}" + + ("" if feed_current else " (current hour not published yet)") ) - - # Wait for next cycle - await asyncio.sleep(interval_seconds) + await asyncio.sleep(max(30.0, (next_run - datetime.now()).total_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() - - # Update stats - app_state["scraping_stats"]["total_runs"] += 1 - 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") + _record_scrape_run(start_time, result) + increment_counter( + "manual_scraping_successful" if result else "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"], - 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"], + **{k: v for k, v in stats.items() if k in ScrapingStatusResponse.model_fields} ) diff --git a/tests/test_scraping_cycle.py b/tests/test_scraping_cycle.py new file mode 100644 index 0000000..e271032 --- /dev/null +++ b/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