diff --git a/src/database_adapters.py b/src/database_adapters.py index f53d389..2e0bc8b 100644 --- a/src/database_adapters.py +++ b/src/database_adapters.py @@ -36,6 +36,26 @@ class DatabaseAdapter(ABC): def get_measurements_for_date(self, target_date: datetime.datetime) -> List[Dict]: pass + def get_measurement_date_range( + self, + ) -> Optional[tuple]: + """Return (min_timestamp, max_timestamp) of stored measurements. + + Returns None when the backend has no data or does not support the query. + """ + return None + + def get_recorded_hours_by_day( + self, start_date: datetime.date, end_date: datetime.date + ) -> Optional[Dict[datetime.date, set]]: + """Map each day in [start_date, end_date] to the set of hours (0-23) + that have at least one measurement. + + Returns None when the backend does not support hour-granular gap + detection (callers should fall back to day-granular checks). + """ + return None + # InfluxDB Adapter class InfluxDBAdapter(DatabaseAdapter): @@ -657,6 +677,93 @@ class SQLAdapter(DatabaseAdapter): ) return [] + @staticmethod + def _coerce_date(value) -> Optional[datetime.date]: + """Normalize a DB-returned day value (str/date/datetime) to a date.""" + if value is None: + return None + if isinstance(value, datetime.datetime): + return value.date() + if isinstance(value, datetime.date): + return value + # SQLite returns strings, e.g. '2024-09-15' + return datetime.datetime.strptime(str(value)[:10], "%Y-%m-%d").date() + + def get_measurement_date_range(self) -> Optional[tuple]: + if not self.engine: + return None + + try: + from sqlalchemy import text + + query = "SELECT MIN(timestamp), MAX(timestamp) FROM water_measurements" + with self.engine.connect() as conn: + row = conn.execute(text(query)).fetchone() + + if not row or row[0] is None: + return None + + def to_datetime(value): + if isinstance(value, datetime.datetime): + return value + return datetime.datetime.fromisoformat(str(value)[:19]) + + return (to_datetime(row[0]), to_datetime(row[1])) + + except Exception as e: + logging.error(f"Error querying {self.db_type.upper()} date range: {e}") + return None + + def get_recorded_hours_by_day( + self, start_date: datetime.date, end_date: datetime.date + ) -> Optional[Dict[datetime.date, set]]: + if not self.engine: + return None + + try: + from sqlalchemy import text + + if self.db_type == "sqlite": + day_expr = "DATE(timestamp)" + hour_expr = "CAST(strftime('%H', timestamp) AS INTEGER)" + elif self.db_type == "postgresql": + day_expr = "CAST(timestamp AS DATE)" + hour_expr = "CAST(EXTRACT(HOUR FROM timestamp) AS INTEGER)" + else: # MySQL + day_expr = "DATE(timestamp)" + hour_expr = "HOUR(timestamp)" + + query = f""" + SELECT {day_expr} AS day, {hour_expr} AS hour + FROM water_measurements + WHERE timestamp >= :start_time AND timestamp < :end_time + GROUP BY {day_expr}, {hour_expr} + """ + + start_time = datetime.datetime.combine(start_date, datetime.time.min) + end_time = datetime.datetime.combine( + end_date + datetime.timedelta(days=1), datetime.time.min + ) + + hours_by_day: Dict[datetime.date, set] = {} + with self.engine.connect() as conn: + result = conn.execute( + text(query), {"start_time": start_time, "end_time": end_time} + ) + for row in result: + day = self._coerce_date(row[0]) + if day is None: + continue + hours_by_day.setdefault(day, set()).add(int(row[1])) + + return hours_by_day + + except Exception as e: + logging.error( + f"Error querying {self.db_type.upper()} recorded hours: {e}" + ) + return None + # VictoriaMetrics Adapter (using Prometheus format) class VictoriaMetricsAdapter(DatabaseAdapter): diff --git a/src/main.py b/src/main.py index fa8e2ea..50dc2a5 100644 --- a/src/main.py +++ b/src/main.py @@ -180,9 +180,12 @@ def run_continuous_monitoring(): return True -def run_gap_filling(days_back: int): - """Run gap filling for missing data""" - logger.info(f"Checking for data gaps in the last {days_back} days...") +def run_gap_filling(days_back: Optional[int]): + """Run gap filling for missing data (days_back=None scans the whole range)""" + if days_back is None: + logger.info("Checking for data gaps across the whole data range...") + else: + logger.info(f"Checking for data gaps in the last {days_back} days...") try: # Validate configuration @@ -445,6 +448,7 @@ Examples: %(prog)s # Run continuous monitoring %(prog)s --web-api # Start web API server %(prog)s --fill-gaps 7 # Fill missing data for last 7 days + %(prog)s --fill-gaps all # Fill missing data across the whole data range %(prog)s --update-data 2 # Update existing data for last 2 days %(prog)s --import-historical 2024-01-01 2024-01-31 # Import historical data %(prog)s --status # Show system status @@ -461,9 +465,11 @@ Examples: parser.add_argument( "--fill-gaps", - type=int, - metavar="DAYS", - help="Fill missing data gaps for the specified number of days back", + metavar="DAYS|all", + help=( + "Fill missing data gaps for the specified number of days back, " + "or 'all' to scan the entire data range in the database" + ), ) parser.add_argument( @@ -529,7 +535,17 @@ Examples: elif args.web_api: success = run_web_api() elif args.fill_gaps is not None: - success = run_gap_filling(args.fill_gaps) + if args.fill_gaps.lower() == "all": + success = run_gap_filling(None) + else: + try: + success = run_gap_filling(int(args.fill_gaps)) + except ValueError: + logger.error( + f"Invalid --fill-gaps value '{args.fill_gaps}': " + "expected a number of days or 'all'" + ) + sys.exit(1) elif args.update_data is not None: success = run_data_update(args.update_data) elif args.import_historical is not None: diff --git a/src/water_scraper_v3.py b/src/water_scraper_v3.py index c374da7..cb5a52e 100644 --- a/src/water_scraper_v3.py +++ b/src/water_scraper_v3.py @@ -579,53 +579,129 @@ class EnhancedWaterMonitorScraper: increment_counter("scraping_cycles_failed") return False - def fill_data_gaps(self, days_back: int) -> int: - """Fill gaps in data for the specified number of days back""" + def fill_data_gaps(self, days_back: Optional[int] = None) -> int: + """Fill gaps in stored data by re-fetching incomplete days. + + Args: + days_back: How many days back to scan. None scans the whole data + range, from the earliest measurement in the database to now. + + Detection is hour-granular where the backend supports it: any calendar + day missing one or more hourly slots gets its source date(s) re-fetched. + Because the API reports hour 24 as midnight of the next day, a missing + 00:00 slot on day D is repaired by re-fetching day D-1. + """ logger = get_logger(__name__) filled_count = 0 try: - # Calculate date range - end_date = datetime.datetime.now() - start_date = end_date - datetime.timedelta(days=days_back) + now = datetime.datetime.now() + end_date = now.date() - logger.info( - f"Checking for gaps from {start_date.date()} to {end_date.date()}" - ) + if days_back is not None: + start_date = end_date - datetime.timedelta(days=days_back) + else: + if not self.db_adapter: + logger.error("Database adapter not initialized") + return 0 + date_range = self.db_adapter.get_measurement_date_range() + if not date_range: + logger.error( + "Database is empty or does not support range queries; " + "use --import-historical to seed data first" + ) + return 0 + start_date = date_range[0].date() - # Iterate through each date in the range - current_date = start_date - while current_date <= end_date: - # Check if we have data for this date - has_data = self._check_data_exists_for_date(current_date) + logger.info(f"Checking for gaps from {start_date} to {end_date}") - if not has_data: - logger.info(f"Filling gap for date: {current_date.date()}") + fetch_dates = self._find_gap_fetch_dates(start_date, end_date, now) - # Fetch data for this specific date - data = self.fetch_water_data_for_date(current_date) + if not fetch_dates: + logger.info("No gaps found") + return 0 - if data: - # Save the data - if self.save_to_database(data): - filled_count += len(data) - logger.info( - f"Filled {len(data)} measurements for {current_date.date()}" - ) - else: - logger.warning( - f"Failed to save data for {current_date.date()}" - ) + logger.info(f"Found {len(fetch_dates)} day(s) needing a re-fetch") + + for fetch_date in sorted(fetch_dates): + fetch_dt = datetime.datetime.combine(fetch_date, datetime.time.min) + logger.info(f"Filling gap for date: {fetch_date}") + + data = self.fetch_water_data_for_date(fetch_dt) + + if data: + if self.save_to_database(data): + filled_count += len(data) + logger.info( + f"Filled {len(data)} measurements for {fetch_date}" + ) else: - logger.warning(f"No data available for {current_date.date()}") + logger.warning(f"Failed to save data for {fetch_date}") + else: + logger.warning(f"No data available for {fetch_date}") - current_date += datetime.timedelta(days=1) + # Be respectful to the API + time.sleep(1) except Exception as e: logger.error(f"Gap filling error: {e}") return filled_count + def _find_gap_fetch_dates( + self, + start_date: datetime.date, + end_date: datetime.date, + now: datetime.datetime, + ) -> List[datetime.date]: + """Determine which source dates need re-fetching to fill gaps. + + Uses hour-granular detection when the adapter supports it, otherwise + falls back to re-fetching only days with no data at all. + """ + hours_by_day = None + if self.db_adapter: + hours_by_day = self.db_adapter.get_recorded_hours_by_day( + start_date, end_date + ) + + fetch_dates = set() + + if hours_by_day is None: + logger.info( + "Backend does not support hour-granular gap detection; " + "checking for fully-missing days only" + ) + current = start_date + while current <= end_date: + current_dt = datetime.datetime.combine(current, datetime.time.min) + if not self._check_data_exists_for_date(current_dt): + fetch_dates.add(current) + current += datetime.timedelta(days=1) + return sorted(fetch_dates) + + current = start_date + while current <= end_date: + if current == now.date(): + # Today: only expect hours that have already passed + expected_hours = set(range(0, now.hour)) + else: + expected_hours = set(range(24)) + + missing = expected_hours - hours_by_day.get(current, set()) + + if missing: + # Hours 1-23 of day D come from fetching D; hour 0 comes from + # the previous day's fetch (the API's "hour 24") + if any(h >= 1 for h in missing): + fetch_dates.add(current) + if 0 in missing and current > start_date: + fetch_dates.add(current - datetime.timedelta(days=1)) + + current += datetime.timedelta(days=1) + + return sorted(fetch_dates) + def update_existing_data(self, days_back: int) -> int: """Update existing data with latest values for the specified number of days back""" logger = get_logger(__name__)